git.iko.soy/streamly
on 43dda21 help sign in
streamly/src/Streamly/Internal/Data/SVar/Pull.hs
164 lines 5.7 kB history raw
1-- |
2-- Module : Streamly.Internal.Data.SVar.Pull
3-- Copyright : (c) 2017 Composewell Technologies
4-- License : BSD-3-Clause
5-- Maintainer : streamly@composewell.com
6-- Stability : experimental
7-- Portability : GHC
8--
9--
10module Streamly.Internal.Data.SVar.Pull
11 (
12 -- * Read Output
13 readOutputQBasic
14 , readOutputQRaw
15 , readOutputQPaced
16 , readOutputQBounded
18 -- * Postprocess Hook After Reading
19 , postProcessPaced
20 , postProcessBounded
22 -- * Release Resources
23 , cleanupSVar
24 , cleanupSVarFromWorker
25 )
26where
28#include "inline.hs"
30import Control.Concurrent (myThreadId, throwTo)
31import Control.Monad (when, void)
32import Control.Monad.IO.Class (MonadIO(liftIO))
33import Data.IORef (readIORef, writeIORef)
34import Data.IORef (IORef)
35import Streamly.Internal.Control.Concurrent (MonadAsync)
36import Streamly.Internal.Data.Atomics (atomicModifyIORefCAS)
38import qualified Data.Set as S
40import Streamly.Internal.Data.SVar.Type
41import Streamly.Internal.Data.SVar.Dispatch
43-------------------------------------------------------------------------------
44-- Reading from the workers' output queue/buffer
45-------------------------------------------------------------------------------
47{-# INLINE readOutputQBasic #-}
48readOutputQBasic :: IORef ([ChildEvent a], Int) -> IO ([ChildEvent a], Int)
49readOutputQBasic q = atomicModifyIORefCAS q $ \x -> (([],0), x)
51{-# INLINE readOutputQRaw #-}
52readOutputQRaw :: SVar t m a -> IO ([ChildEvent a], Int)
53readOutputQRaw sv = do
54 (list, len) <- readOutputQBasic (outputQueue sv)
55 when (svarInspectMode sv) $ do
56 let ref = maxOutQSize $ svarStats sv
57 oqLen <- readIORef ref
58 when (len > oqLen) $ writeIORef ref len
59 return (list, len)
61readOutputQBounded :: MonadAsync m => SVar t m a -> m [ChildEvent a]
62readOutputQBounded sv = do
63 (list, len) <- liftIO $ readOutputQRaw sv
64 -- When there is no output seen we dispatch more workers to help
65 -- out if there is work pending in the work queue.
66 if len <= 0
67 then blockingRead
68 else do
69 -- send a worker proactively, if needed, even before we start
70 -- processing the output. This may degrade single processor
71 -- perf but improves multi-processor, because of more
72 -- parallelism
73 sendOneWorker
74 return list
76 where
78 sendOneWorker = do
79 cnt <- liftIO $ readIORef $ workerCount sv
80 when (cnt <= 0) $ do
81 done <- liftIO $ isWorkDone sv
82 when (not done) (pushWorker 0 sv)
84 {-# INLINE blockingRead #-}
85 blockingRead = do
86 sendWorkerWait sendWorkerDelay (dispatchWorker 0) sv
87 liftIO (fst `fmap` readOutputQRaw sv)
89readOutputQPaced :: MonadAsync m => SVar t m a -> m [ChildEvent a]
90readOutputQPaced sv = do
91 (list, len) <- liftIO $ readOutputQRaw sv
92 if len <= 0
93 then blockingRead
94 else do
95 -- XXX send a worker proactively, if needed, even before we start
96 -- processing the output.
97 void $ dispatchWorkerPaced sv
98 return list
100 where
102 {-# INLINE blockingRead #-}
103 blockingRead = do
104 sendWorkerWait sendWorkerDelayPaced dispatchWorkerPaced sv
105 liftIO (fst `fmap` readOutputQRaw sv)
107postProcessPaced :: MonadAsync m => SVar t m a -> m Bool
108postProcessPaced sv = do
109 workersDone <- allThreadsDone sv
110 -- XXX If during consumption we figure out we are getting delayed then we
111 -- should trigger dispatch there as well. We should try to check on the
112 -- workers after consuming every n item from the buffer?
113 if workersDone
114 then do
115 r <- liftIO $ isWorkDone sv
116 when (not r) $ do
117 void $ dispatchWorkerPaced sv
118 -- Note that we need to guarantee a worker since the work is not
119 -- finished, therefore we cannot just rely on dispatchWorkerPaced
120 -- which may or may not send a worker.
121 noWorker <- allThreadsDone sv
122 when noWorker $ pushWorker 0 sv
123 return r
124 else return False
126postProcessBounded :: MonadAsync m => SVar t m a -> m Bool
127postProcessBounded sv = do
128 workersDone <- allThreadsDone sv
129 -- There may still be work pending even if there are no workers pending
130 -- because all the workers may return if the outputQueue becomes full. In
131 -- that case send off a worker to kickstart the work again.
132 --
133 -- Note that isWorkDone can only be safely checked if all workers are done.
134 -- When some workers are in progress they may have decremented the yield
135 -- Limit and later ending up incrementing it again. If we look at the yield
136 -- limit in that window we may falsely say that it is 0 and therefore we
137 -- are done.
138 if workersDone
139 then do
140 r <- liftIO $ isWorkDone sv
141 -- Note that we need to guarantee a worker, therefore we cannot just
142 -- use dispatchWorker which may or may not send a worker.
143 when (not r) (pushWorker 0 sv)
144 -- XXX do we need to dispatch many here?
145 -- void $ dispatchWorker sv
146 return r
147 else return False
149-------------------------------------------------------------------------------
150-- Cleanup
151-------------------------------------------------------------------------------
153cleanupSVar :: SVar t m a -> IO ()
154cleanupSVar sv = do
155 workers <- readIORef (workerThreads sv)
156 Prelude.mapM_ (`throwTo` ThreadAbort)
157 workers
159cleanupSVarFromWorker :: SVar t m a -> IO ()
160cleanupSVarFromWorker sv = do
161 workers <- readIORef (workerThreads sv)
162 self <- myThreadId
163 Prelude.mapM_ (`throwTo` ThreadAbort)
164 (Prelude.filter (/= self) $ S.toList workers)