1#ifdef __HADDOCK_VERSION__
6{-# LANGUAGE TemplateHaskell #-}
7{-# OPTIONS_GHC -fplugin Test.Inspection.Plugin #-}
11-- Module : Streamly.Internal.Data.Stream.SVar
12-- Copyright : (c) 2017 Composewell Technologies
15-- Maintainer : streamly@composewell.com
16-- Stability : experimental
20module Streamly.Internal.Data.Stream.SVar
22 -- * Unfold streams from SVar
28 -- * Fold streams to SVar
44import Control.Concurrent (myThreadId, takeMVar)
45import Control.Exception (fromException)
46import Control.Monad (when, void)
47import Control.Monad.Catch (throwM)
48import Control.Monad.IO.Class (MonadIO(liftIO))
49import Data.IORef (newIORef, readIORef, mkWeakIORef, writeIORef)
50import Data.Maybe (isNothing)
51import Streamly.Internal.Data.Atomics (atomicModifyIORefCAS_)
52import Streamly.Internal.Data.Fold.Type (Fold(..))
53import Streamly.Internal.Data.Time.Clock (Clock(Monotonic), getTime)
54import System.Mem (performMajorGC)
56import qualified Streamly.Internal.Data.Fold.Type as FL
57import qualified Streamly.Internal.Data.Stream.StreamD.Type as D
58import qualified Streamly.Internal.Data.Stream.StreamK.Type as K
60import Streamly.Internal.Data.SVar
62#if __GLASGOW_HASKELL__ < 810
64import Control.Exception (Exception)
65import Control.Monad.Catch (MonadThrow)
66import Control.Monad.Trans.Control (MonadBaseControl)
67import Data.Typeable (Typeable)
68import Test.Inspection (inspect, hasNoTypeClassesExcept)
72------------------------------------------------------------------------------
73-- Generating streams from SVar
74------------------------------------------------------------------------------
78-- Usually the SVar is used to concurrently evaluate multiple actions in a
79-- stream using many worker threads that push the results to the SVar and a
80-- single puller that pulls them from SVar generating the evaluated stream.
85-- <-----------------|<--------worker
87-- output stream <------SVar<------worker
93-- The puller itself schedules the worker threads based on demand.
94-- Exceptions are propagated from the worker threads to the puller.
96-- | Pull a stream from an SVar.
97{-# NOINLINE fromStreamVar #-}
98fromStreamVar :: MonadAsync m => SVar K.Stream m a -> K.Stream m a
99fromStreamVar sv = K.MkStream $ \st yld sng stp -> do
100 list <- readOutputQ sv
101 -- Reversing the output is important to guarantee that we process the
102 -- outputs in the same order as they were generated by the constituent
104 K.foldStream st yld sng stp $ processEvents $ reverse list
109 when (svarInspectMode sv) $ do
110 t <- liftIO $ getTime Monotonic
111 liftIO $ writeIORef (svarStopTime (svarStats sv)) (Just t)
112 liftIO $ printSVar sv "SVar Done"
115 {-# INLINE processEvents #-}
116 processEvents [] = K.MkStream $ \st yld sng stp -> do
117 done <- postProcess sv
120 else K.foldStream st yld sng stp $ fromStreamVar sv
122 processEvents (ev : es) = K.MkStream $ \st yld sng stp -> do
123 let rest = processEvents es
125 ChildYield a -> yld a rest
126 ChildStop tid e -> do
130 stop <- shouldStop tid
132 then liftIO (cleanupSVar sv) >> allDone stp
133 else K.foldStream st yld sng stp rest
135 case fromException ex of
137 K.foldStream st yld sng stp rest
138 Nothing -> liftIO (cleanupSVar sv) >> throwM ex
140 case svarStopStyle sv of
141 StopNone -> return False
142 StopAny -> return True
144 sid <- liftIO $ readIORef (svarStopBy sv)
147#if __GLASGOW_HASKELL__ < 810
149-- Use of GHC constraint tuple (GHC.Classes.(%,,%)) in fromStreamVar leads to
150-- space leak because the tuple gets allocated in every recursive call and each
151-- allocation holds on to the previous allocation. This test is to make sure
152-- that we do not use the constraint tuple type class.
154inspect $ hasNoTypeClassesExcept 'fromStreamVar
167-- | Generate a stream from an SVar. An unevaluated stream can be pushed to an
168-- SVar using 'toSVar'. As we pull a stream from the SVar the input stream
169-- gets evaluated concurrently. The evaluation depends on the SVar style and
170-- the configuration parameters e.g. using the maxBuffer/maxThreads
173{-# INLINE fromSVar #-}
174fromSVar :: (MonadAsync m, K.IsStream t) => SVar K.Stream m a -> t m a
176 K.mkStream $ \st yld sng stp -> do
177 ref <- liftIO $ newIORef ()
178 _ <- liftIO $ mkWeakIORef ref hook
179 -- We pass a copy of sv to fromStreamVar, so that we know that it has
180 -- no other references, when that copy gets garbage collected "ref"
181 -- will get garbage collected and our hook will be called.
182 K.foldStreamShared st yld sng stp $
183 K.fromStream $ fromStreamVar sv{svarRef = Just ref}
187 when (svarInspectMode sv) $ do
188 r <- liftIO $ readIORef (svarStopTime (svarStats sv))
190 printSVar sv "SVar Garbage Collected"
192 -- If there are any SVars referenced by this SVar a GC will prompt
193 -- them to be cleaned up quickly.
194 when (svarInspectMode sv) performMajorGC
196data FromSVarState t m a =
198 | FromSVarRead (SVar t m a)
199 | FromSVarLoop (SVar t m a) [ChildEvent a]
200 | FromSVarDone (SVar t m a)
202-- | Like 'fromSVar' but generates a StreamD style stream instead of CPS.
204{-# INLINE_NORMAL fromSVarD #-}
205fromSVarD :: (MonadAsync m) => SVar t m a -> D.Stream m a
206fromSVarD svar = D.Stream step FromSVarInit
209 {-# INLINE_LATE step #-}
210 step _ FromSVarInit = do
211 ref <- liftIO $ newIORef ()
212 _ <- liftIO $ mkWeakIORef ref hook
213 -- when this copy of svar gets garbage collected "ref" will get
214 -- garbage collected and our GC hook will be called.
215 let sv = svar{svarRef = Just ref}
216 return $ D.Skip (FromSVarRead sv)
220 {-# NOINLINE hook #-}
222 when (svarInspectMode svar) $ do
223 r <- liftIO $ readIORef (svarStopTime (svarStats svar))
225 printSVar svar "SVar Garbage Collected"
227 -- If there are any SVars referenced by this SVar a GC will prompt
228 -- them to be cleaned up quickly.
229 when (svarInspectMode svar) performMajorGC
231 step _ (FromSVarRead sv) = do
232 list <- readOutputQ sv
233 -- Reversing the output is important to guarantee that we process the
234 -- outputs in the same order as they were generated by the constituent
236 return $ D.Skip $ FromSVarLoop sv (Prelude.reverse list)
238 step _ (FromSVarLoop sv []) = do
239 done <- postProcess sv
240 return $ D.Skip $ if done
244 step _ (FromSVarLoop sv (ev : es)) = do
246 ChildYield a -> return $ D.Yield a (FromSVarLoop sv es)
247 ChildStop tid e -> do
251 stop <- shouldStop tid
254 liftIO (cleanupSVar sv)
255 return $ D.Skip (FromSVarDone sv)
256 else return $ D.Skip (FromSVarLoop sv es)
258 case fromException ex of
260 return $ D.Skip (FromSVarLoop sv es)
261 Nothing -> liftIO (cleanupSVar sv) >> throwM ex
265 case svarStopStyle sv of
266 StopNone -> return False
267 StopAny -> return True
269 sid <- liftIO $ readIORef (svarStopBy sv)
272 step _ (FromSVarDone sv) = do
273 when (svarInspectMode sv) $ do
274 t <- liftIO $ getTime Monotonic
275 liftIO $ writeIORef (svarStopTime (svarStats sv)) (Just t)
276 liftIO $ printSVar sv "SVar Done"
279------------------------------------------------------------------------------
280-- Writing streams to SVar
281------------------------------------------------------------------------------
283-- | Write a stream to an 'SVar' in a non-blocking manner. The stream is
284-- evaluated concurrently as it is read back from the SVar using 'fromSVar'.
286toSVar :: (K.IsStream t, MonadAsync m) => SVar K.Stream m a -> t m a -> m ()
287toSVar sv m = toStreamVar sv $ K.toStream m
289-------------------------------------------------------------------------------
290-- Concurrent application
291-------------------------------------------------------------------------------
293-- | A fold to write a stream to an SVar. Unlike 'toSVar' this does not allow
294-- for concurrent evaluation of the stream, as the fold receives the input one
295-- element at a time, it just forwards the elements to the SVar. However, we
296-- can safely execute the fold in an independent thread, the SVar can act as a
297-- buffer decoupling the sender from the receiver. Also, we can have multiple
298-- folds running concurrently pusing the streams to the SVar.
301write :: MonadIO m => SVar t m a -> Maybe WorkerInfo -> Fold m a ()
302write svar winfo = Fold step initial extract
306 initial = return $ FL.Partial ()
308 -- XXX we can have a separate fold for unlimited buffer case to avoid a
309 -- branch in the step here.
312 decrementBufferLimit svar
313 void $ send svar (ChildYield x)
314 return $ FL.Partial ()
316 extract () = liftIO $ sendStop svar winfo
318-- | Like write, but applies a yield limit.
320{-# INLINE writeLimited #-}
321writeLimited :: MonadIO m
322 => SVar t m a -> Maybe WorkerInfo -> Fold m a ()
323writeLimited svar winfo = Fold step initial extract
327 initial = return $ FL.Partial True
331 yieldLimitOk <- decrementYieldLimit svar
334 decrementBufferLimit svar
335 void $ send svar (ChildYield x)
336 return $ FL.Partial True
338 cleanupSVarFromWorker svar
341 step False _ = return $ FL.Done ()
343 extract True = liftIO $ sendStop svar winfo
344 extract False = return ()
346-- Using StreamD the worker stream producing code can fuse with the code to
347-- queue output to the SVar giving some perf boost.
349-- Note that StreamD can only be used in limited situations, specifically, we
350-- cannot implement joinStreamVarPar using this.
352-- XXX make sure that the SVar passed is a Parallel style SVar.
354-- | Fold the supplied stream to the SVar asynchronously using Parallel
356-- {-# INLINE_NORMAL toSVarParallel #-}
357{-# INLINE toSVarParallel #-}
358toSVarParallel :: MonadAsync m
359 => State t m a -> SVar t m a -> D.Stream m a -> m ()
360toSVarParallel st sv xs =
361 if svarInspectMode sv
365 case getYieldLimit st of
366 Nothing -> doFork (work Nothing)
368 (handleChildException sv)
369 Just _ -> doFork (workLim Nothing)
371 (handleChildException sv)
376 {-# NOINLINE work #-}
377 work info = D.fold (write sv info) xs
379 {-# NOINLINE workLim #-}
380 workLim info = D.fold (writeLimited sv info) xs
382 {-# NOINLINE forkWithDiag #-}
384 -- We do not use workerCount in case of ParallelVar but still there is
385 -- no harm in maintaining it correctly.
386 liftIO $ atomicModifyIORefCAS_ (workerCount sv) $ \n -> n + 1
388 -- This allocation matters when significant number of workers are being
389 -- sent. We allocate it only when needed. The overhead increases by 4x.
391 case yieldRateInfo sv of
392 Nothing -> return Nothing
393 Just _ -> liftIO $ do
395 t <- getTime Monotonic
396 lat <- newIORef (0, t)
397 return $ Just WorkerInfo
399 , workerYieldCount = cntRef
400 , workerLatencyStart = lat
403 case getYieldLimit st of
404 Nothing -> doFork (work winfo)
406 (handleChildException sv)
407 Just _ -> doFork (workLim winfo)
409 (handleChildException sv)
412-------------------------------------------------------------------------------
413-- Support for running folds concurrently
414-------------------------------------------------------------------------------
418-- To run folds concurrently, we need to decouple the fold execution from the
419-- stream production. We use the SVar to do that, we have a single worker
420-- pushing the stream elements to the SVar and on the consumer side a fold
421-- driver pulls the values and folds them.
425-- Fold worker <------SVar<------input stream
431-- We need a channel for pushing exceptions from the fold worker to the stream
432-- pusher. The stream may be pushed to multiple folds at the same time. For
433-- that we need one SVar per fold:
437-- Fold worker <------SVar<---
439-- Fold worker <------SVar<------input stream
441-- Fold worker <------SVar<---
445-- Unlike in case concurrent stream evaluation, the puller does not drive the
446-- scheduling and concurrent execution of the stream. The stream is simply
447-- pushed by the stream producer at its own rate. The fold worker just pulls it
450-- Note: If the stream pusher terminates due to an exception, we do not
451-- actively terminate the fold. It gets cleaned up by the GC.
453-------------------------------------------------------------------------------
454-- Process events received by a fold consumer from a stream producer
455-------------------------------------------------------------------------------
457-- | Pull a stream from an SVar to fold it. Like 'fromSVar' except that it does
458-- not drive the evaluation of the stream. It just pulls whatever is available
459-- on the SVar. Also, when the fold stops it sends a notification to the stream
460-- pusher/producer. No exceptions are expected to be propagated from the stream
461-- pusher to the fold puller.
463{-# NOINLINE fromProducer #-}
464fromProducer :: forall m a . MonadAsync m => SVar K.Stream m a -> K.Stream m a
465fromProducer sv = K.mkStream $ \st yld sng stp -> do
466 list <- readOutputQ sv
467 -- Reversing the output is important to guarantee that we process the
468 -- outputs in the same order as they were generated by the constituent
470 K.foldStream st yld sng stp $ processEvents $ reverse list
474 allDone :: m r -> m r
476 when (svarInspectMode sv) $ do
477 t <- liftIO $ getTime Monotonic
478 liftIO $ writeIORef (svarStopTime (svarStats sv)) (Just t)
479 liftIO $ printSVar sv "SVar Done"
480 sendStopToProducer sv
483 {-# INLINE processEvents #-}
484 processEvents :: [ChildEvent a] -> K.Stream m a
485 processEvents [] = K.mkStream $ \st yld sng stp -> do
486 K.foldStream st yld sng stp $ fromProducer sv
488 processEvents (ev : es) = K.mkStream $ \_ yld _ stp -> do
489 let rest = processEvents es
491 ChildYield a -> yld a rest
492 ChildStop tid e -> do
495 Nothing -> allDone stp
496 Just _ -> error "Bug: fromProducer: received exception"
498-- | Create a Fold style SVar that runs a supplied fold function as the
499-- consumer. Any elements sent to the SVar are consumed by the supplied fold
502{-# INLINE newFoldSVar #-}
503newFoldSVar :: (K.IsStream t, MonadAsync m)
504 => State K.Stream m a -> (t m a -> m b) -> m (SVar K.Stream m a)
505newFoldSVar stt f = do
506 -- Buffer size for the SVar is derived from the current state
507 sv <- newParallelVar StopAny (adaptState stt)
509 -- Add the producer thread-id to the SVar.
510 liftIO myThreadId >>= modifyThread sv
512 void $ doFork (void $ f $ K.fromStream $ fromProducer sv)
514 (handleFoldException sv)
517-- | Like 'fromProducer' but generates a StreamD style stream instead of
520{-# INLINE_NORMAL fromProducerD #-}
521fromProducerD :: (MonadAsync m) => SVar t m a -> D.Stream m a
522fromProducerD svar = D.Stream step (FromSVarRead svar)
525 {-# INLINE_LATE step #-}
526 step _ (FromSVarRead sv) = do
527 list <- readOutputQ sv
528 -- Reversing the output is important to guarantee that we process the
529 -- outputs in the same order as they were generated by the constituent
531 return $ D.Skip $ FromSVarLoop sv (Prelude.reverse list)
533 step _ (FromSVarLoop sv []) = return $ D.Skip $ FromSVarRead sv
534 step _ (FromSVarLoop sv (ev : es)) = do
536 ChildYield a -> return $ D.Yield a (FromSVarLoop sv es)
537 ChildStop tid e -> do
541 sendStopToProducer sv
542 return $ D.Skip (FromSVarDone sv)
543 Just _ -> error "Bug: fromProducer: received exception"
545 step _ (FromSVarDone sv) = do
546 when (svarInspectMode sv) $ do
547 t <- liftIO $ getTime Monotonic
548 liftIO $ writeIORef (svarStopTime (svarStats sv)) (Just t)
549 liftIO $ printSVar sv "SVar Done"
552 step _ FromSVarInit = undefined
554-- | Like 'newFoldSVar' except that it uses a 'Fold' instead of a fold
557{-# INLINE newFoldSVarF #-}
558newFoldSVarF :: MonadAsync m => State t m a -> Fold m a b -> m (SVar t m a)
559newFoldSVarF stt f = do
560 -- Buffer size for the SVar is derived from the current state
561 sv <- newParallelVar StopAny (adaptState stt)
562 -- Add the producer thread-id to the SVar.
563 liftIO myThreadId >>= modifyThread sv
564 void $ doFork (work sv) (svarMrun sv) (handleFoldException sv)
569 {-# NOINLINE work #-}
570 work sv = void $ D.fold f $ fromProducerD sv
572-------------------------------------------------------------------------------
573-- Process events received by the producer thread from the consumer side
574-------------------------------------------------------------------------------
576-- XXX currently only one event is sent by a fold consumer to the stream
577-- producer. But we can potentially have multiple events e.g. the fold step can
578-- generate exception more than once and the producer can ignore those
579-- exceptions or handle them and still keep driving the fold.
581-- | Poll for events sent by the fold consumer to the stream pusher. The fold
582-- consumer can send a "Stop" event or an exception. When a "Stop" is received
583-- this function returns 'True'. If an exception is recieved then it throws the
586{-# NOINLINE fromConsumer #-}
587fromConsumer :: MonadAsync m => SVar K.Stream m a -> m Bool
589 (list, _) <- liftIO $ readOutputQBasic (outputQueueFromConsumer sv)
590 -- Reversing the output is important to guarantee that we process the
591 -- outputs in the same order as they were generated by the constituent
593 processEvents $ reverse list
597 {-# INLINE processEvents #-}
598 processEvents [] = return False
599 processEvents (ev : _) = do
603 Nothing -> return True
605 ChildYield _ -> error "Bug: fromConsumer: invalid ChildYield event"
607-- | Push values from a stream to a fold worker via an SVar. Before pushing a
608-- value to the SVar it polls for events received from the fold consumer. If a
609-- stop event is received then it returns 'True' otherwise false. Propagates
610-- exceptions received from the fold consumer.
612{-# INLINE pushToFold #-}
613pushToFold :: MonadAsync m => SVar K.Stream m a -> a -> m Bool
615 -- Check for exceptions before decrement so that we do not
616 -- block forever if the child already exited with an exception.
618 -- We avoid a race between the consumer fold sending an event and we
619 -- blocking on decrementBufferLimit by waking up the producer thread in
620 -- sendToProducer before any event is sent by the fold to the producer
622 let qref = outputQueueFromConsumer sv
624 (_, n) <- liftIO $ readIORef qref
631 decrementBufferLimit sv
632 void $ send sv (ChildYield a)
635------------------------------------------------------------------------------
636-- Clone and distribute a stream in parallel
637------------------------------------------------------------------------------
639-- XXX this could be written in StreamD style for better efficiency with fusion.
641-- | Tap a stream and send the elements to the specified SVar in addition to
642-- yielding them again. The SVar runs a fold consumer. Elements are tapped and
643-- sent to the SVar until the fold finishes. Any exceptions from the fold
644-- evaluation are propagated in the current thread.
648-- ------input stream---------output stream----->
650-- exceptions | | input
658{-# INLINE teeToSVar #-}
659teeToSVar :: (K.IsStream t, MonadAsync m) =>
660 SVar K.Stream m a -> t m a -> t m a
661teeToSVar svr m = K.mkStream $ \st yld sng stp -> do
662 K.foldStreamShared st yld sng stp (go False m)
666 go False m0 = K.mkStream $ \st yld _ stp -> do
668 -- In general, a Stop event would come equipped with the result
669 -- of the fold. It is not used here but it would be useful in
670 -- applicative and distribute.
671 done <- fromConsumer svr
673 liftIO $ withDiagMVar svr "teeToSVar: waiting to drain"
674 $ takeMVar (outputDoorBellFromConsumer svr)
678 liftIO $ sendStop svr Nothing
679 -- drain/wait until a stop event arrives from the fold.
682 stop = stopFold >> stp
684 done <- pushToFold svr a
685 yld a (go done (K.nilM stopFold))
686 yieldk a r = pushToFold svr a >>= \done -> yld a (go done r)
687 in K.foldStreamShared st yieldk single stop m0
keysesc to close
- j k
- move down, up
- enter
- open
- esc
- back, or up a level
- 1 2 3 4
- readme, files, commits, refs
- /
- filter
- y
- copy the clone command
- c
- compare the chosen commit with another (or shift-click both)
- ?
- show or hide this list