git.iko.soy/streamly
on 56569bc help sign in
streamly/src/Streamly/Internal/Data/Fold/Concurrent/Channel.hs
91 lines 3.1 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 , inspect
19 -- * Fold operations
20 , evalWith
21 )
22where
24import Control.Concurrent (takeMVar)
25import Control.Monad (void, when)
26import Control.Monad.IO.Class (MonadIO(liftIO))
27import Data.IORef (writeIORef)
28import Streamly.Internal.Control.Concurrent (MonadAsync)
29import Streamly.Internal.Data.Fold (Fold(..), Step (..))
30import Streamly.Internal.Data.Stream.Channel.Worker (sendWithDoorBell)
31import Streamly.Internal.Data.Time.Clock (Clock(Monotonic), getTime)
33import Streamly.Internal.Data.Fold.Concurrent.Channel.Type
34import Streamly.Internal.Data.Stream.Channel.Types
36-------------------------------------------------------------------------------
37-- Evaluating a Fold
38-------------------------------------------------------------------------------
40-- XXX Cleanup the fold if the stream is interrupted. Add a GC hook.
42-- | Evaluate the fold asynchronously in a worker thread separate from the
43-- driver thread.
44--
45{-# INLINABLE evalWith #-}
46evalWith :: MonadAsync m => (Config -> Config) -> Fold m a b -> Fold m a b
47evalWith modifier f =
48 Fold step initial extract
50 where
52 initial = Partial <$> newChannel modifier f
54 -- XXX This is not truly asynchronous. If the fold is done we only get to
55 -- know when we send the next input unless the stream ends. We could
56 -- potentially throw an async exception to the driver to inform it
57 -- asynchronously. Alternatively, the stream should not block forever, it
58 -- should keep polling the fold status. We can insert a timer tick in the
59 -- input stream to do that.
60 --
61 -- A polled stream abstraction may be useful, it would consist of normal
62 -- events and tick events, latter are guaranteed to arrive.
63 step chan a = do
64 status <- sendToWorker chan a
65 return $ case status of
66 Nothing -> Partial chan
67 Just b -> Done b
69 extract chan = do
70 liftIO $ void
71 $ sendWithDoorBell
72 (outputQueue chan)
73 (outputDoorBell chan)
74 ChildStopChannel
75 status <- checkFoldStatus chan
76 case status of
77 Nothing -> do
78 liftIO
79 $ withDiagMVar
80 (svarInspectMode chan)
81 (dumpSVar chan)
82 "evalWith: waiting to drain"
83 $ takeMVar (outputDoorBellFromConsumer chan)
84 -- XXX remove recursion
85 extract chan
86 Just b -> do
87 when (svarInspectMode chan) $ liftIO $ do
88 t <- getTime Monotonic
89 writeIORef (svarStopTime (svarStats chan)) (Just t)
90 printSVar (dumpSVar chan) "SVar Done"
91 return b