@@ -51,17 +51,18 @@ module Streamly.Internal.Data.Fold.Window
5151 )
5252where
5353
54+import Control.Monad.IO.Class (MonadIO (liftIO))
5455import Data.Bifunctor(bimap)
55−import Data.Function ((&))
56−import Data.Maybe (fromMaybe)
56+import Foreign (Storable(..))
5757
5858import 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'))
6061
6162import Prelude hiding (length, sum, minimum, maximum)
6263
63−import qualified Deque.Strict as DQ
6464import qualified Streamly.Data.Fold as Fold
65+import qualified Streamly.Internal.Data.Ring.Foreign as Ring
6566
6667-- $setup
6768-- >>> import Data.Bifunctor(bimap)
@@ -213,111 +214,90 @@ powerSumFrac p = lmap (** p) sum
213214-- Location
214215-------------------------------------------------------------------------------
215216
216−-- Theoretically, we can approximate minimum in a rolling window by using a
217−-- 'powerMean' with sufficiently large negative power.
218−--
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
221−-- time.
217+-- XXX Remove MonadIO constraint
218+
219+-- | Determine the maximum and minimum in a rolling window.
222220--
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.
224223--
225−-- If you want to compute the minimum of the entire stream Fold.minimum from
226−-- streamly package would be much faster.
224+-- /Space/: \(\mathcal{O}(n)\) where @n@ is the window size.
227225--
228226-- /Time/: \(\mathcal{O}(n*w)\) where \(w\) is the window size.
229227--
230−{-# INLINE minimum #-}
231−minimum :: (Monad m, Ord a) => Fold m (a, Maybe a) a
232−minimum = Fold step initial extract
228+{-# INLINE range #-}
229+range :: (MonadIO m, Storable a, Ord a) => Int -> Fold m a (Maybe (a, a))
230+range n = Fold step initial extract
233231
234232 where
235233
236− initial = return $ Partial $ Tuple3' (0 :: Int) (0 :: Int)
237− (mempty :: DQ.Deque (Int, a))
238−
239− step (Tuple3' i w q) (a, ma) =
240− case ma of
241− Nothing ->
242− return $ Partial $ Tuple3' (i + 1) (w + 1)
243− (headCheck i q (w + 1) & dqloop (i, a))
244− Just _ ->
245− return $ Partial $ Tuple3' (i + 1) w
246− (headCheck i q w & dqloop (i,a))
247−
248− {-# INLINE headCheck #-}
249− headCheck i q w =
250− case DQ.uncons q of
251− Nothing -> q
252− Just (ia', q') ->
253− if fst ia' <= i - w
254− then q'
255− else q
256−
257− dqloop ia q =
258− case DQ.unsnoc q of
259− Nothing -> DQ.snoc ia q
260− -- XXX This can be improved for the case of `=`
261− Just (ia', q') ->
262− if snd ia <= snd ia'
263− then dqloop ia q'
264− else DQ.snoc ia q
265−
266− extract (Tuple3' _ _ q) = return $ snd
267− $ fromMaybe (0, error "min: Empty stream")
268− $ DQ.head q
269−
270−-- Theoretically, we can approximate maximum in a rolling window by using a
271−-- 'powerMean' with sufficiently large positive power.
234+ -- XXX Use Ring unfold and then fold for composing maximum and minimum to
235+ -- get the range.
236+
237+ initial =
238+ if n <= 0
239+ then error "range: window size must be > 0"
240+ else
241+ let f (a, b) = Partial $ Tuple3Fused' a b (0 :: Int)
242+ in fmap f $ liftIO $ Ring.new n
243+
244+ step (Tuple3Fused' rb rh i) a = do
245+ rh1 <- liftIO $ Ring.unsafeInsert rb rh a
246+ return $ Partial $ Tuple3Fused' rb rh1 (i + 1)
247+
248+ -- XXX We need better Ring array APIs so that we can unfold the ring to a
249+ -- stream and fold the stream using a fold of our choice.
250+ --
251+ -- We could just scan the stream to get a stream of ring buffers and then
252+ -- map required folds over those, but we need to be careful that all those
253+ -- rings refer to the same mutable ring, therefore, downstream needs to
254+ -- process those strictly before it can change.
255+ foldFunc i
256+ | i < n = Ring.unsafeFoldRingM
257+ | otherwise = Ring.unsafeFoldRingFullM
258+
259+ extract (Tuple3Fused' rb rh i) =
260+ if i == 0
261+ then return Nothing
262+ else do
263+ x <- liftIO $ peek rh
264+ let accum (mn, mx) a = return (min mn a, max mx a)
265+ fmap Just $ foldFunc i rh accum (x, x) rb
266+
267+-- | Find the minimum element in a rolling window.
268+--
269+-- This implementation traverses the entire window buffer to compute the
270+-- minimum whenever we demand it. It performs better than the dequeue based
271+-- implementation in @streamly-statistics@ package when any of the following
272+-- holds:
273+--
274+-- * window size is small (< 30)
275+-- * we are using this as a fold instead of a scan.
276+--
277+-- For other cases the implementation in the @streamly-statistics@ package
278+-- performs better.
279+--
280+-- If you want to compute the minimum of the entire stream
281+-- 'Streamly.Data.Fold.minimum' is much faster.
282+--
283+-- /Time/: \(\mathcal{O}(n*w)\) where \(w\) is the window size.
272284--
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
288+
273289-- | The maximum element in a rolling window.
274290--
275−-- If you want to compute the maximum of the entire stream Fold.maximum from
276−-- streamly package would be much faster.
291+-- See the performance related comments in 'minimum'.
292+--
293+-- If you want to compute the maximum of the entire stream 'Fold.maximum' would
294+-- be much faster.
277295--
278296-- /Time/: \(\mathcal{O}(n*w)\) where \(w\) is the window size.
279297--
280298{-# INLINE maximum #-}
281−maximum :: (Monad m, Ord a) => Fold m (a, Maybe a) a
282−maximum = Fold step initial extract
283−
284− where
285−
286− initial = return $ Partial $ Tuple3' (0 :: Int) (0 :: Int)
287− (mempty :: DQ.Deque (Int, a))
288−
289− step (Tuple3' i w q) (a, ma) =
290− case ma of
291− Nothing ->
292− return $ Partial $ Tuple3' (i + 1) (w + 1)
293− (headCheck i q (w + 1) & dqloop (i, a))
294− Just _ ->
295− return $ Partial $ Tuple3' (i + 1) w
296− (headCheck i q w & dqloop (i,a))
297−
298− {-# INLINE headCheck #-}
299− headCheck i q w =
300− case DQ.uncons q of
301− Nothing -> q
302− Just (ia', q') ->
303− if fst ia' <= i - w
304− then q'
305− else q
306−
307− dqloop ia q =
308− case DQ.unsnoc q of
309− Nothing -> DQ.snoc ia q
310− -- XXX This can be improved for the case of `=`
311− Just (ia', q') ->
312− if snd ia >= snd ia'
313− then dqloop ia q'
314− else DQ.snoc ia q
315−
316− extract (Tuple3' _ _ q) =
317− return
318− $ snd
319− $ fromMaybe (0, error "max: Empty stream")
320− $ DQ.head q
299+maximum :: (MonadIO m, Storable a, Ord a) => Int -> Fold m a (Maybe a)
300+maximum n = fmap (fmap snd) $ range n
321301
322302-- | Arithmetic mean of elements in a sliding window:
323303--
@@ -335,19 +315,3 @@ maximum = Fold step initial extract
335315{-# INLINE mean #-}
336316mean :: forall m a. (Monad m, Fractional a) => Fold m (a, Maybe a) a
337317mean = Fold.teeWith (/) sum length
338−
339−
340−-- | The difference between the maximum and minimum elements of a rolling window.
341−--
342−-- >>> range = Fold.teeWith (-) maximum minimum
343−--
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.
346−--
347−-- /Space/: \(\mathcal{O}(n)\) where @n@ is the window size.
348−--
349−-- /Time/: \(\mathcal{O}(n*w)\) where \(w\) is the window size.
350−--
351−{-# INLINE range #-}
352−range :: (Monad m, Num a, Ord a) => Fold m (a, Maybe a) a
353−range = Fold.teeWith (-) maximum minimum