@@ -21,10 +21,8 @@
2121--
2222module Streamly.Core
2323 (
24− MonadAsync
25−
2624 -- * Streams
27− , Stream (..)
25+ Stream (..)
2826
2927 -- * Construction (pure)
3028 , nil
@@ -57,46 +55,30 @@ module Streamly.Core
5755 , zipAsyncWith
5856
5957 -- * Concurrent Stream Vars (SVars)
60− , SVar
61− , newStreamVarPar
6258 , fromStreamVar
63− , toStreamVar
59+ , mkParallel
60+
6461 )
6562where
6663
67−import Control.Concurrent (ThreadId, myThreadId,
68− threadDelay, getNumCapabilities)
69−import Control.Concurrent.MVar (MVar, newEmptyMVar,
70− tryPutMVar, takeMVar)
71−import Control.Exception (SomeException (..), catch, mask)
72−import Control.Monad (when)
73−import Control.Monad.Catch (MonadThrow, throwM)
74−import Control.Monad.IO.Class (MonadIO(..))
75−import Control.Monad.Trans.Class (MonadTrans (lift))
76−import Control.Monad.Trans.Control (MonadBaseControl, control)
77−import Data.Atomics (casIORef, readForCAS, peekTicket
78− ,atomicModifyIORefCAS_
79− ,writeBarrier,storeLoadBarrier)
80−import Data.Concurrent.Queue.MichaelScott (LinkedQueue, newQ, pushL,
81− tryPopR, nullQ)
82−import Data.Functor (void)
83−import Data.Heap (Heap, Entry(..))
84−import qualified Data.Heap as H
85−import Data.IORef (IORef, modifyIORef, newIORef,
86− readIORef, atomicModifyIORef
87−#ifdef DIAGNOSTICS
88− , writeIORef
89−#endif
90− )
91−import Data.Maybe (fromJust)
92−import Data.Semigroup (Semigroup(..))
93−import Data.Set (Set)
94−import qualified Data.Set as S
95−import Prelude hiding (repeat, zipWith)
96−
97−import GHC.Exts
98−import GHC.Conc (ThreadId(..))
64+import Control.Concurrent.MVar (newEmptyMVar)
65+import Control.Monad.Catch (throwM)
66+import Control.Monad.IO.Class (MonadIO(..))
67+import Control.Monad.Trans.Class (MonadTrans(lift))
68+import Data.Atomics (atomicModifyIORefCAS_)
69+import Data.Concurrent.Queue.MichaelScott
70+ (LinkedQueue, newQ, nullQ)
71+import Data.Functor (void)
72+import Data.Heap (Heap, Entry(..))
73+import Data.IORef (IORef, newIORef, readIORef)
74+import Data.Maybe (fromJust)
75+import Data.Semigroup (Semigroup(..))
9976import GHC.IO (IO(..))
77+import Prelude hiding (repeat, zipWith)
78+
79+import Streamly.SVar
80+import qualified Data.Set as S
81+import qualified Data.Heap as H
10082
10183-- MVar diagnostics has some overhead - around 5% on asyncly null benchmark, we
10284-- can keep it on in production to debug problems quickly if and when they
@@ -104,186 +86,10 @@ import GHC.IO (IO(..))
10486-- until they are GCed because the consumer went away.
10587
10688#ifdef DIAGNOSTICS
107−import Control.Concurrent.MVar (tryTakeMVar)
108−import Control.Exception (catches, throwIO, Handler(..),
109− BlockedIndefinitelyOnMVar(..),
110− BlockedIndefinitelyOnSTM(..))
111−import System.IO (hPutStrLn, stderr)
112−#endif
113−
114−------------------------------------------------------------------------------
115−-- Parent child thread communication type
116−------------------------------------------------------------------------------
117−
118−-- | Events that a child thread may send to a parent thread.
119−data ChildEvent a =
120− ChildYield a
121− | ChildStop ThreadId (Maybe SomeException)
122−
123−-- | Sorting out-of-turn outputs in a heap for Ahead style streams
124−data AheadHeapEntry m a =
125− AheadEntryPure a
126− | AheadEntryStream (Stream m a)
127−
128−------------------------------------------------------------------------------
129−-- State threaded around the monad for thread management
130−------------------------------------------------------------------------------
131−
132−-- | Identify the type of the SVar. Two computations using the same style can
133−-- be scheduled on the same SVar.
134−data SVarStyle =
135− AsyncVar -- depth first concurrent
136− | WAsyncVar -- breadth first concurrent
137− | ParallelVar -- all parallel
138− | AheadVar -- Concurrent look ahead
139− deriving (Eq, Show)
140−
141−-- | An SVar or a Stream Var is a conduit to the output from multiple streams
142−-- running concurrently and asynchronously. An SVar can be thought of as an
143−-- asynchronous IO handle. We can write any number of streams to an SVar in a
144−-- non-blocking manner and then read them back at any time at any pace. The
145−-- SVar would run the streams asynchronously and accumulate results. An SVar
146−-- may not really execute the stream completely and accumulate all the results.
147−-- However, it ensures that the reader can read the results at whatever paces
148−-- it wants to read. The SVar monitors and adapts to the consumer's pace.
149−--
150−-- An SVar is a mini scheduler, it has an associated workLoop that holds the
151−-- stream tasks to be picked and run by a pool of worker threads. It has an
152−-- associated output queue where the output stream elements are placed by the
153−-- worker threads. A outputDoorBell is used by the worker threads to intimate the
154−-- consumer thread about availability of new results in the output queue. More
155−-- workers are added to the SVar by 'fromStreamVar' on demand if the output
156−-- produced is not keeping pace with the consumer. On bounded SVars, workers
157−-- block on the output queue to provide throttling of the producer when the
158−-- consumer is not pulling fast enough. The number of workers may even get
159−-- reduced depending on the consuming pace.
160−--
161−-- New work is enqueued either at the time of creation of the SVar or as a
162−-- result of executing the parallel combinators i.e. '<|' and '<|>' when the
163−-- already enqueued computations get evaluated. See 'joinStreamVarAsync'.
164−--
165−data SVar m a =
166− SVar {
167− -- Read only state
168− svarStyle :: SVarStyle
169−
170− -- Shared output queue (events, length)
171− , outputQueue :: IORef ([ChildEvent a], Int)
172− , outputDoorBell :: MVar () -- signal the consumer about output
173− , readOutputQ :: m [ChildEvent a]
174− , postProcess :: m Bool
175−
176− -- Used only by bounded SVar types
177− , enqueue :: Stream m a -> IO ()
178− , isWorkDone :: IO Bool
179− , needDoorBell :: IORef Bool
180− , workLoop :: m ()
181−
182− -- Shared, thread tracking
183− , workerThreads :: IORef (Set ThreadId)
184− , workerCount :: IORef Int
185− , accountThread :: ThreadId -> m ()
186−#ifdef DIAGNOSTICS
187− , outputHeap :: IORef (Heap (Entry Int (AheadHeapEntry m a))
188− , Int
189− )
190− -- Shared work queue (stream, seqNo)
191− , aheadWorkQueue :: IORef ([Stream m a], Int)
192− , totalDispatches :: IORef Int
193− , maxWorkers :: IORef Int
194− , maxOutQSize :: IORef Int
195− , maxHeapSize :: IORef Int
196− , maxWorkQSize :: IORef Int
197−#endif
198− }
199−
200−#ifdef DIAGNOSTICS
201−{-# NOINLINE dumpSVar #-}
202−dumpSVar :: SVar m a -> IO String
203−dumpSVar sv = do
204− tid <- myThreadId
205− (oqList, oqLen) <- readIORef $ outputQueue sv
206− db <- tryTakeMVar $ outputDoorBell sv
207− aheadDump <-
208− if svarStyle sv == AheadVar
209− then do
210− (oheap, oheapSeq) <- readIORef $ outputHeap sv
211− (wq, wqSeq) <- readIORef $ aheadWorkQueue sv
212− maxHp <- readIORef $ maxHeapSize sv
213− return $ unlines
214− [ "heap length = " ++ show (H.size oheap)
215− , "heap seqeunce = " ++ show oheapSeq
216− , "work queue length = " ++ show (length wq)
217− , "work queue sequence = " ++ show wqSeq
218− , "heap max size = " ++ show maxHp
219− ]
220− else return []
221−
222− waiting <- readIORef $ needDoorBell sv
223− rthread <- readIORef $ workerThreads sv
224− workers <- readIORef $ workerCount sv
225− maxWrk <- readIORef $ maxWorkers sv
226− dispatches <- readIORef $ totalDispatches sv
227− maxOq <- readIORef $ maxOutQSize sv
228−
229− return $ unlines
230− [ "tid = " ++ show tid
231− , "style = " ++ show (svarStyle sv)
232− , "outputQueue length computed = " ++ show (length oqList)
233− , "outputQueue length maintained = " ++ show oqLen
234− , "output outputDoorBell = " ++ show db
235− , "total dispatches = " ++ show dispatches
236− , "max workers = " ++ show maxWrk
237− , "max outQSize = " ++ show maxOq
238− ]
239− ++ aheadDump ++ unlines
240− [ "needDoorBell = " ++ show waiting
241− , "running threads = " ++ show rthread
242− , "running thread count = " ++ show workers
243− ]
244−
245−{-# NOINLINE mvarExcHandler #-}
246−mvarExcHandler :: SVar m a -> String -> BlockedIndefinitelyOnMVar -> IO ()
247−mvarExcHandler sv label e@BlockedIndefinitelyOnMVar = do
248− svInfo <- dumpSVar sv
249− hPutStrLn stderr $ label ++ " " ++ "BlockedIndefinitelyOnMVar\n" ++ svInfo
250− throwIO e
251−
252−{-# NOINLINE stmExcHandler #-}
253−stmExcHandler :: SVar m a -> String -> BlockedIndefinitelyOnSTM -> IO ()
254−stmExcHandler sv label e@BlockedIndefinitelyOnSTM = do
255− svInfo <- dumpSVar sv
256− hPutStrLn stderr $ label ++ " " ++ "BlockedIndefinitelyOnSTM\n" ++ svInfo
257− throwIO e
258−
259−withDBGMVar :: SVar m a -> String -> IO () -> IO ()
260−withDBGMVar sv label action =
261− action `catches` [ Handler (mvarExcHandler sv label)
262− , Handler (stmExcHandler sv label)
263− ]
264−#else
265−withDBGMVar :: SVar m a -> String -> IO () -> IO ()
266−withDBGMVar _ _ action = action
89+import Control.Monad (when)
90+import Data.IORef (writeIORef)
26791#endif
26892
269−-- Slightly faster version of CAS. Gained some improvement by avoiding the use
270−-- of "evaluate" because we know we do not have exceptions in fn.
271−{-# INLINE atomicModifyIORefCAS #-}
272−atomicModifyIORefCAS :: IORef a -> (a -> (a,b)) -> IO b
273−atomicModifyIORefCAS ref fn = do
274− tkt <- readForCAS ref
275− loop tkt retries
276−
277− where
278−
279− retries = 25 :: Int
280− loop _ 0 = atomicModifyIORef ref fn
281− loop old tries = do
282− let (new, result) = fn $ peekTicket old
283− (success, tkt) <- casIORef ref old new
284− if success
285− then return result
286− else loop tkt (tries - 1)
28793
28894------------------------------------------------------------------------------
28995-- The stream type
@@ -307,7 +113,7 @@ atomicModifyIORefCAS ref fn = do
307113newtype Stream m a =
308114 Stream {
309115 runStream :: forall r.
310− Maybe (SVar m a) -- local state
116+ Maybe (SVar Stream m a) -- local state
311117 -> m r -- stop
312118 -> (a -> m r) -- singleton
313119 -> (a -> Stream m a -> m r) -- yield
@@ -375,99 +181,13 @@ wSerial m1 m2 = Stream $ \_ stp sng yld -> do
375181 yield a r = yld a (wSerial m2 r)
376182 (runStream m1) Nothing stop single yield
377183
378−------------------------------------------------------------------------------
379−-- Spawning threads and collecting result in streamed fashion
380−------------------------------------------------------------------------------
381−
382−-- | A monad that can perform concurrent or parallel IO operations. Streams
383−-- that can be composed concurrently require the underlying monad to be
384−-- 'MonadAsync'.
385−--
386−-- @since 0.1.0
387−type MonadAsync m = (MonadIO m, MonadBaseControl IO m, MonadThrow m)
388−
389−-- Stolen from the async package. The perf improvement is modest, 2% on a
390−-- thread heavy benchmark (parallel composition using noop computations).
391−-- A version of forkIO that does not include the outer exception
392−-- handler: saves a bit of time when we will be installing our own
393−-- exception handler.
394−{-# INLINE rawForkIO #-}
395−rawForkIO :: IO () -> IO ThreadId
396−rawForkIO action = IO $ \ s ->
397− case (fork# action s) of (# s1, tid #) -> (# s1, ThreadId tid #)
398−
399−{-# INLINE doFork #-}
400−doFork :: MonadBaseControl IO m
401− => m ()
402− -> (SomeException -> IO ())
403− -> m ThreadId
404−doFork action exHandler =
405− control $ \runInIO ->
406− mask $ \restore -> do
407− tid <- rawForkIO $ catch (restore $ void $ runInIO action)
408− exHandler
409− runInIO (return tid)
410−
411−-- XXX exception safety of all atomic/MVar operations
412−
413−-- TBD Each worker can have their own queue and the consumer can empty one
414−-- queue at a time, that way contention can be reduced.
415−
416−maxOutputQLen :: Int
417−maxOutputQLen = 1500
418−
419−-- | This function is used by the producer threads to queue output for the
420−-- consumer thread to consume. Returns whether the queue has more space.
421−{-# NOINLINE send #-}
422−send :: SVar m a -> ChildEvent a -> IO Bool
423−send sv msg = do
424− len <- atomicModifyIORefCAS (outputQueue sv) $ \(es, n) ->
425− ((msg : es, n + 1), n)
426− when (len <= 0) $ do
427− -- The wake up must happen only after the store has finished otherwise
428− -- we can have lost wakeup problems.
429− writeBarrier
430− -- Since multiple workers can try this at the same time, it is possible
431− -- that we may put a spurious MVar after the consumer has already seen
432− -- the output. But that's harmless, at worst it may cause the consumer
433− -- to read the queue again and find it empty.
434− -- The important point is that the consumer is guaranteed to receive a
435− -- doorbell if something was added to the queue after it empties it.
436− void $ tryPutMVar (outputDoorBell sv) ()
437− return (len < maxOutputQLen)
438−
439−{-# NOINLINE sendStop #-}
440−sendStop :: SVar m a -> IO ()
441−sendStop sv = do
442− liftIO $ atomicModifyIORefCAS_ (workerCount sv) $ \n -> n - 1
443− myThreadId >>= \tid -> void $ send sv (ChildStop tid Nothing)
444−
445184-------------------------------------------------------------------------------
446185-- Async
447186-------------------------------------------------------------------------------
448187
449−-- Note: For purely right associated expressions this queue should have at most
450−-- one element. It grows to more than one when we have left associcated
451−-- expressions. Large left associated compositions can grow this to a
452−-- large size
453−{-# INLINE enqueueLIFO #-}
454−enqueueLIFO :: SVar m a -> IORef [Stream m a] -> Stream m a -> IO ()
455−enqueueLIFO sv q m = do
456− atomicModifyIORefCAS_ q $ \ms -> m : ms
457− storeLoadBarrier
458− w <- readIORef $ needDoorBell sv
459− when w $ do
460− -- Note: the sequence of operations is important for correctness here.
461− -- We need to set the flag to false strictly before sending the
462− -- outputDoorBell, otherwise the outputDoorBell may get processed too early and
463− -- then we may set the flag to False to later making the consumer lose
464− -- the flag, even without receiving a outputDoorBell.
465− atomicModifyIORefCAS_ (needDoorBell sv) (const False)
466− void $ tryPutMVar (outputDoorBell sv) ()
467−
468188{-# INLINE runStreamLIFO #-}
469189runStreamLIFO :: MonadIO m
470− => SVar m a -> IORef [Stream m a] -> Stream m a -> m () -> m ()
190+ => SVar Stream m a -> IORef [Stream m a] -> Stream m a -> m () -> m ()
471191runStreamLIFO sv q m stop = runStream m (Just sv) stop single yield
472192 where
473193 single a = do
@@ -479,47 +199,13 @@ runStreamLIFO sv q m stop = runStream m (Just sv) stop single yield
479199 then (runStream r) (Just sv) stop single yield
480200 else liftIO $ enqueueLIFO sv q r >> sendStop sv
481201
482−workLoopLIFO :: MonadIO m => SVar m a -> IORef [Stream m a] -> m ()
483−workLoopLIFO sv q = run
484−
485− where
486−
487− run = do
488− work <- dequeue
489− case work of
490− Nothing -> liftIO $ sendStop sv
491− Just m -> runStreamLIFO sv q m run
492−
493− dequeue = liftIO $ atomicModifyIORefCAS q $ \case
494− [] -> ([], Nothing)
495− x : xs -> (xs, Just x)
496−
497202-------------------------------------------------------------------------------
498203-- WAsync
499204-------------------------------------------------------------------------------
500205
501−-- XXX we can use the Ahead style sequence/heap mechanism to make the best
502−-- effort to always try to finish the streams on the left side of an expression
503−-- first as long as possible.
504−
505−{-# INLINE enqueueFIFO #-}
506−enqueueFIFO :: SVar m a -> LinkedQueue (Stream m a) -> Stream m a -> IO ()
507−enqueueFIFO sv q m = do
508− pushL q m
509− storeLoadBarrier
510− w <- readIORef $ needDoorBell sv
511− when w $ do
512− -- Note: the sequence of operations is important for correctness here.
513− -- We need to set the flag to false strictly before sending the
514− -- outputDoorBell, otherwise the outputDoorBell may get processed too early and
515− -- then we may set the flag to False to later making the consumer lose
516− -- the flag, even without receiving a outputDoorBell.
517− atomicModifyIORefCAS_ (needDoorBell sv) (const False)
518− void $ tryPutMVar (outputDoorBell sv) ()
519−
520206{-# INLINE runStreamFIFO #-}
521207runStreamFIFO :: MonadIO m
522− => SVar m a -> LinkedQueue (Stream m a) -> Stream m a -> m () -> m ()
208+ => SVar Stream m a -> LinkedQueue (Stream m a) -> Stream m a -> m () -> m ()
523209runStreamFIFO sv q m stop = runStream m (Just sv) stop single yield
524210 where
525211 single a = do
@@ -530,23 +216,12 @@ runStreamFIFO sv q m stop = runStream m (Just sv) stop single yield
530216 liftIO (enqueueFIFO sv q r)
531217 if res then stop else liftIO $ sendStop sv
532218
533−workLoopFIFO :: MonadIO m => SVar m a -> LinkedQueue (Stream m a) -> m ()
534−workLoopFIFO sv q = run
535−
536− where
537−
538− run = do
539− work <- liftIO $ tryPopR q
540− case work of
541− Nothing -> liftIO $ sendStop sv
542− Just m -> runStreamFIFO sv q m run
543−
544219-------------------------------------------------------------------------------
545220-- Parallel
546221-------------------------------------------------------------------------------
547222
548223{-# NOINLINE runOne #-}
549−runOne :: MonadIO m => SVar m a -> Stream m a -> m ()
224+runOne :: MonadIO m => SVar Stream m a -> Stream m a -> m ()
550225runOne sv m = (runStream m) (Just sv) stop single yield
551226
552227 where
@@ -616,76 +291,11 @@ runOne sv m = (runStream m) (Just sv) stop single yield
616291-- each left associative expression. The queue is used only for right
617292-- associated expression, we queue the right expression and execute the left.
618293-- Thererefore the queue never has more than on item in it.
619−{-# INLINE enqueueAhead #-}
620−enqueueAhead :: SVar m a -> IORef ([Stream m a], Int) -> Stream m a -> IO ()
621−enqueueAhead sv q m = do
622− atomicModifyIORefCAS_ q $ \ case
623− ([], n) -> ([m], n + 1) -- increment sequence
624− _ -> error "not empty"
625− storeLoadBarrier
626− w <- readIORef $ needDoorBell sv
627− when w $ do
628− -- Note: the sequence of operations is important for correctness here.
629− -- We need to set the flag to false strictly before sending the
630− -- outputDoorBell, otherwise the outputDoorBell may get processed too early and
631− -- then we may set the flag to False to later making the consumer lose
632− -- the flag, even without receiving a outputDoorBell.
633− atomicModifyIORefCAS_ (needDoorBell sv) (const False)
634− void $ tryPutMVar (outputDoorBell sv) ()
635−
636−-- Normally the thread that has the token should never go away. The token gets
637−-- handed over to another thread, but someone or the other has the token at any
638−-- point of time. But if the task that has the token finds that the outputQueue
639−-- is full, in that case it can go away without even handing over the token to
640−-- another thread. In that case it sets the nextSequence number in the heap its
641−-- own sequence number before going away. To handle this case, any task that
642−-- does not have the token tries to dequeue from the heap first before
643−-- dequeuing from the work queue. If it finds that the task at the top of the
644−-- heap is the one that owns the current sequence number then it grabs the
645−-- token and starts with that.
646−--
647−-- XXX instead of queueing just the head element and the remaining computation
648−-- on the heap, evaluate as many as we can and place them on the heap. But we
649−-- need to give higher priority to the lower sequence numbers so that lower
650−-- priority tasks do not fill up the heap making higher priority tasks block
651−-- due to full heap. Maybe we can have a weighted space for them in the heap.
652−-- The weight is inversely proportional to the sequence number.
653−--
654−-- XXX review for livelock
655−--
656−maxHeap :: Int
657−maxHeap = 1500
658−
659−queueEmptyAhead :: MonadIO m => IORef ([Stream m a], Int) -> m Bool
660−queueEmptyAhead q = liftIO $ do
661− (xs, _) <- readIORef q
662− return $ null xs
663−
664−dequeueAhead :: MonadIO m
665− => IORef ([Stream m a], Int) -> m (Maybe (Stream m a, Int))
666−dequeueAhead q = liftIO $ do
667− atomicModifyIORefCAS q $ \case
668− ([], n) -> (([], n), Nothing)
669− (x : [], n) -> (([], n), Just (x, n))
670− _ -> error "more than one item on queue"
671−
672−dequeueFromHeap
673− :: IORef (Heap (Entry Int (AheadHeapEntry m a)), Int)
674− -> IO (Maybe (Entry Int (AheadHeapEntry m a)))
675−dequeueFromHeap hpRef = do
676− atomicModifyIORefCAS hpRef $ \hp@(h, snum) -> do
677− let r = H.uncons h
678− case r of
679− Nothing -> (hp, Nothing)
680− Just (ent@(Entry seqNo _ev), hp') ->
681− if (seqNo == snum)
682− then ((hp', seqNo), Just ent)
683− else (hp, Nothing)
684294
685295workLoopAhead :: MonadIO m
686− => SVar m a
296+ => SVar Stream m a
687297 -> IORef ([Stream m a], Int)
688− -> IORef (Heap (Entry Int (AheadHeapEntry m a)) , Int)
298+ -> IORef (Heap (Entry Int (AheadHeapEntry Stream m a)) , Int)
689299 -> m ()
690300workLoopAhead sv q heap = runHeap
691301
@@ -786,236 +396,9 @@ workLoopAhead sv q heap = runHeap
786396-- remaining computation at the back of the queue instead of the heap, and
787397-- increment the sequence number.
788398
789−-- Thread tracking is needed for two reasons:
790−--
791−-- 1) Killing threads on exceptions. Threads may not be left to go away by
792−-- themselves because they may run for significant times before going away or
793−-- worse they may be stuck in IO and never go away.
794−--
795−-- 2) To know when all threads are done and the stream has ended.
796−
797−{-# NOINLINE addThread #-}
798−addThread :: MonadIO m => SVar m a -> ThreadId -> m ()
799−addThread sv tid =
800− liftIO $ modifyIORef (workerThreads sv) (S.insert tid)
801−
802−-- This is cheaper than modifyThread because we do not have to send a
803−-- outputDoorBell This can make a difference when more workers are being
804−-- dispatched.
805−{-# INLINE delThread #-}
806−delThread :: MonadIO m => SVar m a -> ThreadId -> m ()
807−delThread sv tid =
808− liftIO $ modifyIORef (workerThreads sv) $ (\s -> S.delete tid s)
809−
810−-- If present then delete else add. This takes care of out of order add and
811−-- delete i.e. a delete arriving before we even added a thread.
812−-- This occurs when the forked thread is done even before the 'addThread' right
813−-- after the fork gets a chance to run.
814−{-# INLINE modifyThread #-}
815−modifyThread :: MonadIO m => SVar m a -> ThreadId -> m ()
816−modifyThread sv tid = do
817− changed <- liftIO $ atomicModifyIORefCAS (workerThreads sv) $ \old ->
818− if (S.member tid old)
819− then let new = (S.delete tid old) in (new, new)
820− else let new = (S.insert tid old) in (new, old)
821− if null changed
822− then liftIO $ do
823− writeBarrier
824− void $ tryPutMVar (outputDoorBell sv) ()
825− else return ()
826−
827−-- | This is safe even if we are adding more threads concurrently because if
828−-- a child thread is adding another thread then anyway 'workerThreads' will
829−-- not be empty.
830−{-# INLINE allThreadsDone #-}
831−allThreadsDone :: MonadIO m => SVar m a -> m Bool
832−allThreadsDone sv = liftIO $ S.null <$> readIORef (workerThreads sv)
833−
834−{-# NOINLINE handleChildException #-}
835−handleChildException :: SVar m a -> SomeException -> IO ()
836−handleChildException sv e = do
837− tid <- myThreadId
838− void $ send sv (ChildStop tid (Just e))
839−
840−#ifdef DIAGNOSTICS
841−recordMaxWorkers :: MonadIO m => SVar m a -> m ()
842−recordMaxWorkers sv = liftIO $ do
843− active <- readIORef (workerCount sv)
844− maxWrk <- readIORef (maxWorkers sv)
845− when (active > maxWrk) $ writeIORef (maxWorkers sv) active
846− modifyIORef (totalDispatches sv) (+1)
847−#endif
848−
849−{-# NOINLINE pushWorker #-}
850−pushWorker :: MonadAsync m => SVar m a -> m ()
851−pushWorker sv = do
852− liftIO $ atomicModifyIORefCAS_ (workerCount sv) $ \n -> n + 1
853−#ifdef DIAGNOSTICS
854− recordMaxWorkers sv
855−#endif
856− doFork (workLoop sv) (handleChildException sv) >>= addThread sv
857−
858−-- | In contrast to pushWorker which always happens only from the consumer
859−-- thread, a pushWorkerPar can happen concurrently from multiple threads on the
860−-- producer side. So we need to use a thread safe modification of
861−-- workerThreads. Alternatively, we can use a CreateThread event to avoid
862−-- using a CAS based modification.
863−{-# NOINLINE pushWorkerPar #-}
864−pushWorkerPar :: MonadAsync m => SVar m a -> Stream m a -> m ()
865−pushWorkerPar sv m = do
866− -- We do not use workerCount in case of ParallelVar but still there is no
867− -- harm in maintaining it correctly.
868−#ifdef DIAGNOSTICS
869− liftIO $ atomicModifyIORefCAS_ (workerCount sv) $ \n -> n + 1
870− recordMaxWorkers sv
871−#endif
872− doFork (runOne sv m) (handleChildException sv) >>= modifyThread sv
873−
874−maxWorkerLimit :: Int
875−maxWorkerLimit = 1500
876−
877−dispatchWorker :: MonadAsync m => SVar m a -> m ()
878−dispatchWorker sv = do
879− done <- liftIO $ isWorkDone sv
880− when (not done) $ do
881− -- Note that the worker count is only decremented during event
882− -- processing in fromStreamVar and therefore it is safe to read and
883− -- use it without a lock.
884− cnt <- liftIO $ readIORef $ workerCount sv
885− -- Note that we may deadlock if the previous workers (tasks in the
886− -- stream) wait/depend on the future workers (tasks in the stream)
887− -- executing. In that case we should either configure the maxWorker
888− -- count to higher or use parallel style instead of ahead or async
889− -- style.
890− when (cnt < maxWorkerLimit) $ pushWorker sv
891−
892−{-# NOINLINE sendWorkerWait #-}
893−sendWorkerWait :: MonadAsync m => SVar m a -> m ()
894−sendWorkerWait sv = do
895− -- Note that we are guaranteed to have at least one outstanding worker when
896− -- we enter this function. So if we sleep we are guaranteed to be woken up
897− -- by a outputDoorBell, when the worker exits.
898−
899− -- XXX we need a better way to handle this than hardcoded delays. The
900− -- delays may be different for different systems.
901− ncpu <- liftIO $ getNumCapabilities
902− if ncpu <= 1
903− then
904− if (svarStyle sv == AheadVar)
905− then liftIO $ threadDelay 100
906− else liftIO $ threadDelay 25
907− else
908− if (svarStyle sv == AheadVar)
909− then liftIO $ threadDelay 100
910− else liftIO $ threadDelay 10
911−
912− (_, n) <- liftIO $ readIORef (outputQueue sv)
913− when (n <= 0) $ do
914− -- The queue may be empty temporarily if the worker has dequeued the
915− -- work item but has not enqueued the remaining part yet. For the same
916− -- reason, a worker may come back if it tries to dequeue and finds the
917− -- queue empty, even though the whole work has not finished yet.
918−
919− -- If we find that the queue is empty, but it may be empty
920− -- temporarily, when we checked it. If that's the case we might
921− -- sleep indefinitely unless the active workers produce some
922− -- output. We may deadlock specially if the otuput from the active
923− -- workers depends on the future workers that we may never send.
924− -- So in case the queue was temporarily empty set a flag to inform
925− -- the enqueue to send us a doorbell.
926−
927− -- Note that this is just a best effort mechanism to avoid a
928− -- deadlock. Deadlocks may still happen if for some weird reason
929− -- the consuming computation shares an MVar or some other resource
930− -- with the producing computation and gets blocked on that resource
931− -- and therefore cannot do any pushworker to add more threads to
932− -- the producer. In such cases the programmer should use a parallel
933− -- style so that all the producers are scheduled immediately and
934− -- unconditionally. We can also use a separate monitor thread to
935− -- push workers instead of pushing them from the consumer, but then
936− -- we are no longer using pull based concurrency rate adaptation.
937− --
938− -- XXX update this in the tutorial.
939−
940− -- register for the outputDoorBell before we check the queue so that if we
941− -- sleep because the queue was empty we are guaranteed to get a
942− -- doorbell on the next enqueue.
943−
944− liftIO $ atomicModifyIORefCAS_ (needDoorBell sv) $ const True
945− liftIO $ storeLoadBarrier
946− dispatchWorker sv
947−
948− -- XXX test for the case when we miss sending a worker when the worker
949− -- count is more than 1500.
950− --
951− -- XXX Assert here that if the heap is not empty then there is at
952− -- least one outstanding worker. Otherwise we could be sleeping
953− -- forever.
954−
955− done <- liftIO $ isWorkDone sv
956− if done
957− then do
958− liftIO $ withDBGMVar sv "sendWorkerWait: nothing to do"
959− $ takeMVar (outputDoorBell sv)
960− (_, len) <- liftIO $ readIORef (outputQueue sv)
961− when (len <= 0) $ sendWorkerWait sv
962− else sendWorkerWait sv
963−
964−{-# INLINE readOutputQRaw #-}
965−readOutputQRaw :: SVar m a -> IO ([ChildEvent a], Int)
966−readOutputQRaw sv = do
967− (list, len) <- atomicModifyIORefCAS (outputQueue sv) $ \x -> (([],0), x)
968−#ifdef DIAGNOSTICS
969− oqLen <- readIORef (maxOutQSize sv)
970− when (len > oqLen) $ writeIORef (maxOutQSize sv) len
971−#endif
972− return (list, len)
973−
974−readOutputQBounded :: MonadAsync m => SVar m a -> m [ChildEvent a]
975−readOutputQBounded sv = do
976− (list, len) <- liftIO $ readOutputQRaw sv
977− -- When there is no output seen we dispatch more workers to help
978− -- out if there is work pending in the work queue.
979− if len <= 0
980− then blockingRead
981− else do
982− -- send a worker proactively, if needed, even before we start
983− -- processing the output. This may degrade single processor
984− -- perf but improves multi-processor, because of more
985− -- parallelism
986− sendWorker
987− return list
988−
989− where
990−
991− sendWorker = do
992− cnt <- liftIO $ readIORef $ workerCount sv
993− when (cnt <= 0) $ do
994− done <- liftIO $ isWorkDone sv
995− when (not done) $ pushWorker sv
996−
997− {-# INLINE blockingRead #-}
998− blockingRead = do
999− sendWorkerWait sv
1000− liftIO $ (readOutputQRaw sv >>= return . fst)
1001−
1002−postProcessBounded :: MonadAsync m => SVar m a -> m Bool
1003−postProcessBounded sv = do
1004− workersDone <- allThreadsDone sv
1005− -- There may still be work pending even if there are no workers
1006− -- pending because all the workers may return if the
1007− -- outputQueue becomes full. In that case send off a worker to
1008− -- kickstart the work again.
1009− if workersDone
1010− then do
1011− r <- liftIO $ isWorkDone sv
1012− when (not r) $ pushWorker sv
1013− return r
1014− else return False
1015−
1016399-- | Pull a stream from an SVar.
1017400{-# NOINLINE fromStreamVar #-}
1018−fromStreamVar :: MonadAsync m => SVar m a -> Stream m a
401+fromStreamVar :: MonadAsync m => SVar Stream m a -> Stream m a
1019402fromStreamVar sv = Stream $ \_ stp sng yld -> do
1020403 list <- readOutputQ sv
1021404 -- Reversing the output is important to guarantee that we process the
@@ -1051,228 +434,6 @@ fromStreamVar sv = Stream $ \_ stp sng yld -> do
1051434 Nothing -> runStream rest Nothing stp sng yld
1052435 Just ex -> throwM ex
1053436
1054−getFifoSVar :: MonadAsync m => SVarStyle -> IO (SVar m a)
1055−getFifoSVar ctype = do
1056− outQ <- newIORef ([], 0)
1057− outQMv <- newEmptyMVar
1058− active <- newIORef 0
1059− wfw <- newIORef False
1060− running <- newIORef S.empty
1061− q <- newQ
1062−#ifdef DIAGNOSTICS
1063− disp <- newIORef 0
1064− maxWrk <- newIORef 0
1065− maxOq <- newIORef 0
1066− maxHs <- newIORef 0
1067− maxWq <- newIORef 0
1068−#endif
1069− let sv =
1070− SVar { outputQueue = outQ
1071− , outputDoorBell = outQMv
1072− , readOutputQ = readOutputQBounded sv
1073− , postProcess = postProcessBounded sv
1074− , workerThreads = running
1075− , workLoop = workLoopFIFO sv q
1076− , enqueue = enqueueFIFO sv q
1077− , isWorkDone = nullQ q
1078− , needDoorBell = wfw
1079− , svarStyle = ctype
1080− , workerCount = active
1081− , accountThread = delThread sv
1082−#ifdef DIAGNOSTICS
1083− , aheadWorkQueue = undefined
1084− , outputHeap = undefined
1085− , totalDispatches = disp
1086− , maxWorkers = maxWrk
1087− , maxOutQSize = maxOq
1088− , maxHeapSize = maxHs
1089− , maxWorkQSize = maxWq
1090−#endif
1091− }
1092− in return sv
1093−
1094−getLifoSVar :: MonadAsync m => SVarStyle -> IO (SVar m a)
1095−getLifoSVar ctype = do
1096− outQ <- newIORef ([], 0)
1097− outQMv <- newEmptyMVar
1098− active <- newIORef 0
1099− wfw <- newIORef False
1100− running <- newIORef S.empty
1101− q <- newIORef []
1102−#ifdef DIAGNOSTICS
1103− disp <- newIORef 0
1104− maxWrk <- newIORef 0
1105− maxOq <- newIORef 0
1106− maxHs <- newIORef 0
1107− maxWq <- newIORef 0
1108−#endif
1109− let checkEmpty = null <$> readIORef q
1110− let sv =
1111− SVar { outputQueue = outQ
1112− , outputDoorBell = outQMv
1113− , readOutputQ = readOutputQBounded sv
1114− , postProcess = postProcessBounded sv
1115− , workerThreads = running
1116− , workLoop = workLoopLIFO sv q
1117− , enqueue = enqueueLIFO sv q
1118− , isWorkDone = checkEmpty
1119− , needDoorBell = wfw
1120− , svarStyle = ctype
1121− , workerCount = active
1122− , accountThread = delThread sv
1123−#ifdef DIAGNOSTICS
1124− , aheadWorkQueue = undefined
1125− , outputHeap = undefined
1126− , maxWorkers = maxWrk
1127− , totalDispatches = disp
1128− , maxOutQSize = maxOq
1129− , maxHeapSize = maxHs
1130− , maxWorkQSize = maxWq
1131−#endif
1132− }
1133− in return sv
1134−
1135−getAheadSVar :: MonadAsync m => SVarStyle -> IO (SVar m a)
1136−getAheadSVar style = do
1137− outQ <- newIORef ([], 0)
1138− outH <- newIORef (H.empty, 0)
1139− outQMv <- newEmptyMVar
1140− active <- newIORef 0
1141− wfw <- newIORef False
1142− running <- newIORef S.empty
1143− q <- newIORef ([], -1)
1144−
1145−#ifdef DIAGNOSTICS
1146− disp <- newIORef 0
1147− maxWrk <- newIORef 0
1148− maxOq <- newIORef 0
1149− maxHs <- newIORef 0
1150− maxWq <- newIORef 0
1151−#endif
1152− let sv =
1153− SVar { outputQueue = outQ
1154− , outputDoorBell = outQMv
1155− , readOutputQ = readOutputQBounded sv
1156− , postProcess = postProcessBounded sv
1157− , workerThreads = running
1158− , workLoop = workLoopAhead sv q outH
1159− , enqueue = enqueueAhead sv q
1160− , isWorkDone = isWorkDoneAhead q outH
1161− , needDoorBell = wfw
1162− , svarStyle = style
1163− , workerCount = active
1164− , accountThread = delThread sv
1165−#ifdef DIAGNOSTICS
1166− , aheadWorkQueue = q
1167− , outputHeap = outH
1168− , totalDispatches = disp
1169− , maxWorkers = maxWrk
1170− , maxOutQSize = maxOq
1171− , maxHeapSize = maxHs
1172− , maxWorkQSize = maxWq
1173−#endif
1174− }
1175− in return sv
1176−
1177− where
1178−
1179− {-# INLINE isWorkDoneAhead #-}
1180− isWorkDoneAhead q ref = do
1181− heapDone <- do
1182− (hp, _) <- readIORef ref
1183− return (H.size hp <= 0)
1184− queueDone <- checkEmpty q
1185− return $ queueDone && heapDone
1186−
1187− checkEmpty q = do
1188− (xs, _) <- readIORef q
1189− return $ null xs
1190−
1191−getParSVar :: MonadIO m => SVarStyle -> IO (SVar m a)
1192−getParSVar style = do
1193− outQ <- newIORef ([], 0)
1194− outQMv <- newEmptyMVar
1195− active <- newIORef 0
1196− running <- newIORef S.empty
1197−#ifdef DIAGNOSTICS
1198− disp <- newIORef 0
1199− maxWrk <- newIORef 0
1200− maxOq <- newIORef 0
1201− maxHs <- newIORef 0
1202− maxWq <- newIORef 0
1203−#endif
1204− let sv =
1205− SVar { outputQueue = outQ
1206− , outputDoorBell = outQMv
1207− , readOutputQ = readOutputQPar sv
1208− , postProcess = allThreadsDone sv
1209− , workerThreads = running
1210− , workLoop = undefined
1211− , enqueue = undefined
1212− , isWorkDone = undefined
1213− , needDoorBell = undefined
1214− , svarStyle = style
1215− , workerCount = active
1216− , accountThread = modifyThread sv
1217−#ifdef DIAGNOSTICS
1218− , aheadWorkQueue = undefined
1219− , outputHeap = undefined
1220− , totalDispatches = disp
1221− , maxWorkers = maxWrk
1222− , maxOutQSize = maxOq
1223− , maxHeapSize = maxHs
1224− , maxWorkQSize = maxWq
1225−#endif
1226− }
1227− in return sv
1228−
1229− where
1230−
1231− readOutputQPar sv = liftIO $ do
1232− withDBGMVar sv "fromStreamVar: doorbell" $ takeMVar (outputDoorBell sv)
1233− readOutputQRaw sv >>= return . fst
1234−
1235−-- | Create a new empty SVar.
1236−newEmptySVar :: MonadAsync m => SVarStyle -> m (SVar m a)
1237−newEmptySVar style = do
1238− liftIO $
1239− case style of
1240− WAsyncVar -> getFifoSVar style
1241− AsyncVar -> getLifoSVar style
1242− ParallelVar -> getParSVar style
1243− AheadVar -> getAheadSVar style
1244−
1245−-- | Create a new SVar and enqueue one stream computation on it.
1246−{-# INLINABLE newStreamVar1 #-}
1247−newStreamVar1 :: MonadAsync m => SVarStyle -> Stream m a -> m (SVar m a)
1248−newStreamVar1 style m = do
1249− sv <- newEmptySVar style
1250− -- Note: We must have all the work on the queue before sending the
1251− -- pushworker, otherwise the pushworker may exit before we even get a
1252− -- chance to push.
1253− liftIO $ enqueue sv m
1254− pushWorker sv
1255− return sv
1256−
1257−{-# INLINABLE newStreamVarPar #-}
1258−newStreamVarPar :: MonadAsync m => Stream m a -> m (SVar m a)
1259−newStreamVarPar m = do
1260− sv <- newEmptySVar ParallelVar
1261− pushWorkerPar sv m
1262− return sv
1263−
1264−-- XXX this errors out for Parallel/Ahead SVars
1265−-- | Write a stream to an 'SVar' in a non-blocking manner. The stream can then
1266−-- be read back from the SVar using 'fromSVar'.
1267−toStreamVar :: MonadAsync m => SVar m a -> Stream m a -> m ()
1268−toStreamVar sv m = do
1269− liftIO $ (enqueue sv) m
1270− done <- allThreadsDone sv
1271− -- XXX This is safe only when called from the consumer thread or when no
1272− -- consumer is present. There may be a race if we are not running in the
1273− -- consumer thread.
1274− when done $ pushWorker sv
1275−
1276437------------------------------------------------------------------------------
1277438-- Running streams concurrently
1278439------------------------------------------------------------------------------
@@ -1339,7 +500,10 @@ toStreamVar sv m = do
1339500
1340501forkSVarAsync :: MonadAsync m => SVarStyle -> Stream m a -> Stream m a -> Stream m a
1341502forkSVarAsync style m1 m2 = Stream $ \_ stp sng yld -> do
1342− sv <- newStreamVar1 style (concurrently m1 m2)
503+ sv <- case style of
504+ AsyncVar -> newAsyncVar (concurrently m1 m2)
505+ WAsyncVar -> newWAsyncVar (concurrently m1 m2)
506+ _ -> error "illegal svar type"
1343507 runStream (fromStreamVar sv) Nothing stp sng yld
1344508 where
1345509 concurrently ma mb = Stream $ \svr stp sng yld -> do
@@ -1359,7 +523,7 @@ joinStreamVarAsync style m1 m2 = Stream $ \svr stp sng yld ->
1359523-- computation without a shared SVar.
1360524forkSVarAhead :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
1361525forkSVarAhead m1 m2 = Stream $ \_ stp sng yld -> do
1362− sv <- newStreamVar1 AheadVar (concurrently m1 m2)
526+ sv <- newAheadVar (concurrently m1 m2) workLoopAhead
1363527 (runStream (fromStreamVar sv)) Nothing stp sng yld
1364528 where
1365529 concurrently ma mb = Stream $ \svr stp sng yld -> do
@@ -1381,9 +545,9 @@ ahead m1 m2 = Stream $ \svr stp sng yld -> do
1381545{-# NOINLINE forkSVarPar #-}
1382546forkSVarPar :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
1383547forkSVarPar m r = Stream $ \_ stp sng yld -> do
1384− sv <- newEmptySVar ParallelVar
1385− pushWorkerPar sv m
1386− pushWorkerPar sv r
548+ sv <- newParallelVar
549+ pushWorkerPar sv (runOne sv m)
550+ pushWorkerPar sv (runOne sv r)
1387551 (runStream (fromStreamVar sv)) Nothing stp sng yld
1388552
1389553{-# INLINE joinStreamVarPar #-}
@@ -1392,7 +556,7 @@ joinStreamVarPar :: MonadAsync m
1392556joinStreamVarPar style m1 m2 = Stream $ \svr stp sng yld ->
1393557 case svr of
1394558 Just sv | svarStyle sv == style -> do
1395− pushWorkerPar sv m1 >> (runStream m2) svr stp sng yld
559+ pushWorkerPar sv (runOne sv m1) >> (runStream m2) svr stp sng yld
1396560 _ -> runStream (forkSVarPar m1 m2) Nothing stp sng yld
1397561
1398562------------------------------------------------------------------------------
@@ -1458,10 +622,17 @@ _alt m1 m2 = Stream $ \_ stp sng yld ->
1458622-- Stream to stream function application
1459623------------------------------------------------------------------------------
1460624
625+mkParallel :: MonadAsync m => Stream m a -> m (Stream m a)
626+mkParallel m = do
627+ sv <- newParallelVar
628+ pushWorkerPar sv (runOne sv m)
629+ return $ fromStreamVar sv
630+
1461631applyWith :: MonadAsync m
1462632 => (Stream m a -> Stream m b) -> Stream m a -> Stream m b
1463633applyWith f m = Stream $ \svr stp sng yld -> do
1464− sv <- newStreamVarPar m
634+ sv <- newParallelVar
635+ pushWorkerPar sv (runOne sv m)
1465636 runStream (f $ fromStreamVar sv) svr stp sng yld
1466637
1467638------------------------------------------------------------------------------
@@ -1470,7 +641,8 @@ applyWith f m = Stream $ \svr stp sng yld -> do
1470641
1471642runWith :: MonadAsync m => (Stream m a -> m b) -> Stream m a -> m b
1472643runWith f m = do
1473− sv <- newStreamVarPar m
644+ sv <- newParallelVar
645+ pushWorkerPar sv (runOne sv m)
1474646 f $ fromStreamVar sv
1475647
1476648------------------------------------------------------------------------------
@@ -1501,7 +673,7 @@ zipAsyncWith f m1 m2 = Stream $ \_ stp sng yld -> do
1501673 where
1502674
1503675 mkAsync :: MonadAsync m => Stream m a -> m (Stream m a)
1504− mkAsync m = newStreamVar1 AsyncVar m
676+ mkAsync m = newAsyncVar m
1505677 >>= return . fromStreamVar
1506678
1507679-------------------------------------------------------------------------------
@@ -1510,3 +682,106 @@ zipAsyncWith f m1 m2 = Stream $ \_ stp sng yld -> do
1510682
1511683instance MonadTrans Stream where
1512684 lift = once
685+
686+-------------------------------------------------------------------------------
687+-- SVar creation
688+-- This code belongs in SVar.hs but is kept here for perf reasons
689+-------------------------------------------------------------------------------
690+
691+-- XXX we have this function in this file because passing runStreamLIFO as a
692+-- function argument to this function results in a perf degradation of more
693+-- than 10%. Need to investigate what the root cause is.
694+-- Interestingly, the same thing does not make any difference for Ahead.
695+getLifoSVar :: MonadAsync m => IO (SVar Stream m a)
696+getLifoSVar = do
697+ outQ <- newIORef ([], 0)
698+ outQMv <- newEmptyMVar
699+ active <- newIORef 0
700+ wfw <- newIORef False
701+ running <- newIORef S.empty
702+ q <- newIORef []
703+#ifdef DIAGNOSTICS
704+ disp <- newIORef 0
705+ maxWrk <- newIORef 0
706+ maxOq <- newIORef 0
707+ maxHs <- newIORef 0
708+ maxWq <- newIORef 0
709+#endif
710+ let checkEmpty = null <$> readIORef q
711+ let sv =
712+ SVar { outputQueue = outQ
713+ , outputDoorBell = outQMv
714+ , readOutputQ = readOutputQBounded sv
715+ , postProcess = postProcessBounded sv
716+ , workerThreads = running
717+ , workLoop = workLoopLIFO runStreamLIFO sv q
718+ , enqueue = enqueueLIFO sv q
719+ , isWorkDone = checkEmpty
720+ , needDoorBell = wfw
721+ , svarStyle = AsyncVar
722+ , workerCount = active
723+ , accountThread = delThread sv
724+#ifdef DIAGNOSTICS
725+ , aheadWorkQueue = undefined
726+ , outputHeap = undefined
727+ , maxWorkers = maxWrk
728+ , totalDispatches = disp
729+ , maxOutQSize = maxOq
730+ , maxHeapSize = maxHs
731+ , maxWorkQSize = maxWq
732+#endif
733+ }
734+ in return sv
735+
736+getFifoSVar :: MonadAsync m => IO (SVar Stream m a)
737+getFifoSVar = do
738+ outQ <- newIORef ([], 0)
739+ outQMv <- newEmptyMVar
740+ active <- newIORef 0
741+ wfw <- newIORef False
742+ running <- newIORef S.empty
743+ q <- newQ
744+#ifdef DIAGNOSTICS
745+ disp <- newIORef 0
746+ maxWrk <- newIORef 0
747+ maxOq <- newIORef 0
748+ maxHs <- newIORef 0
749+ maxWq <- newIORef 0
750+#endif
751+ let sv =
752+ SVar { outputQueue = outQ
753+ , outputDoorBell = outQMv
754+ , readOutputQ = readOutputQBounded sv
755+ , postProcess = postProcessBounded sv
756+ , workerThreads = running
757+ , workLoop = workLoopFIFO runStreamFIFO sv q
758+ , enqueue = enqueueFIFO sv q
759+ , isWorkDone = nullQ q
760+ , needDoorBell = wfw
761+ , svarStyle = WAsyncVar
762+ , workerCount = active
763+ , accountThread = delThread sv
764+#ifdef DIAGNOSTICS
765+ , aheadWorkQueue = undefined
766+ , outputHeap = undefined
767+ , totalDispatches = disp
768+ , maxWorkers = maxWrk
769+ , maxOutQSize = maxOq
770+ , maxHeapSize = maxHs
771+ , maxWorkQSize = maxWq
772+#endif
773+ }
774+ in return sv
775+
776+{-# INLINABLE newAsyncVar #-}
777+newAsyncVar :: MonadAsync m => Stream m a -> m (SVar Stream m a)
778+newAsyncVar m = do
779+ sv <- liftIO getLifoSVar
780+ sendWorker sv m
781+
782+-- | Create a new SVar and enqueue one stream computation on it.
783+{-# INLINABLE newWAsyncVar #-}
784+newWAsyncVar :: MonadAsync m => Stream m a -> m (SVar Stream m a)
785+newWAsyncVar m = do
786+ sv <- liftIO getFifoSVar
787+ sendWorker sv m