@@ -58,8 +58,7 @@ module Streamly.Core
59 -- * Concurrent Stream Vars (SVars)
59 -- * Concurrent Stream Vars (SVars)
61− , SVarStyle (..)
61+ , newStreamVarPar
63 , fromStreamVar
62 , fromStreamVar
64 , toStreamVar
63 , toStreamVar
@@ -130,7 +129,6 @@ data AheadHeapEntry m a =
130-- State threaded around the monad for thread management
129-- State threaded around the monad for thread management
131------------------------------------------------------------------------------
130------------------------------------------------------------------------------
133−-- XXX use a separate data structure for each type of SVar
134-- | Identify the type of the SVar. Two computations using the same style can
132-- | Identify the type of the SVar. Two computations using the same style can
135-- be scheduled on the same SVar.
133-- be scheduled on the same SVar.
136data SVarStyle =
134data SVarStyle =
@@ -149,10 +147,10 @@ data SVarStyle =
149-- However, it ensures that the reader can read the results at whatever paces
147-- However, it ensures that the reader can read the results at whatever paces
150-- it wants to read. The SVar monitors and adapts to the consumer's pace.
148-- it wants to read. The SVar monitors and adapts to the consumer's pace.
152−-- An SVar is a mini scheduler, it has an associated runqueue that holds the
150+-- An SVar is a mini scheduler, it has an associated workLoop that holds the
153-- stream tasks to be picked and run by a pool of worker threads. It has an
151-- stream tasks to be picked and run by a pool of worker threads. It has an
154-- associated output queue where the output stream elements are placed by the
152-- associated output queue where the output stream elements are placed by the
155−-- worker threads. A doorBell is used by the worker threads to intimate the
153+-- worker threads. A outputDoorBell is used by the worker threads to intimate the
156-- consumer thread about availability of new results in the output queue. More
154-- consumer thread about availability of new results in the output queue. More
157-- workers are added to the SVar by 'fromStreamVar' on demand if the output
155-- workers are added to the SVar by 'fromStreamVar' on demand if the output
158-- produced is not keeping pace with the consumer. On bounded SVars, workers
156-- produced is not keeping pace with the consumer. On bounded SVars, workers
@@ -171,34 +169,31 @@ data SVar m a =
172 -- Shared output queue (events, length)
170 -- Shared output queue (events, length)
173 , outputQueue :: IORef ([ChildEvent a], Int)
171 , outputQueue :: IORef ([ChildEvent a], Int)
174− , doorBell :: MVar () -- signal the consumer about output
172+ , outputDoorBell :: MVar () -- signal the consumer about output
175−
173+ , readOutputQ :: m [ChildEvent a]
176− -- Output synchronization mechanism for Ahead streams (Ahead and
174+ , postProcess :: m Bool
177− -- wAhead). We maintain a heap of out of sequence ahead of time
178− -- generated outputs and the sequence number of the task that is
179− -- currently at the head of the stream. Concurrent execute ahead
180− -- tasks that have a sequence number greater than the task at the
181− -- head should add their output to the heap.
182− , outputHeap :: IORef (Heap (Entry Int (AheadHeapEntry m a))
186− -- Shared work queue (stream, seqNo)
176+ -- Used only by bounded SVar types
187− , workQueue :: IORef ([Stream m a], Int)
188 , enqueue :: Stream m a -> IO ()
177 , enqueue :: Stream m a -> IO ()
189− , queueEmpty :: m Bool
178+ , isWorkDone :: IO Bool
190− , waitingForWork :: IORef Bool
179+ , needDoorBell :: IORef Bool
191− , runqueue :: m ()
180+ , workLoop :: m ()
193 -- Shared, thread tracking
182 -- Shared, thread tracking
194− , runningThreads :: IORef (Set ThreadId)
183+ , workerThreads :: IORef (Set ThreadId)
195− , activeWorkers :: IORef Int
184+ , workerCount :: IORef Int
185+ , accountThread :: ThreadId -> m ()
196#ifdef DIAGNOSTICS
186#ifdef DIAGNOSTICS
187+ , outputHeap :: IORef (Heap (Entry Int (AheadHeapEntry m a))
190+ -- Shared work queue (stream, seqNo)
191+ , aheadWorkQueue :: IORef ([Stream m a], Int)
197 , totalDispatches :: IORef Int
192 , totalDispatches :: IORef Int
198− , maxWorkers :: IORef Int
193+ , maxWorkers :: IORef Int
199− , maxOutQSize :: IORef Int
194+ , maxOutQSize :: IORef Int
200− , maxHeapSize :: IORef Int
195+ , maxHeapSize :: IORef Int
201− , maxWorkQSize :: IORef Int
196+ , maxWorkQSize :: IORef Int
@@ -208,12 +203,12 @@ dumpSVar :: SVar m a -> IO String
208dumpSVar sv = do
203dumpSVar sv = do
209 tid <- myThreadId
204 tid <- myThreadId
210 (oqList, oqLen) <- readIORef $ outputQueue sv
205 (oqList, oqLen) <- readIORef $ outputQueue sv
211− db <- tryTakeMVar $ doorBell sv
206+ db <- tryTakeMVar $ outputDoorBell sv
212 aheadDump <-
207 aheadDump <-
213 if svarStyle sv == AheadVar
208 if svarStyle sv == AheadVar
215 (oheap, oheapSeq) <- readIORef $ outputHeap sv
210 (oheap, oheapSeq) <- readIORef $ outputHeap sv
216− (wq, wqSeq) <- readIORef $ workQueue sv
211+ (wq, wqSeq) <- readIORef $ aheadWorkQueue sv
217 maxHp <- readIORef $ maxHeapSize sv
212 maxHp <- readIORef $ maxHeapSize sv
218 return $ unlines
213 return $ unlines
219 [ "heap length = " ++ show (H.size oheap)
214 [ "heap length = " ++ show (H.size oheap)
@@ -224,26 +219,25 @@ dumpSVar sv = do
225 else return []
220 else return []
227− waiting <- readIORef $ waitingForWork sv
222+ waiting <- readIORef $ needDoorBell sv
228− rthread <- readIORef $ runningThreads sv
223+ rthread <- readIORef $ workerThreads sv
229− workers <- readIORef $ activeWorkers sv
224+ workers <- readIORef $ workerCount sv
230 maxWrk <- readIORef $ maxWorkers sv
225 maxWrk <- readIORef $ maxWorkers sv
231 dispatches <- readIORef $ totalDispatches sv
226 dispatches <- readIORef $ totalDispatches sv
232 maxOq <- readIORef $ maxOutQSize sv
227 maxOq <- readIORef $ maxOutQSize sv
233− -- XXX queueEmpty should be made IO return type
235 return $ unlines
229 return $ unlines
236 [ "tid = " ++ show tid
230 [ "tid = " ++ show tid
237 , "style = " ++ show (svarStyle sv)
231 , "style = " ++ show (svarStyle sv)
238 , "outputQueue length computed = " ++ show (length oqList)
232 , "outputQueue length computed = " ++ show (length oqList)
239 , "outputQueue length maintained = " ++ show oqLen
233 , "outputQueue length maintained = " ++ show oqLen
240− , "output doorBell = " ++ show db
234+ , "output outputDoorBell = " ++ show db
241 , "total dispatches = " ++ show dispatches
235 , "total dispatches = " ++ show dispatches
242 , "max workers = " ++ show maxWrk
236 , "max workers = " ++ show maxWrk
243 , "max outQSize = " ++ show maxOq
237 , "max outQSize = " ++ show maxOq
245 ++ aheadDump ++ unlines
239 ++ aheadDump ++ unlines
246− [ "waitingForWork = " ++ show waiting
240+ [ "needDoorBell = " ++ show waiting
247 , "running threads = " ++ show rthread
241 , "running threads = " ++ show rthread
248 , "running thread count = " ++ show workers
242 , "running thread count = " ++ show workers
@@ -439,13 +433,13 @@ send sv msg = do
439 -- to read the queue again and find it empty.
433 -- to read the queue again and find it empty.
440 -- The important point is that the consumer is guaranteed to receive a
434 -- The important point is that the consumer is guaranteed to receive a
441 -- doorbell if something was added to the queue after it empties it.
435 -- doorbell if something was added to the queue after it empties it.
442− void $ tryPutMVar (doorBell sv) ()
436+ void $ tryPutMVar (outputDoorBell sv) ()
443 return (len < maxOutputQLen)
437 return (len < maxOutputQLen)
445{-# NOINLINE sendStop #-}
439{-# NOINLINE sendStop #-}
446sendStop :: SVar m a -> IO ()
440sendStop :: SVar m a -> IO ()
447sendStop sv = do
441sendStop sv = do
448− liftIO $ atomicModifyIORefCAS_ (activeWorkers sv) $ \n -> n - 1
442+ liftIO $ atomicModifyIORefCAS_ (workerCount sv) $ \n -> n - 1
449 myThreadId >>= \tid -> void $ send sv (ChildStop tid Nothing)
443 myThreadId >>= \tid -> void $ send sv (ChildStop tid Nothing)
451-------------------------------------------------------------------------------
445-------------------------------------------------------------------------------
@@ -461,18 +455,18 @@ enqueueLIFO :: SVar m a -> IORef [Stream m a] -> Stream m a -> IO ()
461enqueueLIFO sv q m = do
455enqueueLIFO sv q m = do
462 atomicModifyIORefCAS_ q $ \ms -> m : ms
456 atomicModifyIORefCAS_ q $ \ms -> m : ms
463 storeLoadBarrier
457 storeLoadBarrier
464− w <- readIORef $ waitingForWork sv
458+ w <- readIORef $ needDoorBell sv
465 when w $ do
459 when w $ do
466 -- Note: the sequence of operations is important for correctness here.
460 -- Note: the sequence of operations is important for correctness here.
467 -- We need to set the flag to false strictly before sending the
461 -- We need to set the flag to false strictly before sending the
468− -- doorBell, otherwise the doorBell may get processed too early and
462+ -- outputDoorBell, otherwise the outputDoorBell may get processed too early and
469 -- then we may set the flag to False to later making the consumer lose
463 -- then we may set the flag to False to later making the consumer lose
470− -- the flag, even without receiving a doorBell.
464+ -- the flag, even without receiving a outputDoorBell.
471− atomicModifyIORefCAS_ (waitingForWork sv) (const False)
465+ atomicModifyIORefCAS_ (needDoorBell sv) (const False)
472− void $ tryPutMVar (doorBell sv) ()
466+ void $ tryPutMVar (outputDoorBell sv) ()
474−runqueueLIFO :: MonadIO m => SVar m a -> IORef [Stream m a] -> m ()
468+workLoopLIFO :: MonadIO m => SVar m a -> IORef [Stream m a] -> m ()
475−runqueueLIFO sv q = run
469+workLoopLIFO sv q = run
@@ -508,18 +502,18 @@ enqueueFIFO :: SVar m a -> LinkedQueue (Stream m a) -> Stream m a -> IO ()
508enqueueFIFO sv q m = do
502enqueueFIFO sv q m = do
509 pushL q m
503 pushL q m
510 storeLoadBarrier
504 storeLoadBarrier
511− w <- readIORef $ waitingForWork sv
505+ w <- readIORef $ needDoorBell sv
512 when w $ do
506 when w $ do
513 -- Note: the sequence of operations is important for correctness here.
507 -- Note: the sequence of operations is important for correctness here.
514 -- We need to set the flag to false strictly before sending the
508 -- We need to set the flag to false strictly before sending the
515− -- doorBell, otherwise the doorBell may get processed too early and
509+ -- outputDoorBell, otherwise the outputDoorBell may get processed too early and
516 -- then we may set the flag to False to later making the consumer lose
510 -- then we may set the flag to False to later making the consumer lose
517− -- the flag, even without receiving a doorBell.
511+ -- the flag, even without receiving a outputDoorBell.
518− atomicModifyIORefCAS_ (waitingForWork sv) (const False)
512+ atomicModifyIORefCAS_ (needDoorBell sv) (const False)
519− void $ tryPutMVar (doorBell sv) ()
513+ void $ tryPutMVar (outputDoorBell sv) ()
521−runqueueFIFO :: MonadIO m => SVar m a -> LinkedQueue (Stream m a) -> m ()
515+workLoopFIFO :: MonadIO m => SVar m a -> LinkedQueue (Stream m a) -> m ()
522−runqueueFIFO sv q = run
516+workLoopFIFO sv q = run
@@ -620,15 +614,15 @@ enqueueAhead sv q m = do
620 ([], n) -> ([m], n + 1) -- increment sequence
614 ([], n) -> ([m], n + 1) -- increment sequence
621 _ -> error "not empty"
615 _ -> error "not empty"
622 storeLoadBarrier
616 storeLoadBarrier
623− w <- readIORef $ waitingForWork sv
617+ w <- readIORef $ needDoorBell sv
624 when w $ do
618 when w $ do
625 -- Note: the sequence of operations is important for correctness here.
619 -- Note: the sequence of operations is important for correctness here.
626 -- We need to set the flag to false strictly before sending the
620 -- We need to set the flag to false strictly before sending the
627− -- doorBell, otherwise the doorBell may get processed too early and
621+ -- outputDoorBell, otherwise the outputDoorBell may get processed too early and
628 -- then we may set the flag to False to later making the consumer lose
622 -- then we may set the flag to False to later making the consumer lose
629− -- the flag, even without receiving a doorBell.
623+ -- the flag, even without receiving a outputDoorBell.
630− atomicModifyIORefCAS_ (waitingForWork sv) (const False)
624+ atomicModifyIORefCAS_ (needDoorBell sv) (const False)
631− void $ tryPutMVar (doorBell sv) ()
625+ void $ tryPutMVar (outputDoorBell sv) ()
633-- Normally the thread that has the token should never go away. The token gets
627-- Normally the thread that has the token should never go away. The token gets
634-- handed over to another thread, but someone or the other has the token at any
628-- handed over to another thread, but someone or the other has the token at any
@@ -650,15 +644,19 @@ enqueueAhead sv q m = do
651-- XXX review for livelock
645-- XXX review for livelock
647+workLoopAhead :: MonadIO m
653−runqueueAhead :: MonadIO m => SVar m a -> IORef ([Stream m a], Int) -> m ()
649+ -> IORef ([Stream m a], Int)
650+ -> IORef (Heap (Entry Int (AheadHeapEntry m a)) , Int)
654−runqueueAhead sv q = runHeap
652+workLoopAhead sv q heap = runHeap
658 maxHeap = 1500
656 maxHeap = 1500
660 toHeap seqNo ent = do
658 toHeap seqNo ent = do
661− hp <- liftIO $ atomicModifyIORefCAS (outputHeap sv) $ \(h, snum) ->
659+ hp <- liftIO $ atomicModifyIORefCAS heap $ \(h, snum) ->
662 ((H.insert (Entry seqNo ent) h, snum), h)
660 ((H.insert (Entry seqNo ent) h, snum), h)
663 if H.size hp <= maxHeap
661 if H.size hp <= maxHeap
664 then runHeap
662 then runHeap
@@ -672,7 +670,7 @@ runqueueAhead sv q = runHeap
672 if continue
670 if continue
673 then runQueueToken seqNo
671 then runQueueToken seqNo
674 else liftIO $ do
672 else liftIO $ do
675− atomicModifyIORefCAS_ (outputHeap sv) $ \(h, _) -> (h, seqNo + 1)
673+ atomicModifyIORefCAS_ heap $ \(h, _) -> (h, seqNo + 1)
676 sendStop sv
674 sendStop sv
678 yieldOutput seqNo a r = do
676 yieldOutput seqNo a r = do
@@ -682,7 +680,7 @@ runqueueAhead sv q = runHeap
682 (singleOutput seqNo)
680 (singleOutput seqNo)
683 (yieldOutput seqNo)
681 (yieldOutput seqNo)
684 else liftIO $ do
682 else liftIO $ do
685− atomicModifyIORefCAS_ (outputHeap sv) $ \(h, _) ->
683+ atomicModifyIORefCAS_ heap $ \(h, _) ->
686 (H.insert (Entry seqNo (AheadEntryStream r)) h, seqNo)
684 (H.insert (Entry seqNo (AheadEntryStream r)) h, seqNo)
687 sendStop sv
685 sendStop sv
@@ -691,7 +689,7 @@ runqueueAhead sv q = runHeap
691 work <- dequeue
689 work <- dequeue
692 case work of
690 case work of
693 Nothing -> do
691 Nothing -> do
694− liftIO $ atomicModifyIORefCAS_ (outputHeap sv) $ \(h, _) ->
692+ liftIO $ atomicModifyIORefCAS_ heap $ \(h, _) ->
695 (h, prevSeqNo + 1)
693 (h, prevSeqNo + 1)
697 Just (m, seqNo) -> do
695 Just (m, seqNo) -> do
@@ -701,7 +699,7 @@ runqueueAhead sv q = runHeap
701 (singleOutput seqNo)
699 (singleOutput seqNo)
702 (yieldOutput seqNo)
700 (yieldOutput seqNo)
704− liftIO $ atomicModifyIORefCAS_ (outputHeap sv) $ \(h, _) ->
702+ liftIO $ atomicModifyIORefCAS_ heap $ \(h, _) ->
705 (h, prevSeqNo + 1)
703 (h, prevSeqNo + 1)
706 (runStream m) (Just sv) runHeap
704 (runStream m) (Just sv) runHeap
707 (singleToHeap seqNo)
705 (singleToHeap seqNo)
@@ -726,13 +724,13 @@ runqueueAhead sv q = runHeap
726#ifdef DIAGNOSTICS
724#ifdef DIAGNOSTICS
727 liftIO $ do
725 liftIO $ do
728 maxHp <- readIORef (maxHeapSize sv)
726 maxHp <- readIORef (maxHeapSize sv)
729− (hp, _) <- readIORef (outputHeap sv)
727+ (hp, _) <- readIORef heap
730 when (H.size hp > maxHp) $ writeIORef (maxHeapSize sv) (H.size hp)
728 when (H.size hp > maxHp) $ writeIORef (maxHeapSize sv) (H.size hp)
732− ent <- liftIO $ dequeueFromHeap (outputHeap sv)
730+ ent <- liftIO $ dequeueFromHeap heap
733 case ent of
731 case ent of
734 Nothing -> do
732 Nothing -> do
735− done <- queueEmpty sv
733+ done <- queueEmpty q
737 then liftIO $ sendStop sv
735 then liftIO $ sendStop sv
738 else runQueueNoToken
736 else runQueueNoToken
@@ -744,6 +742,10 @@ runqueueAhead sv q = runHeap
744 (singleOutput seqNo)
742 (singleOutput seqNo)
745 (yieldOutput seqNo)
743 (yieldOutput seqNo)
745+ queueEmpty qu = liftIO $ do
746+ (xs, _) <- readIORef qu
747 dequeue = liftIO $ do
749 dequeue = liftIO $ do
748 atomicModifyIORefCAS q $ \case
750 atomicModifyIORefCAS q $ \case
749 ([], n) -> (([], n), Nothing)
751 ([], n) -> (([], n), Nothing)
@@ -782,14 +784,15 @@ runqueueAhead sv q = runHeap
782{-# NOINLINE addThread #-}
784{-# NOINLINE addThread #-}
783addThread :: MonadIO m => SVar m a -> ThreadId -> m ()
785addThread :: MonadIO m => SVar m a -> ThreadId -> m ()
784addThread sv tid =
786addThread sv tid =
785− liftIO $ modifyIORef (runningThreads sv) (S.insert tid)
787+ liftIO $ modifyIORef (workerThreads sv) (S.insert tid)
787−-- This is cheaper than modifyThread because we do not have to send a doorBell
789+-- This is cheaper than modifyThread because we do not have to send a
788−-- This can make a difference when more workers are being dispatched.
790+-- outputDoorBell This can make a difference when more workers are being
789{-# INLINE delThread #-}
792{-# INLINE delThread #-}
790delThread :: MonadIO m => SVar m a -> ThreadId -> m ()
793delThread :: MonadIO m => SVar m a -> ThreadId -> m ()
791delThread sv tid =
794delThread sv tid =
792− liftIO $ modifyIORef (runningThreads sv) $ (\s -> S.delete tid s)
795+ liftIO $ modifyIORef (workerThreads sv) $ (\s -> S.delete tid s)
794-- If present then delete else add. This takes care of out of order add and
797-- If present then delete else add. This takes care of out of order add and
795-- delete i.e. a delete arriving before we even added a thread.
798-- delete i.e. a delete arriving before we even added a thread.
@@ -798,22 +801,22 @@ delThread sv tid =
798{-# INLINE modifyThread #-}
801{-# INLINE modifyThread #-}
799modifyThread :: MonadIO m => SVar m a -> ThreadId -> m ()
802modifyThread :: MonadIO m => SVar m a -> ThreadId -> m ()
800modifyThread sv tid = do
803modifyThread sv tid = do
801− changed <- liftIO $ atomicModifyIORefCAS (runningThreads sv) $ \old ->
804+ changed <- liftIO $ atomicModifyIORefCAS (workerThreads sv) $ \old ->
802 if (S.member tid old)
805 if (S.member tid old)
803 then let new = (S.delete tid old) in (new, new)
806 then let new = (S.delete tid old) in (new, new)
804 else let new = (S.insert tid old) in (new, old)
807 else let new = (S.insert tid old) in (new, old)
805 if null changed
808 if null changed
806 then liftIO $ do
809 then liftIO $ do
807 writeBarrier
810 writeBarrier
808− void $ tryPutMVar (doorBell sv) ()
811+ void $ tryPutMVar (outputDoorBell sv) ()
809 else return ()
812 else return ()
811-- | This is safe even if we are adding more threads concurrently because if
814-- | This is safe even if we are adding more threads concurrently because if
812−-- a child thread is adding another thread then anyway 'runningThreads' will
815+-- a child thread is adding another thread then anyway 'workerThreads' will
813-- not be empty.
816-- not be empty.
814{-# INLINE allThreadsDone #-}
817{-# INLINE allThreadsDone #-}
815allThreadsDone :: MonadIO m => SVar m a -> m Bool
818allThreadsDone :: MonadIO m => SVar m a -> m Bool
816−allThreadsDone sv = liftIO $ S.null <$> readIORef (runningThreads sv)
819+allThreadsDone sv = liftIO $ S.null <$> readIORef (workerThreads sv)
818{-# NOINLINE handleChildException #-}
821{-# NOINLINE handleChildException #-}
819handleChildException :: SVar m a -> SomeException -> IO ()
822handleChildException :: SVar m a -> SomeException -> IO ()
@@ -824,7 +827,7 @@ handleChildException sv e = do
824#ifdef DIAGNOSTICS
827#ifdef DIAGNOSTICS
825recordMaxWorkers :: MonadIO m => SVar m a -> m ()
828recordMaxWorkers :: MonadIO m => SVar m a -> m ()
826recordMaxWorkers sv = liftIO $ do
829recordMaxWorkers sv = liftIO $ do
827− active <- readIORef (activeWorkers sv)
830+ active <- readIORef (workerCount sv)
828 maxWrk <- readIORef (maxWorkers sv)
831 maxWrk <- readIORef (maxWorkers sv)
829 when (active > maxWrk) $ writeIORef (maxWorkers sv) active
832 when (active > maxWrk) $ writeIORef (maxWorkers sv) active
830 modifyIORef (totalDispatches sv) (+1)
833 modifyIORef (totalDispatches sv) (+1)
@@ -833,51 +836,39 @@ recordMaxWorkers sv = liftIO $ do
833{-# NOINLINE pushWorker #-}
836{-# NOINLINE pushWorker #-}
834pushWorker :: MonadAsync m => SVar m a -> m ()
837pushWorker :: MonadAsync m => SVar m a -> m ()
835pushWorker sv = do
838pushWorker sv = do
836− liftIO $ atomicModifyIORefCAS_ (activeWorkers sv) $ \n -> n + 1
839+ liftIO $ atomicModifyIORefCAS_ (workerCount sv) $ \n -> n + 1
837#ifdef DIAGNOSTICS
840#ifdef DIAGNOSTICS
838 recordMaxWorkers sv
841 recordMaxWorkers sv
840− doFork (runqueue sv) (handleChildException sv) >>= addThread sv
843+ doFork (workLoop sv) (handleChildException sv) >>= addThread sv
842-- | In contrast to pushWorker which always happens only from the consumer
845-- | In contrast to pushWorker which always happens only from the consumer
843-- thread, a pushWorkerPar can happen concurrently from multiple threads on the
846-- thread, a pushWorkerPar can happen concurrently from multiple threads on the
844-- producer side. So we need to use a thread safe modification of
847-- producer side. So we need to use a thread safe modification of
845−-- runningThreads. Alternatively, we can use a CreateThread event to avoid
848+-- workerThreads. Alternatively, we can use a CreateThread event to avoid
846-- using a CAS based modification.
849-- using a CAS based modification.
847{-# NOINLINE pushWorkerPar #-}
850{-# NOINLINE pushWorkerPar #-}
848pushWorkerPar :: MonadAsync m => SVar m a -> Stream m a -> m ()
851pushWorkerPar :: MonadAsync m => SVar m a -> Stream m a -> m ()
849pushWorkerPar sv m = do
852pushWorkerPar sv m = do
850− -- We do not use activeWorkers in case of ParallelVar but still there is no
853+ -- We do not use workerCount in case of ParallelVar but still there is no
851 -- harm in maintaining it correctly.
854 -- harm in maintaining it correctly.
852#ifdef DIAGNOSTICS
855#ifdef DIAGNOSTICS
853− liftIO $ atomicModifyIORefCAS_ (activeWorkers sv) $ \n -> n + 1
856+ liftIO $ atomicModifyIORefCAS_ (workerCount sv) $ \n -> n + 1
854 recordMaxWorkers sv
857 recordMaxWorkers sv
856 doFork (runOne sv m) (handleChildException sv) >>= modifyThread sv
859 doFork (runOne sv m) (handleChildException sv) >>= modifyThread sv
858−{-# INLINE workDone #-}
859−workDone :: MonadIO m => SVar m a -> m Bool
862− if (svarStyle sv == AheadVar)
864− (hp, _) <- liftIO $ readIORef (outputHeap sv)
865− return (H.size hp <= 0)
867− queueDone <- queueEmpty sv
868− return $ queueDone && heapDone
870maxWorkerLimit :: Int
861maxWorkerLimit :: Int
871maxWorkerLimit = 1500
862maxWorkerLimit = 1500
873dispatchWorker :: MonadAsync m => SVar m a -> m ()
864dispatchWorker :: MonadAsync m => SVar m a -> m ()
874dispatchWorker sv = do
865dispatchWorker sv = do
875− done <- workDone sv
866+ done <- liftIO $ isWorkDone sv
876 when (not done) $ do
867 when (not done) $ do
877 -- Note that the worker count is only decremented during event
868 -- Note that the worker count is only decremented during event
878 -- processing in fromStreamVar and therefore it is safe to read and
869 -- processing in fromStreamVar and therefore it is safe to read and
879 -- use it without a lock.
870 -- use it without a lock.
880− cnt <- liftIO $ readIORef $ activeWorkers sv
871+ cnt <- liftIO $ readIORef $ workerCount sv
881 -- Note that we may deadlock if the previous workers (tasks in the
872 -- Note that we may deadlock if the previous workers (tasks in the
882 -- stream) wait/depend on the future workers (tasks in the stream)
873 -- stream) wait/depend on the future workers (tasks in the stream)
883 -- executing. In that case we should either configure the maxWorker
874 -- executing. In that case we should either configure the maxWorker
@@ -890,7 +881,7 @@ sendWorkerWait :: MonadAsync m => SVar m a -> m ()
890sendWorkerWait sv = do
881sendWorkerWait sv = do
891 -- Note that we are guaranteed to have at least one outstanding worker when
882 -- Note that we are guaranteed to have at least one outstanding worker when
892 -- we enter this function. So if we sleep we are guaranteed to be woken up
883 -- we enter this function. So if we sleep we are guaranteed to be woken up
893− -- by a doorBell, when the worker exits.
884+ -- by a outputDoorBell, when the worker exits.
895 -- XXX we need a better way to handle this than hardcoded delays. The
886 -- XXX we need a better way to handle this than hardcoded delays. The
896 -- delays may be different for different systems.
887 -- delays may be different for different systems.
@@ -933,11 +924,11 @@ sendWorkerWait sv = do
934 -- XXX update this in the tutorial.
925 -- XXX update this in the tutorial.
936− -- register for the doorBell before we check the queue so that if we
927+ -- register for the outputDoorBell before we check the queue so that if we
937 -- sleep because the queue was empty we are guaranteed to get a
928 -- sleep because the queue was empty we are guaranteed to get a
938 -- doorbell on the next enqueue.
929 -- doorbell on the next enqueue.
940− liftIO $ atomicModifyIORefCAS_ (waitingForWork sv) $ const True
931+ liftIO $ atomicModifyIORefCAS_ (needDoorBell sv) $ const True
941 liftIO $ storeLoadBarrier
932 liftIO $ storeLoadBarrier
942 dispatchWorker sv
933 dispatchWorker sv
@@ -948,67 +939,78 @@ sendWorkerWait sv = do
948 -- least one outstanding worker. Otherwise we could be sleeping
939 -- least one outstanding worker. Otherwise we could be sleeping
949 -- forever.
940 -- forever.
951− done <- workDone sv
942+ done <- liftIO $ isWorkDone sv
954 liftIO $ withDBGMVar sv "sendWorkerWait: nothing to do"
945 liftIO $ withDBGMVar sv "sendWorkerWait: nothing to do"
955− $ takeMVar (doorBell sv)
946+ $ takeMVar (outputDoorBell sv)
956 (_, len) <- liftIO $ readIORef (outputQueue sv)
947 (_, len) <- liftIO $ readIORef (outputQueue sv)
957 when (len <= 0) $ sendWorkerWait sv
948 when (len <= 0) $ sendWorkerWait sv
958 else sendWorkerWait sv
949 else sendWorkerWait sv
960−-- | Pull a stream from an SVar.
951+{-# INLINE readOutputQRaw #-}
961−{-# NOINLINE fromStreamVar #-}
952+readOutputQRaw :: SVar m a -> IO ([ChildEvent a], Int)
962−fromStreamVar :: MonadAsync m => SVar m a -> Stream m a
953+readOutputQRaw sv = do
963−fromStreamVar sv = Stream $ \_ stp sng yld -> do
965− -- XXX we can set this in SVar
966− if svarStyle sv == ParallelVar
968− liftIO $ withDBGMVar sv "fromStreamVar: doorbell"
969− $ takeMVar (doorBell sv)
972− res@(_, len) <- readOutputQ sv
973− -- When there is no output seen we dispatch more workers to help
974− -- out if there is work pending in the work queue.
978− -- send a worker proactively, if needed, even before we start
979− -- processing the output. This may degrade single processor
980− -- perf but improves multi-processor, because of more
985− -- Reversing the output is important to guarantee that we process the
986− -- outputs in the same order as they were generated by the constituent
988− runStream (processEvents $ reverse list) Nothing stp sng yld
992− {-# INLINE readOutputQ #-}
993− readOutputQ svr = liftIO $ do
994− (list, len) <- atomicModifyIORefCAS (outputQueue svr) $
954+ (list, len) <- atomicModifyIORefCAS (outputQueue sv) $ \x -> (([],0), x)
996#ifdef DIAGNOSTICS
955#ifdef DIAGNOSTICS
997− oqLen <- readIORef (maxOutQSize svr)
956+ oqLen <- readIORef (maxOutQSize sv)
998− when (len > oqLen) $ writeIORef (maxOutQSize svr) len
957+ when (len > oqLen) $ writeIORef (maxOutQSize sv) len
1000− return (list, len)
959+ return (list, len)
961+readOutputQBounded :: MonadAsync m => SVar m a -> m [ChildEvent a]
962+readOutputQBounded sv = do
963+ (list, len) <- liftIO $ readOutputQRaw sv
964+ -- When there is no output seen we dispatch more workers to help
965+ -- out if there is work pending in the work queue.
969+ -- send a worker proactively, if needed, even before we start
970+ -- processing the output. This may degrade single processor
971+ -- perf but improves multi-processor, because of more
1002 sendWorker = do
978 sendWorker = do
1003− cnt <- liftIO $ readIORef $ activeWorkers sv
979+ cnt <- liftIO $ readIORef $ workerCount sv
1004 when (cnt <= 0) $ do
980 when (cnt <= 0) $ do
1005− done <- workDone sv
981+ done <- liftIO $ isWorkDone sv
1006 when (not done) $ pushWorker sv
982 when (not done) $ pushWorker sv
1008 {-# INLINE blockingRead #-}
984 {-# INLINE blockingRead #-}
1009 blockingRead = do
985 blockingRead = do
1010 sendWorkerWait sv
986 sendWorkerWait sv
987+ liftIO $ (readOutputQRaw sv >>= return . fst)
989+postProcessBounded :: MonadAsync m => SVar m a -> m Bool
990+postProcessBounded sv = do
991+ workersDone <- allThreadsDone sv
992+ -- There may still be work pending even if there are no workers
993+ -- pending because all the workers may return if the
994+ -- outputQueue becomes full. In that case send off a worker to
995+ -- kickstart the work again.
998+ r <- liftIO $ isWorkDone sv
999+ when (not r) $ pushWorker sv
1003+-- | Pull a stream from an SVar.
1004+{-# NOINLINE fromStreamVar #-}
1005+fromStreamVar :: MonadAsync m => SVar m a -> Stream m a
1006+fromStreamVar sv = Stream $ \_ stp sng yld -> do
1011− readOutputQ sv
1007+ list <- readOutputQ sv
1008+ -- Reversing the output is important to guarantee that we process the
1009+ -- outputs in the same order as they were generated by the constituent
1011+ runStream (processEvents $ reverse list) Nothing stp sng yld
1013 allDone stp = do
1015 allDone stp = do
1014#ifdef DIAGNOSTICS
1016#ifdef DIAGNOSTICS
@@ -1021,23 +1023,7 @@ fromStreamVar sv = Stream $ \_ stp sng yld -> do
1022 {-# INLINE processEvents #-}
1024 {-# INLINE processEvents #-}
1023 processEvents [] = Stream $ \_ stp sng yld -> do
1025 processEvents [] = Stream $ \_ stp sng yld -> do
1024− workersDone <- allThreadsDone sv
1026+ done <- postProcess sv
1026− -- XXX we can set this in SVar
1027− if svarStyle sv == ParallelVar
1028− then return workersDone
1030− -- There may still be work pending even if there are no workers
1031− -- pending because all the workers may return if the
1032− -- outputQueue becomes full. In that case send off a worker to
1033− -- kickstart the work again.
1037− when (not r) $ pushWorker sv
1042 then allDone stp
1028 then allDone stp
1043 else runStream (fromStreamVar sv) Nothing stp sng yld
1029 else runStream (fromStreamVar sv) Nothing stp sng yld
@@ -1047,14 +1033,12 @@ fromStreamVar sv = Stream $ \_ stp sng yld -> do
1047 case ev of
1033 case ev of
1048 ChildYield a -> yld a rest
1034 ChildYield a -> yld a rest
1049 ChildStop tid e -> do
1035 ChildStop tid e -> do
1050− if svarStyle sv == ParallelVar
1036+ accountThread sv tid
1051− then modifyThread sv tid
1052− else delThread sv tid
1053 case e of
1037 case e of
1054 Nothing -> runStream rest Nothing stp sng yld
1038 Nothing -> runStream rest Nothing stp sng yld
1055 Just ex -> throwM ex
1039 Just ex -> throwM ex
1057−getFifoSVar :: MonadIO m => SVarStyle -> IO (SVar m a)
1041+getFifoSVar :: MonadAsync m => SVarStyle -> IO (SVar m a)
1058getFifoSVar ctype = do
1042getFifoSVar ctype = do
1059 outQ <- newIORef ([], 0)
1043 outQ <- newIORef ([], 0)
1060 outQMv <- newEmptyMVar
1044 outQMv <- newEmptyMVar
@@ -1070,28 +1054,31 @@ getFifoSVar ctype = do
1070 maxWq <- newIORef 0
1054 maxWq <- newIORef 0
1072 let sv =
1056 let sv =
1073− SVar { outputQueue = outQ
1057+ SVar { outputQueue = outQ
1074− , doorBell = outQMv
1058+ , outputDoorBell = outQMv
1075− , outputHeap = undefined
1059+ , readOutputQ = readOutputQBounded sv
1076− , runningThreads = running
1060+ , postProcess = postProcessBounded sv
1077− , workQueue = undefined
1061+ , workerThreads = running
1078− , runqueue = runqueueFIFO sv q
1062+ , workLoop = workLoopFIFO sv q
1079− , enqueue = enqueueFIFO sv q
1063+ , enqueue = enqueueFIFO sv q
1080− , queueEmpty = liftIO $ nullQ q
1064+ , isWorkDone = nullQ q
1081− , waitingForWork = wfw
1065+ , needDoorBell = wfw
1082− , svarStyle = ctype
1066+ , svarStyle = ctype
1083− , activeWorkers = active
1067+ , workerCount = active
1068+ , accountThread = delThread sv
1084#ifdef DIAGNOSTICS
1069#ifdef DIAGNOSTICS
1070+ , aheadWorkQueue = undefined
1071+ , outputHeap = undefined
1085− , totalDispatches = disp
1072+ , totalDispatches = disp
1086− , maxWorkers = maxWrk
1073+ , maxWorkers = maxWrk
1087− , maxOutQSize = maxOq
1074+ , maxOutQSize = maxOq
1088− , maxHeapSize = maxHs
1075+ , maxHeapSize = maxHs
1089− , maxWorkQSize = maxWq
1076+ , maxWorkQSize = maxWq
1092 in return sv
1079 in return sv
1094−getLifoSVar :: MonadIO m => SVarStyle -> IO (SVar m a)
1081+getLifoSVar :: MonadAsync m => SVarStyle -> IO (SVar m a)
1095getLifoSVar ctype = do
1082getLifoSVar ctype = do
1096 outQ <- newIORef ([], 0)
1083 outQ <- newIORef ([], 0)
1097 outQMv <- newEmptyMVar
1084 outQMv <- newEmptyMVar
@@ -1106,36 +1093,42 @@ getLifoSVar ctype = do
1106 maxHs <- newIORef 0
1093 maxHs <- newIORef 0
1107 maxWq <- newIORef 0
1094 maxWq <- newIORef 0
1109− let checkEmpty = null <$> liftIO (readIORef q)
1096+ let checkEmpty = null <$> readIORef q
1110 let sv =
1097 let sv =
1111− SVar { outputQueue = outQ
1098+ SVar { outputQueue = outQ
1112− , doorBell = outQMv
1099+ , outputDoorBell = outQMv
1113− , outputHeap = undefined
1100+ , readOutputQ = readOutputQBounded sv
1114− , runningThreads = running
1101+ , postProcess = postProcessBounded sv
1115− , workQueue = undefined
1102+ , workerThreads = running
1116− , runqueue = runqueueLIFO sv q
1103+ , workLoop = workLoopLIFO sv q
1117− , enqueue = enqueueLIFO sv q
1104+ , enqueue = enqueueLIFO sv q
1118− , queueEmpty = checkEmpty
1105+ , isWorkDone = checkEmpty
1119− , waitingForWork = wfw
1106+ , needDoorBell = wfw
1120− , svarStyle = ctype
1107+ , svarStyle = ctype
1121− , activeWorkers = active
1108+ , workerCount = active
1109+ , accountThread = delThread sv
1122#ifdef DIAGNOSTICS
1110#ifdef DIAGNOSTICS
1111+ , aheadWorkQueue = undefined
1112+ , outputHeap = undefined
1123− , maxWorkers = maxWrk
1113+ , maxWorkers = maxWrk
1124− , totalDispatches = disp
1114+ , totalDispatches = disp
1125− , maxOutQSize = maxOq
1115+ , maxOutQSize = maxOq
1126− , maxHeapSize = maxHs
1116+ , maxHeapSize = maxHs
1127− , maxWorkQSize = maxWq
1117+ , maxWorkQSize = maxWq
1130 in return sv
1120 in return sv
1132−getParSVar :: SVarStyle -> IO (SVar m a)
1122+getAheadSVar :: MonadAsync m => SVarStyle -> IO (SVar m a)
1133−getParSVar style = do
1123+getAheadSVar style = do
1134 outQ <- newIORef ([], 0)
1124 outQ <- newIORef ([], 0)
1125+ outH <- newIORef (H.empty, 0)
1135 outQMv <- newEmptyMVar
1126 outQMv <- newEmptyMVar
1136 active <- newIORef 0
1127 active <- newIORef 0
1137 wfw <- newIORef False
1128 wfw <- newIORef False
1138 running <- newIORef S.empty
1129 running <- newIORef S.empty
1130+ q <- newIORef ([], -1)
1139#ifdef DIAGNOSTICS
1132#ifdef DIAGNOSTICS
1140 disp <- newIORef 0
1133 disp <- newIORef 0
1141 maxWrk <- newIORef 0
1134 maxWrk <- newIORef 0
@@ -1144,37 +1137,50 @@ getParSVar style = do
1144 maxWq <- newIORef 0
1137 maxWq <- newIORef 0
1146 let sv =
1139 let sv =
1147− SVar { outputQueue = outQ
1140+ SVar { outputQueue = outQ
1148− , doorBell = outQMv
1141+ , outputDoorBell = outQMv
1149− , outputHeap = undefined
1142+ , readOutputQ = readOutputQBounded sv
1150− , runningThreads = running
1143+ , postProcess = postProcessBounded sv
1151− , workQueue = undefined
1144+ , workerThreads = running
1152− , runqueue = undefined
1145+ , workLoop = workLoopAhead sv q outH
1153− , enqueue = undefined
1146+ , enqueue = enqueueAhead sv q
1154− , queueEmpty = undefined
1147+ , isWorkDone = isWorkDoneAhead q outH
1155− , waitingForWork = wfw
1148+ , needDoorBell = wfw
1156− , svarStyle = style
1149+ , svarStyle = style
1157− , activeWorkers = active
1150+ , workerCount = active
1151+ , accountThread = delThread sv
1158#ifdef DIAGNOSTICS
1152#ifdef DIAGNOSTICS
1153+ , aheadWorkQueue = q
1154+ , outputHeap = outH
1159− , totalDispatches = disp
1155+ , totalDispatches = disp
1160− , maxWorkers = maxWrk
1156+ , maxWorkers = maxWrk
1161− , maxOutQSize = maxOq
1157+ , maxOutQSize = maxOq
1162− , maxHeapSize = maxHs
1158+ , maxHeapSize = maxHs
1163− , maxWorkQSize = maxWq
1159+ , maxWorkQSize = maxWq
1166 in return sv
1162 in return sv
1166+ {-# INLINE isWorkDoneAhead #-}
1167+ isWorkDoneAhead q ref = do
1169+ (hp, _) <- readIORef ref
1170+ return (H.size hp <= 0)
1171+ queueDone <- checkEmpty q
1172+ return $ queueDone && heapDone
1175+ (xs, _) <- readIORef q
1168−getAheadSVar :: MonadIO m => SVarStyle -> IO (SVar m a)
1178+getParSVar :: MonadIO m => SVarStyle -> IO (SVar m a)
1169−getAheadSVar style = do
1179+getParSVar style = do
1170 outQ <- newIORef ([], 0)
1180 outQ <- newIORef ([], 0)
1171− outH <- newIORef (H.empty, 0)
1172 outQMv <- newEmptyMVar
1181 outQMv <- newEmptyMVar
1173 active <- newIORef 0
1182 active <- newIORef 0
1174− wfw <- newIORef False
1175 running <- newIORef S.empty
1183 running <- newIORef S.empty
1176− q <- newIORef ([], -1)
1178#ifdef DIAGNOSTICS
1184#ifdef DIAGNOSTICS
1179 disp <- newIORef 0
1185 disp <- newIORef 0
1180 maxWrk <- newIORef 0
1186 maxWrk <- newIORef 0
@@ -1182,33 +1188,37 @@ getAheadSVar style = do
1182 maxHs <- newIORef 0
1188 maxHs <- newIORef 0
1183 maxWq <- newIORef 0
1189 maxWq <- newIORef 0
1186− let checkEmpty = liftIO $ do
1187− (xs, _) <- readIORef q
1189 let sv =
1191 let sv =
1190− SVar { outputQueue = outQ
1192+ SVar { outputQueue = outQ
1191− , doorBell = outQMv
1193+ , outputDoorBell = outQMv
1192− , outputHeap = outH
1194+ , readOutputQ = readOutputQPar sv
1193− , runningThreads = running
1195+ , postProcess = allThreadsDone sv
1194− , workQueue = q
1196+ , workerThreads = running
1195− , runqueue = runqueueAhead sv q
1197+ , workLoop = undefined
1196− , enqueue = undefined
1198+ , enqueue = undefined
1197− , queueEmpty = checkEmpty
1199+ , isWorkDone = undefined
1198− , waitingForWork = wfw
1200+ , needDoorBell = undefined
1199− , svarStyle = style
1201+ , svarStyle = style
1200− , activeWorkers = active
1202+ , workerCount = active
1201−
1203+ , accountThread = modifyThread sv
1202#ifdef DIAGNOSTICS
1204#ifdef DIAGNOSTICS
1205+ , aheadWorkQueue = undefined
1206+ , outputHeap = undefined
1203− , totalDispatches = disp
1207+ , totalDispatches = disp
1204− , maxWorkers = maxWrk
1208+ , maxWorkers = maxWrk
1205− , maxOutQSize = maxOq
1209+ , maxOutQSize = maxOq
1206− , maxHeapSize = maxHs
1210+ , maxHeapSize = maxHs
1207− , maxWorkQSize = maxWq
1211+ , maxWorkQSize = maxWq
1210 in return sv
1214 in return sv
1218+ readOutputQPar sv = liftIO $ do
1219+ withDBGMVar sv "fromStreamVar: doorbell" $ takeMVar (outputDoorBell sv)
1220+ readOutputQRaw sv >>= return . fst
1212-- | Create a new empty SVar.
1222-- | Create a new empty SVar.
1213newEmptySVar :: MonadAsync m => SVarStyle -> m (SVar m a)
1223newEmptySVar :: MonadAsync m => SVarStyle -> m (SVar m a)
1214newEmptySVar style = do
1224newEmptySVar style = do
@@ -1227,25 +1237,18 @@ newStreamVar1 style m = do
1227 -- Note: We must have all the work on the queue before sending the
1237 -- Note: We must have all the work on the queue before sending the
1228 -- pushworker, otherwise the pushworker may exit before we even get a
1238 -- pushworker, otherwise the pushworker may exit before we even get a
1229 -- chance to push.
1239 -- chance to push.
1230− if style == ParallelVar
1231− then pushWorkerPar sv m
1233− liftIO $ (enqueue sv) m
1240+ liftIO $ enqueue sv m
1234− pushWorker sv
1241+ pushWorker sv
1235 return sv
1242 return sv
1237−-- | Create a new SVar and enqueue one stream computation on it.
1244+{-# INLINABLE newStreamVarPar #-}
1238−{-# INLINABLE newStreamVarAhead #-}
1239−newStreamVarAhead :: MonadAsync m => Stream m a -> m (SVar m a)
1245+newStreamVarPar :: MonadAsync m => Stream m a -> m (SVar m a)
1240−newStreamVarAhead m = do
1246+newStreamVarPar m = do
1241− sv <- newEmptySVar AheadVar
1247+ sv <- newEmptySVar ParallelVar
1242− -- Note: We must have all the work on the queue before sending the
1248+ pushWorkerPar sv m
1243− -- pushworker, otherwise the pushworker may exit before we even get a
1245− liftIO $ enqueueAhead sv (workQueue sv) m
1247 return sv
1249 return sv
1251+-- XXX this errors out for Parallel/Ahead SVars
1249-- | Write a stream to an 'SVar' in a non-blocking manner. The stream can then
1252-- | Write a stream to an 'SVar' in a non-blocking manner. The stream can then
1250-- be read back from the SVar using 'fromSVar'.
1253-- be read back from the SVar using 'fromSVar'.
1251toStreamVar :: MonadAsync m => SVar m a -> Stream m a -> m ()
1254toStreamVar :: MonadAsync m => SVar m a -> Stream m a -> m ()
@@ -1324,7 +1327,7 @@ toStreamVar sv m = do
1324forkSVarAsync :: MonadAsync m => SVarStyle -> Stream m a -> Stream m a -> Stream m a
1327forkSVarAsync :: MonadAsync m => SVarStyle -> Stream m a -> Stream m a -> Stream m a
1325forkSVarAsync style m1 m2 = Stream $ \_ stp sng yld -> do
1328forkSVarAsync style m1 m2 = Stream $ \_ stp sng yld -> do
1326 sv <- newStreamVar1 style (concurrently m1 m2)
1329 sv <- newStreamVar1 style (concurrently m1 m2)
1327− (runStream (fromStreamVar sv)) Nothing stp sng yld
1330+ runStream (fromStreamVar sv) Nothing stp sng yld
1329 concurrently ma mb = Stream $ \svr stp sng yld -> do
1332 concurrently ma mb = Stream $ \svr stp sng yld -> do
1330 liftIO $ enqueue (fromJust svr) mb
1333 liftIO $ enqueue (fromJust svr) mb
@@ -1336,33 +1339,18 @@ joinStreamVarAsync :: MonadAsync m
1336joinStreamVarAsync style m1 m2 = Stream $ \svr stp sng yld ->
1339joinStreamVarAsync style m1 m2 = Stream $ \svr stp sng yld ->
1337 case svr of
1340 case svr of
1338 Just sv | svarStyle sv == style ->
1341 Just sv | svarStyle sv == style ->
1339− liftIO ((enqueue sv) m2) >> (runStream m1) svr stp sng yld
1342+ liftIO (enqueue sv m2) >> (runStream m1) svr stp sng yld
1340 _ -> runStream (forkSVarAsync style m1 m2) Nothing stp sng yld
1343 _ -> runStream (forkSVarAsync style m1 m2) Nothing stp sng yld
1342−{-# NOINLINE forkSVarPar #-}
1345+-- The only difference between forkSVarAsync and this is that we run the left
1343−forkSVarPar :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
1346+-- computation without a shared SVar.
1344−forkSVarPar m r = Stream $ \_ stp sng yld -> do
1345− sv <- newEmptySVar ParallelVar
1348− (runStream (fromStreamVar sv)) Nothing stp sng yld
1350−{-# INLINE joinStreamVarPar #-}
1351−joinStreamVarPar :: MonadAsync m
1352− => SVarStyle -> Stream m a -> Stream m a -> Stream m a
1353−joinStreamVarPar style m1 m2 = Stream $ \svr stp sng yld ->
1355− Just sv | svarStyle sv == style -> do
1356− pushWorkerPar sv m1 >> (runStream m2) svr stp sng yld
1357− _ -> runStream (forkSVarPar m1 m2) Nothing stp sng yld
1359forkSVarAhead :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
1347forkSVarAhead :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
1360forkSVarAhead m1 m2 = Stream $ \_ stp sng yld -> do
1348forkSVarAhead m1 m2 = Stream $ \_ stp sng yld -> do
1361− sv <- newStreamVarAhead (concurrently m1 m2)
1349+ sv <- newStreamVar1 AheadVar (concurrently m1 m2)
1362 (runStream (fromStreamVar sv)) Nothing stp sng yld
1350 (runStream (fromStreamVar sv)) Nothing stp sng yld
1364 concurrently ma mb = Stream $ \svr stp sng yld -> do
1352 concurrently ma mb = Stream $ \svr stp sng yld -> do
1365− liftIO $ enqueueAhead (fromJust svr) (workQueue (fromJust svr)) mb
1353+ liftIO $ enqueue (fromJust svr) mb
1366 (runStream ma) Nothing stp sng yld
1354 (runStream ma) Nothing stp sng yld
1368{-# INLINE ahead #-}
1356{-# INLINE ahead #-}
@@ -1370,18 +1358,29 @@ ahead :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
1370ahead m1 m2 = Stream $ \svr stp sng yld -> do
1358ahead m1 m2 = Stream $ \svr stp sng yld -> do
1371 case svr of
1359 case svr of
1372 Just sv | svarStyle sv == AheadVar -> do
1360 Just sv | svarStyle sv == AheadVar -> do
1373− liftIO $ enqueueAhead sv (workQueue sv) m2
1361+ liftIO $ enqueue sv m2
1374 -- Always run the left side on a new SVar to avoid complexity in
1362 -- Always run the left side on a new SVar to avoid complexity in
1375 -- sequencing results. This means the left side cannot further
1363 -- sequencing results. This means the left side cannot further
1376 -- split into more ahead computations on the same SVar.
1364 -- split into more ahead computations on the same SVar.
1377 (runStream m1) Nothing stp sng yld
1365 (runStream m1) Nothing stp sng yld
1378 _ -> runStream (forkSVarAhead m1 m2) Nothing stp sng yld
1366 _ -> runStream (forkSVarAhead m1 m2) Nothing stp sng yld
1380−-- | XXX we can implement it more efficienty by directly implementing instead
1368+{-# NOINLINE forkSVarPar #-}
1381−-- of combining streams using ahead.
1369+forkSVarPar :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
1382−{-# INLINE consMAhead #-}
1370+forkSVarPar m r = Stream $ \_ stp sng yld -> do
1383−consMAhead :: MonadAsync m => m a -> Stream m a -> Stream m a
1371+ sv <- newEmptySVar ParallelVar
1384−consMAhead m r = once m `ahead` r
1372+ pushWorkerPar sv m
1374+ (runStream (fromStreamVar sv)) Nothing stp sng yld
1376+{-# INLINE joinStreamVarPar #-}
1377+joinStreamVarPar :: MonadAsync m
1378+ => SVarStyle -> Stream m a -> Stream m a -> Stream m a
1379+joinStreamVarPar style m1 m2 = Stream $ \svr stp sng yld ->
1381+ Just sv | svarStyle sv == style -> do
1382+ pushWorkerPar sv m1 >> (runStream m2) svr stp sng yld
1383+ _ -> runStream (forkSVarPar m1 m2) Nothing stp sng yld
1386------------------------------------------------------------------------------
1385------------------------------------------------------------------------------
1387-- Semigroup and Monoid style compositions for parallel actions
1386-- Semigroup and Monoid style compositions for parallel actions
@@ -1407,6 +1406,12 @@ wAsync = joinStreamVarAsync WAsyncVar
1407consMWAsync :: MonadAsync m => m a -> Stream m a -> Stream m a
1406consMWAsync :: MonadAsync m => m a -> Stream m a -> Stream m a
1408consMWAsync m r = once m `wAsync` r
1407consMWAsync m r = once m `wAsync` r
1409+-- | XXX we can implement it more efficienty by directly implementing instead
1410+-- of combining streams using ahead.
1411+{-# INLINE consMAhead #-}
1412+consMAhead :: MonadAsync m => m a -> Stream m a -> Stream m a
1413+consMAhead m r = once m `ahead` r
1410{-# INLINE parallel #-}
1415{-# INLINE parallel #-}
1411parallel :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
1416parallel :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
1412parallel = joinStreamVarPar ParallelVar
1417parallel = joinStreamVarPar ParallelVar
@@ -1441,19 +1446,18 @@ _alt m1 m2 = Stream $ \_ stp sng yld ->
1441------------------------------------------------------------------------------
1446------------------------------------------------------------------------------
1443applyWith :: MonadAsync m
1448applyWith :: MonadAsync m
1444− => SVarStyle -> (Stream m a -> Stream m b) -> Stream m a -> Stream m b
1449+ => (Stream m a -> Stream m b) -> Stream m a -> Stream m b
1445−applyWith style f m = Stream $ \svr stp sng yld -> do
1450+applyWith f m = Stream $ \svr stp sng yld -> do
1446− sv <- newStreamVar1 style m
1451+ sv <- newStreamVarPar m
1447 runStream (f $ fromStreamVar sv) svr stp sng yld
1452 runStream (f $ fromStreamVar sv) svr stp sng yld
1449------------------------------------------------------------------------------
1454------------------------------------------------------------------------------
1450-- Stream runner function application
1455-- Stream runner function application
1451------------------------------------------------------------------------------
1456------------------------------------------------------------------------------
1453−runWith :: MonadAsync m
1458+runWith :: MonadAsync m => (Stream m a -> m b) -> Stream m a -> m b
1454− => SVarStyle -> (Stream m a -> m b) -> Stream m a -> m b
1455−runWith style f m = do
1459+runWith f m = do
1456− sv <- newStreamVar1 style m
1460+ sv <- newStreamVarPar m
1457 f $ fromStreamVar sv
1461 f $ fromStreamVar sv
1459------------------------------------------------------------------------------
1463------------------------------------------------------------------------------