git.iko.soy/streamly
on e856425 help sign in
streamly/src/Streamly/Internal/Data/Stream/Concurrent/Channel/Consumer.hs
130 lines 4.9 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.Trans.Control (MonadBaseControl)
22import Control.Monad (when, void)
23import Control.Monad.IO.Class (MonadIO(liftIO))
24import Data.IORef (readIORef)
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 :: (MonadIO m, MonadBaseControl IO m) =>
42 Bool -> Channel m a -> m [ChildEvent a]
43readOutputQBounded eagerEval sv = do
44 (list, len) <- liftIO $ readOutputQChan sv
45 -- When there is no output seen we dispatch more workers to help
46 -- out if there is work pending in the work queue.
47 if len <= 0
48 then blockingRead
49 else do
50 -- send a worker proactively, if needed, even before we start
51 -- processing the output. This may degrade single processor
52 -- perf but improves multi-processor, because of more
53 -- parallelism
54 sendOneWorker
55 return list
57 where
59 sendOneWorker = do
60 cnt <- liftIO $ readIORef $ workerCount sv
61 when (cnt <= 0) $ do
62 done <- liftIO $ isWorkDone sv
63 when (not done) (pushWorker 0 sv)
65 {-# INLINE blockingRead #-}
66 blockingRead = do
67 sendWorkerWait eagerEval sendWorkerDelay (dispatchWorker 0) sv
68 liftIO (fst `fmap` readOutputQChan sv)
70readOutputQPaced :: (MonadIO m, MonadBaseControl IO m) =>
71 Channel m a -> m [ChildEvent a]
72readOutputQPaced sv = do
73 (list, len) <- liftIO $ readOutputQChan sv
74 if len <= 0
75 then blockingRead
76 else do
77 -- XXX send a worker proactively, if needed, even before we start
78 -- processing the output.
79 void $ dispatchWorkerPaced sv
80 return list
82 where
84 {-# INLINE blockingRead #-}
85 blockingRead = do
86 sendWorkerWait False sendWorkerDelayPaced dispatchWorkerPaced sv
87 liftIO (fst `fmap` readOutputQChan sv)
89postProcessPaced :: (MonadIO m, MonadBaseControl IO m) => Channel m a -> m Bool
90postProcessPaced sv = do
91 workersDone <- allThreadsDone (workerThreads sv)
92 -- XXX If during consumption we figure out we are getting delayed then we
93 -- should trigger dispatch there as well. We should try to check on the
94 -- workers after consuming every n item from the buffer?
95 if workersDone
96 then do
97 r <- liftIO $ isWorkDone sv
98 when (not r) $ do
99 void $ dispatchWorkerPaced sv
100 -- Note that we need to guarantee a worker since the work is not
101 -- finished, therefore we cannot just rely on dispatchWorkerPaced
102 -- which may or may not send a worker.
103 noWorker <- allThreadsDone (workerThreads sv)
104 when noWorker $ pushWorker 0 sv
105 return r
106 else return False
108postProcessBounded :: (MonadIO m, MonadBaseControl IO m) =>
109 Channel m a -> m Bool
110postProcessBounded sv = do
111 workersDone <- allThreadsDone (workerThreads sv)
112 -- There may still be work pending even if there are no workers pending
113 -- because all the workers may return if the outputQueue becomes full. In
114 -- that case send off a worker to kickstart the work again.
115 --
116 -- Note that isWorkDone can only be safely checked if all workers are done.
117 -- When some workers are in progress they may have decremented the yield
118 -- Limit and later ending up incrementing it again. If we look at the yield
119 -- limit in that window we may falsely say that it is 0 and therefore we
120 -- are done.
121 if workersDone
122 then do
123 r <- liftIO $ isWorkDone sv
124 -- Note that we need to guarantee a worker, therefore we cannot just
125 -- use dispatchWorker which may or may not send a worker.
126 when (not r) (pushWorker 0 sv)
127 -- XXX do we need to dispatch many here?
128 -- void $ dispatchWorker sv
129 return r
130 else return False