git.iko.soy/streamly
on e856425 help sign in
streamly/src/Streamly/Internal/Data/Stream/SVar/Eliminate.hs
404 lines 15 kB history raw
1-- |
2-- Module : Streamly.Internal.Data.Stream.SVar.Eliminate
3-- Copyright : (c) 2017 Composewell Technologies
4-- License : BSD-3-Clause
5-- Maintainer : streamly@composewell.com
6-- Stability : experimental
7-- Portability : GHC
8--
9-- Eliminate a stream by distributing it to multiple SVars concurrently.
10--
11module Streamly.Internal.Data.Stream.SVar.Eliminate
12 (
13 -- * Concurrent Function Application
14 toSVarParallel
16 -- * Concurrent folds
17 -- $concurrentFolds
18 , newFoldSVar
19 , newFoldSVarF
21 , fromConsumer
22 , pushToFold
23 , teeToSVar
24 )
25where
27#include "inline.hs"
29import Control.Concurrent (myThreadId, takeMVar)
30import Control.Monad (when, void)
31import Control.Monad.Catch (throwM)
32import Control.Monad.IO.Class (MonadIO(liftIO))
33import Data.IORef (newIORef, readIORef, writeIORef)
34import Streamly.Internal.Control.Concurrent (MonadAsync)
35import Streamly.Internal.Control.ForkLifted (doFork)
36import Streamly.Internal.Data.Atomics (atomicModifyIORefCAS_)
37import Streamly.Internal.Data.Fold.SVar (write, writeLimited)
38import Streamly.Internal.Data.Fold.Type (Fold(..))
39import Streamly.Data.Stream (Stream)
40import Streamly.Internal.Data.Time.Clock (Clock(Monotonic), getTime)
42import qualified Streamly.Internal.Data.Stream.StreamD.Type as D
43 (Stream(..), Step(..), fold)
44import qualified Streamly.Internal.Data.Stream.StreamK.Type as K
45 (Stream(..), mkStream, foldStream, foldStreamShared, nilM)
46import qualified Streamly.Internal.Data.Stream.Type as Stream
47 (fromStreamK, toStreamK)
49import Streamly.Internal.Data.SVar
51-------------------------------------------------------------------------------
52-- Concurrent function application
53-------------------------------------------------------------------------------
55-- Using StreamD the worker stream producing code can fuse with the code to
56-- queue output to the SVar giving some perf boost.
57--
58-- Note that StreamD can only be used in limited situations, specifically, we
59-- cannot implement joinStreamVarPar using this.
60--
61-- XXX make sure that the SVar passed is a Parallel style SVar.
63-- | Fold the supplied stream to the SVar asynchronously using Parallel
64-- concurrency style.
65-- {-# INLINE_NORMAL toSVarParallel #-}
66{-# INLINE toSVarParallel #-}
67toSVarParallel :: MonadAsync m
68 => State t m a -> SVar t m a -> D.Stream m a -> m ()
69toSVarParallel st sv xs =
70 if svarInspectMode sv
71 then forkWithDiag
72 else do
73 tid <-
74 case getYieldLimit st of
75 Nothing -> doFork (work Nothing)
76 (svarMrun sv)
77 (handleChildException sv)
78 Just _ -> doFork (workLim Nothing)
79 (svarMrun sv)
80 (handleChildException sv)
81 modifyThread sv tid
83 where
85 {-# NOINLINE work #-}
86 work info = D.fold (write sv info) xs
88 {-# NOINLINE workLim #-}
89 workLim info = D.fold (writeLimited sv info) xs
91 {-# NOINLINE forkWithDiag #-}
92 forkWithDiag = do
93 -- We do not use workerCount in case of ParallelVar but still there is
94 -- no harm in maintaining it correctly.
95 liftIO $ atomicModifyIORefCAS_ (workerCount sv) $ \n -> n + 1
96 recordMaxWorkers sv
97 -- This allocation matters when significant number of workers are being
98 -- sent. We allocate it only when needed. The overhead increases by 4x.
99 winfo <-
100 case yieldRateInfo sv of
101 Nothing -> return Nothing
102 Just _ -> liftIO $ do
103 cntRef <- newIORef 0
104 t <- getTime Monotonic
105 lat <- newIORef (0, t)
106 return $ Just WorkerInfo
107 { workerYieldMax = 0
108 , workerYieldCount = cntRef
109 , workerLatencyStart = lat
110 }
111 tid <-
112 case getYieldLimit st of
113 Nothing -> doFork (work winfo)
114 (svarMrun sv)
115 (handleChildException sv)
116 Just _ -> doFork (workLim winfo)
117 (svarMrun sv)
118 (handleChildException sv)
119 modifyThread sv tid
121-------------------------------------------------------------------------------
122-- Support for running folds concurrently
123-------------------------------------------------------------------------------
125-- $concurrentFolds
126--
127-- To run folds concurrently, we need to decouple the fold execution from the
128-- stream production. We use the SVar to do that, we have a single worker
129-- pushing the stream elements to the SVar and on the consumer side a fold
130-- driver pulls the values and folds them.
131--
132-- @
133--
134-- Fold worker <------SVar<------input stream
135-- | exceptions |
136-- --------------->
137--
138-- @
139--
140-- We need a channel for pushing exceptions from the fold worker to the stream
141-- pusher. The stream may be pushed to multiple folds at the same time. For
142-- that we need one SVar per fold:
143--
144-- @
145--
146-- Fold worker <------SVar<---
147-- | |
148-- Fold worker <------SVar<------input stream
149-- | |
150-- Fold worker <------SVar<---
151--
152-- @
153--
154-- Unlike in case concurrent stream evaluation, the puller does not drive the
155-- scheduling and concurrent execution of the stream. The stream is simply
156-- pushed by the stream producer at its own rate. The fold worker just pulls it
157-- and folds it.
158--
159-- Note: If the stream pusher terminates due to an exception, we do not
160-- actively terminate the fold. It gets cleaned up by the GC.
162-------------------------------------------------------------------------------
163-- Process events received by a fold consumer from a stream producer
164-------------------------------------------------------------------------------
166-- | Pull a stream from an SVar to fold it. Like 'fromSVar' except that it does
167-- not drive the evaluation of the stream. It just pulls whatever is available
168-- on the SVar. Also, when the fold stops it sends a notification to the stream
169-- pusher/producer. No exceptions are expected to be propagated from the stream
170-- pusher to the fold puller.
171--
172{-# NOINLINE fromProducer #-}
173fromProducer :: forall m a . MonadAsync m => SVar K.Stream m a -> K.Stream m a
174fromProducer sv = K.mkStream $ \st yld sng stp -> do
175 list <- readOutputQ sv
176 -- Reversing the output is important to guarantee that we process the
177 -- outputs in the same order as they were generated by the constituent
178 -- streams.
179 K.foldStream st yld sng stp $ processEvents $ reverse list
181 where
183 allDone :: m r -> m r
184 allDone stp = do
185 when (svarInspectMode sv) $ do
186 t <- liftIO $ getTime Monotonic
187 liftIO $ writeIORef (svarStopTime (svarStats sv)) (Just t)
188 liftIO $ printSVar sv "SVar Done"
189 sendStopToProducer sv
190 stp
192 {-# INLINE processEvents #-}
193 processEvents :: [ChildEvent a] -> K.Stream m a
194 processEvents [] = K.mkStream $ \st yld sng stp -> do
195 K.foldStream st yld sng stp $ fromProducer sv
197 processEvents (ev : es) = K.mkStream $ \_ yld _ stp -> do
198 let rest = processEvents es
199 case ev of
200 ChildYield a -> yld a rest
201 ChildStop tid e -> do
202 accountThread sv tid
203 case e of
204 Nothing -> allDone stp
205 Just _ -> error "Bug: fromProducer: received exception"
207-- | Create a Fold style SVar that runs a supplied fold function as the
208-- consumer. Any elements sent to the SVar are consumed by the supplied fold
209-- function.
210--
211{-# INLINE newFoldSVar #-}
212newFoldSVar :: MonadAsync m
213 => State K.Stream m a -> (Stream m a -> m b) -> m (SVar K.Stream m a)
214newFoldSVar stt f = do
215 -- Buffer size for the SVar is derived from the current state
216 sv <- newParallelVar StopAny (adaptState stt)
218 -- Add the producer thread-id to the SVar.
219 liftIO myThreadId >>= modifyThread sv
221 void $ doFork (void $ f $ Stream.fromStreamK $ fromProducer sv)
222 (svarMrun sv)
223 (handleFoldException sv)
224 return sv
226data FromSVarState t m a =
227 FromSVarInit
228 | FromSVarRead (SVar t m a)
229 | FromSVarLoop (SVar t m a) [ChildEvent a]
230 | FromSVarDone (SVar t m a)
232-- | Like 'fromProducer' but generates a StreamD style stream instead of
233-- StreamK.
234--
235{-# INLINE_NORMAL fromProducerD #-}
236fromProducerD :: (MonadAsync m) => SVar t m a -> D.Stream m a
237fromProducerD svar = D.Stream step (FromSVarRead svar)
238 where
240 {-# INLINE_LATE step #-}
241 step _ (FromSVarRead sv) = do
242 list <- readOutputQ sv
243 -- Reversing the output is important to guarantee that we process the
244 -- outputs in the same order as they were generated by the constituent
245 -- streams.
246 return $ D.Skip $ FromSVarLoop sv (Prelude.reverse list)
248 step _ (FromSVarLoop sv []) = return $ D.Skip $ FromSVarRead sv
249 step _ (FromSVarLoop sv (ev : es)) = do
250 case ev of
251 ChildYield a -> return $ D.Yield a (FromSVarLoop sv es)
252 ChildStop tid e -> do
253 accountThread sv tid
254 case e of
255 Nothing -> do
256 sendStopToProducer sv
257 return $ D.Skip (FromSVarDone sv)
258 Just _ -> error "Bug: fromProducer: received exception"
260 step _ (FromSVarDone sv) = do
261 when (svarInspectMode sv) $ do
262 t <- liftIO $ getTime Monotonic
263 liftIO $ writeIORef (svarStopTime (svarStats sv)) (Just t)
264 liftIO $ printSVar sv "SVar Done"
265 return D.Stop
267 step _ FromSVarInit = undefined
269-- | Like 'newFoldSVar' except that it uses a 'Fold' instead of a fold
270-- function.
271--
272{-# INLINE newFoldSVarF #-}
273newFoldSVarF :: MonadAsync m => State t m a -> Fold m a b -> m (SVar t m a)
274newFoldSVarF stt f = do
275 -- Buffer size for the SVar is derived from the current state
276 sv <- newParallelVar StopAny (adaptState stt)
277 -- Add the producer thread-id to the SVar.
278 liftIO myThreadId >>= modifyThread sv
279 void $ doFork (work sv) (svarMrun sv) (handleFoldException sv)
280 return sv
282 where
284 {-# NOINLINE work #-}
285 work sv = void $ D.fold f $ fromProducerD sv
287-------------------------------------------------------------------------------
288-- Process events received by the producer thread from the consumer side
289-------------------------------------------------------------------------------
291-- XXX currently only one event is sent by a fold consumer to the stream
292-- producer. But we can potentially have multiple events e.g. the fold step can
293-- generate exception more than once and the producer can ignore those
294-- exceptions or handle them and still keep driving the fold.
295--
296-- | Poll for events sent by the fold consumer to the stream pusher. The fold
297-- consumer can send a "Stop" event or an exception. When a "Stop" is received
298-- this function returns 'True'. If an exception is recieved then it throws the
299-- exception.
300--
301{-# NOINLINE fromConsumer #-}
302fromConsumer :: MonadAsync m => SVar K.Stream m a -> m Bool
303fromConsumer sv = do
304 (list, _) <- liftIO $ readOutputQBasic (outputQueueFromConsumer sv)
305 -- Reversing the output is important to guarantee that we process the
306 -- outputs in the same order as they were generated by the constituent
307 -- streams.
308 processEvents $ reverse list
310 where
312 {-# INLINE processEvents #-}
313 processEvents [] = return False
314 processEvents (ev : _) = do
315 case ev of
316 ChildStop _ e -> do
317 case e of
318 Nothing -> return True
319 Just ex -> throwM ex
320 ChildYield _ -> error "Bug: fromConsumer: invalid ChildYield event"
322-- | Push values from a stream to a fold worker via an SVar. Before pushing a
323-- value to the SVar it polls for events received from the fold consumer. If a
324-- stop event is received then it returns 'True' otherwise false. Propagates
325-- exceptions received from the fold consumer.
326--
327{-# INLINE pushToFold #-}
328pushToFold :: MonadAsync m => SVar K.Stream m a -> a -> m Bool
329pushToFold sv a = do
330 -- Check for exceptions before decrement so that we do not
331 -- block forever if the child already exited with an exception.
332 --
333 -- We avoid a race between the consumer fold sending an event and we
334 -- blocking on decrementBufferLimit by waking up the producer thread in
335 -- sendToProducer before any event is sent by the fold to the producer
336 -- stream.
337 let qref = outputQueueFromConsumer sv
338 done <- do
339 (_, n) <- liftIO $ readIORef qref
340 if n > 0
341 then fromConsumer sv
342 else return False
343 if done
344 then return True
345 else liftIO $ do
346 decrementBufferLimit sv
347 void $ send sv (ChildYield a)
348 return False
350------------------------------------------------------------------------------
351-- Clone and distribute a stream in parallel
352------------------------------------------------------------------------------
354-- XXX this could be written in StreamD style for better efficiency with fusion.
355--
356-- | Tap a stream and send the elements to the specified SVar in addition to
357-- yielding them again. The SVar runs a fold consumer. Elements are tapped and
358-- sent to the SVar until the fold finishes. Any exceptions from the fold
359-- evaluation are propagated in the current thread.
360--
361-- @
362--
363-- ------input stream---------output stream----->
364-- /|\\ |
365-- exceptions | | input
366-- | \\|/
367-- ----SVar
368-- |
369-- Fold
370--
371-- @
372--
373{-# INLINE teeToSVar #-}
374teeToSVar :: MonadAsync m =>
375 SVar K.Stream m a -> Stream m a -> Stream m a
376teeToSVar svr m = Stream.fromStreamK $ K.mkStream $ \st yld sng stp -> do
377 K.foldStreamShared st yld sng stp (go False $ Stream.toStreamK m)
379 where
381 go False m0 = K.mkStream $ \st yld _ stp -> do
382 let drain = do
383 -- In general, a Stop event would come equipped with the result
384 -- of the fold. It is not used here but it would be useful in
385 -- applicative and distribute.
386 done <- fromConsumer svr
387 when (not done) $ do
388 liftIO $ withDiagMVar svr "teeToSVar: waiting to drain"
389 $ takeMVar (outputDoorBellFromConsumer svr)
390 drain
392 stopFold = do
393 liftIO $ sendStop svr Nothing
394 -- drain/wait until a stop event arrives from the fold.
395 drain
397 stop = stopFold >> stp
398 single a = do
399 done <- pushToFold svr a
400 yld a (go done (K.nilM stopFold))
401 yieldk a r = pushToFold svr a >>= \done -> yld a (go done r)
402 in K.foldStreamShared st yieldk single stop m0
404 go True m0 = m0