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