git.iko.soy/streamly
on 0e73809 help sign in
streamly/src/Streamly/Internal/Data/Fold/Concurrent/Channel.hs
112 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 module Streamly.Internal.Data.Fold.Concurrent.Channel.Type
13 -- * Configuration
14 , maxBuffer
15 , boundThreads
16 , inspect
18 -- * Fold operations
19 , parEval
20 )
21where
23import Control.Concurrent (takeMVar)
24import Control.Monad (void, when)
25import Control.Monad.IO.Class (MonadIO(liftIO))
26import Data.IORef (writeIORef)
27import Streamly.Internal.Control.Concurrent (MonadAsync)
28import Streamly.Internal.Data.Fold (Fold(..), Step (..))
29import Streamly.Internal.Data.Channel.Worker (sendWithDoorBell)
30import Streamly.Internal.Data.Time.Clock (Clock(Monotonic), getTime)
32import Streamly.Internal.Data.Fold.Concurrent.Channel.Type
33import Streamly.Internal.Data.Channel.Types
35-------------------------------------------------------------------------------
36-- Evaluating a Fold
37-------------------------------------------------------------------------------
39-- XXX Cleanup the fold if the stream is interrupted. Add a GC hook.
41-- | Evaluate a fold asynchronously using a concurrent channel. The driver just
42-- queues the input stream values to the fold channel buffer and returns. The
43-- fold evaluates the queued values asynchronously. On finalization, 'parEval'
44-- waits for the asynchronous fold to complete before it returns.
45--
46{-# INLINABLE parEval #-}
47parEval :: MonadAsync m => (Config -> Config) -> Fold m a b -> Fold m a b
48parEval modifier f =
49 Fold step initial extract final
51 where
53 initial = Partial <$> newChannel modifier f
55 -- XXX This is not truly asynchronous. If the fold is done we only get to
56 -- know when we send the next input unless the stream ends. We could
57 -- potentially throw an async exception to the driver to inform it
58 -- asynchronously. Alternatively, the stream should not block forever, it
59 -- should keep polling the fold status. We can insert a timer tick in the
60 -- input stream to do that.
61 --
62 -- A polled stream abstraction may be useful, it would consist of normal
63 -- events and tick events, latter are guaranteed to arrive.
64 --
65 -- XXX We can use the config to indicate if the fold is a scanning type or
66 -- one-shot, or use a separate parEvalScan for scanning. For a scanning
67 -- type fold the worker would always send the intermediate values back to
68 -- the driver. An intermediate value can be returned on an input, or the
69 -- driver can poll even without input, if we have the Skip input support.
70 -- When the buffer is full we can return "Skip" and then the next step
71 -- without input can wait for an output to arrive. Similarly, when "final"
72 -- is called it can return "Skip" to continue or "Done" to indicate
73 -- termination.
74 step chan a = do
75 status <- sendToWorker chan a
76 return $ case status of
77 Nothing -> Partial chan
78 Just b -> Done b
80 -- XXX We can use a separate type for non-scanning folds that will
81 -- introduce a lot of complexity. Are there combinators that rely on the
82 -- "extract" function even in non-scanning use cases?
83 -- Instead of making such folds partial we can also make them return a
84 -- Maybe type.
85 extract _ = error "Concurrent folds do not support scanning"
87 -- XXX depending on the use case we may want to either wait for the result
88 -- or cancel the ongoing work. We can use the config to control that?
89 -- Currently it waits for the work to complete.
90 final chan = do
91 liftIO $ void
92 $ sendWithDoorBell
93 (outputQueue chan)
94 (outputDoorBell chan)
95 ChildStopChannel
96 status <- checkFoldStatus chan
97 case status of
98 Nothing -> do
99 liftIO
100 $ withDiagMVar
101 (svarInspectMode chan)
102 (dumpSVar chan)
103 "parEval: waiting to drain"
104 $ takeMVar (outputDoorBellFromConsumer chan)
105 -- XXX remove recursion
106 final chan
107 Just b -> do
108 when (svarInspectMode chan) $ liftIO $ do
109 t <- getTime Monotonic
110 writeIORef (svarStopTime (svarStats chan)) (Just t)
111 printSVar (dumpSVar chan) "SVar Done"
112 return b