From 9fe069302397eb2cd7a4b0c1b1d384bbae7d6861 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marcin=20W=C3=B3jtowicz?= Date: Wed, 17 Jun 2026 22:15:09 +0200 Subject: [PATCH 1/8] window-stats: sliding windows with fixed sizes These modules provide a fingertree-backed sliding window with fixed element count. The fingertree backend caches a user-supplied monoidal measure `v` which updates incrementally as elements are added and evicted, which allows for constant time lookup of its value regardless of window size. The Window.Count module exposes a public API, and the private internal implementation is exposed by the Internal.Count module without any guarantees that it stays compatible between any releases. The Measures module exposes some common and practical prebuilt measures. --- window-stats/lib/Data/Window/Count.hs | 94 +++++ .../lib/Data/Window/Internal/Count.hs | 352 ++++++++++++++++++ .../lib/Data/Window/Internal/Measures.hs | 121 ++++++ 3 files changed, 567 insertions(+) create mode 100644 window-stats/lib/Data/Window/Count.hs create mode 100644 window-stats/lib/Data/Window/Internal/Count.hs create mode 100644 window-stats/lib/Data/Window/Internal/Measures.hs 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 From fca9c6e1c1e50f629e62f160de034a59c1b671e4 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marcin=20W=C3=B3jtowicz?= Date: Wed, 17 Jun 2026 22:24:24 +0200 Subject: [PATCH 2/8] window-stats: Time-based sliding windows These modules keep samples within a configured duration of the newest sample's timestamp. These variants abstract over the timestamp type via the 'TimeLike' class, which have two implementations defined: - `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. --- .../lib/Data/Window/Internal/Timed.hs | 444 ++++++++++++++++++ window-stats/lib/Data/Window/TimeLike.hs | 73 +++ window-stats/lib/Data/Window/Timed.hs | 81 ++++ 3 files changed, 598 insertions(+) create mode 100644 window-stats/lib/Data/Window/Internal/Timed.hs create mode 100644 window-stats/lib/Data/Window/TimeLike.hs create mode 100644 window-stats/lib/Data/Window/Timed.hs 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 From c674eb28868e8c76b51a55c5b908b55659f12875 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marcin=20W=C3=B3jtowicz?= Date: Wed, 17 Jun 2026 22:27:15 +0200 Subject: [PATCH 3/8] window-stats: approximate-quantile backend via tdigest These modules expose sliding windows backed by an approximate t-digest measure, which permit efficient computations of quantiles in bounded memory. --- .../tdigest/Data/Window/DigestTimeBatched.hs | 64 +++++ .../Data/Window/Internal/DigestTimeBatched.hs | 228 ++++++++++++++++++ 2 files changed, 292 insertions(+) create mode 100644 window-stats/tdigest/Data/Window/DigestTimeBatched.hs create mode 100644 window-stats/tdigest/Data/Window/Internal/DigestTimeBatched.hs 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..7016bb1427 --- /dev/null +++ b/window-stats/tdigest/Data/Window/Internal/DigestTimeBatched.hs @@ -0,0 +1,228 @@ +{-# 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)) + } + +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 + } + + +-- | 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 } + + +-- | 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 +windowDigest :: KnownNat comp => TimedDigestWindow t comp -> Maybe (TDigest comp) +windowDigest TimedDigestWindow { tdwTree, tdwBucket } = + case tdwBucket of + Nothing -> Nothing + Just OpenBucket { obDigest } -> + let treeDigest = bmDigest $ FT.measure tdwTree + in Just $! treeDigest <> obDigest + + +-- | 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) -> tdw { tdwBucket = Just $! bucket } + (Just sealed, bucket) -> + let sample = SealedBucket (obFinish sealed) (obCount sealed) (obDigest sealed) + tdwTree' = tdwTree FT.|> sample + cutoff = negate tdwDuration `addT` t + -- evictBefore will force the measure once after it's done + 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' } + 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) + -- we will not have to combine t-digests of the buckets + -- which we have dropped from the window + !_ = bmDigest $ FT.measure tdwTree' + +-- | 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 TimedDigestWindow { tdwTree, tdwBucket } = + case tdwBucket of + Nothing -> Nothing + Just bucket -> + let !bucketDigest = obDigest bucket + !treeDigest = bmDigest $ FT.measure tdwTree + combined = treeDigest <> bucketDigest + in Just (TD.minimumValue combined, TD.maximumValue combined) + + +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 From dc435a4992c9cc264e98c202e6576e3e3a3b5cb7 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marcin=20W=C3=B3jtowicz?= Date: Thu, 9 Jul 2026 13:00:11 +0200 Subject: [PATCH 4/8] window-stats: Add cache to TimedDigestWindow --- .../Data/Window/Internal/DigestTimeBatched.hs | 60 ++++++++++++------- 1 file changed, 38 insertions(+), 22 deletions(-) diff --git a/window-stats/tdigest/Data/Window/Internal/DigestTimeBatched.hs b/window-stats/tdigest/Data/Window/Internal/DigestTimeBatched.hs index 7016bb1427..c0d55bbd94 100644 --- a/window-stats/tdigest/Data/Window/Internal/DigestTimeBatched.hs +++ b/window-stats/tdigest/Data/Window/Internal/DigestTimeBatched.hs @@ -75,6 +75,12 @@ data TimedDigestWindow t (comp :: Nat) = TimedDigestWindow , 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) @@ -89,12 +95,13 @@ empty tdwBucketDuration r = , 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 } +reset tdw = tdw { tdwBucket = Nothing, tdwTree = FT.empty, tdwCacheDigest = Nothing } -- | Returns the number of samples which were inserted into the fingertree @@ -107,14 +114,13 @@ sampleCount TimedDigestWindow { tdwBucket, tdwTree } = in count + obCount bucket --- | Returns the digest of the sliding window, or Nothing if there are no samples +-- | 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 TimedDigestWindow { tdwTree, tdwBucket } = - case tdwBucket of - Nothing -> Nothing - Just OpenBucket { obDigest } -> - let treeDigest = bmDigest $ FT.measure tdwTree - in Just $! treeDigest <> obDigest +windowDigest tdw = case tdwCacheDigest tdw of + Nothing -> Nothing + Just !d -> Just d -- | Insert a new timestamped sample, then evict any buckets @@ -127,12 +133,16 @@ insert :: (TimeLike t, KnownNat comp) -> TimedDigestWindow t comp insert (t, a) tdw@TimedDigestWindow { tdwBucketDuration, tdwDuration, tdwBucket, tdwTree} = case fillBucket of - (Nothing, bucket) -> tdw { tdwBucket = Just $! bucket } + (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 will force the measure once after it's done + -- evictBefore forces the tree measure and refreshes the cache w/o forcing in evictBefore cutoff tdw { tdwBucket = Just $! bucket , tdwTree = tdwTree' } where @@ -161,7 +171,10 @@ evictBefore :: (KnownNat comp, TimeLike t) => t -> TimedDigestWindow t comp -> TimedDigestWindow t comp -evictBefore cutoff win = win { tdwTree = tdwTree', tdwBucket = tdwBucket' } +evictBefore cutoff win = win { tdwTree = tdwTree' + , tdwBucket = tdwBucket' + , tdwCacheDigest = cache + } where tdwBucket' = case tdwBucket win of Nothing -> Nothing @@ -171,9 +184,16 @@ evictBefore cutoff win = win { tdwTree = tdwTree', tdwBucket = tdwBucket' } tdwTree' = FT.dropUntil (maybe False (> cutoff) . bmFinish) (tdwTree win) - -- we will not have to combine t-digests of the buckets - -- which we have dropped from the window - !_ = bmDigest $ FT.measure tdwTree' + -- 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 @@ -198,14 +218,10 @@ windowDuration TimedDigestWindow { tdwBucket, tdwBucketDuration, tdwTree } = -- | Returns the means of the outermost centroids - appoximations, not the actual min/max samples windowMinMaxValues :: KnownNat comp => TimedDigestWindow t comp -> Maybe (Mean, Mean) -windowMinMaxValues TimedDigestWindow { tdwTree, tdwBucket } = - case tdwBucket of - Nothing -> Nothing - Just bucket -> - let !bucketDigest = obDigest bucket - !treeDigest = bmDigest $ FT.measure tdwTree - combined = treeDigest <> bucketDigest - in Just (TD.minimumValue combined, TD.maximumValue combined) +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 From 81557f610dfb4b93cc5f6c83e6b12ccfb580267f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marcin=20W=C3=B3jtowicz?= Date: Wed, 17 Jun 2026 22:31:58 +0200 Subject: [PATCH 5/8] window-stats: expose integration with the foldl package These expose some combinators which allow for summarising rolling statistics across a long stream in constant memory. --- .../fold/lib/Data/Window/Fold/Count.hs | 206 +++++++++++++++++ .../fold/lib/Data/Window/Fold/Timed.hs | 215 ++++++++++++++++++ 2 files changed, 421 insertions(+) create mode 100644 window-stats/fold/lib/Data/Window/Fold/Count.hs create mode 100644 window-stats/fold/lib/Data/Window/Fold/Timed.hs 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 From 5d6fec37d7c468238cd4cd99278cf2bb2e269132 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marcin=20W=C3=B3jtowicz?= Date: Wed, 17 Jun 2026 22:34:41 +0200 Subject: [PATCH 6/8] window-stats: add property tests --- window-stats/test/Main.hs | 338 ++++++++++++++++++++++++++++++++++++++ 1 file changed, 338 insertions(+) create mode 100644 window-stats/test/Main.hs diff --git a/window-stats/test/Main.hs b/window-stats/test/Main.hs new file mode 100644 index 0000000000..1bb36d0060 --- /dev/null +++ b/window-stats/test/Main.hs @@ -0,0 +1,338 @@ +{-# 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) + + +-- 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) +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 + ] + ] From b29bdd32d5efbb3e01e2cb9c8edf9c7a37348004 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marcin=20W=C3=B3jtowicz?= Date: Wed, 8 Jul 2026 13:18:01 +0200 Subject: [PATCH 7/8] window-stats: add benchmarks --- window-stats/bench/Main.hs | 147 +++++++++++++++++++++++++++++++++++++ window-stats/test/Main.hs | 67 +++++++++++++++++ 2 files changed, 214 insertions(+) create mode 100644 window-stats/bench/Main.hs 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/test/Main.hs b/window-stats/test/Main.hs index 1bb36d0060..790b1b58a5 100644 --- a/window-stats/test/Main.hs +++ b/window-stats/test/Main.hs @@ -161,6 +161,28 @@ 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; @@ -299,6 +321,47 @@ prop_dtb_windowDurationApproximate = (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 @@ -334,5 +397,9 @@ tests = 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 ] ] From a45580f3ac958cca5f9b02c9b0abd56abfc27aa9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marcin=20W=C3=B3jtowicz?= Date: Wed, 8 Jul 2026 13:20:30 +0200 Subject: [PATCH 8/8] window-stats: meta data --- cabal.project | 1 + window-stats/CHANGELOG.md | 9 ++ window-stats/LICENSE | 202 ++++++++++++++++++++++++ window-stats/NOTICE | 14 ++ window-stats/README.md | 245 +++++++++++++++++++++++++++++ window-stats/changelog.d/scriv.ini | 15 ++ window-stats/window-stats.cabal | 134 ++++++++++++++++ 7 files changed, 620 insertions(+) create mode 100644 window-stats/CHANGELOG.md create mode 100644 window-stats/LICENSE create mode 100644 window-stats/NOTICE create mode 100644 window-stats/README.md create mode 100644 window-stats/changelog.d/scriv.ini create mode 100644 window-stats/window-stats.cabal 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/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/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,