@@ -67,7 +67,7 @@ where
6767
6868import Control.Concurrent (ThreadId, myThreadId, threadDelay)
6969import Control.Concurrent.MVar (MVar, newEmptyMVar, tryTakeMVar,
70− tryPutMVar, takeMVar, putMVar)
70+ tryPutMVar, takeMVar)
7171import Control.Exception (SomeException (..), catch, mask)
7272import Control.Monad (when)
7373import Control.Monad.Catch (MonadThrow, throwM)
@@ -118,10 +118,10 @@ data ChildEvent a =
118118 ChildYield a
119119 | ChildStop ThreadId (Maybe SomeException)
120120
121−-- | Sorting out of turn outputs in a heap for Ahead style streams
122−data AheadHeapEntry a =
123− AheadYieldSingle a
124− | AheadYieldMany a (MVar ()) -- value, MVar to wakeup for more values
121+-- | Sorting out-of-turn outputs in a heap for Ahead style streams
122+data AheadHeapEntry m a =
123+ AheadEntryPure a
124+ | AheadEntryStream (Stream m a)
125125
126126------------------------------------------------------------------------------
127127-- State threaded around the monad for thread management
@@ -169,7 +169,6 @@ data SVar m a =
169169 -- Shared output queue (events, length)
170170 , outputQueue :: IORef ([ChildEvent a], Int)
171171 , doorBell :: MVar () -- signal the consumer about output
172− , siren :: MVar () -- hooter for workers to begin work
173172
174173 -- Output synchronization mechanism for Ahead streams (Ahead and
175174 -- wAhead). We maintain a heap of out of sequence ahead of time
@@ -177,9 +176,9 @@ data SVar m a =
177176 -- currently at the head of the stream. Concurrent execute ahead
178177 -- tasks that have a sequence number greater than the task at the
179178 -- head should add their output to the heap.
180− , outputHeap :: IORef (Heap (Entry Int (AheadHeapEntry a)), Int)
181− , heapSiren :: MVar () -- hooter for workers to continue
182− -- adding to the heap
179+ , outputHeap :: IORef (Heap (Entry Int (AheadHeapEntry m a))
180+ , Int
181+ )
183182
184183 -- Shared work queue
185184 , workQueue :: IORef ([Stream m a], Int)
@@ -192,6 +191,7 @@ data SVar m a =
192191 , runningThreads :: IORef (Set ThreadId)
193192 , activeWorkers :: IORef Int
194193#ifdef DIAGNOSTICS
194+ , totalDispatches :: IORef Int
195195 , maxWorkers :: IORef Int
196196 , maxOutQSize :: IORef Int
197197 , maxHeapSize :: IORef Int
@@ -206,18 +206,15 @@ dumpSVar sv = do
206206 tid <- myThreadId
207207 (oqList, oqLen) <- readIORef $ outputQueue sv
208208 db <- tryTakeMVar $ doorBell sv
209− srn <- tryTakeMVar $ siren sv
210209 aheadDump <-
211210 if svarStyle sv == AheadVar
212211 then do
213212 (oheap, oheapSeq) <- readIORef $ outputHeap sv
214− hsrn <- tryTakeMVar $ heapSiren sv
215213 (wq, wqSeq) <- readIORef $ workQueue sv
216214 maxHp <- readIORef $ maxHeapSize sv
217215 return $ unlines
218216 [ "heap length = " ++ show (H.size oheap)
219217 , "heap seqeunce = " ++ show oheapSeq
220− , "heap siren = " ++ show hsrn
221218 , "work queue length = " ++ show (length wq)
222219 , "work queue sequence = " ++ show wqSeq
223220 , "heap max size = " ++ show maxHp
@@ -228,6 +225,7 @@ dumpSVar sv = do
228225 rthread <- readIORef $ runningThreads sv
229226 workers <- readIORef $ activeWorkers sv
230227 maxWrk <- readIORef $ maxWorkers sv
228+ dispatches <- readIORef $ totalDispatches sv
231229 maxOq <- readIORef $ maxOutQSize sv
232230 -- XXX queueEmpty should be made IO return type
233231
@@ -237,7 +235,7 @@ dumpSVar sv = do
237235 , "outputQueue length computed = " ++ show (length oqList)
238236 , "outputQueue length maintained = " ++ show oqLen
239237 , "output doorBell = " ++ show db
240− , "worker siren = " ++ show srn
238+ , "total dispatches = " ++ show dispatches
241239 , "max workers = " ++ show maxWrk
242240 , "max outQSize = " ++ show maxOq
243241 ]
@@ -419,13 +417,13 @@ doFork action exHandler =
419417-- queue at a time, that way contention can be reduced.
420418
421419-- | This function is used by the producer threads to queue output for the
422−-- consumer thread to consume.
420+-- consumer thread to consume. Returns whether the queue has more space.
423421{-# NOINLINE send #-}
424−send :: SVar m a -> ChildEvent a -> IO ()
422+send :: SVar m a -> ChildEvent a -> IO Bool
425423send sv msg = do
426424 len <- atomicModifyIORefCAS (outputQueue sv) $ \(es, n) ->
427425 ((msg : es, n + 1), n)
428− if (len <= 0) then do
426+ when (len <= 0) $ do
429427 -- XXX need a memory barrier? The wake up must happen only after the
430428 -- store has finished otherwise we can have lost wakeup problems.
431429 --
@@ -436,21 +434,11 @@ send sv msg = do
436434 -- The important point is that the consumer is guaranteed to receive a
437435 -- doorbell if something was added to the queue after it empties it.
438436 void $ tryPutMVar (doorBell sv) ()
439− -- those who were waiting because of buffer limit should now go ahead
440− void $ tryPutMVar (siren sv) ()
441− -- The only thread that reduces the length of the queue is the consumer
442− -- thread. It is possible that the queue may have been emptied in the
443− -- window when we acquired "len" and now when we are using it. In that case
444− -- we will be woken up by the consumer blowing the siren.
445− -- Note: current length after adding our output is len + 1
446− else when (len + 1 >= 1500) $ do
447− withDBGMVar sv "send: wait for siren" $ takeMVar (siren sv)
448− (_, n) <- readIORef (outputQueue sv)
449− when (n <= 1500) $ void $ tryPutMVar (siren sv) ()
437+ return (len < 1500)
450438
451439{-# NOINLINE sendStop #-}
452440sendStop :: SVar m a -> IO ()
453−sendStop sv = myThreadId >>= \tid -> send sv (ChildStop tid Nothing)
441+sendStop sv = myThreadId >>= \tid -> void $ send sv (ChildStop tid Nothing)
454442
455443-------------------------------------------------------------------------------
456444-- Async
@@ -489,9 +477,14 @@ runqueueLIFO sv q = run
489477 Nothing -> liftIO $ sendStop sv
490478 Just m -> (runStream m) (Just sv) run single yield
491479
492− sendit a = liftIO $ send sv (ChildYield a)
493− single a = sendit a >> run
494− yield a r = sendit a >> (runStream r) (Just sv) run single yield
480+ single a = do
481+ res <- liftIO $ send sv (ChildYield a)
482+ if res then run else liftIO $ sendStop sv
483+ yield a r = do
484+ res <- liftIO $ send sv (ChildYield a)
485+ if res
486+ then (runStream r) (Just sv) run single yield
487+ else liftIO $ enqueueLIFO sv q r >> sendStop sv
495488
496489 dequeue = liftIO $ atomicModifyIORefCAS q $ \case
497490 [] -> ([], Nothing)
@@ -528,9 +521,13 @@ runqueueFIFO sv q = run
528521 Just m -> (runStream m) (Just sv) run single yield
529522
530523 dequeue = liftIO $ tryPopR q
531− sendit a = liftIO $ send sv (ChildYield a)
532− single a = sendit a >> run
533− yield a r = sendit a >> liftIO (enqueueFIFO sv q r) >> run
524+ single a = do
525+ res <- liftIO $ send sv (ChildYield a)
526+ if res then run else liftIO $ sendStop sv
527+ yield a r = do
528+ res <- liftIO $ send sv (ChildYield a)
529+ liftIO (enqueueFIFO sv q r)
530+ if res then run else liftIO $ sendStop sv
534531
535532-------------------------------------------------------------------------------
536533-- Parallel
@@ -545,32 +542,37 @@ runOne sv m = (runStream m) (Just sv) stop single yield
545542 stop = liftIO $ sendStop sv
546543 sendit a = liftIO $ send sv (ChildYield a)
547544 single a = sendit a >> stop
548− yield a r = sendit a >> runOne sv r
545+ -- XXX there is no flow control in parallel case. We should perhaps use a
546+ -- queue and queue it back on that and exit the thread when the outputQueue
547+ -- overflows. Parallel is dangerous because it can accumulate unbounded
548+ -- output in the buffer.
549+ yield a r = void (sendit a) >> runOne sv r
549550
550551-------------------------------------------------------------------------------
551552-- Ahead
552553-------------------------------------------------------------------------------
553554
554−-- Lookahead streams can execute multiple tasks concurrently, ahead of time but
555−-- always serves them in the same order as they appear in the stream. To
555+-- Lookahead streams can execute multiple tasks concurrently, ahead of time,
556+-- but always serve them in the same order as they appear in the stream. To
556557-- implement lookahead streams efficiently we assign a sequence number to each
557−-- task when the task is picked up for execution. When the task finishes the
558+-- task when the task is picked up for execution. When the task finishes, the
558559-- output is tagged with the same sequence number and we rearrange the outputs
559560-- in sequence based on that number.
560561--
561562-- To explain the mechanism imagine that the current task at the head of the
562−-- stream has a "token" to yield to the output. The ownership of the token is
563−-- determined by the current sequence number maintained in outputHeap. Sequence
564−-- number is assigned when a task is queued. When a thread dequeues a task it
565−-- picks up the sequence number as well and when the output is ready it uses
566−-- the sequence number to queue the output to the outputQueue.
563+-- stream has a "token" to yield to the outputQueue. The ownership of the token
564+-- is determined by the current sequence number is maintained in outputHeap.
565+-- Sequence number is assigned when a task is queued. When a thread dequeues a
566+-- task it picks up the sequence number as well and when the output is ready it
567+-- uses the sequence number to queue the output to the outputQueue.
567568--
568569-- The thread with current sequence number sends the output directly to the
569570-- outputQueue. Other threads push the output to the outputHeap. When the task
570−-- being queued on the heap is a stream of many elements we yield the first
571−-- element on the heap and an MVar to wakeup for more elements. When such a
572−-- task gets the "token" for outputQueue it can directly keep yielding all the
573−-- elements to the outputQueue without checking for the "token".
571+-- being queued on the heap is a stream of many elements we evaluate only the
572+-- first element and keep the rest of the unevaluated computation in the heap.
573+-- When such a task gets the "token" for outputQueue it evaluates and directly
574+-- yields all the elements to the outputQueue without checking for the
575+-- "token".
574576--
575577-- Note that no two outputs in the heap can have the same sequence numbers and
576578-- therefore we do not need a stable heap. We have also separated the buffer
@@ -587,124 +589,17 @@ runOne sv m = (runStream m) (Just sv) stop single yield
587589-- reason we have a single element buffer limitation for the streams being
588590-- executed in advance.
589591--
590−-- This scheme works pretty efficiently with only 50% more overhead compared to
591−-- the Async streams where we do not have any kind of sequencing of the
592−-- outputs. It is especially devised so that we are most efficient when we have
593−-- short tasks and need just a single thread. Also when a thread yields many
594−-- items it can hold lockfree access to the outputQueue and do it efficiently.
592+-- This scheme works pretty efficiently with less than 40% extra overhead
593+-- compared to the Async streams where we do not have any kind of sequencing of
594+-- the outputs. It is especially devised so that we are most efficient when we
595+-- have short tasks and need just a single thread. Also when a thread yields
596+-- many items it can hold lockfree access to the outputQueue and do it
597+-- efficiently.
595598--
596599-- XXX Maybe we can start the ahead threads at a lower cpu and IO priority so
597600-- that they do not hog the resources and hinder the progress of the threads in
598601-- front of them.
599602
600−{-# INLINE sendToHeap #-}
601−sendToHeap :: SVar m a -> AheadHeapEntry a -> Int -> IO Bool
602−sendToHeap sv msg seqNo = do
603− let maxHeap = 1500
604−
605− -- Maybe we can use a lock instead of using CAS
606− -- insertions in the heap may be expensive if the heap is big
607− -- Note: len is valid only when added is True
608− (len, added) <- atomicModifyIORefCAS (outputHeap sv) $
609− \ref@(h, snum) ->
610− if seqNo > snum
611− then ((H.insert (Entry seqNo msg) h, snum), (H.size h, True))
612− else (ref, (0, False))
613−
614− -- this is only for the case when the consumer generated siren can get
615− -- missed due to race.
616− when (len < maxHeap) $ void $ tryPutMVar (heapSiren sv) ()
617−
618− if added
619− then do
620− when (len + 1 >= maxHeap) $ do
621− withDBGMVar sv "sendToHeap: wait for heapSiren"
622− $ takeMVar (heapSiren sv)
623− (h, _) <- readIORef (outputHeap sv)
624− when (H.size h <= maxHeap) $ void $ tryPutMVar (siren sv) ()
625− return False
626− else do
627− case msg of
628− AheadYieldSingle a -> send sv (ChildYield a)
629− AheadYieldMany a mv -> do
630− send sv (ChildYield a)
631− withDBGMVar sv "sendToHeap: wakeup stream" $ putMVar mv ()
632− return True
633−
634−popFromHeap
635− :: IORef (Heap (Entry Int (AheadHeapEntry a)), Int)
636− -> Int
637− -> IO (Maybe (AheadHeapEntry a))
638−popFromHeap hp nextSeqNo =
639− atomicModifyIORefCAS hp $ \(h, snum) -> do
640− let r = H.uncons h
641− let n = max nextSeqNo snum
642− case r of
643− Nothing -> ((h, n), Nothing)
644− Just (Entry seqNo ev, hp') ->
645− if (seqNo <= n)
646− then ((hp', seqNo), Just ev)
647− else ((h, n), Nothing)
648−
649−-- | As soon as we set the nextSeqNo in outputHeap we have transferred the
650−-- output "token" to the thread executing the task with that sequence number.
651−-- We have to make sure that there is no entry in the heap that has nextSeqNo
652−-- before we set nextSeqNo as the expected sequence number in the heap. It is
653−-- possible that when we are clearing the heap entry of the current sequence
654−-- number the next sequence number gets added and then we have to clear that,
655−-- this can keep going on and on.
656−drainHeap :: SVar m a -> Int -> IO ()
657−drainHeap sv nextSeqNo = do
658− r <- popFromHeap (outputHeap sv) nextSeqNo
659− case r of
660− Nothing -> return ()
661− Just ev -> do
662− case ev of
663− AheadYieldSingle a -> do
664− send sv (ChildYield a)
665− atomicModifyIORefCAS_ (outputHeap sv) $ \(h, snum) ->
666− (h, snum + 1)
667− drainHeap sv nextSeqNo
668− AheadYieldMany a mv -> do
669− send sv (ChildYield a)
670− withDBGMVar sv "drainHeap: wakeup stream" $ putMVar mv ()
671−
672−-- | The thread that has the token has the responsibility to move any pending
673−-- entries in the heap to the outputQueue. When the token is assigned to a
674−-- sequence number we must not have any entries for that sequence number in the
675−-- heap.
676−releaseToken :: SVar m a -> Int -> IO ()
677−releaseToken sv nextSeqNo = do
678−#ifdef DIAGNOSTICS
679− maxHp <- readIORef (maxHeapSize sv)
680− (hp, _) <- readIORef (outputHeap sv)
681− when (H.size hp > maxHp) $ writeIORef (maxHeapSize sv) (H.size hp)
682−#endif
683− drainHeap sv nextSeqNo
684− void $ tryPutMVar (heapSiren sv) ()
685−
686−-- | When we have the output "token" we are allowed to append to the
687−-- outputQueue otherwise we have to insert the output in the outputHeap.
688−-- Returns whether we still have the token.
689−handleOutput :: SVar m a
690− -> Bool
691− -> Int
692− -> Int
693− -> a
694− -> AheadHeapEntry a
695− -> IO Bool
696−handleOutput sv haveToken prevSeqNo seqNo a ev = do
697− if haveToken
698− then do
699− let nextSeqNo = prevSeqNo + 1
700− if seqNo == nextSeqNo
701− then send sv (ChildYield a) >> return True
702− else do
703− releaseToken sv nextSeqNo
704− toHeap
705− else toHeap
706− where toHeap = sendToHeap sv ev seqNo
707−
708603-- Left associated ahead expressions are expensive. We start a new SVar for
709604-- each left associative expression. The queue is used only for right
710605-- associated expression, we queue the right expression and execute the left.
@@ -720,58 +615,109 @@ enqueueAhead sv q m = do
720615 void $ tryPutMVar (doorBell sv) ()
721616 atomicModifyIORefCAS_ (waitingForWork sv) (const False)
722617
618+-- Normally the thread that has the token should never go away. The token gets
619+-- handed over to another thread, but someone or the other has the token at any
620+-- point of time. But if the task that has the token finds that the outputQueue
621+-- is full, in that case it can go away without even handing over the token to
622+-- another thread. In that case it sets the nextSequence number in the heap its
623+-- own sequence number before going away. To handle this case, any task that
624+-- does not have the token tries to dequeue from the heap first before
625+-- dequeuing from the work queue. If it finds that the task at the top of the
626+-- heap is the one that owns the current sequence number then it grabs the
627+-- token and starts with that.
723628runqueueAhead :: MonadIO m => SVar m a -> IORef ([Stream m a], Int) -> m ()
724−runqueueAhead sv q = run False (-1)
629+runqueueAhead sv q = runHeap
725630
726631 where
727632
728− run haveToken prevSeqNo = do
633+ maxHeap = 1500
634+
635+ toHeap seqNo ent = do
636+ hp <- liftIO $ atomicModifyIORefCAS (outputHeap sv) $ \(h, snum) ->
637+ ((H.insert (Entry seqNo ent) h, snum), h)
638+ if H.size hp <= maxHeap
639+ then runHeap
640+ else liftIO $ sendStop sv
641+
642+ singleToHeap seqNo a = toHeap seqNo (AheadEntryPure a)
643+ yieldToHeap seqNo a r = toHeap seqNo (AheadEntryStream (a `cons` r))
644+
645+ singleOutput seqNo a = do
646+ continue <- liftIO $ send sv (ChildYield a)
647+ if continue
648+ then runQueueToken seqNo
649+ else liftIO $ do
650+ atomicModifyIORefCAS_ (outputHeap sv) $ \(h, _) -> (h, seqNo + 1)
651+ sendStop sv
652+
653+ yieldOutput seqNo a r = do
654+ continue <- liftIO $ send sv (ChildYield a)
655+ if continue
656+ then (runStream r) (Just sv) (runQueueToken seqNo)
657+ (singleOutput seqNo)
658+ (yieldOutput seqNo)
659+ else liftIO $ do
660+ atomicModifyIORefCAS_ (outputHeap sv) $ \(h, _) ->
661+ (H.insert (Entry seqNo (AheadEntryStream r)) h, seqNo)
662+ sendStop sv
663+
664+ {-# INLINE runQueueToken #-}
665+ runQueueToken prevSeqNo = do
729666 work <- dequeue
730667 case work of
731− Nothing -> liftIO $ do
732− when haveToken $ releaseToken sv (prevSeqNo + 1)
733− sendStop sv
668+ Nothing -> do
669+ liftIO $ atomicModifyIORefCAS_ (outputHeap sv) $ \(h, _) ->
670+ (h, prevSeqNo + 1)
671+ runHeap
734672 Just (m, seqNo) -> do
735− let tStatus = haveToken || seqNo == 0
736− let stop = do
737− let nextSeqNo = prevSeqNo + 1
738− if tStatus && seqNo /= nextSeqNo
739− then do
740− liftIO $ releaseToken sv nextSeqNo
741− run False seqNo
742− else run tStatus seqNo
743− let single a = do
744− tokenStatus <- liftIO $
745− handleOutput sv tStatus prevSeqNo seqNo a
746− (AheadYieldSingle a)
747− run tokenStatus seqNo
748−
749− -- If we are yielding more than one item and we do not have the
750− -- token then we insert the current item in the pending
751− -- outputHeap and wait for our turn.
752− let
753− stopNext = run True seqNo
754− singleNext a = do
755− liftIO $ send sv (ChildYield a)
756− run True seqNo
757− yieldNext a r = do
758− liftIO $ send sv (ChildYield a)
759− (runStream r) (Just sv) stopNext singleNext yieldNext
760−
761− yield a r = do
762− -- XXX we can allocate the MVar when actually inserting
763− -- in the heap and then return it.
764− liftIO $ do
765− mv <- newEmptyMVar
766− gotToken <-
767− handleOutput sv tStatus prevSeqNo seqNo a
768− (AheadYieldMany a mv)
769− when (not gotToken) $ void $
770− withDBGMVar sv "runqueueAhead: stream waiting"
771− $ takeMVar mv
772− (runStream r) (Just sv) stopNext singleNext yieldNext
773−
774− (runStream m) (Just sv) stop single yield
673+ if seqNo == prevSeqNo + 1
674+ then
675+ (runStream m) (Just sv) (runQueueToken seqNo)
676+ (singleOutput seqNo)
677+ (yieldOutput seqNo)
678+ else do
679+ liftIO $ atomicModifyIORefCAS_ (outputHeap sv) $ \(h, _) ->
680+ (h, prevSeqNo + 1)
681+ (runStream m) (Just sv) runHeap
682+ (singleToHeap seqNo)
683+ (yieldToHeap seqNo)
684+ runQueueNoToken = do
685+ work <- dequeue
686+ case work of
687+ Nothing -> runHeap
688+ Just (m, seqNo) -> do
689+ if seqNo == 0
690+ then
691+ (runStream m) (Just sv) (runQueueToken seqNo)
692+ (singleOutput seqNo)
693+ (yieldOutput seqNo)
694+ else
695+ (runStream m) (Just sv) runHeap
696+ (singleToHeap seqNo)
697+ (yieldToHeap seqNo)
698+
699+ {-# NOINLINE runHeap #-}
700+ runHeap = do
701+#ifdef DIAGNOSTICS
702+ liftIO $ do
703+ maxHp <- readIORef (maxHeapSize sv)
704+ (hp, _) <- readIORef (outputHeap sv)
705+ when (H.size hp > maxHp) $ writeIORef (maxHeapSize sv) (H.size hp)
706+#endif
707+ ent <- liftIO $ dequeueFromHeap (outputHeap sv)
708+ case ent of
709+ Nothing -> do
710+ done <- queueEmpty sv
711+ if done
712+ then liftIO $ sendStop sv
713+ else runQueueNoToken
714+ Just (Entry seqNo hent) -> do
715+ case hent of
716+ AheadEntryPure a -> singleOutput seqNo a
717+ AheadEntryStream r ->
718+ (runStream r) (Just sv) (runQueueToken seqNo)
719+ (singleOutput seqNo)
720+ (yieldOutput seqNo)
775721
776722 dequeue = liftIO $ do
777723 atomicModifyIORefCAS q $ \case
@@ -779,6 +725,19 @@ runqueueAhead sv q = run False (-1)
779725 (x : [], n) -> (([], n), Just (x, n))
780726 _ -> error "more than one item on queue"
781727
728+ dequeueFromHeap
729+ :: IORef (Heap (Entry Int (AheadHeapEntry m a)), Int)
730+ -> IO (Maybe (Entry Int (AheadHeapEntry m a)))
731+ dequeueFromHeap hpRef = do
732+ atomicModifyIORefCAS hpRef $ \hp@(h, snum) -> do
733+ let r = H.uncons h
734+ case r of
735+ Nothing -> (hp, Nothing)
736+ Just (ent@(Entry seqNo _ev), hp') ->
737+ if (seqNo == snum)
738+ then ((hp', seqNo), Just ent)
739+ else (hp, Nothing)
740+
782741-- Thread tracking is needed for two reasons:
783742--
784743-- 1) Killing threads on exceptions. Threads may not be left to go away by
@@ -825,7 +784,7 @@ allThreadsDone sv = liftIO $ S.null <$> readIORef (runningThreads sv)
825784handleChildException :: SVar m a -> SomeException -> IO ()
826785handleChildException sv e = do
827786 tid <- myThreadId
828− send sv (ChildStop tid (Just e))
787+ void $ send sv (ChildStop tid (Just e))
829788
830789#ifdef DIAGNOSTICS
831790recordMaxWorkers :: MonadIO m => SVar m a -> m ()
@@ -833,6 +792,7 @@ recordMaxWorkers sv = liftIO $ do
833792 active <- readIORef (activeWorkers sv)
834793 maxWrk <- readIORef (maxWorkers sv)
835794 when (active > maxWrk) $ writeIORef (maxWorkers sv) active
795+ modifyIORef (totalDispatches sv) (+1)
836796#endif
837797
838798{-# NOINLINE pushWorker #-}
@@ -905,6 +865,10 @@ sendWorkerWait sv = do
905865 else liftIO $ withDBGMVar sv "sendWorkerWait: wait for workers"
906866 $ takeMVar (doorBell sv)
907867 else do
868+ -- XXX Assert here that if the heap is not empty then there is at
869+ -- least one outstanding worker. Otherwise we could be sleeping
870+ -- forever.
871+ --
908872 -- We found that the queue was empty, but it may be empty
909873 -- temporarily, when we checked it. If that's the case we might
910874 -- sleep indefinitely unless the active workers produce some
@@ -954,7 +918,6 @@ fromStreamVar sv = Stream $ \_ stp sng yld -> do
954918#else
955919 (list, _) <- liftIO $ atomicModifyIORefCAS (outputQueue sv) $ \x -> (([],0), x)
956920#endif
957− void $ liftIO $ tryPutMVar (siren sv) ()
958921 -- Reversing the output is important to guarantee that we process the
959922 -- outputs in the same order as they were generated by the constituent
960923 -- streams.
@@ -969,19 +932,50 @@ fromStreamVar sv = Stream $ \_ stp sng yld -> do
969932 -- liftIO $ readIORef (runningThreads sv) >>= mapM_ killThread
970933 throwM e
971934
972− {-# INLINE processEvents #-}
973− processEvents [] = Stream $ \_ stp sng yld -> do
974− done <- allThreadsDone sv
975− if not done
976− then (runStream (fromStreamVar sv)) Nothing stp sng yld
935+ allDone stp = do
977936#ifdef DIAGNOSTICS
978− else do
937+#ifdef DIAGNOSTICS_VERBOSE
979938 svInfo <- liftIO $ dumpSVar sv
980939 liftIO $ hPutStrLn stderr $ "fromStreamVar done\n" ++ svInfo
981− stp
982−#else
983− else stp
984940#endif
941+#endif
942+ stp
943+
944+ {-# INLINE processEvents #-}
945+ processEvents [] = Stream $ \_ stp sng yld -> do
946+ workersDone <- allThreadsDone sv
947+ if not workersDone
948+ then (runStream (fromStreamVar sv)) Nothing stp sng yld
949+ else
950+ if svarStyle sv == ParallelVar
951+ then allDone stp
952+ else do
953+ -- All the workers may come back if the outputQueue is full.
954+ -- So check if there is work pending and kick off a worker to
955+ -- finish it.
956+ -- XXX we should perhaps kickoff even before we start
957+ -- processing if the active workers are zero, rather than at
958+ -- the end of processing the outputQueue. But some thread stop
959+ -- events may be in the outputQueue itself, so we may not be
960+ -- able to determine the correct number of workers before we
961+ -- process it.
962+ -- XXX It is necessary to do this before the processing to keep
963+ -- minimal concurrency always.
964+ -- We should have a separate queue for Stop events so that we
965+ -- can process them before the outputs and send workers
966+ -- promptly when needed.
967+ heapDone <-
968+ if (svarStyle sv == AheadVar)
969+ then do
970+ (hp, _) <- liftIO $ readIORef (outputHeap sv)
971+ return (H.size hp <= 0)
972+ else return True
973+ queueDone <- queueEmpty sv
974+ if queueDone && heapDone
975+ then allDone stp
976+ else do
977+ pushWorker sv
978+ (runStream (fromStreamVar sv)) Nothing stp sng yld
985979
986980 processEvents (ev : es) = Stream $ \_ stp sng yld -> do
987981 let continue = (runStream (processEvents es)) Nothing stp sng yld
@@ -1001,12 +995,12 @@ getFifoSVar :: MonadIO m => SVarStyle -> IO (SVar m a)
1001995getFifoSVar ctype = do
1002996 outQ <- newIORef ([], 0)
1003997 outQMv <- newEmptyMVar
1004− hooter <- newEmptyMVar
1005998 active <- newIORef 0
1006999 wfw <- newIORef False
10071000 running <- newIORef S.empty
10081001 q <- newQ
10091002#ifdef DIAGNOSTICS
1003+ disp <- newIORef 0
10101004 maxWrk <- newIORef 0
10111005 maxOq <- newIORef 0
10121006 maxHs <- newIORef 0
@@ -1015,9 +1009,7 @@ getFifoSVar ctype = do
10151009 let sv =
10161010 SVar { outputQueue = outQ
10171011 , doorBell = outQMv
1018− , siren = hooter
10191012 , outputHeap = undefined
1020− , heapSiren = undefined
10211013 , runningThreads = running
10221014 , workQueue = undefined
10231015 , runqueue = runqueueFIFO sv q
@@ -1027,6 +1019,7 @@ getFifoSVar ctype = do
10271019 , svarStyle = ctype
10281020 , activeWorkers = active
10291021#ifdef DIAGNOSTICS
1022+ , totalDispatches = disp
10301023 , maxWorkers = maxWrk
10311024 , maxOutQSize = maxOq
10321025 , maxHeapSize = maxHs
@@ -1039,12 +1032,12 @@ getLifoSVar :: MonadIO m => SVarStyle -> IO (SVar m a)
10391032getLifoSVar ctype = do
10401033 outQ <- newIORef ([], 0)
10411034 outQMv <- newEmptyMVar
1042− hooter <- newEmptyMVar
10431035 active <- newIORef 0
10441036 wfw <- newIORef False
10451037 running <- newIORef S.empty
10461038 q <- newIORef []
10471039#ifdef DIAGNOSTICS
1040+ disp <- newIORef 0
10481041 maxWrk <- newIORef 0
10491042 maxOq <- newIORef 0
10501043 maxHs <- newIORef 0
@@ -1054,9 +1047,7 @@ getLifoSVar ctype = do
10541047 let sv =
10551048 SVar { outputQueue = outQ
10561049 , doorBell = outQMv
1057− , siren = hooter
10581050 , outputHeap = undefined
1059− , heapSiren = undefined
10601051 , runningThreads = running
10611052 , workQueue = undefined
10621053 , runqueue = runqueueLIFO sv q
@@ -1067,6 +1058,7 @@ getLifoSVar ctype = do
10671058 , activeWorkers = active
10681059#ifdef DIAGNOSTICS
10691060 , maxWorkers = maxWrk
1061+ , totalDispatches = disp
10701062 , maxOutQSize = maxOq
10711063 , maxHeapSize = maxHs
10721064 , maxWorkQSize = maxWq
@@ -1078,11 +1070,11 @@ getParSVar :: SVarStyle -> IO (SVar m a)
10781070getParSVar style = do
10791071 outQ <- newIORef ([], 0)
10801072 outQMv <- newEmptyMVar
1081− hooter <- newEmptyMVar
10821073 active <- newIORef 0
10831074 wfw <- newIORef False
10841075 running <- newIORef S.empty
10851076#ifdef DIAGNOSTICS
1077+ disp <- newIORef 0
10861078 maxWrk <- newIORef 0
10871079 maxOq <- newIORef 0
10881080 maxHs <- newIORef 0
@@ -1091,9 +1083,7 @@ getParSVar style = do
10911083 let sv =
10921084 SVar { outputQueue = outQ
10931085 , doorBell = outQMv
1094− , siren = hooter
10951086 , outputHeap = undefined
1096− , heapSiren = undefined
10971087 , runningThreads = running
10981088 , workQueue = undefined
10991089 , runqueue = undefined
@@ -1103,6 +1093,7 @@ getParSVar style = do
11031093 , svarStyle = style
11041094 , activeWorkers = active
11051095#ifdef DIAGNOSTICS
1096+ , totalDispatches = disp
11061097 , maxWorkers = maxWrk
11071098 , maxOutQSize = maxOq
11081099 , maxHeapSize = maxHs
@@ -1116,14 +1107,13 @@ getAheadSVar style = do
11161107 outQ <- newIORef ([], 0)
11171108 outH <- newIORef (H.empty, 0)
11181109 outQMv <- newEmptyMVar
1119− hooter <- newEmptyMVar
1120− hsiren <- newEmptyMVar
11211110 active <- newIORef 0
11221111 wfw <- newIORef False
11231112 running <- newIORef S.empty
11241113 q <- newIORef ([], -1)
11251114
11261115#ifdef DIAGNOSTICS
1116+ disp <- newIORef 0
11271117 maxWrk <- newIORef 0
11281118 maxOq <- newIORef 0
11291119 maxHs <- newIORef 0
@@ -1136,9 +1126,7 @@ getAheadSVar style = do
11361126 let sv =
11371127 SVar { outputQueue = outQ
11381128 , doorBell = outQMv
1139− , siren = hooter
11401129 , outputHeap = outH
1141− , heapSiren = hsiren
11421130 , runningThreads = running
11431131 , workQueue = q
11441132 , runqueue = runqueueAhead sv q
@@ -1149,6 +1137,7 @@ getAheadSVar style = do
11491137 , activeWorkers = active
11501138
11511139#ifdef DIAGNOSTICS
1140+ , totalDispatches = disp
11521141 , maxWorkers = maxWrk
11531142 , maxOutQSize = maxOq
11541143 , maxHeapSize = maxHs