@@ -58,8 +58,7 @@ module Streamly.Core
5858
5959 -- * Concurrent Stream Vars (SVars)
6060 , SVar
61− , SVarStyle (..)
62− , newStreamVar1
61+ , newStreamVarPar
6362 , fromStreamVar
6463 , toStreamVar
6564 )
@@ -130,7 +129,6 @@ data AheadHeapEntry m a =
130129-- State threaded around the monad for thread management
131130------------------------------------------------------------------------------
132131
133−-- XXX use a separate data structure for each type of SVar
134132-- | Identify the type of the SVar. Two computations using the same style can
135133-- be scheduled on the same SVar.
136134data SVarStyle =
@@ -149,10 +147,10 @@ data SVarStyle =
149147-- However, it ensures that the reader can read the results at whatever paces
150148-- it wants to read. The SVar monitors and adapts to the consumer's pace.
151149--
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
153151-- stream tasks to be picked and run by a pool of worker threads. It has an
154152-- 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
156154-- consumer thread about availability of new results in the output queue. More
157155-- workers are added to the SVar by 'fromStreamVar' on demand if the output
158156-- produced is not keeping pace with the consumer. On bounded SVars, workers
@@ -171,34 +169,31 @@ data SVar m a =
171169
172170 -- Shared output queue (events, length)
173171 , outputQueue :: IORef ([ChildEvent a], Int)
174− , doorBell :: MVar () -- signal the consumer about output
175−
176− -- Output synchronization mechanism for Ahead streams (Ahead and
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))
183− , Int
184− )
172+ , outputDoorBell :: MVar () -- signal the consumer about output
173+ , readOutputQ :: m [ChildEvent a]
174+ , postProcess :: m Bool
185175
186− -- Shared work queue (stream, seqNo)
187− , workQueue :: IORef ([Stream m a], Int)
176+ -- Used only by bounded SVar types
188177 , enqueue :: Stream m a -> IO ()
189− , queueEmpty :: m Bool
190− , waitingForWork :: IORef Bool
191− , runqueue :: m ()
178+ , isWorkDone :: IO Bool
179+ , needDoorBell :: IORef Bool
180+ , workLoop :: m ()
192181
193182 -- Shared, thread tracking
194− , runningThreads :: IORef (Set ThreadId)
195− , activeWorkers :: IORef Int
183+ , workerThreads :: IORef (Set ThreadId)
184+ , workerCount :: IORef Int
185+ , accountThread :: ThreadId -> m ()
196186#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)
197192 , totalDispatches :: IORef Int
198− , maxWorkers :: IORef Int
199− , maxOutQSize :: IORef Int
200− , maxHeapSize :: IORef Int
201− , maxWorkQSize :: IORef Int
193+ , maxWorkers :: IORef Int
194+ , maxOutQSize :: IORef Int
195+ , maxHeapSize :: IORef Int
196+ , maxWorkQSize :: IORef Int
202197#endif
203198 }
204199
@@ -208,12 +203,12 @@ dumpSVar :: SVar m a -> IO String
208203dumpSVar sv = do
209204 tid <- myThreadId
210205 (oqList, oqLen) <- readIORef $ outputQueue sv
211− db <- tryTakeMVar $ doorBell sv
206+ db <- tryTakeMVar $ outputDoorBell sv
212207 aheadDump <-
213208 if svarStyle sv == AheadVar
214209 then do
215210 (oheap, oheapSeq) <- readIORef $ outputHeap sv
216− (wq, wqSeq) <- readIORef $ workQueue sv
211+ (wq, wqSeq) <- readIORef $ aheadWorkQueue sv
217212 maxHp <- readIORef $ maxHeapSize sv
218213 return $ unlines
219214 [ "heap length = " ++ show (H.size oheap)
@@ -224,26 +219,25 @@ dumpSVar sv = do
224219 ]
225220 else return []
226221
227− waiting <- readIORef $ waitingForWork sv
228− rthread <- readIORef $ runningThreads sv
229− workers <- readIORef $ activeWorkers sv
222+ waiting <- readIORef $ needDoorBell sv
223+ rthread <- readIORef $ workerThreads sv
224+ workers <- readIORef $ workerCount sv
230225 maxWrk <- readIORef $ maxWorkers sv
231226 dispatches <- readIORef $ totalDispatches sv
232227 maxOq <- readIORef $ maxOutQSize sv
233− -- XXX queueEmpty should be made IO return type
234228
235229 return $ unlines
236230 [ "tid = " ++ show tid
237231 , "style = " ++ show (svarStyle sv)
238232 , "outputQueue length computed = " ++ show (length oqList)
239233 , "outputQueue length maintained = " ++ show oqLen
240− , "output doorBell = " ++ show db
234+ , "output outputDoorBell = " ++ show db
241235 , "total dispatches = " ++ show dispatches
242236 , "max workers = " ++ show maxWrk
243237 , "max outQSize = " ++ show maxOq
244238 ]
245239 ++ aheadDump ++ unlines
246− [ "waitingForWork = " ++ show waiting
240+ [ "needDoorBell = " ++ show waiting
247241 , "running threads = " ++ show rthread
248242 , "running thread count = " ++ show workers
249243 ]
@@ -439,13 +433,13 @@ send sv msg = do
439433 -- to read the queue again and find it empty.
440434 -- The important point is that the consumer is guaranteed to receive a
441435 -- doorbell if something was added to the queue after it empties it.
442− void $ tryPutMVar (doorBell sv) ()
436+ void $ tryPutMVar (outputDoorBell sv) ()
443437 return (len < maxOutputQLen)
444438
445439{-# NOINLINE sendStop #-}
446440sendStop :: SVar m a -> IO ()
447441sendStop sv = do
448− liftIO $ atomicModifyIORefCAS_ (activeWorkers sv) $ \n -> n - 1
442+ liftIO $ atomicModifyIORefCAS_ (workerCount sv) $ \n -> n - 1
449443 myThreadId >>= \tid -> void $ send sv (ChildStop tid Nothing)
450444
451445-------------------------------------------------------------------------------
@@ -461,18 +455,18 @@ enqueueLIFO :: SVar m a -> IORef [Stream m a] -> Stream m a -> IO ()
461455enqueueLIFO sv q m = do
462456 atomicModifyIORefCAS_ q $ \ms -> m : ms
463457 storeLoadBarrier
464− w <- readIORef $ waitingForWork sv
458+ w <- readIORef $ needDoorBell sv
465459 when w $ do
466460 -- Note: the sequence of operations is important for correctness here.
467461 -- 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
469463 -- then we may set the flag to False to later making the consumer lose
470− -- the flag, even without receiving a doorBell.
471− atomicModifyIORefCAS_ (waitingForWork sv) (const False)
472− void $ tryPutMVar (doorBell sv) ()
464+ -- the flag, even without receiving a outputDoorBell.
465+ atomicModifyIORefCAS_ (needDoorBell sv) (const False)
466+ void $ tryPutMVar (outputDoorBell sv) ()
473467
474−runqueueLIFO :: MonadIO m => SVar m a -> IORef [Stream m a] -> m ()
475−runqueueLIFO sv q = run
468+workLoopLIFO :: MonadIO m => SVar m a -> IORef [Stream m a] -> m ()
469+workLoopLIFO sv q = run
476470
477471 where
478472
@@ -508,18 +502,18 @@ enqueueFIFO :: SVar m a -> LinkedQueue (Stream m a) -> Stream m a -> IO ()
508502enqueueFIFO sv q m = do
509503 pushL q m
510504 storeLoadBarrier
511− w <- readIORef $ waitingForWork sv
505+ w <- readIORef $ needDoorBell sv
512506 when w $ do
513507 -- Note: the sequence of operations is important for correctness here.
514508 -- 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
516510 -- then we may set the flag to False to later making the consumer lose
517− -- the flag, even without receiving a doorBell.
518− atomicModifyIORefCAS_ (waitingForWork sv) (const False)
519− void $ tryPutMVar (doorBell sv) ()
511+ -- the flag, even without receiving a outputDoorBell.
512+ atomicModifyIORefCAS_ (needDoorBell sv) (const False)
513+ void $ tryPutMVar (outputDoorBell sv) ()
520514
521−runqueueFIFO :: MonadIO m => SVar m a -> LinkedQueue (Stream m a) -> m ()
522−runqueueFIFO sv q = run
515+workLoopFIFO :: MonadIO m => SVar m a -> LinkedQueue (Stream m a) -> m ()
516+workLoopFIFO sv q = run
523517
524518 where
525519
@@ -620,15 +614,15 @@ enqueueAhead sv q m = do
620614 ([], n) -> ([m], n + 1) -- increment sequence
621615 _ -> error "not empty"
622616 storeLoadBarrier
623− w <- readIORef $ waitingForWork sv
617+ w <- readIORef $ needDoorBell sv
624618 when w $ do
625619 -- Note: the sequence of operations is important for correctness here.
626620 -- 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
628622 -- then we may set the flag to False to later making the consumer lose
629− -- the flag, even without receiving a doorBell.
630− atomicModifyIORefCAS_ (waitingForWork sv) (const False)
631− void $ tryPutMVar (doorBell sv) ()
623+ -- the flag, even without receiving a outputDoorBell.
624+ atomicModifyIORefCAS_ (needDoorBell sv) (const False)
625+ void $ tryPutMVar (outputDoorBell sv) ()
632626
633627-- Normally the thread that has the token should never go away. The token gets
634628-- handed over to another thread, but someone or the other has the token at any
@@ -650,15 +644,19 @@ enqueueAhead sv q m = do
650644--
651645-- XXX review for livelock
652646--
653−runqueueAhead :: MonadIO m => SVar m a -> IORef ([Stream m a], Int) -> m ()
654−runqueueAhead sv q = runHeap
647+workLoopAhead :: MonadIO m
648+ => SVar m a
649+ -> IORef ([Stream m a], Int)
650+ -> IORef (Heap (Entry Int (AheadHeapEntry m a)) , Int)
651+ -> m ()
652+workLoopAhead sv q heap = runHeap
655653
656654 where
657655
658656 maxHeap = 1500
659657
660658 toHeap seqNo ent = do
661− hp <- liftIO $ atomicModifyIORefCAS (outputHeap sv) $ \(h, snum) ->
659+ hp <- liftIO $ atomicModifyIORefCAS heap $ \(h, snum) ->
662660 ((H.insert (Entry seqNo ent) h, snum), h)
663661 if H.size hp <= maxHeap
664662 then runHeap
@@ -672,7 +670,7 @@ runqueueAhead sv q = runHeap
672670 if continue
673671 then runQueueToken seqNo
674672 else liftIO $ do
675− atomicModifyIORefCAS_ (outputHeap sv) $ \(h, _) -> (h, seqNo + 1)
673+ atomicModifyIORefCAS_ heap $ \(h, _) -> (h, seqNo + 1)
676674 sendStop sv
677675
678676 yieldOutput seqNo a r = do
@@ -682,7 +680,7 @@ runqueueAhead sv q = runHeap
682680 (singleOutput seqNo)
683681 (yieldOutput seqNo)
684682 else liftIO $ do
685− atomicModifyIORefCAS_ (outputHeap sv) $ \(h, _) ->
683+ atomicModifyIORefCAS_ heap $ \(h, _) ->
686684 (H.insert (Entry seqNo (AheadEntryStream r)) h, seqNo)
687685 sendStop sv
688686
@@ -691,7 +689,7 @@ runqueueAhead sv q = runHeap
691689 work <- dequeue
692690 case work of
693691 Nothing -> do
694− liftIO $ atomicModifyIORefCAS_ (outputHeap sv) $ \(h, _) ->
692+ liftIO $ atomicModifyIORefCAS_ heap $ \(h, _) ->
695693 (h, prevSeqNo + 1)
696694 runHeap
697695 Just (m, seqNo) -> do
@@ -701,7 +699,7 @@ runqueueAhead sv q = runHeap
701699 (singleOutput seqNo)
702700 (yieldOutput seqNo)
703701 else do
704− liftIO $ atomicModifyIORefCAS_ (outputHeap sv) $ \(h, _) ->
702+ liftIO $ atomicModifyIORefCAS_ heap $ \(h, _) ->
705703 (h, prevSeqNo + 1)
706704 (runStream m) (Just sv) runHeap
707705 (singleToHeap seqNo)
@@ -726,13 +724,13 @@ runqueueAhead sv q = runHeap
726724#ifdef DIAGNOSTICS
727725 liftIO $ do
728726 maxHp <- readIORef (maxHeapSize sv)
729− (hp, _) <- readIORef (outputHeap sv)
727+ (hp, _) <- readIORef heap
730728 when (H.size hp > maxHp) $ writeIORef (maxHeapSize sv) (H.size hp)
731729#endif
732− ent <- liftIO $ dequeueFromHeap (outputHeap sv)
730+ ent <- liftIO $ dequeueFromHeap heap
733731 case ent of
734732 Nothing -> do
735− done <- queueEmpty sv
733+ done <- queueEmpty q
736734 if done
737735 then liftIO $ sendStop sv
738736 else runQueueNoToken
@@ -744,6 +742,10 @@ runqueueAhead sv q = runHeap
744742 (singleOutput seqNo)
745743 (yieldOutput seqNo)
746744
745+ queueEmpty qu = liftIO $ do
746+ (xs, _) <- readIORef qu
747+ return $ null xs
748+
747749 dequeue = liftIO $ do
748750 atomicModifyIORefCAS q $ \case
749751 ([], n) -> (([], n), Nothing)
@@ -782,14 +784,15 @@ runqueueAhead sv q = runHeap
782784{-# NOINLINE addThread #-}
783785addThread :: MonadIO m => SVar m a -> ThreadId -> m ()
784786addThread sv tid =
785− liftIO $ modifyIORef (runningThreads sv) (S.insert tid)
787+ liftIO $ modifyIORef (workerThreads sv) (S.insert tid)
786788
787−-- This is cheaper than modifyThread because we do not have to send a doorBell
788−-- This can make a difference when more workers are being dispatched.
789+-- This is cheaper than modifyThread because we do not have to send a
790+-- outputDoorBell This can make a difference when more workers are being
791+-- dispatched.
789792{-# INLINE delThread #-}
790793delThread :: MonadIO m => SVar m a -> ThreadId -> m ()
791794delThread sv tid =
792− liftIO $ modifyIORef (runningThreads sv) $ (\s -> S.delete tid s)
795+ liftIO $ modifyIORef (workerThreads sv) $ (\s -> S.delete tid s)
793796
794797-- If present then delete else add. This takes care of out of order add and
795798-- delete i.e. a delete arriving before we even added a thread.
@@ -798,22 +801,22 @@ delThread sv tid =
798801{-# INLINE modifyThread #-}
799802modifyThread :: MonadIO m => SVar m a -> ThreadId -> m ()
800803modifyThread sv tid = do
801− changed <- liftIO $ atomicModifyIORefCAS (runningThreads sv) $ \old ->
804+ changed <- liftIO $ atomicModifyIORefCAS (workerThreads sv) $ \old ->
802805 if (S.member tid old)
803806 then let new = (S.delete tid old) in (new, new)
804807 else let new = (S.insert tid old) in (new, old)
805808 if null changed
806809 then liftIO $ do
807810 writeBarrier
808− void $ tryPutMVar (doorBell sv) ()
811+ void $ tryPutMVar (outputDoorBell sv) ()
809812 else return ()
810813
811814-- | 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
813816-- not be empty.
814817{-# INLINE allThreadsDone #-}
815818allThreadsDone :: 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)
817820
818821{-# NOINLINE handleChildException #-}
819822handleChildException :: SVar m a -> SomeException -> IO ()
@@ -824,7 +827,7 @@ handleChildException sv e = do
824827#ifdef DIAGNOSTICS
825828recordMaxWorkers :: MonadIO m => SVar m a -> m ()
826829recordMaxWorkers sv = liftIO $ do
827− active <- readIORef (activeWorkers sv)
830+ active <- readIORef (workerCount sv)
828831 maxWrk <- readIORef (maxWorkers sv)
829832 when (active > maxWrk) $ writeIORef (maxWorkers sv) active
830833 modifyIORef (totalDispatches sv) (+1)
@@ -833,51 +836,39 @@ recordMaxWorkers sv = liftIO $ do
833836{-# NOINLINE pushWorker #-}
834837pushWorker :: MonadAsync m => SVar m a -> m ()
835838pushWorker sv = do
836− liftIO $ atomicModifyIORefCAS_ (activeWorkers sv) $ \n -> n + 1
839+ liftIO $ atomicModifyIORefCAS_ (workerCount sv) $ \n -> n + 1
837840#ifdef DIAGNOSTICS
838841 recordMaxWorkers sv
839842#endif
840− doFork (runqueue sv) (handleChildException sv) >>= addThread sv
843+ doFork (workLoop sv) (handleChildException sv) >>= addThread sv
841844
842845-- | In contrast to pushWorker which always happens only from the consumer
843846-- thread, a pushWorkerPar can happen concurrently from multiple threads on the
844847-- 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
846849-- using a CAS based modification.
847850{-# NOINLINE pushWorkerPar #-}
848851pushWorkerPar :: MonadAsync m => SVar m a -> Stream m a -> m ()
849852pushWorkerPar 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
851854 -- harm in maintaining it correctly.
852855#ifdef DIAGNOSTICS
853− liftIO $ atomicModifyIORefCAS_ (activeWorkers sv) $ \n -> n + 1
856+ liftIO $ atomicModifyIORefCAS_ (workerCount sv) $ \n -> n + 1
854857 recordMaxWorkers sv
855858#endif
856859 doFork (runOne sv m) (handleChildException sv) >>= modifyThread sv
857860
858−{-# INLINE workDone #-}
859−workDone :: MonadIO m => SVar m a -> m Bool
860−workDone sv = do
861− heapDone <-
862− if (svarStyle sv == AheadVar)
863− then do
864− (hp, _) <- liftIO $ readIORef (outputHeap sv)
865− return (H.size hp <= 0)
866− else return True
867− queueDone <- queueEmpty sv
868− return $ queueDone && heapDone
869−
870861maxWorkerLimit :: Int
871862maxWorkerLimit = 1500
872863
873864dispatchWorker :: MonadAsync m => SVar m a -> m ()
874865dispatchWorker sv = do
875− done <- workDone sv
866+ done <- liftIO $ isWorkDone sv
876867 when (not done) $ do
877868 -- Note that the worker count is only decremented during event
878869 -- processing in fromStreamVar and therefore it is safe to read and
879870 -- use it without a lock.
880− cnt <- liftIO $ readIORef $ activeWorkers sv
871+ cnt <- liftIO $ readIORef $ workerCount sv
881872 -- Note that we may deadlock if the previous workers (tasks in the
882873 -- stream) wait/depend on the future workers (tasks in the stream)
883874 -- executing. In that case we should either configure the maxWorker
@@ -890,7 +881,7 @@ sendWorkerWait :: MonadAsync m => SVar m a -> m ()
890881sendWorkerWait sv = do
891882 -- Note that we are guaranteed to have at least one outstanding worker when
892883 -- 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.
894885
895886 -- XXX we need a better way to handle this than hardcoded delays. The
896887 -- delays may be different for different systems.
@@ -933,11 +924,11 @@ sendWorkerWait sv = do
933924 --
934925 -- XXX update this in the tutorial.
935926
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
937928 -- sleep because the queue was empty we are guaranteed to get a
938929 -- doorbell on the next enqueue.
939930
940− liftIO $ atomicModifyIORefCAS_ (waitingForWork sv) $ const True
931+ liftIO $ atomicModifyIORefCAS_ (needDoorBell sv) $ const True
941932 liftIO $ storeLoadBarrier
942933 dispatchWorker sv
943934
@@ -948,67 +939,78 @@ sendWorkerWait sv = do
948939 -- least one outstanding worker. Otherwise we could be sleeping
949940 -- forever.
950941
951− done <- workDone sv
942+ done <- liftIO $ isWorkDone sv
952943 if done
953944 then do
954945 liftIO $ withDBGMVar sv "sendWorkerWait: nothing to do"
955− $ takeMVar (doorBell sv)
946+ $ takeMVar (outputDoorBell sv)
956947 (_, len) <- liftIO $ readIORef (outputQueue sv)
957948 when (len <= 0) $ sendWorkerWait sv
958949 else sendWorkerWait sv
959950
960−-- | Pull a stream from an SVar.
961−{-# NOINLINE fromStreamVar #-}
962−fromStreamVar :: MonadAsync m => SVar m a -> Stream m a
963−fromStreamVar sv = Stream $ \_ stp sng yld -> do
964− (list, _) <-
965− -- XXX we can set this in SVar
966− if svarStyle sv == ParallelVar
967− then do
968− liftIO $ withDBGMVar sv "fromStreamVar: doorbell"
969− $ takeMVar (doorBell sv)
970− readOutputQ sv
971− else do
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.
975− if len <= 0
976− then blockingRead
977− else do
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
981− -- parallelism
982− sendWorker
983− return res
984−
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
987− -- streams.
988− runStream (processEvents $ reverse list) Nothing stp sng yld
989−
990− where
991−
992− {-# INLINE readOutputQ #-}
993− readOutputQ svr = liftIO $ do
994− (list, len) <- atomicModifyIORefCAS (outputQueue svr) $
995− \x -> (([],0), x)
951+{-# INLINE readOutputQRaw #-}
952+readOutputQRaw :: SVar m a -> IO ([ChildEvent a], Int)
953+readOutputQRaw sv = do
954+ (list, len) <- atomicModifyIORefCAS (outputQueue sv) $ \x -> (([],0), x)
996955#ifdef DIAGNOSTICS
997− oqLen <- readIORef (maxOutQSize svr)
998− when (len > oqLen) $ writeIORef (maxOutQSize svr) len
956+ oqLen <- readIORef (maxOutQSize sv)
957+ when (len > oqLen) $ writeIORef (maxOutQSize sv) len
999958#endif
1000− return (list, len)
959+ return (list, len)
960+
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.
966+ if len <= 0
967+ then blockingRead
968+ else do
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
972+ -- parallelism
973+ sendWorker
974+ return list
975+
976+ where
1001977
1002978 sendWorker = do
1003− cnt <- liftIO $ readIORef $ activeWorkers sv
979+ cnt <- liftIO $ readIORef $ workerCount sv
1004980 when (cnt <= 0) $ do
1005− done <- workDone sv
981+ done <- liftIO $ isWorkDone sv
1006982 when (not done) $ pushWorker sv
1007983
1008984 {-# INLINE blockingRead #-}
1009985 blockingRead = do
1010986 sendWorkerWait sv
1011− readOutputQ sv
987+ liftIO $ (readOutputQRaw sv >>= return . fst)
988+
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.
996+ if workersDone
997+ then do
998+ r <- liftIO $ isWorkDone sv
999+ when (not r) $ pushWorker sv
1000+ return r
1001+ else return False
1002+
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
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
1010+ -- streams.
1011+ runStream (processEvents $ reverse list) Nothing stp sng yld
1012+
1013+ where
10121014
10131015 allDone stp = do
10141016#ifdef DIAGNOSTICS
@@ -1021,23 +1023,7 @@ fromStreamVar sv = Stream $ \_ stp sng yld -> do
10211023
10221024 {-# INLINE processEvents #-}
10231025 processEvents [] = Stream $ \_ stp sng yld -> do
1024− workersDone <- allThreadsDone sv
1025− done <-
1026− -- XXX we can set this in SVar
1027− if svarStyle sv == ParallelVar
1028− then return workersDone
1029− else
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.
1034− if workersDone
1035− then do
1036− r <- workDone sv
1037− when (not r) $ pushWorker sv
1038− return r
1039− else return False
1040−
1026+ done <- postProcess sv
10411027 if done
10421028 then allDone stp
10431029 else runStream (fromStreamVar sv) Nothing stp sng yld
@@ -1047,14 +1033,12 @@ fromStreamVar sv = Stream $ \_ stp sng yld -> do
10471033 case ev of
10481034 ChildYield a -> yld a rest
10491035 ChildStop tid e -> do
1050− if svarStyle sv == ParallelVar
1051− then modifyThread sv tid
1052− else delThread sv tid
1036+ accountThread sv tid
10531037 case e of
10541038 Nothing -> runStream rest Nothing stp sng yld
10551039 Just ex -> throwM ex
10561040
1057−getFifoSVar :: MonadIO m => SVarStyle -> IO (SVar m a)
1041+getFifoSVar :: MonadAsync m => SVarStyle -> IO (SVar m a)
10581042getFifoSVar ctype = do
10591043 outQ <- newIORef ([], 0)
10601044 outQMv <- newEmptyMVar
@@ -1070,28 +1054,31 @@ getFifoSVar ctype = do
10701054 maxWq <- newIORef 0
10711055#endif
10721056 let sv =
1073− SVar { outputQueue = outQ
1074− , doorBell = outQMv
1075− , outputHeap = undefined
1076− , runningThreads = running
1077− , workQueue = undefined
1078− , runqueue = runqueueFIFO sv q
1079− , enqueue = enqueueFIFO sv q
1080− , queueEmpty = liftIO $ nullQ q
1081− , waitingForWork = wfw
1082− , svarStyle = ctype
1083− , activeWorkers = active
1057+ SVar { outputQueue = outQ
1058+ , outputDoorBell = outQMv
1059+ , readOutputQ = readOutputQBounded sv
1060+ , postProcess = postProcessBounded sv
1061+ , workerThreads = running
1062+ , workLoop = workLoopFIFO sv q
1063+ , enqueue = enqueueFIFO sv q
1064+ , isWorkDone = nullQ q
1065+ , needDoorBell = wfw
1066+ , svarStyle = ctype
1067+ , workerCount = active
1068+ , accountThread = delThread sv
10841069#ifdef DIAGNOSTICS
1085− , totalDispatches = disp
1086− , maxWorkers = maxWrk
1087− , maxOutQSize = maxOq
1088− , maxHeapSize = maxHs
1089− , maxWorkQSize = maxWq
1070+ , aheadWorkQueue = undefined
1071+ , outputHeap = undefined
1072+ , totalDispatches = disp
1073+ , maxWorkers = maxWrk
1074+ , maxOutQSize = maxOq
1075+ , maxHeapSize = maxHs
1076+ , maxWorkQSize = maxWq
10901077#endif
10911078 }
10921079 in return sv
10931080
1094−getLifoSVar :: MonadIO m => SVarStyle -> IO (SVar m a)
1081+getLifoSVar :: MonadAsync m => SVarStyle -> IO (SVar m a)
10951082getLifoSVar ctype = do
10961083 outQ <- newIORef ([], 0)
10971084 outQMv <- newEmptyMVar
@@ -1106,36 +1093,42 @@ getLifoSVar ctype = do
11061093 maxHs <- newIORef 0
11071094 maxWq <- newIORef 0
11081095#endif
1109− let checkEmpty = null <$> liftIO (readIORef q)
1096+ let checkEmpty = null <$> readIORef q
11101097 let sv =
1111− SVar { outputQueue = outQ
1112− , doorBell = outQMv
1113− , outputHeap = undefined
1114− , runningThreads = running
1115− , workQueue = undefined
1116− , runqueue = runqueueLIFO sv q
1117− , enqueue = enqueueLIFO sv q
1118− , queueEmpty = checkEmpty
1119− , waitingForWork = wfw
1120− , svarStyle = ctype
1121− , activeWorkers = active
1098+ SVar { outputQueue = outQ
1099+ , outputDoorBell = outQMv
1100+ , readOutputQ = readOutputQBounded sv
1101+ , postProcess = postProcessBounded sv
1102+ , workerThreads = running
1103+ , workLoop = workLoopLIFO sv q
1104+ , enqueue = enqueueLIFO sv q
1105+ , isWorkDone = checkEmpty
1106+ , needDoorBell = wfw
1107+ , svarStyle = ctype
1108+ , workerCount = active
1109+ , accountThread = delThread sv
11221110#ifdef DIAGNOSTICS
1123− , maxWorkers = maxWrk
1124− , totalDispatches = disp
1125− , maxOutQSize = maxOq
1126− , maxHeapSize = maxHs
1127− , maxWorkQSize = maxWq
1111+ , aheadWorkQueue = undefined
1112+ , outputHeap = undefined
1113+ , maxWorkers = maxWrk
1114+ , totalDispatches = disp
1115+ , maxOutQSize = maxOq
1116+ , maxHeapSize = maxHs
1117+ , maxWorkQSize = maxWq
11281118#endif
11291119 }
11301120 in return sv
11311121
1132−getParSVar :: SVarStyle -> IO (SVar m a)
1133−getParSVar style = do
1122+getAheadSVar :: MonadAsync m => SVarStyle -> IO (SVar m a)
1123+getAheadSVar style = do
11341124 outQ <- newIORef ([], 0)
1125+ outH <- newIORef (H.empty, 0)
11351126 outQMv <- newEmptyMVar
11361127 active <- newIORef 0
11371128 wfw <- newIORef False
11381129 running <- newIORef S.empty
1130+ q <- newIORef ([], -1)
1131+
11391132#ifdef DIAGNOSTICS
11401133 disp <- newIORef 0
11411134 maxWrk <- newIORef 0
@@ -1144,37 +1137,50 @@ getParSVar style = do
11441137 maxWq <- newIORef 0
11451138#endif
11461139 let sv =
1147− SVar { outputQueue = outQ
1148− , doorBell = outQMv
1149− , outputHeap = undefined
1150− , runningThreads = running
1151− , workQueue = undefined
1152− , runqueue = undefined
1153− , enqueue = undefined
1154− , queueEmpty = undefined
1155− , waitingForWork = wfw
1156− , svarStyle = style
1157− , activeWorkers = active
1140+ SVar { outputQueue = outQ
1141+ , outputDoorBell = outQMv
1142+ , readOutputQ = readOutputQBounded sv
1143+ , postProcess = postProcessBounded sv
1144+ , workerThreads = running
1145+ , workLoop = workLoopAhead sv q outH
1146+ , enqueue = enqueueAhead sv q
1147+ , isWorkDone = isWorkDoneAhead q outH
1148+ , needDoorBell = wfw
1149+ , svarStyle = style
1150+ , workerCount = active
1151+ , accountThread = delThread sv
11581152#ifdef DIAGNOSTICS
1159− , totalDispatches = disp
1160− , maxWorkers = maxWrk
1161− , maxOutQSize = maxOq
1162− , maxHeapSize = maxHs
1163− , maxWorkQSize = maxWq
1153+ , aheadWorkQueue = q
1154+ , outputHeap = outH
1155+ , totalDispatches = disp
1156+ , maxWorkers = maxWrk
1157+ , maxOutQSize = maxOq
1158+ , maxHeapSize = maxHs
1159+ , maxWorkQSize = maxWq
11641160#endif
11651161 }
11661162 in return sv
11671163
1168−getAheadSVar :: MonadIO m => SVarStyle -> IO (SVar m a)
1169−getAheadSVar style = do
1164+ where
1165+
1166+ {-# INLINE isWorkDoneAhead #-}
1167+ isWorkDoneAhead q ref = do
1168+ heapDone <- do
1169+ (hp, _) <- readIORef ref
1170+ return (H.size hp <= 0)
1171+ queueDone <- checkEmpty q
1172+ return $ queueDone && heapDone
1173+
1174+ checkEmpty q = do
1175+ (xs, _) <- readIORef q
1176+ return $ null xs
1177+
1178+getParSVar :: MonadIO m => SVarStyle -> IO (SVar m a)
1179+getParSVar style = do
11701180 outQ <- newIORef ([], 0)
1171− outH <- newIORef (H.empty, 0)
11721181 outQMv <- newEmptyMVar
11731182 active <- newIORef 0
1174− wfw <- newIORef False
11751183 running <- newIORef S.empty
1176− q <- newIORef ([], -1)
1177−
11781184#ifdef DIAGNOSTICS
11791185 disp <- newIORef 0
11801186 maxWrk <- newIORef 0
@@ -1182,33 +1188,37 @@ getAheadSVar style = do
11821188 maxHs <- newIORef 0
11831189 maxWq <- newIORef 0
11841190#endif
1185−
1186− let checkEmpty = liftIO $ do
1187− (xs, _) <- readIORef q
1188− return $ null xs
11891191 let sv =
1190− SVar { outputQueue = outQ
1191− , doorBell = outQMv
1192− , outputHeap = outH
1193− , runningThreads = running
1194− , workQueue = q
1195− , runqueue = runqueueAhead sv q
1196− , enqueue = undefined
1197− , queueEmpty = checkEmpty
1198− , waitingForWork = wfw
1199− , svarStyle = style
1200− , activeWorkers = active
1201−
1192+ SVar { outputQueue = outQ
1193+ , outputDoorBell = outQMv
1194+ , readOutputQ = readOutputQPar sv
1195+ , postProcess = allThreadsDone sv
1196+ , workerThreads = running
1197+ , workLoop = undefined
1198+ , enqueue = undefined
1199+ , isWorkDone = undefined
1200+ , needDoorBell = undefined
1201+ , svarStyle = style
1202+ , workerCount = active
1203+ , accountThread = modifyThread sv
12021204#ifdef DIAGNOSTICS
1203− , totalDispatches = disp
1204− , maxWorkers = maxWrk
1205− , maxOutQSize = maxOq
1206− , maxHeapSize = maxHs
1207− , maxWorkQSize = maxWq
1205+ , aheadWorkQueue = undefined
1206+ , outputHeap = undefined
1207+ , totalDispatches = disp
1208+ , maxWorkers = maxWrk
1209+ , maxOutQSize = maxOq
1210+ , maxHeapSize = maxHs
1211+ , maxWorkQSize = maxWq
12081212#endif
12091213 }
12101214 in return sv
12111215
1216+ where
1217+
1218+ readOutputQPar sv = liftIO $ do
1219+ withDBGMVar sv "fromStreamVar: doorbell" $ takeMVar (outputDoorBell sv)
1220+ readOutputQRaw sv >>= return . fst
1221+
12121222-- | Create a new empty SVar.
12131223newEmptySVar :: MonadAsync m => SVarStyle -> m (SVar m a)
12141224newEmptySVar style = do
@@ -1227,25 +1237,18 @@ newStreamVar1 style m = do
12271237 -- Note: We must have all the work on the queue before sending the
12281238 -- pushworker, otherwise the pushworker may exit before we even get a
12291239 -- chance to push.
1230− if style == ParallelVar
1231− then pushWorkerPar sv m
1232− else do
1233− liftIO $ (enqueue sv) m
1234− pushWorker sv
1240+ liftIO $ enqueue sv m
1241+ pushWorker sv
12351242 return sv
12361243
1237−-- | Create a new SVar and enqueue one stream computation on it.
1238−{-# INLINABLE newStreamVarAhead #-}
1239−newStreamVarAhead :: MonadAsync m => Stream m a -> m (SVar m a)
1240−newStreamVarAhead m = do
1241− sv <- newEmptySVar AheadVar
1242− -- Note: We must have all the work on the queue before sending the
1243− -- pushworker, otherwise the pushworker may exit before we even get a
1244− -- chance to push.
1245− liftIO $ enqueueAhead sv (workQueue sv) m
1246− pushWorker sv
1244+{-# INLINABLE newStreamVarPar #-}
1245+newStreamVarPar :: MonadAsync m => Stream m a -> m (SVar m a)
1246+newStreamVarPar m = do
1247+ sv <- newEmptySVar ParallelVar
1248+ pushWorkerPar sv m
12471249 return sv
12481250
1251+-- XXX this errors out for Parallel/Ahead SVars
12491252-- | Write a stream to an 'SVar' in a non-blocking manner. The stream can then
12501253-- be read back from the SVar using 'fromSVar'.
12511254toStreamVar :: MonadAsync m => SVar m a -> Stream m a -> m ()
@@ -1324,7 +1327,7 @@ toStreamVar sv m = do
13241327forkSVarAsync :: MonadAsync m => SVarStyle -> Stream m a -> Stream m a -> Stream m a
13251328forkSVarAsync style m1 m2 = Stream $ \_ stp sng yld -> do
13261329 sv <- newStreamVar1 style (concurrently m1 m2)
1327− (runStream (fromStreamVar sv)) Nothing stp sng yld
1330+ runStream (fromStreamVar sv) Nothing stp sng yld
13281331 where
13291332 concurrently ma mb = Stream $ \svr stp sng yld -> do
13301333 liftIO $ enqueue (fromJust svr) mb
@@ -1336,33 +1339,18 @@ joinStreamVarAsync :: MonadAsync m
13361339joinStreamVarAsync style m1 m2 = Stream $ \svr stp sng yld ->
13371340 case svr of
13381341 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
13401343 _ -> runStream (forkSVarAsync style m1 m2) Nothing stp sng yld
13411344
1342−{-# NOINLINE forkSVarPar #-}
1343−forkSVarPar :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
1344−forkSVarPar m r = Stream $ \_ stp sng yld -> do
1345− sv <- newEmptySVar ParallelVar
1346− pushWorkerPar sv m
1347− pushWorkerPar sv r
1348− (runStream (fromStreamVar sv)) Nothing stp sng yld
1349−
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 ->
1354− case svr of
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
1358−
1345+-- The only difference between forkSVarAsync and this is that we run the left
1346+-- computation without a shared SVar.
13591347forkSVarAhead :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
13601348forkSVarAhead m1 m2 = Stream $ \_ stp sng yld -> do
1361− sv <- newStreamVarAhead (concurrently m1 m2)
1349+ sv <- newStreamVar1 AheadVar (concurrently m1 m2)
13621350 (runStream (fromStreamVar sv)) Nothing stp sng yld
13631351 where
13641352 concurrently ma mb = Stream $ \svr stp sng yld -> do
1365− liftIO $ enqueueAhead (fromJust svr) (workQueue (fromJust svr)) mb
1353+ liftIO $ enqueue (fromJust svr) mb
13661354 (runStream ma) Nothing stp sng yld
13671355
13681356{-# INLINE ahead #-}
@@ -1370,18 +1358,29 @@ ahead :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
13701358ahead m1 m2 = Stream $ \svr stp sng yld -> do
13711359 case svr of
13721360 Just sv | svarStyle sv == AheadVar -> do
1373− liftIO $ enqueueAhead sv (workQueue sv) m2
1361+ liftIO $ enqueue sv m2
13741362 -- Always run the left side on a new SVar to avoid complexity in
13751363 -- sequencing results. This means the left side cannot further
13761364 -- split into more ahead computations on the same SVar.
13771365 (runStream m1) Nothing stp sng yld
13781366 _ -> runStream (forkSVarAhead m1 m2) Nothing stp sng yld
13791367
1380−-- | XXX we can implement it more efficienty by directly implementing instead
1381−-- of combining streams using ahead.
1382−{-# INLINE consMAhead #-}
1383−consMAhead :: MonadAsync m => m a -> Stream m a -> Stream m a
1384−consMAhead m r = once m `ahead` r
1368+{-# NOINLINE forkSVarPar #-}
1369+forkSVarPar :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
1370+forkSVarPar m r = Stream $ \_ stp sng yld -> do
1371+ sv <- newEmptySVar ParallelVar
1372+ pushWorkerPar sv m
1373+ pushWorkerPar sv r
1374+ (runStream (fromStreamVar sv)) Nothing stp sng yld
1375+
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 ->
1380+ case svr of
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
13851384
13861385------------------------------------------------------------------------------
13871386-- Semigroup and Monoid style compositions for parallel actions
@@ -1407,6 +1406,12 @@ wAsync = joinStreamVarAsync WAsyncVar
14071406consMWAsync :: MonadAsync m => m a -> Stream m a -> Stream m a
14081407consMWAsync m r = once m `wAsync` r
14091408
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
1414+
14101415{-# INLINE parallel #-}
14111416parallel :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
14121417parallel = joinStreamVarPar ParallelVar
@@ -1441,19 +1446,18 @@ _alt m1 m2 = Stream $ \_ stp sng yld ->
14411446------------------------------------------------------------------------------
14421447
14431448applyWith :: MonadAsync m
1444− => SVarStyle -> (Stream m a -> Stream m b) -> Stream m a -> Stream m b
1445−applyWith style f m = Stream $ \svr stp sng yld -> do
1446− sv <- newStreamVar1 style m
1449+ => (Stream m a -> Stream m b) -> Stream m a -> Stream m b
1450+applyWith f m = Stream $ \svr stp sng yld -> do
1451+ sv <- newStreamVarPar m
14471452 runStream (f $ fromStreamVar sv) svr stp sng yld
14481453
14491454------------------------------------------------------------------------------
14501455-- Stream runner function application
14511456------------------------------------------------------------------------------
14521457
1453−runWith :: MonadAsync m
1454− => SVarStyle -> (Stream m a -> m b) -> Stream m a -> m b
1455−runWith style f m = do
1456− sv <- newStreamVar1 style m
1458+runWith :: MonadAsync m => (Stream m a -> m b) -> Stream m a -> m b
1459+runWith f m = do
1460+ sv <- newStreamVarPar m
14571461 f $ fromStreamVar sv
14581462
14591463------------------------------------------------------------------------------