git.iko.soy/streamly
on 167ea55 help sign in
streamly/src/Streamly/Internal/Data/Fold/Concurrent/Channel.hs
114 lines 4.6 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. On finalization, 'parEval'
46-- waits for the asynchronous fold to complete before it returns.
47--
48{-# INLINABLE parEval #-}
49parEval :: MonadAsync m => (Config -> Config) -> Fold m a b -> Fold m a b
50parEval modifier f =
51 Fold step initial extract final
53 where
55 initial = Partial <$> newChannel modifier f
57 -- XXX This is not truly asynchronous. If the fold is done we only get to
58 -- know when we send the next input unless the stream ends. We could
59 -- potentially throw an async exception to the driver to inform it
60 -- asynchronously. Alternatively, the stream should not block forever, it
61 -- should keep polling the fold status. We can insert a timer tick in the
62 -- input stream to do that.
63 --
64 -- A polled stream abstraction may be useful, it would consist of normal
65 -- events and tick events, latter are guaranteed to arrive.
66 --
67 -- XXX We can use the config to indicate if the fold is a scanning type or
68 -- one-shot, or use a separate parEvalScan for scanning. For a scanning
69 -- type fold the worker would always send the intermediate values back to
70 -- the driver. An intermediate value can be returned on an input, or the
71 -- driver can poll even without input, if we have the Skip input support.
72 -- When the buffer is full we can return "Skip" and then the next step
73 -- without input can wait for an output to arrive. Similarly, when "final"
74 -- is called it can return "Skip" to continue or "Done" to indicate
75 -- termination.
76 step chan a = do
77 status <- sendToWorker chan a
78 return $ case status of
79 Nothing -> Partial chan
80 Just b -> Done b
82 -- XXX We can use a separate type for non-scanning folds that will
83 -- introduce a lot of complexity. Are there combinators that rely on the
84 -- "extract" function even in non-scanning use cases?
85 -- Instead of making such folds partial we can also make them return a
86 -- Maybe type.
87 extract _ = error "Concurrent folds do not support scanning"
89 -- XXX depending on the use case we may want to either wait for the result
90 -- or cancel the ongoing work. We can use the config to control that?
91 -- Currently it waits for the work to complete.
92 final chan = do
93 liftIO $ void
94 $ sendWithDoorBell
95 (outputQueue chan)
96 (outputDoorBell chan)
97 ChildStopChannel
98 status <- checkFoldStatus chan
99 case status of
100 Nothing -> do
101 liftIO
102 $ withDiagMVar
103 (svarInspectMode chan)
104 (dumpSVar chan)
105 "parEval: waiting to drain"
106 $ takeMVar (outputDoorBellFromConsumer chan)
107 -- XXX remove recursion
108 final chan
109 Just b -> do
110 when (svarInspectMode chan) $ liftIO $ do
111 t <- getTime Monotonic
112 writeIORef (svarStopTime (svarStats chan)) (Just t)
113 printSVar (dumpSVar chan) "SVar Done"
114 return b