@@ -51,17 +51,18 @@ module Streamly.Internal.Data.Fold.Window
54+import Control.Monad.IO.Class (MonadIO (liftIO))
54import Data.Bifunctor(bimap)
55import Data.Bifunctor(bimap)
55−import Data.Function ((&))
56+import Foreign (Storable(..))
56−import Data.Maybe (fromMaybe)
58import Streamly.Internal.Data.Fold.Type (Fold(..), Step(..))
58import Streamly.Internal.Data.Fold.Type (Fold(..), Step(..))
59−import Streamly.Internal.Data.Tuple.Strict (Tuple'(..), Tuple3'(..))
59+import Streamly.Internal.Data.Tuple.Strict
60+ (Tuple'(..), Tuple3Fused' (Tuple3Fused'))
61import Prelude hiding (length, sum, minimum, maximum)
62import Prelude hiding (length, sum, minimum, maximum)
63−import qualified Deque.Strict as DQ
64import qualified Streamly.Data.Fold as Fold
64import qualified Streamly.Data.Fold as Fold
65+import qualified Streamly.Internal.Data.Ring.Foreign as Ring
67-- >>> import Data.Bifunctor(bimap)
68-- >>> import Data.Bifunctor(bimap)
@@ -213,111 +214,90 @@ powerSumFrac p = lmap (** p) sum
213-- Location
214-- Location
214-------------------------------------------------------------------------------
215-------------------------------------------------------------------------------
216−-- Theoretically, we can approximate minimum in a rolling window by using a
217+-- XXX Remove MonadIO constraint
217−-- 'powerMean' with sufficiently large negative power.
218+
218−--
219+-- | Determine the maximum and minimum in a rolling window.
219−-- XXX If we need to know the minimum in the window only once in a while then
220−-- we can use linear search when it is extracted and not pay the cost all the
223−-- | The minimum element in a rolling window.
221+-- If you want to compute the range of the entire stream @Fold.teeWith (,)
222+-- Fold.maximum Fold.minimum@ would be much faster.
225−-- If you want to compute the minimum of the entire stream Fold.minimum from
224+-- /Space/: \(\mathcal{O}(n)\) where @n@ is the window size.
226−-- streamly package would be much faster.
228-- /Time/: \(\mathcal{O}(n*w)\) where \(w\) is the window size.
226-- /Time/: \(\mathcal{O}(n*w)\) where \(w\) is the window size.
230−{-# INLINE minimum #-}
228+{-# INLINE range #-}
231−minimum :: (Monad m, Ord a) => Fold m (a, Maybe a) a
229+range :: (MonadIO m, Storable a, Ord a) => Int -> Fold m a (Maybe (a, a))
232−minimum = Fold step initial extract
230+range n = Fold step initial extract
236− initial = return $ Partial $ Tuple3' (0 :: Int) (0 :: Int)
234+ -- XXX Use Ring unfold and then fold for composing maximum and minimum to
237− (mempty :: DQ.Deque (Int, a))
235+ -- get the range.
239− step (Tuple3' i w q) (a, ma) =
237+ initial =
240− case ma of
238+ if n <= 0
241− Nothing ->
239+ then error "range: window size must be > 0"
241+ let f (a, b) = Partial $ Tuple3Fused' a b (0 :: Int)
242+ in fmap f $ liftIO $ Ring.new n
244+ step (Tuple3Fused' rb rh i) a = do
245+ rh1 <- liftIO $ Ring.unsafeInsert rb rh a
242− return $ Partial $ Tuple3' (i + 1) (w + 1)
246+ return $ Partial $ Tuple3Fused' rb rh1 (i + 1)
243− (headCheck i q (w + 1) & dqloop (i, a))
247+
244− Just _ ->
248+ -- XXX We need better Ring array APIs so that we can unfold the ring to a
245− return $ Partial $ Tuple3' (i + 1) w
249+ -- stream and fold the stream using a fold of our choice.
246− (headCheck i q w & dqloop (i,a))
250+ --
247−
251+ -- We could just scan the stream to get a stream of ring buffers and then
248− {-# INLINE headCheck #-}
252+ -- map required folds over those, but we need to be careful that all those
249− headCheck i q w =
253+ -- rings refer to the same mutable ring, therefore, downstream needs to
250− case DQ.uncons q of
254+ -- process those strictly before it can change.
251− Nothing -> q
255+ foldFunc i
252− Just (ia', q') ->
256+ | i < n = Ring.unsafeFoldRingM
253− if fst ia' <= i - w
257+ | otherwise = Ring.unsafeFoldRingFullM
255− else q
259+ extract (Tuple3Fused' rb rh i) =
257− dqloop ia q =
261+ then return Nothing
258− case DQ.unsnoc q of
262+ else do
259− Nothing -> DQ.snoc ia q
263+ x <- liftIO $ peek rh
260− -- XXX This can be improved for the case of `=`
264+ let accum (mn, mx) a = return (min mn a, max mx a)
261− Just (ia', q') ->
265+ fmap Just $ foldFunc i rh accum (x, x) rb
262− if snd ia <= snd ia'
266+
263− then dqloop ia q'
267+-- | Find the minimum element in a rolling window.
264− else DQ.snoc ia q
268+--
265−
269+-- This implementation traverses the entire window buffer to compute the
266− extract (Tuple3' _ _ q) = return $ snd
270+-- minimum whenever we demand it. It performs better than the dequeue based
267− $ fromMaybe (0, error "min: Empty stream")
271+-- implementation in @streamly-statistics@ package when any of the following
268− $ DQ.head q
272+-- holds:
270−-- Theoretically, we can approximate maximum in a rolling window by using a
274+-- * window size is small (< 30)
271−-- 'powerMean' with sufficiently large positive power.
275+-- * we are using this as a fold instead of a scan.
277+-- For other cases the implementation in the @streamly-statistics@ package
280+-- If you want to compute the minimum of the entire stream
281+-- 'Streamly.Data.Fold.minimum' is much faster.
283+-- /Time/: \(\mathcal{O}(n*w)\) where \(w\) is the window size.
285+{-# INLINE minimum #-}
286+minimum :: (MonadIO m, Storable a, Ord a) => Int -> Fold m a (Maybe a)
287+minimum n = fmap (fmap fst) $ range n
273-- | The maximum element in a rolling window.
289-- | The maximum element in a rolling window.
291+-- See the performance related comments in 'minimum'.
275−-- If you want to compute the maximum of the entire stream Fold.maximum from
293+-- If you want to compute the maximum of the entire stream 'Fold.maximum' would
276−-- streamly package would be much faster.
294+-- be much faster.
278-- /Time/: \(\mathcal{O}(n*w)\) where \(w\) is the window size.
296-- /Time/: \(\mathcal{O}(n*w)\) where \(w\) is the window size.
280{-# INLINE maximum #-}
298{-# INLINE maximum #-}
281−maximum :: (Monad m, Ord a) => Fold m (a, Maybe a) a
299+maximum :: (MonadIO m, Storable a, Ord a) => Int -> Fold m a (Maybe a)
282−maximum = Fold step initial extract
300+maximum n = fmap (fmap snd) $ range n
286− initial = return $ Partial $ Tuple3' (0 :: Int) (0 :: Int)
287− (mempty :: DQ.Deque (Int, a))
289− step (Tuple3' i w q) (a, ma) =
292− return $ Partial $ Tuple3' (i + 1) (w + 1)
293− (headCheck i q (w + 1) & dqloop (i, a))
295− return $ Partial $ Tuple3' (i + 1) w
296− (headCheck i q w & dqloop (i,a))
298− {-# INLINE headCheck #-}
309− Nothing -> DQ.snoc ia q
310− -- XXX This can be improved for the case of `=`
312− if snd ia >= snd ia'
316− extract (Tuple3' _ _ q) =
319− $ fromMaybe (0, error "max: Empty stream")
322-- | Arithmetic mean of elements in a sliding window:
302-- | Arithmetic mean of elements in a sliding window:
@@ -335,19 +315,3 @@ maximum = Fold step initial extract
335{-# INLINE mean #-}
315{-# INLINE mean #-}
336mean :: forall m a. (Monad m, Fractional a) => Fold m (a, Maybe a) a
316mean :: forall m a. (Monad m, Fractional a) => Fold m (a, Maybe a) a
337mean = Fold.teeWith (/) sum length
317mean = Fold.teeWith (/) sum length
340−-- | The difference between the maximum and minimum elements of a rolling window.
342−-- >>> range = Fold.teeWith (-) maximum minimum
344−-- If you want to compute the range of the entire stream @Fold.teeWith (-)
345−-- Fold.maximum Fold.min@ from the streamly package would be much faster.
347−-- /Space/: \(\mathcal{O}(n)\) where @n@ is the window size.
349−-- /Time/: \(\mathcal{O}(n*w)\) where \(w\) is the window size.
352−range :: (Monad m, Num a, Ord a) => Fold m (a, Maybe a) a
353−range = Fold.teeWith (-) maximum minimum