git.iko.soy/streamly
on 08046ff help sign in
streamly/src/Streamly/Internal/Data/Fold/Async.hs
120 lines 4.2 kB history raw
1-- |
2-- Module : Streamly.Internal.Data.Fold.Async
3-- Copyright : (c) 2019 Composewell Technologies
4-- License : BSD-3-Clause
5-- Maintainer : streamly@composewell.com
6-- Stability : experimental
7-- Portability : GHC
8--
9module Streamly.Internal.Data.Fold.Async
10 (
11 -- * Trimming
12 takeInterval
14 -- * Splitting
15 , intervalsOf
16 )
17where
19import Control.Concurrent (threadDelay, forkIO, killThread)
20import Control.Concurrent.MVar (MVar, newMVar, swapMVar, readMVar)
21import Control.Exception (SomeException(..), catch, mask)
22import Control.Monad (void)
23import Control.Monad.IO.Class (MonadIO(..))
24import Control.Monad.Trans.Control (control)
25import Streamly.Internal.Control.Concurrent (MonadAsync)
26import Streamly.Internal.Data.Tuple.Strict (Tuple3'(..))
28import Streamly.Internal.Data.Fold.Type
30-- $setup
31-- >>> :m
32-- >>> import qualified Streamly.Prelude as Stream
33-- >>> import qualified Streamly.Data.Fold as Fold
34-- >>> import qualified Streamly.Internal.Data.Fold.Async as Fold
36-- XXX We can use asyncClock here. A parser can be used to return an input that
37-- arrives after the timeout.
38-- XXX If n is 0 return immediately in initial.
39-- XXX we should probably discard the input received after the timeout like
40-- takeEndBy_.
41--
42-- | @takeInterval n fold@ uses @fold@ to fold the input items arriving within
43-- a window of first @n@ seconds.
44--
45-- >>> input = Stream.delay 0.1 $ Stream.fromList [1..]
46-- >>> Stream.fold (Fold.takeInterval 1.0 Fold.toList) input
47-- [1,2,3,4,5,6,7,8,9,10,11]
48--
49-- Stops when @fold@ stops or when the timeout occurs. Note that the fold needs
50-- an input after the timeout to stop. For example, if no input is pushed to
51-- the fold until one hour after the timeout had occurred, then the fold will
52-- be done only after consuming that input.
53--
54-- /Pre-release/
55--
56{-# INLINE takeInterval #-}
57takeInterval :: MonadAsync m => Double -> Fold m a b -> Fold m a b
58takeInterval n (Fold step initial done) = Fold step' initial' done'
60 where
62 initial' = do
63 res <- initial
64 case res of
65 Partial s -> do
66 mv <- liftIO $ newMVar False
67 t <-
68 control $ \run ->
69 mask $ \restore -> do
70 tid <-
71 forkIO
72 $ catch
73 (restore $ void $ run (timerThread mv))
74 (handleChildException mv)
75 run (return tid)
76 return $ Partial $ Tuple3' s mv t
77 Done b -> return $ Done b
79 step' (Tuple3' s mv t) a = do
80 val <- liftIO $ readMVar mv
81 if val
82 then do
83 res <- step s a
84 case res of
85 Partial sres -> Done <$> done sres
86 Done bres -> return $ Done bres
87 else do
88 res <- step s a
89 case res of
90 Partial fs -> return $ Partial $ Tuple3' fs mv t
91 Done b -> liftIO (killThread t) >> return (Done b)
93 done' (Tuple3' s _ _) = done s
95 timerThread mv = do
96 liftIO $ threadDelay (round $ n * 1000000)
97 -- Use IORef + CAS? instead of MVar since its a Bool?
98 liftIO $ void $ swapMVar mv True
100 handleChildException :: MVar Bool -> SomeException -> IO ()
101 handleChildException mv _ = void $ swapMVar mv True
103-- For example, we can copy and distribute a stream to multiple folds where
104-- each fold can group the input differently e.g. by one second, one minute and
105-- one hour windows respectively and fold each resulting stream of folds.
107-- | Group the input stream into windows of n second each using the first fold
108-- and then fold the resulting groups using the second fold.
109--
110-- >>> intervals = Fold.intervalsOf 0.5 Fold.toList Fold.toList
111-- >>> Stream.fold intervals $ Stream.delay 0.2 $ Stream.fromList [1..10]
112-- [[1,2,3,4],[5,6,7],[8,9,10]]
113--
114-- > intervalsOf n split = many (takeInterval n split)
115--
116-- /Pre-release/
117--
118{-# INLINE intervalsOf #-}
119intervalsOf :: MonadAsync m => Double -> Fold m a b -> Fold m b c -> Fold m a c
120intervalsOf n split = many (takeInterval n split)