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