git.iko.soy/streamly
on e01057f help sign in
streamly/src/Streamly/Internal/Data/Stream/Concurrent/Channel/Consumer.hs
127 lines 4.8 kB history raw
1-- |
2-- Module : Streamly.Internal.Data.Stream.Concurrent.Channel.Consumer
3-- Copyright : (c) 2017 Composewell Technologies
4-- License : BSD-3-Clause
5-- Maintainer : streamly@composewell.com
6-- Stability : experimental
7-- Portability : GHC
8--
9module Streamly.Internal.Data.Stream.Concurrent.Channel.Consumer
10 (
11 -- * Read Output
12 readOutputQPaced
13 , readOutputQBounded
15 -- * Postprocess Hook After Reading
16 , postProcessPaced
17 , postProcessBounded
18 )
19where
21import Control.Monad (when, void)
22import Control.Monad.IO.Class (MonadIO(liftIO))
23import Data.IORef (readIORef)
24import Streamly.Internal.Control.Concurrent (MonadRunInIO)
26import Streamly.Internal.Data.Stream.Concurrent.Channel.Dispatcher
27import Streamly.Internal.Data.Stream.Concurrent.Channel.Type
28import Streamly.Internal.Data.Stream.Channel.Dispatcher
29import Streamly.Internal.Data.Stream.Channel.Types
31-------------------------------------------------------------------------------
32-- Reading from the workers' output queue/buffer
33-------------------------------------------------------------------------------
35{-# INLINE readOutputQChan #-}
36readOutputQChan :: Channel m a -> IO ([ChildEvent a], Int)
37readOutputQChan sv = do
38 let ss = if svarInspectMode sv then Just (svarStats sv) else Nothing
39 in readOutputQRaw (outputQueue sv) ss
41readOutputQBounded :: MonadRunInIO m => Bool -> Channel m a -> m [ChildEvent a]
42readOutputQBounded eagerEval sv = do
43 (list, len) <- liftIO $ readOutputQChan sv
44 -- When there is no output seen we dispatch more workers to help
45 -- out if there is work pending in the work queue.
46 if len <= 0
47 then blockingRead
48 else do
49 -- send a worker proactively, if needed, even before we start
50 -- processing the output. This may degrade single processor
51 -- perf but improves multi-processor, because of more
52 -- parallelism
53 sendOneWorker
54 return list
56 where
58 sendOneWorker = do
59 cnt <- liftIO $ readIORef $ workerCount sv
60 when (cnt <= 0) $ do
61 done <- liftIO $ isWorkDone sv
62 when (not done) (pushWorker 0 sv)
64 {-# INLINE blockingRead #-}
65 blockingRead = do
66 sendWorkerWait eagerEval sendWorkerDelay (dispatchWorker 0) sv
67 liftIO (fst `fmap` readOutputQChan sv)
69readOutputQPaced :: MonadRunInIO m => Channel m a -> m [ChildEvent a]
70readOutputQPaced sv = do
71 (list, len) <- liftIO $ readOutputQChan sv
72 if len <= 0
73 then blockingRead
74 else do
75 -- XXX send a worker proactively, if needed, even before we start
76 -- processing the output.
77 void $ dispatchWorkerPaced sv
78 return list
80 where
82 {-# INLINE blockingRead #-}
83 blockingRead = do
84 sendWorkerWait False sendWorkerDelayPaced dispatchWorkerPaced sv
85 liftIO (fst `fmap` readOutputQChan sv)
87postProcessPaced :: MonadRunInIO m => Channel m a -> m Bool
88postProcessPaced sv = do
89 workersDone <- allThreadsDone (workerThreads sv)
90 -- XXX If during consumption we figure out we are getting delayed then we
91 -- should trigger dispatch there as well. We should try to check on the
92 -- workers after consuming every n item from the buffer?
93 if workersDone
94 then do
95 r <- liftIO $ isWorkDone sv
96 when (not r) $ do
97 void $ dispatchWorkerPaced sv
98 -- Note that we need to guarantee a worker since the work is not
99 -- finished, therefore we cannot just rely on dispatchWorkerPaced
100 -- which may or may not send a worker.
101 noWorker <- allThreadsDone (workerThreads sv)
102 when noWorker $ pushWorker 0 sv
103 return r
104 else return False
106postProcessBounded :: MonadRunInIO m => Channel m a -> m Bool
107postProcessBounded sv = do
108 workersDone <- allThreadsDone (workerThreads sv)
109 -- There may still be work pending even if there are no workers pending
110 -- because all the workers may return if the outputQueue becomes full. In
111 -- that case send off a worker to kickstart the work again.
112 --
113 -- Note that isWorkDone can only be safely checked if all workers are done.
114 -- When some workers are in progress they may have decremented the yield
115 -- Limit and later ending up incrementing it again. If we look at the yield
116 -- limit in that window we may falsely say that it is 0 and therefore we
117 -- are done.
118 if workersDone
119 then do
120 r <- liftIO $ isWorkDone sv
121 -- Note that we need to guarantee a worker, therefore we cannot just
122 -- use dispatchWorker which may or may not send a worker.
123 when (not r) (pushWorker 0 sv)
124 -- XXX do we need to dispatch many here?
125 -- void $ dispatchWorker sv
126 return r
127 else return False