diff --git a/cabal.project b/cabal.project index 32e5cc6667..5a0b4167d6 100644 --- a/cabal.project +++ b/cabal.project @@ -30,6 +30,7 @@ packages: ./monoidal-synchronisation ./cardano-diffusion ./ntp-client ./acts-generic + ./window-stats tests: True benchmarks: True diff --git a/window-stats/CHANGELOG.md b/window-stats/CHANGELOG.md new file mode 100644 index 0000000000..d3fecf5c6b --- /dev/null +++ b/window-stats/CHANGELOG.md @@ -0,0 +1,9 @@ +# window-stats changelog + + + + + +## 0.1.0.0 -- YYYY-mm-dd + +* First version. Released on an unsuspecting world. diff --git a/window-stats/LICENSE b/window-stats/LICENSE new file mode 100644 index 0000000000..d645695673 --- /dev/null +++ b/window-stats/LICENSE @@ -0,0 +1,202 @@ + + Apache License + Version 2.0, January 2004 + http://www.apache.org/licenses/ + + TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION + + 1. Definitions. + + "License" shall mean the terms and conditions for use, reproduction, + and distribution as defined by Sections 1 through 9 of this document. + + "Licensor" shall mean the copyright owner or entity authorized by + the copyright owner that is granting the License. + + "Legal Entity" shall mean the union of the acting entity and all + other entities that control, are controlled by, or are under common + control with that entity. For the purposes of this definition, + "control" means (i) the power, direct or indirect, to cause the + direction or management of such entity, whether by contract or + otherwise, or (ii) ownership of fifty percent (50%) or more of the + outstanding shares, or (iii) beneficial ownership of such entity. + + "You" (or "Your") shall mean an individual or Legal Entity + exercising permissions granted by this License. + + "Source" form shall mean the preferred form for making modifications, + including but not limited to software source code, documentation + source, and configuration files. + + "Object" form shall mean any form resulting from mechanical + transformation or translation of a Source form, including but + not limited to compiled object code, generated documentation, + and conversions to other media types. + + "Work" shall mean the work of authorship, whether in Source or + Object form, made available under the License, as indicated by a + copyright notice that is included in or attached to the work + (an example is provided in the Appendix below). + + "Derivative Works" shall mean any work, whether in Source or Object + form, that is based on (or derived from) the Work and for which the + editorial revisions, annotations, elaborations, or other modifications + represent, as a whole, an original work of authorship. For the purposes + of this License, Derivative Works shall not include works that remain + separable from, or merely link (or bind by name) to the interfaces of, + the Work and Derivative Works thereof. + + "Contribution" shall mean any work of authorship, including + the original version of the Work and any modifications or additions + to that Work or Derivative Works thereof, that is intentionally + submitted to Licensor for inclusion in the Work by the copyright owner + or by an individual or Legal Entity authorized to submit on behalf of + the copyright owner. For the purposes of this definition, "submitted" + means any form of electronic, verbal, or written communication sent + to the Licensor or its representatives, including but not limited to + communication on electronic mailing lists, source code control systems, + and issue tracking systems that are managed by, or on behalf of, the + Licensor for the purpose of discussing and improving the Work, but + excluding communication that is conspicuously marked or otherwise + designated in writing by the copyright owner as "Not a Contribution." + + "Contributor" shall mean Licensor and any individual or Legal Entity + on behalf of whom a Contribution has been received by Licensor and + subsequently incorporated within the Work. + + 2. Grant of Copyright License. Subject to the terms and conditions of + this License, each Contributor hereby grants to You a perpetual, + worldwide, non-exclusive, no-charge, royalty-free, irrevocable + copyright license to reproduce, prepare Derivative Works of, + publicly display, publicly perform, sublicense, and distribute the + Work and such Derivative Works in Source or Object form. + + 3. Grant of Patent License. Subject to the terms and conditions of + this License, each Contributor hereby grants to You a perpetual, + worldwide, non-exclusive, no-charge, royalty-free, irrevocable + (except as stated in this section) patent license to make, have made, + use, offer to sell, sell, import, and otherwise transfer the Work, + where such license applies only to those patent claims licensable + by such Contributor that are necessarily infringed by their + Contribution(s) alone or by combination of their Contribution(s) + with the Work to which such Contribution(s) was submitted. If You + institute patent litigation against any entity (including a + cross-claim or counterclaim in a lawsuit) alleging that the Work + or a Contribution incorporated within the Work constitutes direct + or contributory patent infringement, then any patent licenses + granted to You under this License for that Work shall terminate + as of the date such litigation is filed. + + 4. Redistribution. You may reproduce and distribute copies of the + Work or Derivative Works thereof in any medium, with or without + modifications, and in Source or Object form, provided that You + meet the following conditions: + + (a) You must give any other recipients of the Work or + Derivative Works a copy of this License; and + + (b) You must cause any modified files to carry prominent notices + stating that You changed the files; and + + (c) You must retain, in the Source form of any Derivative Works + that You distribute, all copyright, patent, trademark, and + attribution notices from the Source form of the Work, + excluding those notices that do not pertain to any part of + the Derivative Works; and + + (d) If the Work includes a "NOTICE" text file as part of its + distribution, then any Derivative Works that You distribute must + include a readable copy of the attribution notices contained + within such NOTICE file, excluding those notices that do not + pertain to any part of the Derivative Works, in at least one + of the following places: within a NOTICE text file distributed + as part of the Derivative Works; within the Source form or + documentation, if provided along with the Derivative Works; or, + within a display generated by the Derivative Works, if and + wherever such third-party notices normally appear. The contents + of the NOTICE file are for informational purposes only and + do not modify the License. You may add Your own attribution + notices within Derivative Works that You distribute, alongside + or as an addendum to the NOTICE text from the Work, provided + that such additional attribution notices cannot be construed + as modifying the License. + + You may add Your own copyright statement to Your modifications and + may provide additional or different license terms and conditions + for use, reproduction, or distribution of Your modifications, or + for any such Derivative Works as a whole, provided Your use, + reproduction, and distribution of the Work otherwise complies with + the conditions stated in this License. + + 5. Submission of Contributions. Unless You explicitly state otherwise, + any Contribution intentionally submitted for inclusion in the Work + by You to the Licensor shall be under the terms and conditions of + this License, without any additional terms or conditions. + Notwithstanding the above, nothing herein shall supersede or modify + the terms of any separate license agreement you may have executed + with Licensor regarding such Contributions. + + 6. Trademarks. This License does not grant permission to use the trade + names, trademarks, service marks, or product names of the Licensor, + except as required for reasonable and customary use in describing the + origin of the Work and reproducing the content of the NOTICE file. + + 7. Disclaimer of Warranty. Unless required by applicable law or + agreed to in writing, Licensor provides the Work (and each + Contributor provides its Contributions) on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or + implied, including, without limitation, any warranties or conditions + of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A + PARTICULAR PURPOSE. You are solely responsible for determining the + appropriateness of using or redistributing the Work and assume any + risks associated with Your exercise of permissions under this License. + + 8. Limitation of Liability. In no event and under no legal theory, + whether in tort (including negligence), contract, or otherwise, + unless required by applicable law (such as deliberate and grossly + negligent acts) or agreed to in writing, shall any Contributor be + liable to You for damages, including any direct, indirect, special, + incidental, or consequential damages of any character arising as a + result of this License or out of the use or inability to use the + Work (including but not limited to damages for loss of goodwill, + work stoppage, computer failure or malfunction, or any and all + other commercial damages or losses), even if such Contributor + has been advised of the possibility of such damages. + + 9. Accepting Warranty or Additional Liability. While redistributing + the Work or Derivative Works thereof, You may choose to offer, + and charge a fee for, acceptance of support, warranty, indemnity, + or other liability obligations and/or rights consistent with this + License. However, in accepting such obligations, You may act only + on Your own behalf and on Your sole responsibility, not on behalf + of any other Contributor, and only if You agree to indemnify, + defend, and hold each Contributor harmless for any liability + incurred by, or claims asserted against, such Contributor by reason + of your accepting any such warranty or additional liability. + + END OF TERMS AND CONDITIONS + + APPENDIX: How to apply the Apache License to your work. + + To apply the Apache License to your work, attach the following + boilerplate notice, with the fields enclosed by brackets "[]" + replaced with your own identifying information. (Don't include + the brackets!) The text should be enclosed in the appropriate + comment syntax for the file format. We also recommend that a + file or class name and description of purpose be included on the + same "printed page" as the copyright notice for easier + identification within third-party archives. + + Copyright [yyyy] [name of copyright owner] + + Licensed under the Apache License, Version 2.0 (the "License"); + you may not use this file except in compliance with the License. + You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. diff --git a/window-stats/NOTICE b/window-stats/NOTICE new file mode 100644 index 0000000000..025f8f91dc --- /dev/null +++ b/window-stats/NOTICE @@ -0,0 +1,14 @@ +Copyright 2019-2023 Input Output Global Inc (IOG), 2023-2026 Intersect + + Licensed under the Apache License, Version 2.0 (the "License"); + you may not use this file except in compliance with the License. + You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. + diff --git a/window-stats/README.md b/window-stats/README.md new file mode 100644 index 0000000000..83b324365f --- /dev/null +++ b/window-stats/README.md @@ -0,0 +1,245 @@ +# window-stats + +Efficient sliding-window statistics for Haskell, backed by a finger +tree, with a cached user-supplied measure that updates incrementally +as the window slides. + +## Scope + +The package provides two variants of sliding window: + +- **Count-based** - keeps the most recent _N_ inserted samples. +- **Time-based** - keeps samples within a configured duration of the + newest sample's timestamp. + +Both are generic in the cached measure: a finger-tree node stores the +combination of an element count (or timestamp bounds) and a +user-supplied monoid `v`. Looking the measure up is `O(1)` regardless +of window size. + +Prebuilt sample wrappers and their measures are included for the +common rolling statistics: + +| Sample wrapper | Measure | Statistic | +|-----------------|------------------|----------------------------------------| +| `SumSample` | `Sum` | running sum | +| `MinMaxSample` | `MinMaxV` | running min / max | +| `MomentSample` | `WelfordMeasure` | running mean / sample variance / stddev (Welford) | +| `DigestSample` | `TDigest comp` | approximate quantiles (with-tdigest) | + +Time-based windows abstract over the timestamp type via the +`TimeLike` class. Two stock instances are shipped: + +- `UTCTime` / `NominalDiffTime` (wall-clock, from `time`) +- `Time` / `DiffTime` (monotonic, from `io-classes:si-timers`) + +Monotonic is preferred when sliding-window correctness must not be +perturbed by NTP corrections or wall-clock jumps; wall-clock fits +data that already carries `UTCTime` timestamps. + +## Sublibraries + +The package is split so that downstream consumers only pay for the +dependencies they need. + +``` +window-stats -- core: Window, TimedWindow, prebuilt measures +window-stats:with-tdigest -- approximate-quantile backend via tdigest +window-stats:with-foldl -- Control.Foldl combinators that drive a window +``` + +Sublibraries compose at the user's call site: a project depending on +both `with-tdigest` and `with-foldl` automatically gets foldl-driven +t-digest rolling quantiles. + +## Complexity + +Per-operation costs, for a window of size _w_: + +| operation | amortised | worst-case | +|------------------------------------|-----------|------------| +| `insert` | `O(1)` | `O(log w)` | +| `insert`'s built-in eviction | `O(1)` | `O(log w)` | +| `evictOldest` | `O(1)` | `O(log w)` | +| `evictOldestN` | `O(log w)`| `O(log w)` | +| `trimByMeasure` | `O(log w)`| `O(log w)` | +| `evictBefore` (Timed) | `O(log w)`| `O(log w)` | +| `resize` (shrink) | `O(log w)`| `O(log w)` | +| `resize` (grow / no-op) | `O(1)` | `O(1)` | +| `windowMeasure` and other queries | `O(1)` | `O(1)` | +| `fromListN` / `fromFoldable` | `O(n)` | `O(n log w)` | + +t-digest-backed insertion is the one exception: each insert triggers +up to `O(log w)` TDigest `<>` merges along the finger-tree spine, and +each merge is `O(δ)` in the compression parameter - so an effective +`O(δ log w)` per insert. + +Space is `O(w)` for the window itself plus whatever the user's +measure adds (typically `O(1)` for `Sum`/`MinMaxV`/`WelfordMeasure`; +`O(δ)` for `TDigest comp`, independent of `w`). + +## When to reach for this package + +- You need rolling statistics over a continuously moving window. +- You want both count- and time-based windowing in one API. +- You want approximate quantiles in bounded memory (the t-digest + backend gives `O(δ)` regardless of window size). +- You don't want to commit to a particular streaming library - the + core types are pure values and you compose them however suits your + pipeline; `Control.Foldl` integration is optional. +- You want the cached running statistic - mean, variance, sum, a + digest - as a value, on demand, in `O(1)`. + +## Use-case examples + +- **Network monitoring**. Rolling latency / throughput / drop-rate + windows per peer, with cheap `O(1)` reads at each tick. +- **Service-level analytics**. Rolling p50/p95/p99 of request times + over the last _N_ minutes via the t-digest backend. +- **Budget enforcement**. Variable-cost windows - keep the most + recent samples whose cumulative weight (a `Sum`) stays below a + threshold - via `trimByMeasure`. +- **Cross-window comparison**. Short- vs long-period moving averages + for trend / crossover detection, via the dual-window combinators in + `Data.Window.Fold.Count`. + +## Relation to similar packages + +The cleanest split is between packages that compete at the **windowing** +layer (where `window-stats` sits) and those that compete or compose at +the **measure** layer (the `v` in `Window v a`). Packages in the +second group plug in as alternative measures rather than substituting +for the window itself. + +### `streamly-statistics` + +Provides exact rolling statistics over a streamly stream: +- Exact quantiles via a sorted structure, at `O(w)` memory. +- Tied to streamly's streaming model and types. + +`window-stats` is complementary rather than a replacement: it trades +exactness for `O(δ)` memory on quantiles via t-digest, is independent +of any streaming library, and provides a `Window` value you can hold, +query, and pass around - useful when sliding statistics are part of +some larger data structure rather than a one-shot pipeline result. + +Use `streamly-statistics` when you need exact quantiles and you're +already on streamly; use `window-stats` when you can tolerate +approximate quantiles, want fixed memory, want time-based as well as +count-based windowing, or aren't already in a streamly context. + +### `streaming`'s `slidingWindow` + +`Streaming.Prelude.slidingWindow :: Int -> Stream (Of a) m r -> Stream (Of (Seq a)) m r` +yields each window position as a `Seq a`. It does no measure caching: +to compute a statistic at each step you'd traverse the `Seq` from +scratch. + +`window-stats` carries the running measure inside the window itself +and updates it incrementally on insertion / eviction. For a stream of +length _n_ with windows of size _w_, computing a rolling sum: + +- `streaming` + `slidingWindow`: `O(n · w)` work total (sum each window). +- `window-stats`: `O(n)` work total - each insert updates the cached + measure in `O(1)` amortised; the read is `O(1)`. + +Pick `streaming`'s `slidingWindow` if you genuinely need the window +contents at each step and the windows are small. Pick `window-stats` +if you want an incrementally-maintained statistic and the window +might be large. + +### `conduit`'s `slidingVector` / `slidingWindow` + +Same shape as `streaming`'s `slidingWindow` - yields each window +position as a `Vector a` / `Seq a` over the conduit pipeline, with no +measure caching. Same complexity trade-off, same advice: reach for it +when you actually need the window contents at each step, not just a +rolling statistic. + +### `monoid-statistics` + +Complementary, not competing. Provides numerically careful statistical +monoids (`Mean`, `Variance`, `Min`, `Max`, `KBNSum` Kahan-Babuška +summation, etc.). Any of them can be plugged in as the `v` parameter +of `Window v a` to get rolling versions of those statistics. Our +prebuilt `WelfordMeasure` covers the most common case; reach for +`monoid-statistics` when you need its specific numerical guarantees +or one of its less common stats. + +### `histogram-fill` + +Also complementary. Its `Histogram bin val` is a `Monoid` (bin-wise +addition), so a user can define a `Measured (Histogram bin val) sample` +instance and use `Window (Histogram bin val) sample` directly. The +window then carries a **rolling histogram** as its cached measure, +available in `O(1)` via `windowMeasure`. Cost per insert scales with +the bin count - much like the t-digest backend scales with δ - but +the result is exact rather than approximate. + +### `tdigest` + +We sit on top of it for the `with-tdigest` sublibrary's approximate +quantiles. If you only need a one-shot quantile over a static dataset +(no sliding-window aspect), use `tdigest` directly. Reach for our +`with-tdigest` when you specifically want approximate quantiles that +*roll* with a moving window in bounded memory. + +### `Control.Foldl`-based folds + +The `with-foldl` sublibrary integrates with the `foldl` package via +combinators such as `windowFoldRolling`, which feed each step's +window measure into an arbitrary downstream `Fold`. Memory is bounded +by the inner fold's own state, so summarising rolling statistics +across a long stream stays constant-memory. + +## A small example + +```haskell +import Data.Monoid (Sum) +import Data.Window.Count + +-- A window of the last 100 sample sums: +let w0 = empty 100 :: Window (Sum Double) (SumSample Double) + w1 = insert 1.5 w0 + w2 = insert 2.5 w1 +in (windowSum w2, size w2) +-- => (Just 4.0, 2) +``` + +t-digest quantiles: + +```haskell +import Data.Window.DigestCount +import Data.TDigest qualified as TD + +let w = fromListN 1000 prices :: DigestWindow 100 + p50 = TD.quantile 0.5 (windowMeasure w) + p99 = TD.quantile 0.99 (windowMeasure w) +in (p50, p99) +``` + +Time-based, with monotonic timestamps: + +```haskell +import Control.Monad.Class.MonadTime.SI (Time, getMonotonicTime) +import Data.Window.Timed + +-- 60-second sliding window of running mean +type LatencyWin = TimedWindow Time (WelfordMeasure Double) (MomentSample Double) + +step :: Double -> LatencyWin -> IO LatencyWin +step latency w = do + now <- getMonotonicTime + return (insert (now, latency) w) +``` + +## Building and testing + +```bash +cabal build all +cabal test window-stats +``` + +## License + +Apache-2.0. See [LICENSE](LICENSE). diff --git a/window-stats/bench/Main.hs b/window-stats/bench/Main.hs new file mode 100644 index 0000000000..54e4fdc4e7 --- /dev/null +++ b/window-stats/bench/Main.hs @@ -0,0 +1,147 @@ +{-# LANGUAGE BangPatterns #-} +{-# LANGUAGE CPP #-} +{-# LANGUAGE DataKinds #-} + +-- | Cascade-eviction benchmark for "Data.Window.DigestTimeBatched", +-- used to compare lazy vs. strict 'bmDigest' in the finger-tree +-- measure. +-- +-- Each case streams a 'cascadeSamples' input into a window with 1-second +-- buckets and 60-bucket retention, then queries the median (forcing the +-- top-level measure through 'windowDigest'). Timings are monotonic from +-- 'GHC.Clock.getMonotonicTime'; allocation, GC count and residency +-- come from 'GHC.Stats' when the RTS is built with @-T@. +-- +-- The cases alternate dense stretches with wide idle gaps. The gap +-- puts each next sample's eviction cutoff past many sealed buckets in +-- a single 'FT.dropUntil'; a lazy 'bmDigest' can discard the cached +-- subtree measures for those buckets without forcing their digest +-- combines, while a strict variant has already paid for them at +-- insertion time. +-- +-- Measured lazy vs. strict (bmDigest strict + no manual force in +-- evictBefore), @-O2 +RTS -T@, wall / alloc: +-- +-- * 10x (6000 @ 0.01, gap 60s) — 60k samples — 1.36x / 1.30x +-- * 20x (6000 @ 0.01, gap 60s) — 120k samples — 1.25x / 1.30x +-- * 50x (3000 @ 0.01, gap 60s) — 150k samples — 1.23x / 1.22x +-- * 100x (1000 @ 0.01, gap 60s) — 100k samples — 1.21x / 1.31x +-- +module Main (main) where + +import Control.Exception (evaluate) +#if !MIN_VERSION_base(4,20,0) +import Data.Foldable +#endif +import Data.Time (UTCTime (..), addUTCTime, fromGregorian) +import Data.Word (Word64) +import GHC.Clock (getMonotonicTime) +import GHC.Stats (RTSStats (..), getRTSStats, getRTSStatsEnabled) +import System.Mem (performMajorGC) +import Text.Printf (printf) + +import Data.Window.DigestTimeBatched qualified as DTB + + +type Comp = 100 +type DigestW = DTB.TimedDigestWindow UTCTime Comp + +base :: UTCTime +base = UTCTime (fromGregorian 2024 1 1) 0 + +mkT :: Double -> UTCTime +mkT s = realToFrac s `addUTCTime` base + +-- | Alternating dense/gap stream. +-- +-- Each cycle emits @nDense@ samples spaced by @denseStep@ seconds +-- (duration @(nDense - 1) * denseStep@), then a silent @gapDur@ before +-- the next cycle. The first sample of the next cycle triggers a seal +-- AND — if @gapDur@ is a significant fraction of the retention window — +-- moves the eviction cutoff past many buckets at once, causing a +-- cascade drop that lazy 'bmDigest' can avoid combining. +cascadeSamples + :: Int -- ^ number of cycles + -> Int -- ^ samples per dense stretch + -> Double -- ^ time step within a dense stretch (s) + -> Double -- ^ gap between stretches (s) + -> [(UTCTime, Double)] +cascadeSamples cycles nDense denseStep gapDur = + [ (mkT (fromIntegral k * cycleDur + fromIntegral i * denseStep) + , fromIntegral (k * nDense + i) + ) + | k <- [0 .. cycles - 1] + , i <- [0 .. nDense - 1] + ] + where + stretchDur = fromIntegral (nDense - 1) * denseStep + cycleDur = stretchDur + gapDur + +-- Statistics captured around a measured action. All GHC.Stats counters +-- are Word64; keep them so on all architectures. +data Snapshot = Snapshot + { snapAllocated :: !Word64 + , snapMaxLive :: !Word64 + , snapGC :: !Word64 + } + +zeroSnap :: Snapshot +zeroSnap = Snapshot 0 0 0 + +takeSnap :: IO Snapshot +takeSnap = do + ok <- getRTSStatsEnabled + if ok + then do + s <- getRTSStats + pure Snapshot + { snapAllocated = allocated_bytes s + , snapMaxLive = max_live_bytes s + , snapGC = fromIntegral (gcs s) -- gcs :: Word32 + } + else pure zeroSnap + +benchCase :: String -> IO () -> IO () +benchCase label action = do + performMajorGC + s0 <- takeSnap + t0 <- getMonotonicTime + action + t1 <- getMonotonicTime + s1 <- takeSnap + let dt = t1 - t0 + alloc = fromIntegral (snapAllocated s1 - snapAllocated s0) / (1024 * 1024 :: Double) + gcs' = snapGC s1 - snapGC s0 + maxL = fromIntegral (snapMaxLive s1) / (1024 * 1024 :: Double) + printf "%-46s %8.4f s %8.2f MB alloc %6d GCs %6.2f MB max_live\n" + label dt alloc gcs' maxL + +main :: IO () +main = do + putStrLn "window-stats: DigestTimeBatched cascade-eviction benchmark" + putStrLn "(compile with -O2 and pass +RTS -T for allocation/GC stats)" + putStrLn "" + printf "%-46s %10s %19s %9s %14s\n" + "case" "wall" "alloc" "gcs" "max_live" + + -- Cases use bucketDur = 1.0 s, retention = 60 buckets. + let cascade cycles nDense denseStep gapDur = + benchCase + (printf "cascade: %d× (%d @ %.3fs dense, then %.1fs gap)" + cycles nDense denseStep gapDur) + (do + let !samples = cascadeSamples cycles nDense denseStep gapDur + !w0 = DTB.empty 1.0 60 :: DigestW + !w = foldl' (flip DTB.insert) w0 samples + _ <- evaluate (DTB.windowMedian w) + pure ()) + + -- Full-window cascade: each cycle fills the window then drops it. + cascade 10 6000 0.01 60.0 -- 60k samples, 10 cascades of 60 buckets + cascade 20 6000 0.01 60.0 -- 120k samples, 20 cascades of 60 buckets + + -- Half-window cascade: less accumulation, more cascade events. + cascade 50 3000 0.01 60.0 -- 150k samples, 50 cascades of ~30 buckets + + -- Shallow cascade: small drops per cascade, many events. + cascade 100 1000 0.01 60.0 -- 100k samples, 100 cascades of ~10 buckets diff --git a/window-stats/changelog.d/scriv.ini b/window-stats/changelog.d/scriv.ini new file mode 100644 index 0000000000..0d51d54748 --- /dev/null +++ b/window-stats/changelog.d/scriv.ini @@ -0,0 +1,15 @@ +[scriv] +format = md +insert_marker = Changelog entries +md_header_level = 2 +version = literal: window-stats.cabal: version +categories = Breaking, Non-Breaking, Patch +start_marker = scriv-insert-here +end_marker = scriv-end-here +fragment_directory = changelog.d +ghrel_template = {{body}} +main_branches = main +new_fragment_template = file: new_fragment.${config:format}.j2 +output_file = CHANGELOG.${config:format} +skip_fragments = README.* +entry_title_template = {%% if version %%}{{ version }} -- {%% endif %%}{{ date.strftime('%%Y-%%m-%%d') }} diff --git a/window-stats/fold/lib/Data/Window/Fold/Count.hs b/window-stats/fold/lib/Data/Window/Fold/Count.hs new file mode 100644 index 0000000000..a168120399 --- /dev/null +++ b/window-stats/fold/lib/Data/Window/Fold/Count.hs @@ -0,0 +1,206 @@ +{-# LANGUAGE BangPatterns #-} +{-# LANGUAGE RankNTypes #-} +{-# LANGUAGE ScopedTypeVariables #-} + +-- | +-- Module : Data.Window.Fold.Count +-- Description : @foldl@ combinators driving a count-based sliding window. +-- Stability : experimental +-- +-- 'F.Fold' combinators that maintain a sliding count window +-- ('Data.Window.Count.Window') over the input stream and surface its +-- rolling measure to a downstream fold. +-- +-- All combinators are polymorphic in the measure @v@ and the sample +-- type @a@ behind a 'FT.Measured' constraint, so they work unchanged +-- with any of the built-in sample wrappers +-- ('Data.Window.Count.SumSample', 'Data.Window.Count.MomentSample', +-- 'Data.Window.Count.MinMaxSample') and with user-defined measures. +-- +module Data.Window.Fold.Count + ( windowScan + , windowFoldWith + , windowFoldRolling + , windowFoldRollingFull + , windowFoldRollingM + , windowFoldRolling2 + , windowFoldRolling2Full + , windowFinal + , windowMeasureFinal + ) where + +import Control.Foldl qualified as F +import Data.FingerTree qualified as FT +import Data.Sequence qualified as S + +import Data.Window.Internal.Count qualified as W + +-- | Compute the window measure at each step as the window moves across the +-- input stream, collecting all intermediate measures into a sequence. +-- The sequence has the same length as the input stream, with the i-th element +-- being the measure of the window ending at the i-th input element. +-- +windowScan :: forall v a. FT.Measured v a + => Int + -> F.Fold a (S.Seq v) +windowScan n = F.Fold step initial extract + where + step :: (W.Window v a, S.Seq v) -> a -> (W.Window v a, S.Seq v) + step (w, ms) x = + let !w' = W.insert x w + !ms' = ms S.|> W.windowMeasure w' + in (w', ms') + initial = (W.empty n, S.empty) + extract = snd + +-- | Apply an extraction function to the window measure at each step, +-- producing a rolling sequence of statistics as the window moves across +-- the input stream. A specialisation of 'windowScan' for when you want +-- to post-process the measure at each step. +-- +windowFoldWith :: FT.Measured v a + => Int + -> (v -> b) + -> F.Fold a (S.Seq b) +windowFoldWith n f = fmap f <$> windowScan n + + +-- | Stream the rolling sequence of window measures into an inner +-- 'F.Fold'. Unlike 'windowScan' / 'windowFoldWith', the intermediate +-- measures are __not__ retained — each measure is fed to the inner +-- fold as soon as it is produced. This keeps the memory footprint +-- bounded by the inner fold's own state, independent of the input +-- length. +-- +-- Use this when you want to summarise the rolling statistics +-- (e.g. average them, take their max, fold them into a histogram) +-- rather than materialise every intermediate value. +-- +windowFoldRolling :: forall v a r. FT.Measured v a + => Int + -> F.Fold v r + -> F.Fold a r +windowFoldRolling n = F.purely $ \innerStep innerInit innerExtract -> + let step (w, s) x = + let !w' = W.insert x w + !s' = innerStep s (W.windowMeasure w') + in (w', s') + in F.Fold step (W.empty n :: W.Window v a, innerInit) (innerExtract . snd) + + +-- | Like 'windowFoldRolling' but only feeds a measure to the inner +-- fold once the window has reached its configured size. The first +-- @n - 1@ measures, which would otherwise be over a partially filled +-- window, are dropped. +-- +-- This is usually what you want for statistical measures (rolling +-- mean, variance, quantiles) where partial-window readings are +-- misleading. +-- +windowFoldRollingFull :: forall v a r. FT.Measured v a + => Int + -> F.Fold v r + -> F.Fold a r +windowFoldRollingFull n = F.purely $ \innerStep innerInit innerExtract -> + let step (w, s) x = + let !w' = W.insert x w + !s' = if W.isFull w' + then innerStep s (W.windowMeasure w') + else s + in (w', s') + in F.Fold step (W.empty n :: W.Window v a, innerInit) (innerExtract . snd) + + +-- | Monadic counterpart of 'windowFoldRolling', for combining a sliding +-- window with a 'F.FoldM' (e.g. one that emits each measure to an +-- effectful sink). +-- +windowFoldRollingM :: forall m v a r. (Monad m, FT.Measured v a) + => Int + -> F.FoldM m v r + -> F.FoldM m a r +windowFoldRollingM n = F.impurely $ \innerStep innerInit innerExtract -> + let step (w, s) x = do + let !w' = W.insert x w + !s' <- innerStep s (W.windowMeasure w') + return (w', s') + init' = do + s <- innerInit + return (W.empty n :: W.Window v a, s) + extract' (_, s) = innerExtract s + in F.FoldM step init' extract' + + +-- | Maintain two windows of different sizes over the same input +-- stream, feeding the pair of measures @(short, long)@ to the inner +-- fold at each step. Convenient for cross-window comparisons such as +-- short- versus long-period moving averages. +-- +-- A pair is emitted on every input, starting from one-sample windows. +-- The first @max short long - 1@ pairs therefore include at least one +-- partially-filled window; only from reading @max short long@ onward +-- are both windows full. If those partial readings are unwanted, use +-- 'windowFoldRolling2Full', which suppresses emission until both +-- windows are full. +-- +windowFoldRolling2 :: forall v a r. FT.Measured v a + => Int -- ^ short window size + -> Int -- ^ long window size + -> F.Fold (v, v) r + -> F.Fold a r +windowFoldRolling2 short long = F.purely $ \innerStep innerInit innerExtract -> + let step (ws, wl, s) x = + let !ws' = W.insert x ws + !wl' = W.insert x wl + !s' = innerStep s (W.windowMeasure ws', W.windowMeasure wl') + in (ws', wl', s') + init' = ( W.empty short :: W.Window v a + , W.empty long :: W.Window v a + , innerInit + ) + in F.Fold step init' (\(_, _, s) -> innerExtract s) + + +-- | Like 'windowFoldRolling2' but only emits a pair of measures once +-- __both__ windows are full. Use this when partial-window readings +-- would distort the comparison (typically for any statistical +-- measure). +-- +windowFoldRolling2Full :: forall v a r. FT.Measured v a + => Int -- ^ short window size + -> Int -- ^ long window size + -> F.Fold (v, v) r + -> F.Fold a r +windowFoldRolling2Full short long = F.purely $ \innerStep innerInit innerExtract -> + let step (ws, wl, s) x = + let !ws' = W.insert x ws + !wl' = W.insert x wl + !s' = if W.isFull ws' && W.isFull wl' + then innerStep s (W.windowMeasure ws', W.windowMeasure wl') + else s + in (ws', wl', s') + init' = ( W.empty short :: W.Window v a + , W.empty long :: W.Window v a + , innerInit + ) + in F.Fold step init' (\(_, _, s) -> innerExtract s) + + +-- | Return the window itself at the end of the input stream. Useful +-- when you want to query multiple statistics over the trailing window +-- after consuming a stream. +-- +windowFinal :: forall v a. FT.Measured v a + => Int + -> F.Fold a (W.Window v a) +windowFinal n = F.Fold (\w x -> W.insert x w) (W.empty n) id + + +-- | Return only the measure of the window at the end of the input +-- stream. The most common shorthand when you want a single rolling +-- statistic over the final @n@ samples. +-- +windowMeasureFinal :: forall v a. FT.Measured v a + => Int + -> F.Fold a v +windowMeasureFinal n = W.windowMeasure <$> windowFinal n diff --git a/window-stats/fold/lib/Data/Window/Fold/Timed.hs b/window-stats/fold/lib/Data/Window/Fold/Timed.hs new file mode 100644 index 0000000000..ecc6180470 --- /dev/null +++ b/window-stats/fold/lib/Data/Window/Fold/Timed.hs @@ -0,0 +1,215 @@ +{-# LANGUAGE BangPatterns #-} +{-# LANGUAGE FlexibleContexts #-} +{-# LANGUAGE RankNTypes #-} +{-# LANGUAGE ScopedTypeVariables #-} + +-- | +-- Module : Data.Window.Fold.Timed +-- Description : @foldl@ combinators driving a time-based sliding window. +-- Stability : experimental +-- +-- 'F.Fold' combinators that maintain a sliding time-based window +-- ('Data.Window.Timed.TimedWindow') over the input stream and surface +-- its rolling measure to a downstream fold. +-- +-- All combinators are polymorphic in the time type @t@ (behind a +-- 'TimeLike' constraint), the measure @v@, and the sample type @a@ +-- (behind a 'FT.Measured' constraint), so they work unchanged with +-- the wall-clock ('UTCTime') and monotonic +-- ('Control.Monad.Class.MonadTime.SI.Time') stock instances, with any +-- user-defined 'TimeLike' instance, and with any of the built-in +-- sample wrappers ('Data.Window.Timed.SumSample', +-- 'Data.Window.Timed.MomentSample', 'Data.Window.Timed.MinMaxSample'). +-- +module Data.Window.Fold.Timed + ( windowScan + , windowFoldWith + , windowFoldRolling + , windowFoldRollingFull + , windowFoldRollingM + , windowFoldRolling2 + , windowFoldRolling2Full + , windowFinal + , windowMeasureFinal + ) where + +import Control.Foldl qualified as F +import Data.FingerTree qualified as FT +import Data.Sequence qualified as S + +import Data.Window.Internal.Timed qualified as W +import Data.Window.TimeLike + + +-- | Compute the window measure at each step as the window moves across the +-- input stream, collecting all intermediate measures into a sequence. +-- The sequence has the same length as the input stream, with the i-th element +-- being the measure of the window ending at the i-th input element. +-- +windowScan :: forall t v a. (TimeLike t, FT.Measured v a) + => Dur t + -> F.Fold (t, a) (S.Seq v) +windowScan timedWindowDuration = F.Fold step initial extract + where + step :: (W.TimedWindow t v a, S.Seq v) -> (t, a) -> (W.TimedWindow t v a, S.Seq v) + step (w, ms) (t, a) = + let !w' = W.insert (t, a) w + !ms' = ms S.|> W.windowMeasure w' + in (w', ms') + initial = (W.empty timedWindowDuration, S.empty) + extract = snd + +-- | Apply an extraction function to the window measure at each step, +-- producing a rolling sequence of statistics as the window moves across +-- the input stream. A specialisation of 'windowScan' for when you want +-- to post-process the measure at each step. +-- +windowFoldWith :: (TimeLike t, FT.Measured v a) + => Dur t + -> (v -> b) + -> F.Fold (t, a) (S.Seq b) +windowFoldWith timedWindowDuration f = fmap f <$> windowScan timedWindowDuration + + +-- | Stream the rolling sequence of window measures into an inner +-- 'F.Fold'. Unlike 'windowScan' / 'windowFoldWith', the intermediate +-- measures are __not__ retained — each measure is fed to the inner +-- fold as soon as it is produced. This keeps the memory footprint +-- bounded by the inner fold's own state, independent of the input +-- length. +-- +-- Use this when you want to summarise the rolling statistics +-- (e.g. average them, take their max, fold them into a histogram) +-- rather than materialise every intermediate value. +-- +windowFoldRolling :: forall t v a r. (TimeLike t, FT.Measured v a) + => Dur t + -> F.Fold v r + -> F.Fold (t, a) r +windowFoldRolling timedWindowDuration = F.purely $ \innerStep innerInit innerExtract -> + let step (!w, !s) (t, a) = + let !w' = W.insert (t, a) w + !s' = innerStep s (W.windowMeasure w') + in (w', s') + in F.Fold step (W.empty timedWindowDuration :: W.TimedWindow t v a, innerInit) (innerExtract . snd) + + +-- | Like 'windowFoldRolling' but only feeds a measure to the inner +-- fold once the timestamps span at least the configured duration. +-- Measures over a window that has not yet been "filled" by the +-- duration are dropped. +-- +-- This is usually what you want for statistical measures (rolling +-- mean, variance, quantiles) where partial-window readings are +-- misleading. +-- +windowFoldRollingFull :: forall t v a r. (TimeLike t, FT.Measured v a) + => Dur t + -> F.Fold v r + -> F.Fold (t, a) r +windowFoldRollingFull timedWindowDuration = F.purely $ \innerStep innerInit innerExtract -> + let step (!w, !s) (t, a) = + let !w' = W.insert (t, a) w + !s' = if maybe False (>= timedWindowDuration) (W.windowDuration w') + then innerStep s (W.windowMeasure w') + else s + in (w', s') + in F.Fold step (W.empty timedWindowDuration :: W.TimedWindow t v a, innerInit) (innerExtract . snd) + + +-- | Monadic counterpart of 'windowFoldRolling', for combining a sliding +-- window with a 'F.FoldM' (e.g. one that emits each measure to an +-- effectful sink). +-- +windowFoldRollingM :: forall m t v a r. (Monad m, TimeLike t, FT.Measured v a) + => Dur t + -> F.FoldM m v r + -> F.FoldM m (t, a) r +windowFoldRollingM timedWindowDuration = F.impurely $ \innerStep innerInit innerExtract -> + let step (w, s) (t, a) = do + let !w' = W.insert (t, a) w + !s' <- innerStep s (W.windowMeasure w') + return (w', s') + init' = do + s <- innerInit + return (W.empty timedWindowDuration :: W.TimedWindow t v a, s) + extract' (_, s) = innerExtract s + in F.FoldM step init' extract' + + +-- | Maintain two windows of different durations over the same input +-- stream, feeding the pair of measures @(short, long)@ to the inner +-- fold at each step. Convenient for cross-window comparisons such as +-- short- versus long-period moving averages. +-- +-- A pair is emitted on every input, starting from a one-sample +-- window (zero duration). Until the input timestamps span at least +-- @max short long@, every emitted pair includes at least one +-- not-yet-spanned window. If those partial readings are unwanted, use +-- 'windowFoldRolling2Full', which suppresses emission until both +-- windows have been spanned by their configured duration. +-- +windowFoldRolling2 :: forall t v a r. (TimeLike t, FT.Measured v a) + => Dur t -- ^ short window duration + -> Dur t -- ^ long window duration + -> F.Fold (v, v) r + -> F.Fold (t, a) r +windowFoldRolling2 short long = F.purely $ \innerStep innerInit innerExtract -> + let step (ws, wl, s) (t, a) = + let !ws' = W.insert (t, a) ws + !wl' = W.insert (t, a) wl + !s' = innerStep s (W.windowMeasure ws', W.windowMeasure wl') + in (ws', wl', s') + init' = ( W.empty short :: W.TimedWindow t v a + , W.empty long :: W.TimedWindow t v a + , innerInit + ) + in F.Fold step init' (\(_, _, s) -> innerExtract s) + + +-- | Like 'windowFoldRolling2' but only emits a pair of measures once +-- __both__ windows have been spanned by their configured duration +-- (i.e. @windowDuration w >= configured duration@ for each window). +-- Use this when partial-window readings would distort the comparison +-- (typically for any statistical measure). +-- +windowFoldRolling2Full :: forall t v a r. (TimeLike t, FT.Measured v a) + => Dur t -- ^ short window duration + -> Dur t -- ^ long window duration + -> F.Fold (v, v) r + -> F.Fold (t, a) r +windowFoldRolling2Full short long = F.purely $ \innerStep innerInit innerExtract -> + let step (ws, wl, s) (t, a) = + let !ws' = W.insert (t, a) ws + !wl' = W.insert (t, a) wl + !s' = if maybe False (>= short) (W.windowDuration ws') + && maybe False (>= long) (W.windowDuration wl') + then innerStep s (W.windowMeasure ws', W.windowMeasure wl') + else s + in (ws', wl', s') + init' = ( W.empty short :: W.TimedWindow t v a + , W.empty long :: W.TimedWindow t v a + , innerInit + ) + in F.Fold step init' (\(_, _, s) -> innerExtract s) + + +-- | Return the window itself at the end of the input stream. Useful +-- when you want to query multiple statistics over the trailing window +-- after consuming a stream. +-- +windowFinal :: forall t v a. (TimeLike t, FT.Measured v a) + => Dur t + -> F.Fold (t, a) (W.TimedWindow t v a) +windowFinal timedWindowDuration = + F.Fold (\w ta -> W.insert ta w) (W.empty timedWindowDuration) id + + +-- | Return only the measure of the window at the end of the input +-- stream. The most common shorthand when you want a single rolling +-- statistic over the trailing duration. +-- +windowMeasureFinal :: forall t v a. (TimeLike t, FT.Measured v a) + => Dur t + -> F.Fold (t, a) v +windowMeasureFinal d = W.windowMeasure <$> windowFinal d diff --git a/window-stats/lib/Data/Window/Count.hs b/window-stats/lib/Data/Window/Count.hs new file mode 100644 index 0000000000..9876b776d6 --- /dev/null +++ b/window-stats/lib/Data/Window/Count.hs @@ -0,0 +1,94 @@ +-- | +-- Module : Data.Window.Count +-- Description : Count-based sliding window backed by a finger tree. +-- Stability : experimental +-- +-- This module provides the core sliding window data structure, 'Window', +-- which retains a fixed number of the most recently inserted elements. +-- Elements are evicted from the oldest end of the window automatically +-- on insertion once the window reaches its maximum size. +-- +-- The window is backed by a finger tree ('Data.FingerTree.FingerTree'), +-- giving O(1) amortised insertion (and on-insert eviction); the bulk +-- eviction primitives ('evictOldestN', 'trimByMeasure', 'resize') run +-- in O(log w). The window measure is a product of an element count +-- and a user-supplied monoid @v@, cached at the root of the finger +-- tree and accessible in O(1) via 'windowMeasure'. +-- +-- The user-supplied measure @v@ is updated incrementally as elements +-- are inserted and evicted — no recomputation of the entire window is +-- needed. Combine this module with "Data.Window.DigestCount" from the +-- @with-tdigest@ sublibrary for approximate-quantile statistics, or +-- define your own measure type for other rolling statistics. +-- +-- This module is intended to be imported qualified. +-- +-- = Ordering convention +-- +-- The window's underlying finger tree always has the newest element +-- at the left end and the oldest at the right. The construction +-- functions differ in the expected input order — see the per-function +-- haddocks — so refer to those when feeding pre-collected data. +-- +-- = Example +-- +-- @ +-- -- Build a window of size 5 tracking a running sum: +-- let samples = [10.0, 9.0 .. 1.0] :: [Double] +-- window = fromListN 5 samples :: Window (Sum Double) (SumSample Double) +-- in windowSum window +-- -- => Just 40.0 +-- +-- let window' = insert (11.0 :: Double) window +-- in windowSum window' +-- -- => Just 45.0 +-- @ +-- +-- = Integration with other libraries +-- +-- For approximate quantile statistics, see "Data.Window.DigestCount". +-- For time-based windowing, see "Data.Window.Timed". +-- For a combination of the above two, see "Data.Window.DigestTimed". +-- For integration with the @foldl@ package, see the @with-foldl@ sublibrary. +-- +module Data.Window.Count + ( -- * Types + Window + -- * Construction + , empty + , singleton + -- ** From collections + , fromListN + , fromFoldable + -- * Insertion + , insert + , insertMany + -- * Querying + , size + , windowMaxSize + , isFull + , isEmpty + , windowMeasure + -- ** Statistics + , SumSample (..) + , windowSum + , MinMaxV + , MinMaxSample (..) + , windowMinMax + , MomentSample (..) + , WelfordMeasure + , windowMean + , windowVariance + , windowStdDev + -- * Splitting + , evictOldest + , evictOldestN + , trimByMeasure + , resize + -- * Conversions + , toNewestFirst + , toOldestFirst + ) where + +import Data.Window.Internal.Count +import Data.Window.Internal.Measures diff --git a/window-stats/lib/Data/Window/Internal/Count.hs b/window-stats/lib/Data/Window/Internal/Count.hs new file mode 100644 index 0000000000..fed9c233ba --- /dev/null +++ b/window-stats/lib/Data/Window/Internal/Count.hs @@ -0,0 +1,352 @@ +{-# LANGUAGE BangPatterns #-} +{-# LANGUAGE DeriveAnyClass #-} +{-# LANGUAGE DeriveGeneric #-} +{-# LANGUAGE DerivingStrategies #-} +{-# LANGUAGE FlexibleInstances #-} +{-# LANGUAGE MultiParamTypeClasses #-} +{-# LANGUAGE NamedFieldPuns #-} +{-# LANGUAGE RankNTypes #-} +{-# LANGUAGE UndecidableInstances #-} + +module Data.Window.Internal.Count where + +import Control.DeepSeq +import Data.Coerce +import Data.FingerTree qualified as FT +import Data.Foldable +import Data.Monoid +import GHC.Generics + +import Data.Window.Internal.Measures + +-- | A measure that tracks the count of elements, +-- and the user-supplied measure. +-- +data WindowMeasure v = WindowMeasure + { wmCount :: {-# UNPACK #-} !Int + , wmV :: v + } + deriving (Generic, NFData, Show) + +instance Semigroup v => Semigroup (WindowMeasure v) where + WindowMeasure c1 v1 <> WindowMeasure c2 v2 = + WindowMeasure (c1 + c2) (v1 <> v2) + +instance Monoid v => Monoid (WindowMeasure v) where + {-# INLINE mempty #-} + mempty = WindowMeasure 0 mempty + +-- | A sample wrapper +-- +newtype Sample a = Sample { sampleValue :: a } + deriving (Generic, Show) + deriving anyclass NFData + +instance FT.Measured v a => FT.Measured (WindowMeasure v) (Sample a) where + {-# INLINE measure #-} + measure = WindowMeasure 1 . FT.measure . sampleValue + + +-- | A window over a stream of samples, evicting elements +-- spilling past 'windowMaxSize' +-- +data Window v a = Window + { windowMaxSize :: {-# UNPACK #-} !Int + , windowTree :: !(FT.FingerTree (WindowMeasure v) (Sample a)) + } + deriving (Generic, NFData, Show) + +instance Foldable (Window v) where + foldMap f Window { windowTree } = foldMap (f . sampleValue) windowTree + + +-- | The user-supplied measure of the current window contents. +-- +windowMeasure :: FT.Measured v a + => Window v a + -> v +windowMeasure = wmV . FT.measure . windowTree +{-# INLINE windowMeasure #-} + + +-- | retrieve the sum over the sliding window +-- \(O(1)\) +-- +windowSum :: Num a => Window (Sum a) (SumSample a) -> Maybe a +windowSum w + | size w == 0 = Nothing + | otherwise = Just . getSum . windowMeasure $ w +{-# INLINE windowSum #-} + + +-- | retrieve min/max values over the sliding window +-- \(O(1)\) +-- +windowMinMax :: Ord a => Window (MinMaxV a) (MinMaxSample a) -> Maybe (a, a) +windowMinMax = getMinMaxV . windowMeasure +{-# INLINE windowMinMax #-} + + +-- | retrieve the running mean (via Welford's algorithm) over the +-- sliding window. +-- \(O(1)\) +-- +windowMean :: Fractional a => Window (WelfordMeasure a) (MomentSample a) -> Maybe a +windowMean w + | size w == 0 = Nothing + | otherwise = Just . welfordMean . windowMeasure $ w +{-# INLINE windowMean #-} + + +-- | retrieve the running sample variance (denominator @n - 1@) over +-- the sliding window. +-- \(O(1)\) +-- +windowVariance :: Fractional a => Window (WelfordMeasure a) (MomentSample a) -> Maybe a +windowVariance w + | welfordN m < 2 = Nothing + | otherwise = Just (welfordM2 m / fromIntegral (welfordN m - 1)) + where m = windowMeasure w +{-# INLINE windowVariance #-} + + +-- | retrieve the running sample standard deviation over the sliding +-- window. +-- \(O(1)\) +-- +windowStdDev :: Floating a => Window (WelfordMeasure a) (MomentSample a) -> Maybe a +windowStdDev = fmap sqrt . windowVariance +{-# INLINE windowStdDev #-} + + +-- | Returns the count of elements in the window +-- \(O(1)\) +-- +size :: FT.Measured v a + => Window v a + -> Int +size = wmCount . FT.measure . windowTree +{-# INLINE size #-} + + +-- | Constructs an empty window with capacity of @windowMaxSize@ +-- +empty :: FT.Measured v a => Int -> Window v a +empty windowMaxSize = Window { windowMaxSize, windowTree = FT.empty } + + +-- | Constructs a window with capacity of @windowMaxSize@ containing a single sample +-- +singleton :: (FT.Measured v b, Coercible a b) => Int -> a -> Window v b +singleton windowMaxSize a = + Window { windowMaxSize, windowTree = FT.singleton (Sample (coerce a)) } + + +-- | Insert a new sample, then evict if the window has +-- exceeded @windowMaxSize@. +-- +-- @a@ must be the unwrapped version of b. +-- +-- /Invariant:/ assumes the input window satisfies +-- @'size' w '<=' 'windowMaxSize' w@. Insert evicts at most one +-- element, so handing 'insert' an over-full window leaves it +-- over-full. Windows produced by any function in this module satisfy +-- the invariant, so this only matters if you construct or mutate the +-- 'Window' record by hand. +-- +-- \(O(1)\) amortized, \(O(\log w)\) w/c +-- +insert :: (FT.Measured v b, Coercible a b) + => a + -> Window v b + -> Window v b +insert val win@Window { windowMaxSize, windowTree } = win { windowTree = ftree' } + where + ftree = Sample (coerce val) FT.<| windowTree + ftree' = if size win >= windowMaxSize + then case FT.viewr ftree of + FT.EmptyR -> error "impossible" + prefix FT.:> _ -> prefix + else ftree + !_ = FT.measure ftree' + + +-- | Insert many samples into the window, evicting as soon +-- as the size exceeds @windowMaxSize@. For a bulk add followed by single trim +-- operation, see 'fromFoldable'. +-- Elements should be provided in in order of freshest samples in the __tail__. +-- \(O(\n log w)\), use when n >> w +-- +insertMany :: (FT.Measured v b, Foldable f, Coercible a b) + => f a + -> Window v b + -> Window v b +insertMany as w = foldl' (flip insert) w as + + +-- | Discard the oldest sample +-- \(O(1)\) amortized, \(O(\log w)\) w/c +-- +evictOldest :: FT.Measured v a + => Window v a + -> Window v a +evictOldest win@Window { windowTree } = + case FT.viewr windowTree of + FT.EmptyR -> win + prefix FT.:> _ -> + let !_ = FT.measure prefix + in win { windowTree = prefix } + + +-- | Discard N oldest samples +-- \(O(\log w)\) +-- +evictOldestN :: FT.Measured v a + => Int + -> Window v a + -> Window v a +evictOldestN n win + | n <= 0 = win + | n >= total = win { windowTree = FT.empty } + | otherwise = takeUntil (\v -> wmCount v > keep) win + where + total = size win + keep = total - n + + +-- | Returns the longest newest-end prefix of the window whose +-- accumulated 'WindowMeasure' does __not__ satisfy the predicate. +-- Equivalently, walks the window from newest to oldest, accumulating +-- the measure, and stops at the first point where the predicate flips +-- to 'True' — discarding everything from that point on (the oldest +-- tail). +-- +-- /Precondition:/ the predicate must be __monotone__ along the prefix: +-- once it returns 'True' for some prefix, it must continue to return +-- 'True' for every longer prefix. Calling 'takeUntil' with a +-- non-monotone predicate (for example one phrased over the running +-- variance carried by 'WelfordMeasure') yields an unspecified split +-- point and is almost never what you want. +-- +-- This function lives in the internal module as the primitive used by +-- 'evictOldestN' and 'trimByMeasure'. Public callers should prefer +-- 'Data.Window.Count.trimByMeasure' (which hides 'WindowMeasure' and +-- works on the user-supplied measure @v@) or the dedicated helper +-- 'evictOldestN'. +-- \(O(\log w)\) +-- +takeUntil :: FT.Measured v a + => (WindowMeasure v -> Bool) + -> Window v a + -> Window v a +takeUntil p win@Window { windowTree } = win { windowTree = windowTree' } + where + windowTree' = FT.takeUntil p windowTree + !_ = wmV $ FT.measure windowTree' + + +-- | Trim the window from the oldest end, keeping the longest +-- newest-end prefix whose cumulative user-supplied measure @v@ does +-- __not__ satisfy the predicate. +-- +-- For example, to retain only the most recent samples whose cumulative +-- weight (a 'Data.Monoid.Sum') stays under a budget: +-- +-- @ +-- trimByMeasure (\\s -> 'Data.Monoid.getSum' s > budget) window +-- @ +-- +-- /Precondition:/ the predicate must be __monotone__ along the prefix. +-- See 'takeUntil' for the full caveat. +-- \(O(\log w)\) +-- +trimByMeasure :: FT.Measured v a + => (v -> Bool) + -> Window v a + -> Window v a +trimByMeasure p = takeUntil (p . wmV) + + +-- | Set a new maximum size for the window, evicting the oldest +-- samples if the new size is smaller than the current count. +-- \(O(\log w)\) +-- +resize :: FT.Measured v a + => Int + -> Window v a + -> Window v a +resize n win + | n <= 0 = win { windowMaxSize = 0, windowTree = FT.empty } + | size win > n = let win' = win { windowMaxSize = n } + in takeUntil (\v -> wmCount v > n) win' + | otherwise = win { windowMaxSize = n } + + +-- | Tests if window is full +-- \(O(1)\) +-- +isFull :: FT.Measured v a => Window v a -> Bool +isFull w = size w >= windowMaxSize w +{-# INLINE isFull #-} + + +-- | Tests if window is empty +-- \(O(1)\) +-- +isEmpty :: FT.Measured v a => Window v a -> Bool +isEmpty = (== 0) . size +{-# INLINE isEmpty #-} + + +-- | This is a more efficient alternative to 'fromFoldable' for +-- creating a window from a list for cases where the length of the input +-- list is significantly greater than the provided window size: +-- only the first @windowMaxSize@ elements of the input are consumed. +-- +-- Elements should be provided in the order of freshest samples at the __head__ +-- for correct behaviour (__note__: this is opposite from 'insertMany' and 'fromFoldable') +-- \(O(\w log w)\) +-- +fromListN :: forall v a b. (FT.Measured v b, Coercible a b) + => Int -- ^ window size + -> [a] + -> Window v b +fromListN windowMaxSize as = + Window + { windowMaxSize + , windowTree = windowTree' + } + where + windowTree' = FT.fromList . map (Sample . coerce) . take windowMaxSize $ as + !_ = FT.measure windowTree' + + +-- | Construct a window from a 'Foldable' collection, +-- which first creates a window from all available elements and performs +-- a single trim operation at the end. +-- Elements should be provided in the order of freshest samples in the __tail__ +-- \(O(\n log n)\), use when n ~= w +-- +fromFoldable :: forall t v a b. (FT.Measured v b, Foldable t, Coercible a b) + => Int -- ^ window size + -> t a + -> Window v b +fromFoldable n as = takeUntil (\v -> wmCount v > n) win + where + win = Window { windowMaxSize = n, windowTree } + windowTree = foldl' (\ft a -> Sample (coerce a) FT.<| ft) FT.empty as + + +-- | Yields the window in order, ie. newest elements at the head +-- \(O(n)\) +-- +toNewestFirst :: Coercible a b => Window v a -> [b] +toNewestFirst = coerce . toList +{-# INLINE toNewestFirst #-} + + +-- | Yields the window in reverse, ie. oldest elements at the head +-- \(O(n)\) +-- +toOldestFirst :: Coercible a b => Window v a -> [b] +toOldestFirst = coerce . foldl' (flip (:)) [] diff --git a/window-stats/lib/Data/Window/Internal/Measures.hs b/window-stats/lib/Data/Window/Internal/Measures.hs new file mode 100644 index 0000000000..341f34f95d --- /dev/null +++ b/window-stats/lib/Data/Window/Internal/Measures.hs @@ -0,0 +1,121 @@ +{-# LANGUAGE BangPatterns #-} +{-# LANGUAGE DeriveAnyClass #-} +{-# LANGUAGE DeriveGeneric #-} +{-# LANGUAGE DerivingStrategies #-} +{-# LANGUAGE FlexibleInstances #-} +{-# LANGUAGE GeneralizedNewtypeDeriving #-} +{-# LANGUAGE MultiParamTypeClasses #-} +{-# LANGUAGE RankNTypes #-} +{-# LANGUAGE UndecidableInstances #-} + +-- | +-- Module : Data.Window.Internal.Measures +-- Description : Prebuilt sample wrappers and cached measure types. +-- Stability : internal +-- +-- The sample wrappers ('SumSample', 'MinMaxSample', 'MomentSample') +-- and their associated cached-measure types ('MinMaxV', +-- 'WelfordMeasure') used by the prebuilt statistics in +-- 'Data.Window.Count' / 'Data.Window.Timed'. The public modules +-- re-export the sample wrappers (with constructors) and the measure +-- types (opaquely); the measure internals — constructors and field +-- accessors — live here. +-- +-- This module is __internal__. Importing it grants the access needed +-- for things like combining 'WelfordMeasure' values from disjoint +-- windows, computing population variance from the running +-- 'welfordM2' / 'welfordN', serialising the cached measure for +-- telemetry, or seeding a window with a precomputed measure. Use at +-- your own risk: field names and representations may change between +-- minor versions without corresponding changes to the stable public +-- API. +-- +module Data.Window.Internal.Measures + ( -- * Sum + SumSample (..) + -- * Min/Max + , MinMaxSample (..) + , MinMaxV (..) + -- * Welford moments + , MomentSample (..) + , WelfordMeasure (..) + ) where + +import Control.Applicative ((<|>)) +import Control.DeepSeq +import Data.FingerTree +import Data.Monoid +import GHC.Generics + +-- | Wrapper for computing sums over the sliding window +-- +newtype SumSample a = SumSample { getSumSample :: a } + deriving (Generic, Show) + deriving anyclass NFData + deriving newtype Num + +instance Num a => Measured (Sum a) (SumSample a) where + {-# INLINE measure #-} + measure (SumSample s) = Sum s + + +newtype MinMaxSample a = MinMaxSample { getMinMaxSample :: a } + deriving (Generic, Show) + deriving anyclass NFData + +-- | Wrapper for capturing min/max over the sliding window +newtype MinMaxV a = MinMaxV { getMinMaxV :: Maybe (a, a) } + deriving (Eq, Show) + +instance Ord a => Semigroup (MinMaxV a) where + {-# INLINEABLE (<>) #-} + MinMaxV v1 <> MinMaxV v2 = + case (v1, v2) of + (Just (l1, u1), Just (l2, u2)) -> + let !l = min l1 l2 + !u = max u1 u2 + in MinMaxV $ Just (l, u) + _otherwise -> MinMaxV $ v1 <|> v2 + +instance Ord a => Monoid (MinMaxV a) where + {-# INLINE mempty #-} + mempty = MinMaxV Nothing + +instance Ord a => Measured (MinMaxV a) (MinMaxSample a) where + {-# INLINE measure #-} + measure (MinMaxSample s) = MinMaxV $ Just (s, s) + + +newtype MomentSample a = MomentSample { getMomentSample :: a } + deriving (Generic, Show) + deriving anyclass NFData + +data WelfordMeasure a = WelfordMeasure + { welfordN :: {-# UNPACK #-} !Int + , welfordMean :: !a + , welfordM2 :: !a + } + deriving (Generic, NFData, Eq, Show) + +instance (Fractional a) => Semigroup (WelfordMeasure a) where + {-# INLINEABLE (<>) #-} + WelfordMeasure nA mA m2A <> WelfordMeasure nB mB m2B + | nA == 0 = WelfordMeasure nB mB m2B + | nB == 0 = WelfordMeasure nA mA m2A + | otherwise = + let n = nA + nB + delta = mB - mA + mean = mA + delta * fromIntegral nB / fromIntegral n + m2 = m2A + m2B + delta * delta + * fromIntegral nA + * fromIntegral nB + / fromIntegral n + in WelfordMeasure n mean m2 + +instance Fractional a => Monoid (WelfordMeasure a) where + {-# INLINE mempty #-} + mempty = WelfordMeasure 0 0 0 + +instance Fractional a => Measured (WelfordMeasure a) (MomentSample a) where + {-# INLINE measure #-} + measure (MomentSample x) = WelfordMeasure 1 x 0 diff --git a/window-stats/lib/Data/Window/Internal/Timed.hs b/window-stats/lib/Data/Window/Internal/Timed.hs new file mode 100644 index 0000000000..e7273f4bdb --- /dev/null +++ b/window-stats/lib/Data/Window/Internal/Timed.hs @@ -0,0 +1,444 @@ +{-# LANGUAGE BangPatterns #-} +{-# LANGUAGE DeriveAnyClass #-} +{-# LANGUAGE DeriveGeneric #-} +{-# LANGUAGE FlexibleContexts #-} +{-# LANGUAGE FlexibleInstances #-} +{-# LANGUAGE MultiParamTypeClasses #-} +{-# LANGUAGE NamedFieldPuns #-} +{-# LANGUAGE StandaloneDeriving #-} +{-# LANGUAGE TypeFamilies #-} +{-# LANGUAGE UndecidableInstances #-} + +module Data.Window.Internal.Timed where + +import Control.DeepSeq +import Control.Monad.Class.MonadTime.SI (Time) +import Data.Coerce +import Data.FingerTree (FingerTree, ViewL (..), ViewR (..)) +import Data.FingerTree qualified as FT +import Data.Foldable +import Data.Maybe (listToMaybe) +import Data.Monoid (Last(..), Sum(..)) +import Data.Time (UTCTime) +import GHC.Generics + +import Data.Window.Internal.Measures +import Data.Window.TimeLike + + +-- | A measure that tracks the count of elements, the timestamp +-- bounds, and the user-supplied measure. +-- +data TimedMeasure t v = TimedMeasure + { tmStart :: !(Maybe t) + -- ^ oldest timestamp in the window prefix + , tmCount :: {-# UNPACK #-} !Int + -- ^ number of samples in the window prefix + , tmV :: v + -- ^ user supplied measure + } + deriving (Generic, NFData, Show) + +instance Semigroup v => Semigroup (TimedMeasure t v) where + {-# INLINE (<>) #-} + TimedMeasure _s1 c1 v1 <> TimedMeasure s2 c2 v2 = + TimedMeasure s2 (c1 + c2) (v1 <> v2) + +instance Monoid v => Monoid (TimedMeasure t v) where + {-# INLINE mempty #-} + mempty = TimedMeasure Nothing 0 mempty + +instance FT.Measured v a => FT.Measured (TimedMeasure t v) (TimedSample t a) where + {-# INLINE measure #-} + measure (TimedSample t a) = + TimedMeasure + { tmStart = Just t + , tmCount = 1 + , tmV = FT.measure a + } + + +-- | Configured maximum duration of the window. +windowMaxDuration :: TimedWindow t v a -> Dur t +windowMaxDuration = twDuration +{-# INLINE windowMaxDuration #-} + + +-- | A sample wrapper with an associated timestamp. +-- +data TimedSample t a = TimedSample + { tsNow :: !t + , tsValue :: !a + } + deriving (Generic, NFData, Show) + + +-- | A window over a sequence of timestamped samples within the given +-- duration. Values are stored in non-increasing timestamp order, ie. +-- with freshest samples at the head. +data TimedWindow t v a = TimedWindow + { twDuration :: !(Dur t) + , twTree :: !(FingerTree (TimedMeasure t v) (TimedSample t a)) + } + deriving Generic + +deriving instance (Show (Dur t), Show t, Show v, Show a) + => Show (TimedWindow t v a) +deriving instance (NFData (Dur t), NFData t, NFData v, NFData a) + => NFData (TimedWindow t v a) + +instance Foldable (TimedWindow t v) where + foldMap f TimedWindow { twTree } = + foldMap (f . tsValue) twTree + + +-- | The user-supplied measure of the current window contents. +-- +windowMeasure :: FT.Measured v a + => TimedWindow t v a + -> v +windowMeasure = tmV . FT.measure . twTree +{-# INLINE windowMeasure #-} + + +-- | retrieve the sum over the sliding window +-- \(O(1)\) +-- +windowSum :: Num a + => TimedWindow t (Sum a) (SumSample a) -> Maybe a +windowSum w + | size w == 0 = Nothing + | otherwise = Just . getSum . windowMeasure $ w +{-# INLINE windowSum #-} + + +-- | retrieve min/max values over the sliding window +-- \(O(1)\) +-- +windowMinMax :: Ord a + => TimedWindow t (MinMaxV a) (MinMaxSample a) -> Maybe (a, a) +windowMinMax = getMinMaxV . windowMeasure +{-# INLINE windowMinMax #-} + + +-- | retrieve the running mean (via Welford's algorithm) over the +-- sliding window. +-- \(O(1)\) +-- +windowMean :: Fractional a + => TimedWindow t (WelfordMeasure a) (MomentSample a) -> Maybe a +windowMean w + | size w == 0 = Nothing + | otherwise = Just . welfordMean . windowMeasure $ w +{-# INLINE windowMean #-} + + +-- | retrieve the running sample variance (denominator @n - 1@) over +-- the sliding window. +-- \(O(1)\) +-- +windowVariance :: Fractional a + => TimedWindow t (WelfordMeasure a) (MomentSample a) -> Maybe a +windowVariance w + | welfordN m < 2 = Nothing + | otherwise = Just (welfordM2 m / fromIntegral (welfordN m - 1)) + where m = windowMeasure w +{-# INLINE windowVariance #-} + + +-- | retrieve the running sample standard deviation over the sliding +-- window. +-- \(O(1)\) +-- +windowStdDev :: Floating a + => TimedWindow t (WelfordMeasure a) (MomentSample a) -> Maybe a +windowStdDev = fmap sqrt . windowVariance +{-# INLINE windowStdDev #-} + + +-- | The number of elements currently in the window. +-- \(O(1)\) +-- +size :: FT.Measured v a + => TimedWindow t v a + -> Int +size = tmCount . FT.measure . twTree +{-# INLINE size #-} + + +-- | Actual duration covered by the samples in the window. 'Nothing' +-- for an empty window. +-- \(O(1)\) +-- +windowDuration :: (TimeLike t, FT.Measured v a) + => TimedWindow t v a + -> Maybe (Dur t) +windowDuration TimedWindow { twTree } = + do + start <- tmStart $ FT.measure twTree + finish <- finish' + pure $! finish `diffT` start + where + finish' = case FT.viewl twTree of + EmptyL -> error "TimedWindow: empty tree with non-empty measure - internal invariant broken" + TimedSample t _a :< _suffix -> Just t + + +-- | Constructs an empty time window of the given @timedWindowDuration@ duration +-- +empty :: FT.Measured v a + => Dur t + -> TimedWindow t v a +empty twDuration = + TimedWindow { twDuration, twTree = FT.empty } + + +-- | Constructs a window with duration of @timedWindowDuration@ containing a single sample +-- +singleton :: (FT.Measured v b, Coercible a b) + => Dur t + -> (t, a) + -> TimedWindow t v b +singleton twDuration (t, a) = + TimedWindow + { twDuration + , twTree = FT.singleton (TimedSample t (coerce a)) + } + + +-- | Insert a new timestamped sample, then evict any elements +-- that have spilled past the window duration. +-- +-- @a@ must be the unwrapped version of b. +-- \(O(1)\) amortized, \(O(\log w)\) w/c +-- +insert :: (TimeLike t, FT.Measured v b, Coercible a b) + => (t, a) + -> TimedWindow t v b + -> TimedWindow t v b +insert (t, a) win@TimedWindow { twDuration, twTree } = + evictBefore cutoff win { twTree = TimedSample t (coerce a) FT.<| twTree } + where + cutoff = negate twDuration `addT` t +{-# INLINEABLE insert #-} + + +-- | Insert a sequence of timestamped samples into the window, performing +-- a single trim operation at the end. +-- +-- Elements should be provided in __non-decreasing__ timestamp order for +-- correct eviction behaviour. +-- @a@ must be the unwrapped version of b. +-- \(O(\n log w)\), use when n >> w +-- +insertMany :: (TimeLike t, FT.Measured v b, Foldable f, Coercible a b) + => f (t, a) + -> TimedWindow t v b + -> TimedWindow t v b +insertMany as win@TimedWindow { twDuration, twTree } = maybe win (`evictBefore` toWin twTree') cutoff + where + step (ft, finish') (t, a) = (TimedSample t (coerce a) FT.<| ft, finish' <> pure t) + (!twTree', !finish) = foldl' step (twTree, mempty) as + cutoff = do + finish' <- getLast finish + pure $ negate twDuration `addT` finish' + + toWin ft = win { twTree = ft } + + +-- | Build a 'TimedWindow' from a list of timestamped samples, retaining +-- only elements within the given duration of the newest element. +-- +-- Elements should be provided in __non-increasing__ timestamp order +-- for correct behaviour (__note__: this is opposite from 'insertMany' and 'fromFoldable') +-- \(O(\w log w)\) +-- +fromListN :: (TimeLike t, FT.Measured v b, Coercible a b) + => Dur t + -> [(t, a)] + -> TimedWindow t v b +fromListN twDuration as = maybe (empty twDuration) (`evictBefore` win) cutoff + where + step ft (t, a) = ft FT.|> TimedSample t (coerce a) + twTree = foldl' step mempty as + cutoff = do + finish' <- fst <$> listToMaybe as + pure $ negate twDuration `addT` finish' + + win = TimedWindow { twDuration, twTree } + + +-- | Construct a window from a 'Foldable' collection, performing +-- a single trim operation at the end. +-- Elements should be provided in __non-decreasing__ timestamp order for +-- correct eviction behaviour. +-- \(O(\n log n)\), use when n ~= w +-- +fromFoldable :: (TimeLike t, FT.Measured v b, Foldable f, Coercible a b) + => Dur t + -> f (t, a) + -> TimedWindow t v b +fromFoldable twDuration as = maybe (empty twDuration) (`evictBefore` win) cutoff + where + step (ft, finish') (t, a) = (TimedSample t (coerce a) FT.<| ft, finish' <> pure t) + (!twTree, !finish) = foldl' step (mempty, mempty) as + cutoff = do + finish' <- getLast finish + pure $ negate twDuration `addT` finish' + + win = TimedWindow { twDuration, twTree } + + +-- | Trim the window from the oldest end, keeping the longest +-- newest-end prefix whose cumulative user-supplied measure @v@ does +-- __not__ satisfy the predicate. +-- +-- /Precondition:/ the predicate must be __monotone__ along the prefix. +-- See 'takeUntil' for the full caveat. +-- \(O(\log w)\) +-- +trimByMeasure :: FT.Measured v a + => (v -> Bool) + -> TimedWindow t v a + -> TimedWindow t v a +trimByMeasure p = takeUntil (p . tmV) +{-# INLINE trimByMeasure #-} + + +-- | Discard the oldest sample. +-- \(O(1)\) amortized, \(O(\log w)\) w/c +-- +evictOldest :: FT.Measured v a + => TimedWindow t v a + -> TimedWindow t v a +evictOldest win@TimedWindow { twTree } = + case FT.viewr twTree of + EmptyR -> win + prefix :> _ -> win { twTree = prefix } + + +-- | Discard the @n@ oldest samples. +-- \(O(\log w)\) +-- +evictOldestN :: FT.Measured v a + => Int + -> TimedWindow t v a + -> TimedWindow t v a +evictOldestN n win + | n <= 0 = win + | n >= total = win { twTree = FT.empty } + | otherwise = takeUntil (\m -> tmCount m > keep) win + where + total = size win + keep = total - n + + +-- | Evict all elements whose timestamp is older than the given cutoff time. +-- \(O(\log w)\) +-- +evictBefore :: (TimeLike t, FT.Measured v a) + => t + -> TimedWindow t v a + -> TimedWindow t v a +evictBefore cutoff = + takeUntil (maybe False (<= cutoff) . tmStart) + +{-# INLINEABLE evictBefore #-} +{-# SPECIALIZE evictBefore + :: FT.Measured v a + => UTCTime -> TimedWindow UTCTime v a -> TimedWindow UTCTime v a #-} +{-# SPECIALIZE evictBefore + :: FT.Measured v a + => Time -> TimedWindow Time v a -> TimedWindow Time v a #-} + + +-- | Set a new maximum duration for the window, evicting any samples +-- now older than that duration relative to the newest element. +-- \(O(\log w)\) +-- +resize :: (TimeLike t, FT.Measured v a) + => Dur t + -> TimedWindow t v a + -> TimedWindow t v a +resize twDuration win@TimedWindow { twTree } = + case FT.viewl twTree of + EmptyL -> win { twDuration } + TimedSample t _a :< _suffix -> + let cutoff = negate twDuration `addT` t + in evictBefore cutoff win { twDuration } + + +-- | Tests if window is empty +-- \(O(1)\) +-- +isEmpty :: FT.Measured v a => TimedWindow t v a -> Bool +isEmpty = (== 0) . size +{-# INLINE isEmpty #-} + + +-- | Yields the window in order, ie. newest elements at the head +-- \(O(w)\) +-- +toNewestFirst :: Coercible a b => TimedWindow t v a -> [b] +toNewestFirst = coerce . toList +{-# INLINE toNewestFirst #-} + + +-- | Yields the window in reverse, ie. oldest elements at the head +-- +toOldestFirst :: Coercible a b => TimedWindow t v a -> [b] +toOldestFirst = coerce . foldl' (flip (:)) [] + + +-- | Yields the window in order, ie. newest elements at the head, +-- with timestamps. +-- \(O(w)\) +-- +toTimedNewestFirst :: Coercible a b => TimedWindow t v a -> [(t, b)] +toTimedNewestFirst = + foldr (\(TimedSample t a) acc -> (t, coerce a) : acc) [] + . twTree +{-# INLINE toTimedNewestFirst #-} + + +-- | Yields the window in reverse, ie. oldest elements at the head +-- with timestamps +-- \(O(w)\) +-- +toTimedOldestFirst :: Coercible a b => TimedWindow t v a -> [(t, b)] +toTimedOldestFirst = + foldl' (\acc (TimedSample t a) -> (t, coerce a) : acc) [] + . twTree + + +--- INTERNAL --- + +-- | Returns the longest prefix of the window whose +-- accumulated 'TimedMeasure' does __not__ satisfy the predicate. +-- Equivalently, walks the window from newest to oldest, accumulating +-- the measure, and stops at the first point where the predicate flips +-- to 'True' — discarding everything from that point on (the oldest +-- tail). +-- +-- /Precondition:/ the predicate must be __monotone__ along the prefix: +-- once it returns 'True' for some prefix, it must continue to return +-- 'True' for every longer prefix. Calling 'takeUntil' with a +-- non-monotone predicate (for example one phrased over the running +-- variance carried by 'WelfordMeasure') yields an unspecified split +-- point and is almost never what you want. +-- +-- This function lives in the internal module as the primitive used by +-- 'evictBefore' and 'trimByMeasure'. Public callers should prefer +-- 'Data.Window.Timed.trimByMeasure' (which hides 'TimedMeasure' and +-- works on the user-supplied measure @v@) or the dedicated helper +-- 'evictBefore'. +-- \(O(\log w)\) +-- +takeUntil :: FT.Measured v a + => (TimedMeasure t v -> Bool) + -> TimedWindow t v a + -> TimedWindow t v a +takeUntil p win@TimedWindow { twTree } = win { twTree = twTree' } + where + twTree' = FT.takeUntil p twTree + !_ = tmV $ FT.measure twTree' diff --git a/window-stats/lib/Data/Window/TimeLike.hs b/window-stats/lib/Data/Window/TimeLike.hs new file mode 100644 index 0000000000..1ffc9ea61e --- /dev/null +++ b/window-stats/lib/Data/Window/TimeLike.hs @@ -0,0 +1,73 @@ +{-# LANGUAGE FlexibleContexts #-} +{-# LANGUAGE TypeFamilies #-} +{-# LANGUAGE TypeFamilyDependencies #-} + +-- | +-- Module : Data.Window.TimeLike +-- Description : Abstraction over absolute time and its duration. +-- Stability : experimental +-- +-- The time-based window machinery in 'Data.Window.Timed' is +-- parametrised over an absolute time type @t@ via the 'TimeLike' +-- class. Two stock instances are provided: +-- +-- * 'Data.Time.UTCTime' / 'Data.Time.NominalDiffTime' — wall-clock +-- semantics, standard from the @time@ package. +-- +-- * 'Control.Monad.Class.MonadTime.SI.Time' / 'Data.Time.DiffTime' +-- — monotonic semantics from @io-classes@/@si-timers@. Preferred +-- when sliding-window correctness must not be perturbed by NTP +-- adjustments or clock jumps. +-- +-- Users may add their own instance for any time representation that +-- satisfies the laws below. The internal windowing code is marked +-- 'INLINEABLE' and 'SPECIALIZE'd for the two stock instances, so a +-- downstream user wanting dictionary-free code for a custom time +-- type only needs to add a 'SPECIALIZE' pragma on the relevant +-- functions in their own module. +-- +-- = Laws +-- +-- For any @t1, t2 :: t@ and @d :: 'Dur' t@: +-- +-- @ +-- ('diffT' t2 t1) \`'addT'\` t1 == t2 +-- 'diffT' ('addT' d t) t == d +-- @ +-- +module Data.Window.TimeLike (TimeLike (..)) where + +import Control.Monad.Class.MonadTime.SI + +-- | Absolute time type @t@ paired with its associated duration type +-- @'Dur' t@. +-- +-- The 'Dur' type family is __injective__: given the duration GHC can +-- recover the time, so users rarely need explicit type annotations. +-- +class (Ord t, Ord (Dur t), Num (Dur t)) => TimeLike t where + -- | The duration type associated with @t@. + type Dur t = d | d -> t + + -- | @'diffT' t2 t1@ is the duration from @t1@ to @t2@. + diffT :: t -> t -> Dur t + + -- | @'addT' d t@ shifts @t@ forward (or backward, for negative @d@) + -- by @d@. + addT :: Dur t -> t -> t + + +instance TimeLike UTCTime where + type Dur UTCTime = NominalDiffTime + diffT = diffUTCTime + {-# INLINE diffT #-} + addT = addUTCTime + {-# INLINE addT #-} + + +instance TimeLike Time where + type Dur Time = DiffTime + diffT (Time a) (Time b) = a - b + {-# INLINE diffT #-} + addT d (Time t) = Time (d + t) + {-# INLINE addT #-} diff --git a/window-stats/lib/Data/Window/Timed.hs b/window-stats/lib/Data/Window/Timed.hs new file mode 100644 index 0000000000..9db8bdf90c --- /dev/null +++ b/window-stats/lib/Data/Window/Timed.hs @@ -0,0 +1,81 @@ +-- | +-- Module : Data.Window.Timed +-- Description : Time-based sliding window backed by a finger tree. +-- Stability : experimental +-- +-- This module provides a time-based sliding window data structure, +-- 'TimedWindow', which retains elements within a specified duration +-- of the newest element in the window. Elements older than the +-- duration are evicted automatically on insertion. +-- +-- The time representation is abstracted via the 'TimeLike' class. Two +-- stock instances are provided in "Data.Window.TimeLike": +-- +-- * @'TimedWindow' 'Data.Time.UTCTime' v a@ for wall-clock timestamps +-- from the @time@ package. +-- +-- * @'TimedWindow' 'Control.Monad.Class.MonadTime.SI.Time' v a@ for +-- monotonic timestamps from @io-classes@/@si-timers@. +-- +-- The window is backed by a finger tree ('Data.FingerTree.FingerTree'), +-- giving O(1) amortised insertion and O(log w) eviction. The window +-- measure is cached at the root of the finger tree and accessible in +-- O(1) via 'windowMeasure', making it efficient to query rolling +-- statistics. +-- +-- This module is intended to be imported qualified. +-- +-- = Ordering convention +-- +-- The window's underlying finger tree always has the newest element +-- at the left end and the oldest at the right. The construction +-- functions differ in the expected input order — see the per-function +-- haddocks — so refer to those when feeding pre-collected data. +-- +module Data.Window.Timed + ( -- * Types + TimedWindow + -- * Time abstraction + , TimeLike (..) + -- * Construction + , empty + , singleton + -- * From collections + , fromListN + , fromFoldable + -- * Insertion + , insert + , insertMany + -- * Querying + , size + , windowMaxDuration + , isEmpty + , windowDuration + , windowMeasure + -- ** Statistics + , SumSample (..) + , windowSum + , MinMaxV + , MinMaxSample (..) + , windowMinMax + , MomentSample (..) + , WelfordMeasure + , windowMean + , windowVariance + , windowStdDev + -- * Splitting + , evictOldest + , evictOldestN + , evictBefore + , trimByMeasure + , resize + -- * Conversions + , toNewestFirst + , toTimedNewestFirst + , toOldestFirst + , toTimedOldestFirst + ) where + +import Data.Window.Internal.Measures +import Data.Window.Internal.Timed +import Data.Window.TimeLike diff --git a/window-stats/tdigest/Data/Window/DigestTimeBatched.hs b/window-stats/tdigest/Data/Window/DigestTimeBatched.hs new file mode 100644 index 0000000000..6353be06b6 --- /dev/null +++ b/window-stats/tdigest/Data/Window/DigestTimeBatched.hs @@ -0,0 +1,64 @@ +-- | +-- Module : Data.Window.DigestTimeBatched +-- Description : Approximate quantile statistics over a time-based sliding window. +-- Stability : experimental +-- +-- A time-based sliding window backed by a t-digest, supporting +-- approximate quantile queries over the specified duration. +-- The window is quantized by buckets of some shorter duration, and +-- samples are inserted into the newest bucket's t-digest object. +-- Once the bucket is sealed, it is inserted into the window, and buckets +-- which spill past the end of the window are evicted. This implies that +-- the granularity of statistics is determined by the ratio of bucket duration +-- to the overall window duration. The digest is cached at the +-- root of the underlying finger tree and accessible in \(O(1)\), so +-- multiple quantiles can be queried cheaply over the same window +-- without recomputing. +-- +-- The time representation is abstracted via 'TimeLike' from +-- "Data.Window.TimeLike", so the same combinators work for wall-clock +-- ('UTCTime') and monotonic +-- ('Control.Monad.Class.MonadTime.SI.Time') timestamps. +-- +-- = Approximate quantiles and memory +-- +-- Estimates are approximate; error is bounded by the compression +-- parameter @comp@ (higher @comp@ = more accurate, larger digest). +-- The digest size is \(O(\delta)\) in @comp@, independent of the +-- window's element count — attractive for large or long-running +-- windows where an exact sorted structure would need \(O(w)\) memory. +-- +module Data.Window.DigestTimeBatched + ( -- * Types + TimedDigestWindow + -- * Construction + , empty + -- TODO: From collections + -- , fromListN + -- , fromFoldable + -- * Insertion + , insert + -- TODO: , insertMany + -- * Querying + , sampleCount + , windowDuration + , windowDigest + -- ** Statistics + , windowQuantile + , windowMedian + , windowMean + , windowVariance + , windowStdDev + , windowMinMaxValues + -- * Adjustments + , evictBefore + , reset + -- * re-exports + , TimeLike (..) + , TDigest + ) where + +import Data.TDigest (TDigest) + +import Data.Window.Internal.DigestTimeBatched +import Data.Window.TimeLike diff --git a/window-stats/tdigest/Data/Window/Internal/DigestTimeBatched.hs b/window-stats/tdigest/Data/Window/Internal/DigestTimeBatched.hs new file mode 100644 index 0000000000..c0d55bbd94 --- /dev/null +++ b/window-stats/tdigest/Data/Window/Internal/DigestTimeBatched.hs @@ -0,0 +1,244 @@ +{-# LANGUAGE BangPatterns #-} +{-# LANGUAGE DataKinds #-} +{-# LANGUAGE DeriveAnyClass #-} +{-# LANGUAGE DeriveGeneric #-} +{-# LANGUAGE FlexibleContexts #-} +{-# LANGUAGE KindSignatures #-} +{-# LANGUAGE MultiParamTypeClasses #-} +{-# LANGUAGE NamedFieldPuns #-} +{-# LANGUAGE StandaloneDeriving #-} +{-# LANGUAGE UndecidableInstances #-} + +module Data.Window.Internal.DigestTimeBatched where + +import Control.DeepSeq +import Control.Exception (assert) +import Data.FingerTree (FingerTree, ViewL (..)) +import Data.FingerTree qualified as FT +import Data.TDigest (TDigest) +import Data.TDigest qualified as TD +import Data.TDigest.Internal (Mean) +import GHC.Generics +import GHC.TypeLits + +import Data.Window.TimeLike + + +-- | Finger-tree measure over the sealed buckets. +data BucketMeasure t (comp :: Nat) = BucketMeasure + { bmFinish :: !(Maybe t) + -- ^ Time at insertion of a sealed bucket + , bmCount :: {-# UNPACK #-} !Int + -- ^ Number of samples in the prefix + -- the sliding window is bounded and therefore so is space usage, + -- but tactical forcing will even allow us to not compute the measure + -- for the elements which are evicted from the window, which is + -- fortunate as combining digests, which the finger tree does frequently, + -- is \(O(delta)\). + , bmDigest :: TDigest comp + } + deriving (Show, Generic, NFData) + +-- | A sealed bucket +data SealedBucket t (comp :: Nat) = SealedBucket + { sbFinish :: !t + -- ^ newest sample time ie. when bucket is inserted into the finger tree + , sbCount :: {-# UNPACK #-} !Int + -- ^ Number of samples in the bucket + , sbDigest :: !(TDigest comp) + } + deriving (Show, Generic, NFData) + +instance KnownNat comp => Semigroup (BucketMeasure t comp) where + BucketMeasure _f1 c1 d1 <> BucketMeasure f2 c2 d2 = BucketMeasure f2 (c1 + c2) (d1 <> d2) + +instance KnownNat comp => Monoid (BucketMeasure t comp) where + mempty = BucketMeasure Nothing 0 mempty + +instance KnownNat comp => FT.Measured (BucketMeasure t comp) (SealedBucket t comp) where + {-# INLINE measure #-} + measure (SealedBucket sealed count digest) = BucketMeasure (pure sealed) count digest + +-- | The open (in-progress) bucket. +data OpenBucket t (comp :: Nat) = OpenBucket + { obStart :: !t -- ^ first sample time + , obFinish :: !t -- ^ latest sample time + , obCount :: {-# UNPACK #-} !Int + , obDigest :: !(TDigest comp) + } + deriving (Show, Generic, NFData) + +-- | A time-bucketed sliding window. The finger tree holds sealed buckets, +-- newest at the right; 'tdwBucket' is the bucket currently being filled. +data TimedDigestWindow t (comp :: Nat) = TimedDigestWindow + { tdwDuration :: !(Dur t) -- ^ duration of the whole window + , tdwBucketDuration :: !(Dur t) -- ^ duration of a single bucket + , tdwBucket :: !(Maybe (OpenBucket t comp)) + , tdwTree :: !(FingerTree (BucketMeasure t comp) (SealedBucket t comp)) + , tdwCacheDigest :: !(Maybe (TDigest comp)) + -- ^ Cached (treeDigest <> obDigest); 'Nothing' iff 'tdwBucket' is + -- 'Nothing'. The wrapped 'TDigest' is left as a thunk — set on every + -- 'insert'/'evictBefore' so the first 'windowDigest' after a batch + -- of writes materialises it and subsequent reads project the + -- field directly. + } + +deriving instance (Show (Dur t), Show t) => Show (TimedDigestWindow t comp) + + +-- | An empty window with the given bucket width and retention duration (a multiple +-- of bucket width). +empty :: (TimeLike t, KnownNat comp) => Dur t -> Int -> TimedDigestWindow t comp +empty tdwBucketDuration r = + TimedDigestWindow + { tdwDuration = fromIntegral r * tdwBucketDuration + , tdwBucketDuration + , tdwBucket = Nothing + , tdwTree = FT.empty + , tdwCacheDigest = Nothing + } + + +-- | Resets the window, keeping only the retention durations +reset :: KnownNat comp => TimedDigestWindow t comp -> TimedDigestWindow t comp +reset tdw = tdw { tdwBucket = Nothing, tdwTree = FT.empty, tdwCacheDigest = Nothing } + + +-- | Returns the number of samples which were inserted into the fingertree +-- and which are still within the retention duration. +sampleCount :: KnownNat comp => TimedDigestWindow t comp -> Int +sampleCount TimedDigestWindow { tdwBucket, tdwTree } = + case tdwBucket of + Nothing -> 0 + Just bucket -> let count = bmCount $ FT.measure tdwTree + in count + obCount bucket + + +-- | Returns the digest of the sliding window, or 'Nothing' if there +-- are no samples. Forces the cached 'tdwCacheDigest' on the first +-- call after a batch of writes. +windowDigest :: KnownNat comp => TimedDigestWindow t comp -> Maybe (TDigest comp) +windowDigest tdw = case tdwCacheDigest tdw of + Nothing -> Nothing + Just !d -> Just d + + +-- | Insert a new timestamped sample, then evict any buckets +-- that have spilled past the window duration. Note that eviction +-- occurs at the granularity of buckets so only when a new bucket +-- is added. +insert :: (TimeLike t, KnownNat comp) + => (t, Double) + -> TimedDigestWindow t comp + -> TimedDigestWindow t comp +insert (t, a) tdw@TimedDigestWindow { tdwBucketDuration, tdwDuration, tdwBucket, tdwTree} = + case fillBucket of + (Nothing, bucket) -> + -- Fits: tree is unchanged, only the open bucket grew. Refresh + -- the cache thunk so it closes over the new obDigest. + let cache = Just (bmDigest (FT.measure tdwTree) <> obDigest bucket) + in tdw { tdwBucket = Just $! bucket, tdwCacheDigest = cache } + (Just sealed, bucket) -> + let sample = SealedBucket (obFinish sealed) (obCount sealed) (obDigest sealed) + tdwTree' = tdwTree FT.|> sample + cutoff = negate tdwDuration `addT` t + -- evictBefore forces the tree measure and refreshes the cache w/o forcing + in evictBefore cutoff tdw { tdwBucket = Just $! bucket + , tdwTree = tdwTree' } + where + newBucket = OpenBucket + { obStart = t + , obFinish = t + , obCount = 1 + , obDigest = TD.singleton a + } + + fillBucket = case tdwBucket of + Nothing -> (Nothing, newBucket) + Just bucket@OpenBucket { obStart, obFinish, obCount, obDigest } -> + let dt = t `diffT` obStart + in if dt < tdwBucketDuration + then (Nothing, bucket { obFinish = max obFinish t + , obDigest = TD.insert a obDigest + , obCount = succ obCount + }) + else (tdwBucket, newBucket) + + +-- | Drop all buckets whose newest sample timestamp is older than the given value +-- +evictBefore :: (KnownNat comp, TimeLike t) + => t + -> TimedDigestWindow t comp + -> TimedDigestWindow t comp +evictBefore cutoff win = win { tdwTree = tdwTree' + , tdwBucket = tdwBucket' + , tdwCacheDigest = cache + } + where + tdwBucket' = case tdwBucket win of + Nothing -> Nothing + Just bucket -> if obFinish bucket > cutoff + then tdwBucket win + else assert (FT.null tdwTree') Nothing + + tdwTree' = FT.dropUntil (maybe False (> cutoff) . bmFinish) + (tdwTree win) + -- Force the post-eviction tree measure. This bounds the tail + -- latency of the next 'windowDigest' by one '<>' with 'obDigest', + -- and drops thunks for evicted subtrees without ever combining + -- them. + !treeDigest = bmDigest $ FT.measure tdwTree' + + -- Lazy cache thunk closing over the post-eviction state. + cache = case tdwBucket' of + Nothing -> Nothing + Just bucket -> Just (treeDigest <> obDigest bucket) + +-- | Approximate duration covered by the samples in the window. 'Nothing' +-- for an empty window. The reported value may exceed the actual duration of +-- samples held by up to one bucket period over which actually there aren't +-- any samples held. +-- +windowDuration :: (KnownNat comp, TimeLike t) + => TimedDigestWindow t comp + -> Maybe (Dur t) +windowDuration TimedDigestWindow { tdwBucket, tdwBucketDuration, tdwTree } = + do + start <- start' + finish <- obFinish <$> tdwBucket + pure $! finish `diffT` start + where + start' = case FT.viewl tdwTree of + EmptyL -> obStart <$> tdwBucket + SealedBucket sealed _c _s :< _suffix -> Just $ negate tdwBucketDuration `addT` sealed + + +-- Convenience API -- + +-- | Returns the means of the outermost centroids - appoximations, not the actual min/max samples +windowMinMaxValues :: KnownNat comp => TimedDigestWindow t comp -> Maybe (Mean, Mean) +windowMinMaxValues tdw = + case windowDigest tdw of + Nothing -> Nothing + Just d -> Just (TD.minimumValue d, TD.maximumValue d) + + +windowMedian :: KnownNat comp => TimedDigestWindow t comp -> Maybe Double +windowMedian = maybe Nothing TD.median . windowDigest + + +windowQuantile :: KnownNat comp => Double -> TimedDigestWindow t comp -> Maybe Double +windowQuantile q = maybe Nothing (TD.quantile q) . windowDigest + + +windowMean :: KnownNat comp => TimedDigestWindow t comp -> Maybe Double +windowMean = maybe Nothing TD.mean . windowDigest + + +windowVariance :: KnownNat comp => TimedDigestWindow t comp -> Maybe Double +windowVariance = maybe Nothing TD.variance . windowDigest + + +windowStdDev :: KnownNat comp => TimedDigestWindow t comp -> Maybe Double +windowStdDev = maybe Nothing TD.stddev . windowDigest diff --git a/window-stats/test/Main.hs b/window-stats/test/Main.hs new file mode 100644 index 0000000000..790b1b58a5 --- /dev/null +++ b/window-stats/test/Main.hs @@ -0,0 +1,405 @@ +{-# LANGUAGE CPP #-} +{-# LANGUAGE DataKinds #-} + +module Main (main) where + +#if !MIN_VERSION_base(4,20,0) +import Data.Foldable +#endif +import Data.List (sort, sortBy, tails) +import Data.Maybe (isNothing) +import Data.Monoid (Sum (..)) +import Data.Ord (comparing) +import Data.Time (NominalDiffTime, UTCTime (..), addUTCTime, diffUTCTime, + fromGregorian) +import Test.QuickCheck +import Test.Tasty +import Test.Tasty.QuickCheck (testProperty) + +import Data.Window.Count qualified as C +import Data.Window.DigestTimeBatched qualified as DTB +import Data.Window.Timed qualified as T + + +-- Use Int rather than Double for arithmetic invariants — the +-- finger-tree-cached sum and the naive list sum can disagree by +-- floating-point ULPs even though both are "correct". +type CountW = C.Window (Sum Int) (C.SumSample Int) +type TimedW = T.TimedWindow UTCTime (Sum Int) (T.SumSample Int) + + +newtype Cap = Cap Int deriving Show +instance Arbitrary Cap where + arbitrary = Cap <$> choose (1, 32) + +newtype Dur = Dur NominalDiffTime deriving Show +instance Arbitrary Dur where + arbitrary = Dur . fromInteger <$> choose (1, 60) + + +-- Count --------------------------------------------------------------------- + +-- insertMany agrees with repeated insert (orientation invariant). +prop_insertMany_eq_foldl :: Cap -> [Int] -> Property +prop_insertMany_eq_foldl (Cap n) xs = + let w0 = C.empty n :: CountW + w1 = foldl' (flip C.insert) w0 xs + w2 = C.insertMany xs w0 + in (C.toNewestFirst w1 :: [Int]) === C.toNewestFirst w2 + +-- size after inserting xs into a fresh window is min n (length xs). +prop_size_bounded :: Cap -> [Int] -> Property +prop_size_bounded (Cap n) xs = + let w = C.insertMany xs (C.empty n :: CountW) + in C.size w === min n (length xs) + +-- windowSum matches the live elements' sum. +prop_sum_consistent :: Cap -> [Int] -> Property +prop_sum_consistent (Cap n) xs = + let w = C.insertMany xs (C.empty n :: CountW) + live = C.toNewestFirst w :: [Int] + expected = if null live then Nothing else Just (sum live) + in C.windowSum w === expected + +-- fromFoldable (oldest-first input) gives the same window as the +-- equivalent sequence of inserts. +prop_fromFoldable_orientation :: Cap -> [Int] -> Property +prop_fromFoldable_orientation (Cap n) xs = + let w1 = C.fromFoldable n xs :: CountW + w2 = foldl' (flip C.insert) (C.empty n :: CountW) xs + in (C.toNewestFirst w1 :: [Int]) === C.toNewestFirst w2 + +-- fromListN (newest-first input) is the reverse-input mirror of insert. +prop_fromListN_orientation :: Cap -> [Int] -> Property +prop_fromListN_orientation (Cap n) xs = + let w1 = C.fromListN n (reverse xs) :: CountW + w2 = foldl' (flip C.insert) (C.empty n :: CountW) xs + in (C.toNewestFirst w1 :: [Int]) === C.toNewestFirst w2 + +-- evictOldestN reduces the size as expected. +prop_evictOldestN :: Cap -> [Int] -> NonNegative Int -> Property +prop_evictOldestN (Cap n) xs (NonNegative k) = + let w = C.insertMany xs (C.empty n :: CountW) + w' = C.evictOldestN k w + in C.size w' === max 0 (C.size w - k) + + +-- Timed --------------------------------------------------------------------- + +-- evictBefore retains only samples with timestamp >= cutoff. Input is +-- sorted to honour the non-decreasing-timestamp invariant of insertMany. +prop_evictBefore :: Dur -> [(NonNegative Int, Int)] -> Int -> Property +prop_evictBefore (Dur d) tvs offset = + let base = UTCTime (fromGregorian 2024 1 1) 0 + sorted = sortBy (comparing fst) tvs + pairs = [ (fromIntegral s `addUTCTime` base, v) + | (NonNegative s, v) <- sorted ] + w = T.insertMany pairs (T.empty d :: TimedW) + cutoff = fromIntegral offset `addUTCTime` base + w' = T.evictBefore cutoff w + kept = T.toTimedNewestFirst w' :: [(UTCTime, Int)] + in conjoin [ counterexample (show t) (t >= cutoff) | (t, _) <- kept ] + +-- windowDuration on a freshly-built window from a non-decreasing +-- timestamp list equals (newest - oldest), within the configured cap. +prop_windowDuration :: Dur -> [NonNegative Int] -> Property +prop_windowDuration (Dur d) ts0 = + let sorted = sort (map getNonNegative ts0) + pairs = [ (mkT (fromIntegral s), 0 :: Int) | s <- sorted ] + w = T.insertMany pairs (T.empty d :: TimedW) + live = T.toTimedNewestFirst w :: [(UTCTime, Int)] + in case live of + [] -> T.windowDuration w === Nothing + x:xs -> + let newest = fst x + oldest = fst $ foldl' (\_ x' -> x') x xs + dur = newest `diffUTC` oldest + in T.windowDuration w === Just dur + where + diffUTC a b = realToFrac (utcTimeDiff a b) :: NominalDiffTime + utcTimeDiff a b = + let toS (UTCTime _ d') = realToFrac d' :: Double + in toS a - toS b + + +-- DigestTimeBatched --------------------------------------------------------- + +type Comp = 100 +type DigestW = DTB.TimedDigestWindow UTCTime Comp + +digestBase :: UTCTime +digestBase = UTCTime (fromGregorian 2024 1 1) 0 + +mkT :: Double -> UTCTime +mkT s = realToFrac s `addUTCTime` digestBase + +-- Sorted timestamps uniformly within [0, spanSec], sample values in +-- [-100, 100]. +genTimedSamplesIn + :: Double -- ^ maximum offset in seconds + -> Gen [(UTCTime, Double)] +genTimedSamplesIn spanSec = do + n <- choose (10, 200) + offsets <- sort <$> vectorOf n (choose (0, spanSec)) + values <- vectorOf n (choose (-100, 100)) + return + [ (mkT t, v) | (t, v) <- zip offsets values ] + +-- Same, but all sample values in [0, 100] for the accuracy tests. +genPositiveSamplesIn + :: Double + -> Gen [(UTCTime, Double)] +genPositiveSamplesIn spanSec = do + n <- choose (100, 500) + offsets <- sort <$> vectorOf n (choose (0, spanSec)) + values <- vectorOf n (choose (0, 100)) + return + [ (mkT t, v) | (t, v) <- zip offsets values ] + +buildDigestW :: NominalDiffTime -> Int -> [(UTCTime, Double)] -> DigestW +buildDigestW bucketDur retention = + foldl' (flip DTB.insert) (DTB.empty bucketDur retention) + + +-- | Alternating dense/gap stream. Each cycle emits @nDense@ samples +-- spaced by @denseStep@ seconds, then a silent @gapDur@ before the +-- next cycle. Same generator as the benchmark uses; kept in sync by +-- hand. +cascadeSamples + :: Int -- ^ number of cycles + -> Int -- ^ samples per dense stretch + -> Double -- ^ time step within a dense stretch (s) + -> Double -- ^ gap between stretches (s) + -> [(UTCTime, Double)] +cascadeSamples cycles nDense denseStep gapDur = + [ (mkT (fromIntegral k * cycleDur + fromIntegral i * denseStep) + , fromIntegral (k * nDense + i) + ) + | k <- [0 .. cycles - 1] + , i <- [0 .. nDense - 1] + ] + where + stretchDur = fromIntegral (nDense - 1) * denseStep + cycleDur = stretchDur + gapDur + + +-- Bug #1 (direct regression): a hand-picked sequence that spans two +-- bucket periods with multiple samples in the second period. Under +-- correct behaviour the open bucket accumulates the last three samples; +-- windowDuration reports 7. Under the bug, obFinish stays at 1 and +-- windowDuration reports 1. +-- +unit_dtb_openBucketResetsOnSeal :: Property +unit_dtb_openBucketResetsOnSeal = once $ + let samples = [ (mkT 0, 0), (mkT 1, 1) -- bucket [0, 2) + , (mkT 2, 2), (mkT 3, 3) -- bucket [2, 4) + , (mkT 5, 5) -- evicts first bucket + ] + w = buildDigestW 2 2 samples + in DTB.sampleCount w === 3 + .&&. DTB.windowDuration w === Just (4 :: NominalDiffTime) + + +-- Empty window semantics. +unit_dtb_emptySemantics :: Property +unit_dtb_emptySemantics = once $ + let w = DTB.empty (1 :: NominalDiffTime) 60 :: DigestW + in DTB.sampleCount w === 0 + .&&. counterexample "windowDigest" (isNothing (DTB.windowDigest w)) + .&&. DTB.windowDuration w === Nothing + .&&. DTB.windowMedian w === Nothing + .&&. DTB.windowQuantile 0.99 w === Nothing + + +-- reset returns an empty window. +prop_dtb_resetEmpties :: Property +prop_dtb_resetEmpties = + forAll (genTimedSamplesIn 100.0) $ \samples -> + let w = buildDigestW 1 60 samples + w' = DTB.reset w + in DTB.sampleCount w' === 0 + .&&. counterexample "windowDigest after reset" + (isNothing (DTB.windowDigest w')) + + +-- sampleCount counts every inserted sample when nothing has been +-- evicted (all samples fit within the window duration). +prop_dtb_sampleCountConservation :: Property +prop_dtb_sampleCountConservation = + forAll (genTimedSamplesIn 3.0) $ \samples -> + -- bucket_dur = 1s, retention = 5 → window = 5s ; all samples ≤ 3s + let w = buildDigestW 1.0 5 samples + in DTB.sampleCount w === length samples + + +-- A quantile query lies within [min, max] of the input samples that +-- are still in the window. +prop_dtb_quantileInRange :: Property +prop_dtb_quantileInRange = + forAll (genTimedSamplesIn 3.0) $ \samples -> + not (null samples) ==> + let w = buildDigestW 1.0 5 samples + values = map snd samples + lo = minimum values + hi = maximum values + in case DTB.windowMedian w of + Nothing -> counterexample "no median for non-empty window" False + Just m -> counterexample + (show m ++ " outside [" ++ show lo ++ ", " ++ show hi ++ "]") + (m >= lo && m <= hi) + + +-- The batched digest's median is within a tolerance of the exact +-- sample median, for reasonably-sized uniform inputs that all fit in +-- the window. +prop_dtb_quantileAccuracy :: Property +prop_dtb_quantileAccuracy = + forAll (genPositiveSamplesIn 30.0) $ \samples -> + let w = buildDigestW 1.0 60 samples + values = map snd samples + n = length values + sorted = sort values + exactMedian = if odd n + then sorted !! (n `div` 2) + else (sorted !! (n `div` 2 - 1) + sorted !! (n `div` 2)) / 2 + in case DTB.windowMedian w of + Nothing -> counterexample "no median for non-empty window" False + Just m -> counterexample + ("batched " ++ show m ++ " vs exact " ++ show exactMedian) + (abs (m - exactMedian) <= 5.0) -- generous + + +-- Eviction correctness: after inserting samples spread over +-- [0, 6 × windowDur], the sample count should be at most the count of +-- samples whose timestamps fall within the last windowDur (plus two +-- buckets' worth of slack). +prop_dtb_evictionRetainsOnlyRecent :: Property +prop_dtb_evictionRetainsOnlyRecent = + forAll gen $ \samples -> + not (null samples) ==> + let bucketDur = 1.0 :: NominalDiffTime + retention = 5 + windowDur = fromIntegral retention * bucketDur + w = buildDigestW bucketDur retention samples + lastT = maximum (map fst samples) + -- Two bucket durations of slack: eviction only runs on seal + -- events, so the last seal may have fired up to bucketDur before + -- lastT, and a retained bucket's oldest sample can be up to + -- bucketDur older than its sbFinish. + cutoff = negate (windowDur + 2 * bucketDur) `addUTCTime` lastT + liveCount = length (filter (\(t, _) -> t >= cutoff) samples) + in counterexample (show (DTB.sampleCount w, liveCount)) + (DTB.sampleCount w <= liveCount) + where + gen = do + n <- choose (50, 400) + offsets <- sort <$> vectorOf n (choose (0, 30)) -- 30s span, window 5s + values <- vectorOf n (choose (-100, 100)) + return [ (mkT t, v) | (t, v) <- zip offsets values ] + + +-- windowDuration reports a value bounded below by the actual sample +-- duration (newest - oldest) and above by that duration plus one +-- bucket period, per its haddock. The upper slack is the range of +-- possible obStart positions for the leftmost sealed bucket (whose +-- obStart is not tracked; the implementation approximates it as +-- sbFinish - bucketDur). Samples span less than the window so no +-- eviction narrows the "actual" duration. +prop_dtb_windowDurationApproximate :: Property +prop_dtb_windowDurationApproximate = + forAll (genTimedSamplesIn 30.0) $ \samples -> + not (null samples) ==> + let bucketDur = 1.0 :: NominalDiffTime + retention = 60 -- window = 60s > spanSec = 30s + w = buildDigestW bucketDur retention samples + times = map fst samples + actual = maximum times `diffUTCTime` minimum times + in case DTB.windowDuration w of + Nothing -> counterexample "unexpected Nothing for non-empty window" False + Just d -> + counterexample + (show d ++ " outside [" ++ show actual ++ + ", " ++ show (actual + bucketDur) ++ "]") + (d >= actual .&&. d <= actual + bucketDur) + + +-- The cascade generator produces at least one bucket-sized window with +-- many samples, i.e. dense stretches actually exist. +unit_cascadeGeneratorHasDenseStretches :: Property +unit_cascadeGeneratorHasDenseStretches = once $ + let bucketDur = 1.0 :: NominalDiffTime + nDense = 500 + samples = cascadeSamples 4 nDense 0.01 10.0 + times = map fst samples + limit t0 = bucketDur `addUTCTime` t0 + perBucket = + [ length (takeWhile (< limit t0) suffix) + | suffix@(t0:_) <- tails times + ] + isDense = not . null $ dropWhile (< 90) perBucket + in counterexample ("no dense buckets") $ + isDense -- ~100 expected with denseStep = 0.01s + + +-- The cascade generator drives multi-bucket evictions: some insertion +-- causes 'sampleCount' to drop by more than one bucket's worth of +-- samples in a single step. +unit_cascadeGeneratorTriggersMultiBucketEviction :: Property +unit_cascadeGeneratorTriggersMultiBucketEviction = once $ + let bucketDur = 1.0 :: NominalDiffTime + retention = 10 -- window = 10s, gap = 10s -> cascade + nDense = 500 -- 5s of dense samples, ~100/bucket + samples = cascadeSamples 4 nDense 0.01 10.0 + w0 = DTB.empty bucketDur retention :: DigestW + ws = scanl (flip DTB.insert) w0 samples + counts = map DTB.sampleCount ws + deltas = zipWith (-) (drop 1 counts) counts + minDelta = minimum deltas + in counterexample ("smallest sampleCount delta: " ++ show minDelta) $ + -- A single insert should evict several full buckets at once. + -- With ~100 samples/bucket and window = 10s, a 10s gap after a + -- 5s dense stretch drops all ~5 buckets in one dropUntil call. + minDelta <= -200 + + +main :: IO () +main = defaultMain tests + +tests :: TestTree +tests = + testGroup "window-stats" + [ testGroup "Count" + [ testProperty "insertMany ≡ foldl' insert" prop_insertMany_eq_foldl + , testProperty "size = min cap (length xs)" prop_size_bounded + , testProperty "windowSum = sum live" prop_sum_consistent + , testProperty "fromFoldable orientation" prop_fromFoldable_orientation + , testProperty "fromListN orientation" prop_fromListN_orientation + , testProperty "evictOldestN reduces size" prop_evictOldestN + ] + , testGroup "Timed" + [ testProperty "evictBefore keeps t >= cutoff" prop_evictBefore + , testProperty "windowDuration = newest - oldest" prop_windowDuration + ] + , testGroup "DigestTimeBatched" + [ testProperty "open bucket resets on seal (bug #1 regression)" + unit_dtb_openBucketResetsOnSeal + , testProperty "empty window has trivial statistics" + unit_dtb_emptySemantics + , testProperty "reset empties the window" + prop_dtb_resetEmpties + , testProperty "sampleCount conserves inserts (no eviction)" + prop_dtb_sampleCountConservation + , testProperty "median lies in [min, max] of samples" + prop_dtb_quantileInRange + , testProperty "median accuracy vs exact" + prop_dtb_quantileAccuracy + , testProperty "eviction retains only recent samples" + prop_dtb_evictionRetainsOnlyRecent + , testProperty "windowDuration is bounded [actual, actual + bucketDur]" + prop_dtb_windowDurationApproximate + , testProperty "cascade generator has dense stretches" + unit_cascadeGeneratorHasDenseStretches + , testProperty "cascade generator triggers multi-bucket eviction" + unit_cascadeGeneratorTriggersMultiBucketEviction + ] + ] diff --git a/window-stats/window-stats.cabal b/window-stats/window-stats.cabal new file mode 100644 index 0000000000..90df2ac9be --- /dev/null +++ b/window-stats/window-stats.cabal @@ -0,0 +1,134 @@ +cabal-version: 3.4 +name: window-stats +version: 0.1.0.0 +synopsis: Efficient sliding window statistics with finger trees and t-digest +license: Apache-2.0 +license-files: + LICENSE + NOTICE + +copyright: Intersect +author: Marcin Wójtowicz +maintainer: marcin.wojtowicz@iohk.io +category: Data +build-type: Simple +extra-doc-files: CHANGELOG.md + +common ghc-options + default-language: Haskell2010 + default-extensions: ImportQualifiedPost + ghc-options: + -Wall + -Wno-unticked-promoted-constructors + -Wcompat + -Wincomplete-uni-patterns + -Wincomplete-record-updates + -Wpartial-fields + -Widentities + -Wredundant-constraints + -Wunused-packages + + -- In ghc-9.14 the `pattern` namespace specifier is deprecated. + if impl(ghc >=9.14) + ghc-options: + -Wno-pattern-namespace-specifier + + -- GHC-9.14 warns on constraints that it considers redundant + -- but are necessary for earlier compilers. + if impl(ghc >=9.14) + ghc-options: + -Wno-redundant-constraints + +library with-tdigest + import: ghc-options + visibility: public + exposed-modules: + -- TODO implement: + -- Data.Window.Internal.DigestCountBatched, + Data.Window.DigestTimeBatched + Data.Window.Internal.DigestTimeBatched + + build-depends: + base, + deepseq, + fingertree, + tdigest, + window-stats, + + hs-source-dirs: + tdigest + + default-extensions: + ImportQualifiedPost + + default-language: Haskell2010 + +library + import: ghc-options + exposed-modules: + Data.Window.Count + Data.Window.Internal.Count + Data.Window.Internal.Measures + Data.Window.Internal.Timed + Data.Window.TimeLike + Data.Window.Timed + + build-depends: + base >=4.14 && <4.23, + deepseq, + fingertree, + io-classes:si-timers ^>=1.8, + time, + + hs-source-dirs: + lib + +library with-foldl + import: ghc-options + visibility: public + exposed-modules: + Data.Window.Fold.Count + Data.Window.Fold.Timed + + build-depends: + base, + containers, + fingertree, + foldl, + window-stats, + + hs-source-dirs: + fold/lib + +test-suite ws-test + import: ghc-options + type: exitcode-stdio-1.0 + hs-source-dirs: test + main-is: Main.hs + ghc-options: + -rtsopts + -fno-ignore-asserts + + build-depends: + QuickCheck, + base, + tasty, + tasty-quickcheck, + time, + window-stats, + window-stats:with-tdigest, + +benchmark ws-bench + import: ghc-options + type: exitcode-stdio-1.0 + hs-source-dirs: bench + main-is: Main.hs + ghc-options: + -O2 + -rtsopts + -with-rtsopts=-T + + build-depends: + base, + time, + window-stats:with-tdigest,