@@ -269,10 +269,12 @@ doFork action exHandler =
269269{-# INLINE send #-}
270270send :: MonadIO m => SVar m a -> ChildEvent a -> m ()
271271send sv msg = liftIO $ do
272− atomicModifyIORefCAS_ (outputQueue sv) $ \es -> msg : es
273− -- XXX need a memory barrier? The wake up must happen only after the
274− -- store has finished otherwise we can have lost wakeup problems.
275− void $ tryPutMVar (doorBell sv) True
272+ old <- atomicModifyIORefCAS (outputQueue sv) $ \es -> (msg : es, es)
273+ if null old
274+ -- XXX need a memory barrier? The wake up must happen only after the
275+ -- store has finished otherwise we can have lost wakeup problems.
276+ then void $ tryPutMVar (doorBell sv) True
277+ else return ()
276278
277279{-# INLINE sendStop #-}
278280sendStop :: MonadIO m => SVar m a -> m ()
@@ -362,10 +364,13 @@ pushWorker sv =
362364{-# INLINE sendWorkerWait #-}
363365sendWorkerWait :: MonadParallel m => SVar m a -> m ()
364366sendWorkerWait sv = do
365− case svarStyle sv of
366− SVarStyle _ LIFO -> liftIO $ threadDelay 200
367− SVarStyle _ FIFO -> liftIO $ threadDelay 0
367+ -- When there is no output seen we dispatch more workers to help out if
368+ -- there is work pending in the work queue. But we wait a little while
369+ -- and check the output again so that we are not too aggressive.
370+ -- If there is no output pending to process and there is no worker to be
371+ -- sent then we block, so that we do not keep looping fruitlessly.
368372
373+ liftIO $ threadDelay 200
369374 output <- liftIO $ readIORef (outputQueue sv)
370375 when (null output) $ do
371376 done <- queueEmpty sv
@@ -377,19 +382,13 @@ sendWorkerWait sv = do
377382{-# NOINLINE fromStreamVar #-}
378383fromStreamVar :: MonadParallel m => SVar m a -> Stream m a
379384fromStreamVar sv = Stream $ \_ stp sng yld -> do
380− -- XXX if reading the IORef is costly we can use a flag in the SVar to
381− -- indicate we are done.
382− done <- allThreadsDone sv
383− if done
384− then stp
385− else do
386− res <- liftIO $ tryTakeMVar (doorBell sv)
387− when (isNothing res) $ sendWorkerWait sv
388− list <- liftIO $ atomicModifyIORefCAS (outputQueue sv) $ \x -> ([], x)
389− -- To avoid lock overhead we read all events at once instead of reading
390− -- one at a time. We just reverse the list to process the events in the
391− -- order they arrived. Maybe we can use a queue instead?
392− (runStream $ processEvents (reverse list)) Nothing stp sng yld
385+ res <- liftIO $ tryTakeMVar (doorBell sv)
386+ when (isNothing res) $ sendWorkerWait sv
387+ list <- liftIO $ atomicModifyIORefCAS (outputQueue sv) $ \x -> ([], x)
388+ -- Reversing the output is important to guarantee that we process the
389+ -- outputs in the same order as they were generated by the constituent
390+ -- streams.
391+ (runStream $ processEvents (reverse list)) Nothing stp sng yld
393392
394393 where
395394
@@ -424,14 +423,14 @@ getFifoSVar ctype = do
424423 running <- newIORef S.empty
425424 q <- newQ
426425 let sv =
427− SVar { outputQueue = outQ
428− , doorBell = outQMv
429− , runningThreads = running
430− , runqueue = runqueueFIFO sv q
431− , enqueue = pushL q
432− , queueEmpty = liftIO $ nullQ q
433− , svarStyle = ctype
434− }
426+ SVar { outputQueue = outQ
427+ , doorBell = outQMv
428+ , runningThreads = running
429+ , runqueue = runqueueFIFO sv q
430+ , enqueue = pushL q
431+ , queueEmpty = liftIO $ nullQ q
432+ , svarStyle = ctype
433+ }
435434 in return sv
436435
437436getLifoSVar :: MonadIO m => SVarStyle -> IO (SVar m a)
@@ -442,14 +441,14 @@ getLifoSVar ctype = do
442441 q <- newIORef []
443442 let checkEmpty = null <$> liftIO (readIORef q)
444443 let sv =
445− SVar { outputQueue = outQ
446− , doorBell = outQMv
447− , runningThreads = running
448− , runqueue = runqueueLIFO sv q
449− , enqueue = enqueueLIFO q
450− , queueEmpty = checkEmpty
451− , svarStyle = ctype
452− }
444+ SVar { outputQueue = outQ
445+ , doorBell = outQMv
446+ , runningThreads = running
447+ , runqueue = runqueueLIFO sv q
448+ , enqueue = enqueueLIFO q
449+ , queueEmpty = checkEmpty
450+ , svarStyle = ctype
451+ }
453452 in return sv
454453
455454-- | Create a new empty SVar.