@@ -18,26 +18,46 @@
1818--
1919--
2020module Streamly.Internal.Data.Stream.SVar
21− ( fromSVar
22− , fromStreamVar
23− , fromProducer
24− , fromConsumer
21+ (
22+ -- * Unfold streams from SVar
23+ -- $concurrentEval
24+ fromSVar
25+ , fromSVarD
26+ -- , fromStreamVar
27+
28+ -- * Fold streams to SVar
2529 , toSVar
30+ , toSVarParallel
31+
32+ -- * Concurrent folds
33+ -- $concurrentFolds
34+ , fromConsumer
2635 , pushToFold
36+ , teeToSVar
37+ , newFoldSVar
38+ , newFoldSVarF
2739 )
2840where
2941
42+#include "inline.hs"
43+
44+import Control.Concurrent (myThreadId, takeMVar)
3045import Control.Exception (fromException)
3146import Control.Monad (when, void)
3247import Control.Monad.Catch (throwM)
3348import Control.Monad.IO.Class (MonadIO(liftIO))
3449import Data.IORef (newIORef, readIORef, mkWeakIORef, writeIORef)
3550import Data.Maybe (isNothing)
51+import Streamly.Internal.Data.Atomics (atomicModifyIORefCAS_)
52+import Streamly.Internal.Data.Fold.Types (Fold(..))
3653import Streamly.Internal.Data.Time.Clock (Clock(Monotonic), getTime)
3754import System.Mem (performMajorGC)
3855
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
59+
3960import Streamly.Internal.Data.SVar
40−import Streamly.Internal.Data.Stream.StreamK hiding (reverse)
4161
4262#if __GLASGOW_HASKELL__ < 810
4363#ifdef INSPECTION
@@ -49,15 +69,39 @@ import Test.Inspection (inspect, hasNoTypeClassesExcept)
4969#endif
5070#endif
5171
72+------------------------------------------------------------------------------
73+-- Generating streams from SVar
74+------------------------------------------------------------------------------
75+
76+-- $concurrentEval
77+--
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.
81+--
82+-- @
83+-- input stream
84+-- |
85+-- <-----------------|<--------worker
86+-- | exceptions |
87+-- output stream <------SVar<------worker
88+-- |
89+-- |<--------worker
90+--
91+-- @
92+--
93+-- The puller itself schedules the worker threads based on demand.
94+-- Exceptions are propagated from the worker threads to the puller.
95+
5296-- | Pull a stream from an SVar.
5397{-# NOINLINE fromStreamVar #-}
54−fromStreamVar :: MonadAsync m => SVar Stream m a -> Stream m a
55−fromStreamVar sv = MkStream $ \st yld sng stp -> do
98+fromStreamVar :: MonadAsync m => SVar K.Stream m a -> K.Stream m a
99+fromStreamVar sv = K.MkStream $ \st yld sng stp -> do
56100 list <- readOutputQ sv
57101 -- Reversing the output is important to guarantee that we process the
58102 -- outputs in the same order as they were generated by the constituent
59103 -- streams.
60− foldStream st yld sng stp $ processEvents $ reverse list
104+ K.foldStream st yld sng stp $ processEvents $ reverse list
61105
62106 where
63107
@@ -69,13 +113,13 @@ fromStreamVar sv = MkStream $ \st yld sng stp -> do
69113 stp
70114
71115 {-# INLINE processEvents #-}
72− processEvents [] = MkStream $ \st yld sng stp -> do
116+ processEvents [] = K.MkStream $ \st yld sng stp -> do
73117 done <- postProcess sv
74118 if done
75119 then allDone stp
76− else foldStream st yld sng stp $ fromStreamVar sv
120+ else K.foldStream st yld sng stp $ fromStreamVar sv
77121
78− processEvents (ev : es) = MkStream $ \st yld sng stp -> do
122+ processEvents (ev : es) = K.MkStream $ \st yld sng stp -> do
79123 let rest = processEvents es
80124 case ev of
81125 ChildYield a -> yld a rest
@@ -86,11 +130,11 @@ fromStreamVar sv = MkStream $ \st yld sng stp -> do
86130 stop <- shouldStop tid
87131 if stop
88132 then liftIO (cleanupSVar sv) >> allDone stp
89− else foldStream st yld sng stp rest
133+ else K.foldStream st yld sng stp rest
90134 Just ex ->
91135 case fromException ex of
92136 Just ThreadAbort ->
93− foldStream st yld sng stp rest
137+ K.foldStream st yld sng stp rest
94138 Nothing -> liftIO (cleanupSVar sv) >> throwM ex
95139 shouldStop tid =
96140 case svarStopStyle sv of
@@ -98,7 +142,7 @@ fromStreamVar sv = MkStream $ \st yld sng stp -> do
98142 StopAny -> return True
99143 StopBy -> do
100144 sid <- liftIO $ readIORef (svarStopBy sv)
101− return $ if tid == sid then True else False
145+ return $ tid == sid
102146
103147#if __GLASGOW_HASKELL__ < 810
104148#ifdef INSPECTION
@@ -120,17 +164,23 @@ inspect $ hasNoTypeClassesExcept 'fromStreamVar
120164#endif
121165#endif
122166
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
171+-- combinators.
172+--
123173{-# 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
125175fromSVar sv =
126− mkStream $ \st yld sng stp -> do
176+ K.mkStream $ \st yld sng stp -> do
127177 ref <- liftIO $ newIORef ()
128178 _ <- liftIO $ mkWeakIORef ref hook
129179 -- We pass a copy of sv to fromStreamVar, so that we know that it has
130180 -- no other references, when that copy gets garbage collected "ref"
131181 -- will get garbage collected and our hook will be called.
132− foldStreamShared st yld sng stp $
133− fromStream $ fromStreamVar sv{svarRef = Just ref}
182+ K.foldStreamShared st yld sng stp $
183+ K.fromStream $ fromStreamVar sv{svarRef = Just ref}
134184 where
135185
136186 hook = do
@@ -143,24 +193,281 @@ fromSVar sv =
143193 -- them to be cleaned up quickly.
144194 when (svarInspectMode sv) performMajorGC
145195
146−-- | Write a stream to an 'SVar' in a non-blocking manner. The stream can then
147−-- be read back from the SVar using 'fromSVar'.
148−toSVar :: (IsStream t, MonadAsync m) => SVar Stream m a -> t m a -> m ()
149−toSVar sv m = toStreamVar sv $ toStream m
196+data FromSVarState t m a =
197+ FromSVarInit
198+ | FromSVarRead (SVar t m a)
199+ | FromSVarLoop (SVar t m a) [ChildEvent a]
200+ | FromSVarDone (SVar t m a)
201+
202+-- | Like 'fromSVar' but generates a StreamD style stream instead of CPS.
203+--
204+{-# INLINE_NORMAL fromSVarD #-}
205+fromSVarD :: (MonadAsync m) => SVar t m a -> D.Stream m a
206+fromSVarD svar = D.Stream step FromSVarInit
207+ where
208+
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)
217+
218+ where
219+
220+ {-# NOINLINE hook #-}
221+ hook = do
222+ when (svarInspectMode svar) $ do
223+ r <- liftIO $ readIORef (svarStopTime (svarStats svar))
224+ when (isNothing r) $
225+ printSVar svar "SVar Garbage Collected"
226+ cleanupSVar svar
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
230+
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
235+ -- streams.
236+ return $ D.Skip $ FromSVarLoop sv (Prelude.reverse list)
237+
238+ step _ (FromSVarLoop sv []) = do
239+ done <- postProcess sv
240+ return $ D.Skip $ if done
241+ then FromSVarDone sv
242+ else FromSVarRead sv
243+
244+ step _ (FromSVarLoop sv (ev : es)) = do
245+ case ev of
246+ ChildYield a -> return $ D.Yield a (FromSVarLoop sv es)
247+ ChildStop tid e -> do
248+ accountThread sv tid
249+ case e of
250+ Nothing -> do
251+ stop <- shouldStop tid
252+ if stop
253+ then do
254+ liftIO (cleanupSVar sv)
255+ return $ D.Skip (FromSVarDone sv)
256+ else return $ D.Skip (FromSVarLoop sv es)
257+ Just ex ->
258+ case fromException ex of
259+ Just ThreadAbort ->
260+ return $ D.Skip (FromSVarLoop sv es)
261+ Nothing -> liftIO (cleanupSVar sv) >> throwM ex
262+ where
263+
264+ shouldStop tid =
265+ case svarStopStyle sv of
266+ StopNone -> return False
267+ StopAny -> return True
268+ StopBy -> do
269+ sid <- liftIO $ readIORef (svarStopBy sv)
270+ return $ tid == sid
271+
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"
277+ return D.Stop
278+
279+------------------------------------------------------------------------------
280+-- Writing streams to SVar
281+------------------------------------------------------------------------------
282+
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'.
285+--
286+toSVar :: (K.IsStream t, MonadAsync m) => SVar K.Stream m a -> t m a -> m ()
287+toSVar sv m = toStreamVar sv $ K.toStream m
288+
289+-------------------------------------------------------------------------------
290+-- Concurrent application
291+-------------------------------------------------------------------------------
292+
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.
299+--
300+{-# INLINE write #-}
301+write :: MonadIO m => SVar t m a -> Maybe WorkerInfo -> Fold m a ()
302+write svar winfo = Fold step initial extract
303+
304+ where
305+
306+ initial = return $ FL.Partial ()
307+
308+ -- XXX we can have a separate fold for unlimited buffer case to avoid a
309+ -- branch in the step here.
310+ step () x =
311+ liftIO $ do
312+ decrementBufferLimit svar
313+ void $ send svar (ChildYield x)
314+ return $ FL.Partial ()
315+
316+ extract () = liftIO $ sendStop svar winfo
317+
318+-- | Like write, but applies a yield limit.
319+--
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
324+
325+ where
326+
327+ initial = return $ FL.Partial True
328+
329+ step True x =
330+ liftIO $ do
331+ yieldLimitOk <- decrementYieldLimit svar
332+ if yieldLimitOk
333+ then do
334+ decrementBufferLimit svar
335+ void $ send svar (ChildYield x)
336+ return $ FL.Partial True
337+ else do
338+ cleanupSVarFromWorker svar
339+ sendStop svar winfo
340+ return $ FL.Done ()
341+ step False _ = return $ FL.Done ()
342+
343+ extract True = liftIO $ sendStop svar winfo
344+ extract False = return ()
345+
346+-- Using StreamD the worker stream producing code can fuse with the code to
347+-- queue output to the SVar giving some perf boost.
348+--
349+-- Note that StreamD can only be used in limited situations, specifically, we
350+-- cannot implement joinStreamVarPar using this.
351+--
352+-- XXX make sure that the SVar passed is a Parallel style SVar.
353+
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
362+ then forkWithDiag
363+ else do
364+ tid <-
365+ case getYieldLimit st of
366+ Nothing -> doFork (work Nothing)
367+ (svarMrun sv)
368+ (handleChildException sv)
369+ Just _ -> doFork (workLim Nothing)
370+ (svarMrun sv)
371+ (handleChildException sv)
372+ modifyThread sv tid
373+
374+ where
375+
376+ {-# NOINLINE work #-}
377+ work info = D.foldOnce (write sv info) xs
378+
379+ {-# NOINLINE workLim #-}
380+ workLim info = D.foldOnce (writeLimited sv info) xs
381+
382+ {-# NOINLINE forkWithDiag #-}
383+ forkWithDiag = do
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
387+ recordMaxWorkers sv
388+ -- This allocation matters when significant number of workers are being
389+ -- sent. We allocate it only when needed. The overhead increases by 4x.
390+ winfo <-
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
401+ }
402+ tid <-
403+ case getYieldLimit st of
404+ Nothing -> doFork (work winfo)
405+ (svarMrun sv)
406+ (handleChildException sv)
407+ Just _ -> doFork (workLim winfo)
408+ (svarMrun sv)
409+ (handleChildException sv)
410+ modifyThread sv tid
411+
412+-------------------------------------------------------------------------------
413+-- Support for running folds concurrently
414+-------------------------------------------------------------------------------
415+
416+-- $concurrentFolds
417+--
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.
422+--
423+-- @
424+--
425+-- Fold worker <------SVar<------input stream
426+-- | exceptions |
427+-- --------------->
428+--
429+-- @
430+--
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:
434+--
435+-- @
436+--
437+-- Fold worker <------SVar<---
438+-- | |
439+-- Fold worker <------SVar<------input stream
440+-- | |
441+-- Fold worker <------SVar<---
442+--
443+-- @
444+--
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
448+-- and folds it.
449+--
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.
150452
151453-------------------------------------------------------------------------------
152454-- Process events received by a fold consumer from a stream producer
153455-------------------------------------------------------------------------------
154456
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.
462+--
156463{-# NOINLINE fromProducer #-}
157−fromProducer :: forall m a . MonadAsync m => SVar Stream m a -> Stream m a
158−fromProducer sv = mkStream $ \st yld sng stp -> do
464+fromProducer :: forall m a . MonadAsync m => SVar K.Stream m a -> K.Stream m a
465+fromProducer sv = K.mkStream $ \st yld sng stp -> do
159466 list <- readOutputQ sv
160467 -- Reversing the output is important to guarantee that we process the
161468 -- outputs in the same order as they were generated by the constituent
162469 -- streams.
163− foldStream st yld sng stp $ processEvents $ reverse list
470+ K.foldStream st yld sng stp $ processEvents $ reverse list
164471
165472 where
166473
@@ -174,11 +481,11 @@ fromProducer sv = mkStream $ \st yld sng stp -> do
174481 stp
175482
176483 {-# INLINE processEvents #-}
177− processEvents :: [ChildEvent a] -> Stream m a
178− processEvents [] = mkStream $ \st yld sng stp -> do
179− foldStream st yld sng stp $ fromProducer sv
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
180487
181− processEvents (ev : es) = mkStream $ \_ yld _ stp -> do
488+ processEvents (ev : es) = K.mkStream $ \_ yld _ stp -> do
182489 let rest = processEvents es
183490 case ev of
184491 ChildYield a -> yld a rest
@@ -188,6 +495,80 @@ fromProducer sv = mkStream $ \st yld sng stp -> do
188495 Nothing -> allDone stp
189496 Just _ -> error "Bug: fromProducer: received exception"
190497
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
500+-- function.
501+--
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)
508+
509+ -- Add the producer thread-id to the SVar.
510+ liftIO myThreadId >>= modifyThread sv
511+
512+ void $ doFork (void $ f $ K.fromStream $ fromProducer sv)
513+ (svarMrun sv)
514+ (handleFoldException sv)
515+ return sv
516+
517+-- | Like 'fromProducer' but generates a StreamD style stream instead of
518+-- StreamK.
519+--
520+{-# INLINE_NORMAL fromProducerD #-}
521+fromProducerD :: (MonadAsync m) => SVar t m a -> D.Stream m a
522+fromProducerD svar = D.Stream step (FromSVarRead svar)
523+ where
524+
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
530+ -- streams.
531+ return $ D.Skip $ FromSVarLoop sv (Prelude.reverse list)
532+
533+ step _ (FromSVarLoop sv []) = return $ D.Skip $ FromSVarRead sv
534+ step _ (FromSVarLoop sv (ev : es)) = do
535+ case ev of
536+ ChildYield a -> return $ D.Yield a (FromSVarLoop sv es)
537+ ChildStop tid e -> do
538+ accountThread sv tid
539+ case e of
540+ Nothing -> do
541+ sendStopToProducer sv
542+ return $ D.Skip (FromSVarDone sv)
543+ Just _ -> error "Bug: fromProducer: received exception"
544+
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"
550+ return D.Stop
551+
552+ step _ FromSVarInit = undefined
553+
554+-- | Like 'newFoldSVar' except that it uses a 'Fold' instead of a fold
555+-- function.
556+--
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)
565+ return sv
566+
567+ where
568+
569+ {-# NOINLINE work #-}
570+ work sv = void $ D.foldOnce f $ fromProducerD sv
571+
191572-------------------------------------------------------------------------------
192573-- Process events received by the producer thread from the consumer side
193574-------------------------------------------------------------------------------
@@ -197,8 +578,13 @@ fromProducer sv = mkStream $ \st yld sng stp -> do
197578-- generate exception more than once and the producer can ignore those
198579-- exceptions or handle them and still keep driving the fold.
199580--
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
584+-- exception.
585+--
200586{-# NOINLINE fromConsumer #-}
201−fromConsumer :: MonadAsync m => SVar Stream m a -> m Bool
587+fromConsumer :: MonadAsync m => SVar K.Stream m a -> m Bool
202588fromConsumer sv = do
203589 (list, _) <- liftIO $ readOutputQBasic (outputQueueFromConsumer sv)
204590 -- Reversing the output is important to guarantee that we process the
@@ -218,9 +604,13 @@ fromConsumer sv = do
218604 Just ex -> throwM ex
219605 ChildYield _ -> error "Bug: fromConsumer: invalid ChildYield event"
220606
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.
611+--
222612{-# 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
224614pushToFold sv a = do
225615 -- Check for exceptions before decrement so that we do not
226616 -- block forever if the child already exited with an exception.
@@ -241,3 +631,59 @@ pushToFold sv a = do
241631 decrementBufferLimit sv
242632 void $ send sv (ChildYield a)
243633 return False
634+
635+------------------------------------------------------------------------------
636+-- Clone and distribute a stream in parallel
637+------------------------------------------------------------------------------
638+
639+-- XXX this could be written in StreamD style for better efficiency with fusion.
640+--
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.
645+--
646+-- @
647+--
648+-- ------input stream---------output stream----->
649+-- /|\\ |
650+-- exceptions | | input
651+-- | \\|/
652+-- ----SVar
653+-- |
654+-- Fold
655+--
656+-- @
657+--
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)
663+
664+ where
665+
666+ go False m0 = K.mkStream $ \st yld _ stp -> do
667+ let drain = 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)
675+ drain
676+
677+ stopFold = do
678+ liftIO $ sendStop svr Nothing
679+ -- drain/wait until a stop event arrives from the fold.
680+ drain
681+
682+ stop = stopFold >> stp
683+ single a = do
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
688+
689+ go True m0 = m0