@@ -269,10 +269,12 @@ doFork action exHandler =
269{-# INLINE send #-}
269{-# INLINE send #-}
270send :: MonadIO m => SVar m a -> ChildEvent a -> m ()
270send :: MonadIO m => SVar m a -> ChildEvent a -> m ()
271send sv msg = liftIO $ do
271send sv msg = liftIO $ do
272− atomicModifyIORefCAS_ (outputQueue sv) $ \es -> msg : es
272+ old <- atomicModifyIORefCAS (outputQueue sv) $ \es -> (msg : es, es)
273− -- XXX need a memory barrier? The wake up must happen only after the
274+ -- 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+ -- store has finished otherwise we can have lost wakeup problems.
275− void $ tryPutMVar (doorBell sv) True
276+ then void $ tryPutMVar (doorBell sv) True
277{-# INLINE sendStop #-}
279{-# INLINE sendStop #-}
278sendStop :: MonadIO m => SVar m a -> m ()
280sendStop :: MonadIO m => SVar m a -> m ()
@@ -362,10 +364,13 @@ pushWorker sv =
362{-# INLINE sendWorkerWait #-}
364{-# INLINE sendWorkerWait #-}
363sendWorkerWait :: MonadParallel m => SVar m a -> m ()
365sendWorkerWait :: MonadParallel m => SVar m a -> m ()
364sendWorkerWait sv = do
366sendWorkerWait sv = do
365− case svarStyle sv of
367+ -- When there is no output seen we dispatch more workers to help out if
366− SVarStyle _ LIFO -> liftIO $ threadDelay 200
368+ -- there is work pending in the work queue. But we wait a little while
367− SVarStyle _ FIFO -> liftIO $ threadDelay 0
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.
373+ liftIO $ threadDelay 200
369 output <- liftIO $ readIORef (outputQueue sv)
374 output <- liftIO $ readIORef (outputQueue sv)
370 when (null output) $ do
375 when (null output) $ do
371 done <- queueEmpty sv
376 done <- queueEmpty sv
@@ -377,19 +382,13 @@ sendWorkerWait sv = do
377{-# NOINLINE fromStreamVar #-}
382{-# NOINLINE fromStreamVar #-}
378fromStreamVar :: MonadParallel m => SVar m a -> Stream m a
383fromStreamVar :: MonadParallel m => SVar m a -> Stream m a
379fromStreamVar sv = Stream $ \_ stp sng yld -> do
384fromStreamVar 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
386− res <- liftIO $ tryTakeMVar (doorBell sv)
385+ res <- liftIO $ tryTakeMVar (doorBell sv)
387− when (isNothing res) $ sendWorkerWait sv
386+ when (isNothing res) $ sendWorkerWait sv
388− list <- liftIO $ atomicModifyIORefCAS (outputQueue sv) $ \x -> ([], x)
387+ list <- liftIO $ atomicModifyIORefCAS (outputQueue sv) $ \x -> ([], x)
389− -- To avoid lock overhead we read all events at once instead of reading
388+ -- Reversing the output is important to guarantee that we process the
390− -- one at a time. We just reverse the list to process the events in the
389+ -- outputs in the same order as they were generated by the constituent
391− -- order they arrived. Maybe we can use a queue instead?
390+ -- streams.
392− (runStream $ processEvents (reverse list)) Nothing stp sng yld
391+ (runStream $ processEvents (reverse list)) Nothing stp sng yld
@@ -424,14 +423,14 @@ getFifoSVar ctype = do
424 running <- newIORef S.empty
423 running <- newIORef S.empty
425 q <- newQ
424 q <- newQ
427− SVar { outputQueue = outQ
426+ SVar { outputQueue = outQ
428− , doorBell = outQMv
427+ , doorBell = outQMv
429− , runningThreads = running
428+ , runningThreads = running
430− , runqueue = runqueueFIFO sv q
429+ , runqueue = runqueueFIFO sv q
431− , enqueue = pushL q
430+ , enqueue = pushL q
432− , queueEmpty = liftIO $ nullQ q
431+ , queueEmpty = liftIO $ nullQ q
433− , svarStyle = ctype
432+ , svarStyle = ctype
435 in return sv
434 in return sv
437getLifoSVar :: MonadIO m => SVarStyle -> IO (SVar m a)
436getLifoSVar :: MonadIO m => SVarStyle -> IO (SVar m a)
@@ -442,14 +441,14 @@ getLifoSVar ctype = do
442 q <- newIORef []
441 q <- newIORef []
443 let checkEmpty = null <$> liftIO (readIORef q)
442 let checkEmpty = null <$> liftIO (readIORef q)
445− SVar { outputQueue = outQ
444+ SVar { outputQueue = outQ
446− , doorBell = outQMv
445+ , doorBell = outQMv
447− , runningThreads = running
446+ , runningThreads = running
448− , runqueue = runqueueLIFO sv q
447+ , runqueue = runqueueLIFO sv q
449− , enqueue = enqueueLIFO q
448+ , enqueue = enqueueLIFO q
450− , queueEmpty = checkEmpty
449+ , queueEmpty = checkEmpty
451− , svarStyle = ctype
450+ , svarStyle = ctype
453 in return sv
452 in return sv
455-- | Create a new empty SVar.
454-- | Create a new empty SVar.