a0d9419 · streamly · git.iko.soy

52import Control.Monad.Trans.Class (MonadTrans (lift))

53import Control.Monad.Trans.Control (MonadBaseControl, liftBaseWith)

54import Data.Concurrent.Queue.MichaelScott (LinkedQueue, newQ, pushL,

55− tryPopR)

55+ tryPopR, nullQ)

56import Data.Functor (void)

57import Data.IORef (IORef, modifyIORef, newIORef,

58 readIORef)

101 , enqueue :: AsyncT m a -> IO ()

102 , runqueue :: m ()

103 , runningThreads :: IORef (Set ThreadId)

104+ , queueEmpty :: m Bool

104 }

105 }

105

106

106-- The 'Maybe (AsyncT m a)' is redundant as we can use 'stop' value for the

107-- The 'Maybe (AsyncT m a)' is redundant as we can use 'stop' value for the

313sendWorkerWait ctx = do

314sendWorkerWait ctx = do

314 liftIO $ threadDelay 200

315 liftIO $ threadDelay 200

315 output <- liftIO $ readIORef (outputQueue ctx)

316 output <- liftIO $ readIORef (outputQueue ctx)

316− case output of

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

317− [] -> pushWorker ctx

321+ when (not done) (pushWorker ctx)

318− _ -> return ()

319 void $ liftIO $ takeMVar (synchOutQ ctx)

322 void $ liftIO $ takeMVar (synchOutQ ctx)

320

323

321-- Note: This is performance sensitive code.

324-- Note: This is performance sensitive code.

378 , runningThreads = running

381 , runningThreads = running

379 , runqueue = runqueueFIFO ctx q

382 , runqueue = runqueueFIFO ctx q

380 , enqueue = pushL q

383 , enqueue = pushL q

384+ , queueEmpty = liftIO $ nullQ q

381 }

385 }

382 in return ctx

386 in return ctx

383 False -> do

387 False -> do

384 q <- newIORef []

388 q <- newIORef []

385 enqueueLIFO q m2 >> enqueueLIFO q m1

389 enqueueLIFO q m2 >> enqueueLIFO q m1

390+ let checkEmpty = liftIO (readIORef q) >>= return . null

386 let ctx =

391 let ctx =

387 Context { outputQueue = outQ

392 Context { outputQueue = outQ

388 , synchOutQ = outQMv

393 , synchOutQ = outQMv

389 , runningThreads = running

394 , runningThreads = running

390 , runqueue = runqueueLIFO ctx q

395 , runqueue = runqueueLIFO ctx q

391 , enqueue = enqueueLIFO q

396 , enqueue = enqueueLIFO q

397+ , queueEmpty = checkEmpty

392 }

398 }

393 in return ctx

399 in return ctx

394

400