@@ -18,26 +18,46 @@
20module Streamly.Internal.Data.Stream.SVar
20module Streamly.Internal.Data.Stream.SVar
22+ -- * Unfold streams from SVar
21− ( fromSVar
24+ fromSVar
22− , fromStreamVar
26+ -- , fromStreamVar
24− , fromConsumer
28+ -- * Fold streams to SVar
32+ -- * Concurrent folds
26 , pushToFold
35 , pushToFold
44+import Control.Concurrent (myThreadId, takeMVar)
30import Control.Exception (fromException)
45import Control.Exception (fromException)
31import Control.Monad (when, void)
46import Control.Monad (when, void)
32import Control.Monad.Catch (throwM)
47import Control.Monad.Catch (throwM)
33import Control.Monad.IO.Class (MonadIO(liftIO))
48import Control.Monad.IO.Class (MonadIO(liftIO))
34import Data.IORef (newIORef, readIORef, mkWeakIORef, writeIORef)
49import Data.IORef (newIORef, readIORef, mkWeakIORef, writeIORef)
35import Data.Maybe (isNothing)
50import Data.Maybe (isNothing)
51+import Streamly.Internal.Data.Atomics (atomicModifyIORefCAS_)
52+import Streamly.Internal.Data.Fold.Types (Fold(..))
36import Streamly.Internal.Data.Time.Clock (Clock(Monotonic), getTime)
53import Streamly.Internal.Data.Time.Clock (Clock(Monotonic), getTime)
37import System.Mem (performMajorGC)
54import System.Mem (performMajorGC)
56+import qualified Streamly.Internal.Data.Fold.Types as FL
57+import qualified Streamly.Internal.Data.Stream.StreamD.Type as D
58+import qualified Streamly.Internal.Data.Stream.StreamK.Type as K
39import Streamly.Internal.Data.SVar
60import Streamly.Internal.Data.SVar
40−import Streamly.Internal.Data.Stream.StreamK hiding (reverse)
42#if __GLASGOW_HASKELL__ < 810
62#if __GLASGOW_HASKELL__ < 810
43#ifdef INSPECTION
63#ifdef INSPECTION
@@ -49,15 +69,39 @@ import 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.
52-- | Pull a stream from an SVar.
96-- | Pull a stream from an SVar.
53{-# NOINLINE fromStreamVar #-}
97{-# NOINLINE fromStreamVar #-}
54−fromStreamVar :: MonadAsync m => SVar Stream m a -> Stream m a
98+fromStreamVar :: MonadAsync m => SVar K.Stream m a -> K.Stream m a
55−fromStreamVar sv = MkStream $ \st yld sng stp -> do
99+fromStreamVar sv = K.MkStream $ \st yld sng stp -> do
56 list <- readOutputQ sv
100 list <- readOutputQ sv
57 -- Reversing the output is important to guarantee that we process the
101 -- Reversing the output is important to guarantee that we process the
58 -- outputs in the same order as they were generated by the constituent
102 -- outputs in the same order as they were generated by the constituent
59 -- streams.
103 -- streams.
60− foldStream st yld sng stp $ processEvents $ reverse list
104+ K.foldStream st yld sng stp $ processEvents $ reverse list
@@ -69,13 +113,13 @@ fromStreamVar sv = MkStream $ \st yld sng stp -> do
71 {-# INLINE processEvents #-}
115 {-# INLINE processEvents #-}
72− processEvents [] = MkStream $ \st yld sng stp -> do
116+ processEvents [] = K.MkStream $ \st yld sng stp -> do
73 done <- postProcess sv
117 done <- postProcess sv
75 then allDone stp
119 then allDone stp
76− else foldStream st yld sng stp $ fromStreamVar sv
120+ else K.foldStream st yld sng stp $ fromStreamVar sv
78− processEvents (ev : es) = MkStream $ \st yld sng stp -> do
122+ processEvents (ev : es) = K.MkStream $ \st yld sng stp -> do
79 let rest = processEvents es
123 let rest = processEvents es
80 case ev of
124 case ev of
81 ChildYield a -> yld a rest
125 ChildYield a -> yld a rest
@@ -86,11 +130,11 @@ fromStreamVar sv = MkStream $ \st yld sng stp -> do
86 stop <- shouldStop tid
130 stop <- shouldStop tid
88 then liftIO (cleanupSVar sv) >> allDone stp
132 then liftIO (cleanupSVar sv) >> allDone stp
89− else foldStream st yld sng stp rest
133+ else K.foldStream st yld sng stp rest
90 Just ex ->
134 Just ex ->
91 case fromException ex of
135 case fromException ex of
92 Just ThreadAbort ->
136 Just ThreadAbort ->
93− foldStream st yld sng stp rest
137+ K.foldStream st yld sng stp rest
94 Nothing -> liftIO (cleanupSVar sv) >> throwM ex
138 Nothing -> liftIO (cleanupSVar sv) >> throwM ex
95 shouldStop tid =
139 shouldStop tid =
96 case svarStopStyle sv of
140 case svarStopStyle sv of
@@ -98,7 +142,7 @@ fromStreamVar sv = MkStream $ \st yld sng stp -> do
98 StopAny -> return True
142 StopAny -> return True
99 StopBy -> do
143 StopBy -> do
100 sid <- liftIO $ readIORef (svarStopBy sv)
144 sid <- liftIO $ readIORef (svarStopBy sv)
101− return $ if tid == sid then True else False
145+ return $ tid == sid
103#if __GLASGOW_HASKELL__ < 810
147#if __GLASGOW_HASKELL__ < 810
104#ifdef INSPECTION
148#ifdef INSPECTION
@@ -120,17 +164,23 @@ inspect $ 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
123{-# INLINE fromSVar #-}
173{-# INLINE fromSVar #-}
124−fromSVar :: (MonadAsync m, IsStream t) => SVar Stream m a -> t m a
174+fromSVar :: (MonadAsync m, K.IsStream t) => SVar K.Stream m a -> t m a
125fromSVar sv =
175fromSVar sv =
126− mkStream $ \st yld sng stp -> do
176+ K.mkStream $ \st yld sng stp -> do
127 ref <- liftIO $ newIORef ()
177 ref <- liftIO $ newIORef ()
128 _ <- liftIO $ mkWeakIORef ref hook
178 _ <- liftIO $ mkWeakIORef ref hook
129 -- We pass a copy of sv to fromStreamVar, so that we know that it has
179 -- We pass a copy of sv to fromStreamVar, so that we know that it has
130 -- no other references, when that copy gets garbage collected "ref"
180 -- no other references, when that copy gets garbage collected "ref"
131 -- will get garbage collected and our hook will be called.
181 -- will get garbage collected and our hook will be called.
132− foldStreamShared st yld sng stp $
182+ K.foldStreamShared st yld sng stp $
133− fromStream $ fromStreamVar sv{svarRef = Just ref}
183+ K.fromStream $ fromStreamVar sv{svarRef = Just ref}
136 hook = do
186 hook = do
@@ -143,24 +193,281 @@ fromSVar sv =
143 -- them to be cleaned up quickly.
193 -- them to be cleaned up quickly.
144 when (svarInspectMode sv) performMajorGC
194 when (svarInspectMode sv) performMajorGC
196+data 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 #-}
205+fromSVarD :: (MonadAsync m) => SVar t m a -> D.Stream m a
206+fromSVarD 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))
224+ when (isNothing r) $
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
241+ then FromSVarDone sv
242+ else FromSVarRead sv
244+ step _ (FromSVarLoop sv (ev : es)) = do
246+ ChildYield a -> return $ D.Yield a (FromSVarLoop sv es)
247+ ChildStop tid e -> do
248+ accountThread sv tid
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+------------------------------------------------------------------------------
146−-- | Write a stream to an 'SVar' in a non-blocking manner. The stream can then
283+-- | Write a stream to an 'SVar' in a non-blocking manner. The stream is
147−-- be read back from the SVar using 'fromSVar'.
284+-- evaluated concurrently as it is read back from the SVar using 'fromSVar'.
148−toSVar :: (IsStream t, MonadAsync m) => SVar Stream m a -> t m a -> m ()
286+toSVar :: (K.IsStream t, MonadAsync m) => SVar K.Stream m a -> t m a -> m ()
149−toSVar sv m = toStreamVar sv $ toStream m
287+toSVar 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.
301+write :: MonadIO m => SVar t m a -> Maybe WorkerInfo -> Fold m a ()
302+write 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 #-}
321+writeLimited :: MonadIO m
322+ => SVar t m a -> Maybe WorkerInfo -> Fold m a ()
323+writeLimited 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
355+-- concurrency style.
356+-- {-# INLINE_NORMAL toSVarParallel #-}
357+{-# INLINE toSVarParallel #-}
358+toSVarParallel :: MonadAsync m
359+ => State t m a -> SVar t m a -> D.Stream m a -> m ()
360+toSVarParallel 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.foldOnce (write sv info) xs
379+ {-# NOINLINE workLim #-}
380+ workLim info = D.foldOnce (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
394+ cntRef <- newIORef 0
395+ t <- getTime Monotonic
396+ lat <- newIORef (0, t)
397+ return $ Just WorkerInfo
398+ { workerYieldMax = 0
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.
151-------------------------------------------------------------------------------
453-------------------------------------------------------------------------------
152-- Process events received by a fold consumer from a stream producer
454-- Process events received by a fold consumer from a stream producer
153-------------------------------------------------------------------------------
455-------------------------------------------------------------------------------
155−-- | Pull a stream from an SVar.
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.
156{-# NOINLINE fromProducer #-}
463{-# NOINLINE fromProducer #-}
157−fromProducer :: forall m a . MonadAsync m => SVar Stream m a -> Stream m a
464+fromProducer :: forall m a . MonadAsync m => SVar K.Stream m a -> K.Stream m a
158−fromProducer sv = mkStream $ \st yld sng stp -> do
465+fromProducer sv = K.mkStream $ \st yld sng stp -> do
159 list <- readOutputQ sv
466 list <- readOutputQ sv
160 -- Reversing the output is important to guarantee that we process the
467 -- Reversing the output is important to guarantee that we process the
161 -- outputs in the same order as they were generated by the constituent
468 -- outputs in the same order as they were generated by the constituent
162 -- streams.
469 -- streams.
163− foldStream st yld sng stp $ processEvents $ reverse list
470+ K.foldStream st yld sng stp $ processEvents $ reverse list
@@ -174,11 +481,11 @@ fromProducer sv = mkStream $ \st yld sng stp -> do
176 {-# INLINE processEvents #-}
483 {-# INLINE processEvents #-}
177− processEvents :: [ChildEvent a] -> Stream m a
484+ processEvents :: [ChildEvent a] -> K.Stream m a
178− processEvents [] = mkStream $ \st yld sng stp -> do
485+ processEvents [] = K.mkStream $ \st yld sng stp -> do
179− foldStream st yld sng stp $ fromProducer sv
486+ K.foldStream st yld sng stp $ fromProducer sv
181− processEvents (ev : es) = mkStream $ \_ yld _ stp -> do
488+ processEvents (ev : es) = K.mkStream $ \_ yld _ stp -> do
182 let rest = processEvents es
489 let rest = processEvents es
183 case ev of
490 case ev of
184 ChildYield a -> yld a rest
491 ChildYield a -> yld a rest
@@ -188,6 +495,80 @@ fromProducer sv = mkStream $ \st yld sng stp -> do
188 Nothing -> allDone stp
495 Nothing -> allDone stp
189 Just _ -> error "Bug: fromProducer: received exception"
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 #-}
503+newFoldSVar :: (K.IsStream t, MonadAsync m)
504+ => State K.Stream m a -> (t m a -> m b) -> m (SVar K.Stream m a)
505+newFoldSVar 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 #-}
521+fromProducerD :: (MonadAsync m) => SVar t m a -> D.Stream m a
522+fromProducerD 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
538+ accountThread sv tid
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 #-}
558+newFoldSVarF :: MonadAsync m => State t m a -> Fold m a b -> m (SVar t m a)
559+newFoldSVarF 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.foldOnce f $ fromProducerD sv
191-------------------------------------------------------------------------------
572-------------------------------------------------------------------------------
192-- Process events received by the producer thread from the consumer side
573-- Process events received by the producer thread from the consumer side
193-------------------------------------------------------------------------------
574-------------------------------------------------------------------------------
@@ -197,8 +578,13 @@ fromProducer sv = mkStream $ \st yld sng stp -> do
197-- generate exception more than once and the producer can ignore those
578-- generate exception more than once and the producer can ignore those
198-- exceptions or handle them and still keep driving the fold.
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
200{-# NOINLINE fromConsumer #-}
586{-# NOINLINE fromConsumer #-}
201−fromConsumer :: MonadAsync m => SVar Stream m a -> m Bool
587+fromConsumer :: MonadAsync m => SVar K.Stream m a -> m Bool
202fromConsumer sv = do
588fromConsumer sv = do
203 (list, _) <- liftIO $ readOutputQBasic (outputQueueFromConsumer sv)
589 (list, _) <- liftIO $ readOutputQBasic (outputQueueFromConsumer sv)
204 -- Reversing the output is important to guarantee that we process the
590 -- Reversing the output is important to guarantee that we process the
@@ -218,9 +604,13 @@ fromConsumer sv = do
218 Just ex -> throwM ex
604 Just ex -> throwM ex
219 ChildYield _ -> error "Bug: fromConsumer: invalid ChildYield event"
605 ChildYield _ -> error "Bug: fromConsumer: invalid ChildYield event"
221−-- push values to a fold worker via an SVar. Returns whether the fold is done.
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.
222{-# INLINE pushToFold #-}
612{-# INLINE pushToFold #-}
223−pushToFold :: MonadAsync m => SVar Stream m a -> a -> m Bool
613+pushToFold :: MonadAsync m => SVar K.Stream m a -> a -> m Bool
224pushToFold sv a = do
614pushToFold sv a = do
225 -- Check for exceptions before decrement so that we do not
615 -- Check for exceptions before decrement so that we do not
226 -- block forever if the child already exited with an exception.
616 -- block forever if the child already exited with an exception.
@@ -241,3 +631,59 @@ pushToFold sv a = do
241 decrementBufferLimit sv
631 decrementBufferLimit sv
242 void $ send sv (ChildYield a)
632 void $ send sv (ChildYield a)
243 return False
633 return False
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 #-}
659+teeToSVar :: (K.IsStream t, MonadAsync m) =>
660+ SVar K.Stream m a -> t m a -> t m a
661+teeToSVar 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
672+ when (not done) $ do
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