git.iko.soy/streamly
on e856425 help sign in
streamly/src/Streamly/Internal/Data/Stream/Concurrent/Channel/Interleave.hs
251 lines 9.4 kB history raw
1-- |
2-- Module : Streamly.Internal.Data.Stream.Concurrent.Channel.Interleave
3-- Copyright : (c) 2017 Composewell Technologies
4-- License : BSD-3-Clause
5-- Maintainer : streamly@composewell.com
6-- Stability : experimental
7-- Portability : GHC
9module Streamly.Internal.Data.Stream.Concurrent.Channel.Interleave
10 (
11 newChannel
12 )
13where
15#include "inline.hs"
17import Control.Concurrent (myThreadId)
18import Control.Concurrent.MVar (newEmptyMVar)
19import Control.Monad.IO.Class (MonadIO(liftIO))
20import Control.Monad.Trans.Control (MonadBaseControl, restoreM)
21import Data.Concurrent.Queue.MichaelScott (LinkedQueue, newQ, nullQ, tryPopR, pushL)
22import Data.IORef (newIORef, readIORef)
23import Streamly.Internal.Control.Concurrent
24 (MonadRunInIO, MonadAsync, RunInIO(..), askRunInIO)
25import Streamly.Internal.Data.Stream.Channel.Dispatcher (delThread)
27import qualified Data.Set as Set
28import qualified Streamly.Internal.Data.Stream.StreamK.Type as K
30import Streamly.Internal.Data.Stream.Concurrent.Channel.Consumer
31import Streamly.Internal.Data.Stream.Concurrent.Channel.Type
32import Streamly.Internal.Data.Stream.Channel.Types
34------------------------------------------------------------------------------
35-- Creating a channel
36------------------------------------------------------------------------------
38data WorkerStatus = Continue | Suspend
40-- XXX This is not strictly round-robin as the streams that are faster may
41-- yield more elements than the ones that are slower. Also, when streams
42-- suspend due to buffer getting full they get added to the queue in a random
43-- order.
45{-# INLINE enqueueFIFO #-}
46enqueueFIFO ::
47 Channel m a
48 -> LinkedQueue (RunInIO m, K.Stream m a)
49 -> (RunInIO m, K.Stream m a)
50 -> IO ()
51enqueueFIFO sv q m = do
52 pushL q m
53 ringDoorBell (doorBellOnWorkQ sv) (outputDoorBell sv)
55{-# INLINE workLoopFIFO #-}
56workLoopFIFO
57 :: MonadRunInIO m
58 => LinkedQueue (RunInIO m, K.Stream m a)
59 -> Channel m a
60 -> Maybe WorkerInfo
61 -> m ()
62workLoopFIFO q sv winfo = run
64 where
66 run = do
67 work <- liftIO $ tryPopR q
68 case work of
69 Nothing -> liftIO $ stop sv winfo
70 Just (RunInIO runin, m) -> do
71 r <- liftIO
72 $ runin
73 $ K.foldStreamShared
74 undefined yieldk single (return Continue) m
75 res <- restoreM r
76 case res of
77 Continue -> run
78 Suspend -> liftIO $ stop sv winfo
80 single a = do
81 res <- liftIO $ yield sv winfo a
82 return $ if res then Continue else Suspend
84 -- XXX in general we would like to yield "n" elements from a single stream
85 -- before moving on to the next. Single element granularity could be too
86 -- expensive in certain cases. Similarly, we can use time limit for
87 -- yielding.
88 yieldk a r = do
89 res <- liftIO $ yield sv winfo a
90 runInIO <- askRunInIO
91 liftIO $ enqueueFIFO sv q (runInIO, r)
92 return $ if res then Continue else Suspend
94{-# INLINE workLoopFIFOLimited #-}
95workLoopFIFOLimited
96 :: forall m a. MonadRunInIO m
97 => LinkedQueue (RunInIO m, K.Stream m a)
98 -> Channel m a
99 -> Maybe WorkerInfo
100 -> m ()
101workLoopFIFOLimited q sv winfo = run
103 where
105 incrContinue =
106 liftIO (incrementYieldLimit (remainingWork sv)) >> return Continue
108 run = do
109 work <- liftIO $ tryPopR q
110 case work of
111 Nothing -> liftIO $ stop sv winfo
112 Just (RunInIO runin, m) -> do
113 yieldLimitOk <- liftIO $ decrementYieldLimit (remainingWork sv)
114 if yieldLimitOk
115 then do
116 r <- liftIO
117 $ runin
118 $ K.foldStreamShared
119 undefined yieldk single incrContinue m
120 res <- restoreM r
121 case res of
122 Continue -> run
123 Suspend -> liftIO $ stop sv winfo
124 else liftIO $ do
125 enqueueFIFO sv q (RunInIO runin, m)
126 incrementYieldLimit (remainingWork sv)
127 stop sv winfo
129 single a = do
130 res <- liftIO $ yield sv winfo a
131 return $ if res then Continue else Suspend
133 yieldk a r = do
134 res <- liftIO $ yield sv winfo a
135 runInIO <- askRunInIO
136 liftIO $ enqueueFIFO sv q (runInIO, r)
137 yieldLimitOk <- liftIO $ decrementYieldLimit (remainingWork sv)
138 if res && yieldLimitOk
139 then return Continue
140 else liftIO $ do
141 incrementYieldLimit (remainingWork sv)
142 return Suspend
144-------------------------------------------------------------------------------
145-- SVar creation
146-------------------------------------------------------------------------------
148-- XXX we have this function in this file because passing runStreamLIFO as a
149-- function argument to this function results in a perf degradation of more
150-- than 10%. Need to investigate what the root cause is.
151-- Interestingly, the same thing does not make any difference for Ahead.
152getFifoSVar :: forall m a. (MonadIO m, MonadBaseControl IO m) =>
153 RunInIO m -> Config -> IO (Channel m a)
154getFifoSVar mrun cfg = do
155 outQ <- newIORef ([], 0)
156 outQMv <- newEmptyMVar
157 active <- newIORef 0
158 wfw <- newIORef False
159 running <- newIORef Set.empty
160 q <- newQ
161 yl <- case getYieldLimit cfg of
162 Nothing -> return Nothing
163 Just x -> Just <$> newIORef x
164 rateInfo <- newRateInfo cfg
166 stats <- newSVarStats
167 tid <- myThreadId
169 let isWorkFinished _ = nullQ q
170 let isWorkFinishedLimited sv = do
171 yieldsDone <-
172 case remainingWork sv of
173 Just ref -> do
174 n <- readIORef ref
175 return (n <= 0)
176 Nothing -> return False
177 qEmpty <- nullQ q
178 return $ qEmpty || yieldsDone
180 let getSVar :: Channel m a
181 -> (Channel m a -> m [ChildEvent a])
182 -> (Channel m a -> m Bool)
183 -> (Channel m a -> IO Bool)
184 -> (LinkedQueue (RunInIO m, K.Stream m a)
185 -> Channel m a
186 -> Maybe WorkerInfo
187 -> m())
188 -> Channel m a
189 getSVar sv readOutput postProc workDone wloop = Channel
190 { outputQueue = outQ
191 , remainingWork = yl
192 , maxBufferLimit = getMaxBuffer cfg
193 , maxWorkerLimit = min (getMaxThreads cfg) (getMaxBuffer cfg)
194 , yieldRateInfo = rateInfo
195 , outputDoorBell = outQMv
196 , readOutputQ = readOutput sv
197 , postProcess = postProc sv
198 , workerThreads = running
199 , workLoop = wloop q sv
200 , enqueue = \_ -> enqueueFIFO sv q
201 , eagerDispatch = return ()
202 , isWorkDone = workDone sv
203 , isQueueDone = workDone sv
204 , doorBellOnWorkQ = wfw
205 , svarMrun = mrun
206 , workerCount = active
207 , accountThread = delThread running
208 , workerStopMVar = undefined
209 , svarRef = Nothing
210 , svarInspectMode = getInspectMode cfg
211 , svarCreator = tid
212 , svarStats = stats
213 }
215 let sv =
216 case getStreamRate cfg of
217 Nothing ->
218 case getYieldLimit cfg of
219 Nothing -> getSVar sv (readOutputQBounded False)
220 postProcessBounded
221 isWorkFinished
222 workLoopFIFO
223 Just _ -> getSVar sv (readOutputQBounded False)
224 postProcessBounded
225 isWorkFinishedLimited
226 workLoopFIFOLimited
227 Just _ ->
228 case getYieldLimit cfg of
229 Nothing -> getSVar sv readOutputQPaced
230 postProcessPaced
231 isWorkFinished
232 workLoopFIFO
233 Just _ -> getSVar sv readOutputQPaced
234 postProcessPaced
235 isWorkFinishedLimited
236 workLoopFIFOLimited
237 in return sv
239-- XXX GHC: If instead of MonadAsync we use (MonadIO m, MonadBaseControl IO m)
240-- constraint we get a 2x perf regression. Need to look into that.
241--
242-- | Create a new async style concurrent stream evaluation channel. The monad
243-- state used to run the stream actions is taken from the call site of
244-- newChannel.
245{-# INLINABLE newChannel #-}
246{-# SPECIALIZE newChannel :: (Config -> Config) -> IO (Channel IO a) #-}
247newChannel :: MonadAsync m =>
248 (Config -> Config) -> m (Channel m a)
249newChannel modifier = do
250 mrun <- askRunInIO
251 liftIO $ getFifoSVar mrun (modifier defaultConfig)