@@ -52,7 +52,7 @@ import Control.Monad.State.Class (MonadState(..))
5252import Control.Monad.Trans.Class (MonadTrans (lift))
5353import Control.Monad.Trans.Control (MonadBaseControl, liftBaseWith)
5454import Data.Concurrent.Queue.MichaelScott (LinkedQueue, newQ, pushL,
55− tryPopR)
55+ tryPopR, nullQ)
5656import Data.Functor (void)
5757import Data.IORef (IORef, modifyIORef, newIORef,
5858 readIORef)
@@ -101,6 +101,7 @@ data Context m a =
101101 , enqueue :: AsyncT m a -> IO ()
102102 , runqueue :: m ()
103103 , runningThreads :: IORef (Set ThreadId)
104+ , queueEmpty :: m Bool
104105 }
105106
106107-- The 'Maybe (AsyncT m a)' is redundant as we can use 'stop' value for the
@@ -313,9 +314,11 @@ sendWorkerWait :: MonadAsync m => Context m a -> m ()
313314sendWorkerWait ctx = do
314315 liftIO $ threadDelay 200
315316 output <- liftIO $ readIORef (outputQueue ctx)
316− case output of
317− [] -> pushWorker ctx
318− _ -> return ()
317+ when (null output) $ do
318+ -- Send a worker only if workqueue is not empty otherwise we may keep
319+ -- sending a worker in a loop running into a livelock
320+ done <- queueEmpty ctx
321+ when (not done) (pushWorker ctx)
319322 void $ liftIO $ takeMVar (synchOutQ ctx)
320323
321324-- Note: This is performance sensitive code.
@@ -378,17 +381,20 @@ pullFork m1 m2 fifo = AsyncT $ \_ stp yld -> do
378381 , runningThreads = running
379382 , runqueue = runqueueFIFO ctx q
380383 , enqueue = pushL q
384+ , queueEmpty = liftIO $ nullQ q
381385 }
382386 in return ctx
383387 False -> do
384388 q <- newIORef []
385389 enqueueLIFO q m2 >> enqueueLIFO q m1
390+ let checkEmpty = liftIO (readIORef q) >>= return . null
386391 let ctx =
387392 Context { outputQueue = outQ
388393 , synchOutQ = outQMv
389394 , runningThreads = running
390395 , runqueue = runqueueLIFO ctx q
391396 , enqueue = enqueueLIFO q
397+ , queueEmpty = checkEmpty
392398 }
393399 in return ctx
394400