git.iko.soy/streamly
on 2e489ea help sign in
streamly/src/Streamly/Internal/Data/Fold/Concurrent/Channel.hs
93 lines 3.3 kB history raw
1-- |
2-- Module : Streamly.Internal.Data.Fold.Concurrent.Channel
3-- Copyright : (c) 2022 Composewell Technologies
4-- License : BSD-3-Clause
5-- Maintainer : streamly@composewell.com
6-- Stability : experimental
7-- Portability : GHC
9module Streamly.Internal.Data.Fold.Concurrent.Channel
10 (
11 -- * Channel
12 Channel
14 -- * Configuration
15 , Config
16 , maxBuffer
17 , bound
18 , inspect
20 -- * Fold operations
21 , parEval
22 )
23where
25import Control.Concurrent (takeMVar)
26import Control.Monad (void, when)
27import Control.Monad.IO.Class (MonadIO(liftIO))
28import Data.IORef (writeIORef)
29import Streamly.Internal.Control.Concurrent (MonadAsync)
30import Streamly.Internal.Data.Fold (Fold(..), Step (..))
31import Streamly.Internal.Data.Stream.Channel.Worker (sendWithDoorBell)
32import Streamly.Internal.Data.Time.Clock (Clock(Monotonic), getTime)
34import Streamly.Internal.Data.Fold.Concurrent.Channel.Type
35import Streamly.Internal.Data.Stream.Channel.Types
37-------------------------------------------------------------------------------
38-- Evaluating a Fold
39-------------------------------------------------------------------------------
41-- XXX Cleanup the fold if the stream is interrupted. Add a GC hook.
43-- | Evaluate a fold asynchronously using a concurrent channel. The driver just
44-- queues the input stream values to the fold channel buffer and returns. The
45-- fold evaluates the queued values asynchronously.
46--
47{-# INLINABLE parEval #-}
48parEval :: MonadAsync m => (Config -> Config) -> Fold m a b -> Fold m a b
49parEval modifier f =
50 Fold step initial extract
52 where
54 initial = Partial <$> newChannel modifier f
56 -- XXX This is not truly asynchronous. If the fold is done we only get to
57 -- know when we send the next input unless the stream ends. We could
58 -- potentially throw an async exception to the driver to inform it
59 -- asynchronously. Alternatively, the stream should not block forever, it
60 -- should keep polling the fold status. We can insert a timer tick in the
61 -- input stream to do that.
62 --
63 -- A polled stream abstraction may be useful, it would consist of normal
64 -- events and tick events, latter are guaranteed to arrive.
65 step chan a = do
66 status <- sendToWorker chan a
67 return $ case status of
68 Nothing -> Partial chan
69 Just b -> Done b
71 extract chan = do
72 liftIO $ void
73 $ sendWithDoorBell
74 (outputQueue chan)
75 (outputDoorBell chan)
76 ChildStopChannel
77 status <- checkFoldStatus chan
78 case status of
79 Nothing -> do
80 liftIO
81 $ withDiagMVar
82 (svarInspectMode chan)
83 (dumpSVar chan)
84 "parEval: waiting to drain"
85 $ takeMVar (outputDoorBellFromConsumer chan)
86 -- XXX remove recursion
87 extract chan
88 Just b -> do
89 when (svarInspectMode chan) $ liftIO $ do
90 t <- getTime Monotonic
91 writeIORef (svarStopTime (svarStats chan)) (Just t)
92 printSVar (dumpSVar chan) "SVar Done"
93 return b