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