a4e4c07 · streamly · git.iko.soy

66

67import Control.Concurrent (ThreadId, myThreadId, threadDelay)

68import Control.Concurrent.MVar (MVar, newEmptyMVar, tryTakeMVar,

69− tryPutMVar, takeMVar, readMVar)

69+ tryPutMVar, takeMVar)

70import Control.Exception (SomeException (..))

71import qualified Control.Exception.Lifted as EL

72import Control.Monad (when)

324 -- The important point is that the consumer is guaranteed to receive a

325 -- doorbell if something was added to the queue after it empties it.

326 void $ tryPutMVar (doorBell sv) ()

327− -- The first worker who notices the output queue was emptied puts the

327+ -- those who were waiting because of buffer limit should now go ahead

328− -- siren off.

328+ void $ tryPutMVar (siren sv) ()

329− void $ tryTakeMVar (siren sv)

329+ -- The only thread that reduces the length of the queue is the consumer

330− else if (len + 1 >= 1500) then do

330+ -- thread. It is possible that the queue may have been emptied in the

331− -- We are guaranteed to receive the siren if the consumer reads the

331+ -- window when we acquired "len" and now when we are using it. In that case

332− -- queue because the consumer puts the siren on before reading the

332+ -- we will be woken up by the consumer blowing the siren.

333− -- queue.

333+ -- Note: current length after adding our output is len + 1

334− --

334+ else when (len + 1 >= 1500) $ do

335− -- We may get the siren between the siren being swicthed on and the

336− -- queue getting read but that's harmless, at amost everyone will go

337− -- back to work and will have to sleep again if queue was still not

338− -- emptied.

339− --

340− -- If even before a worker could read the MVar, the queue gets emptied

341− -- and another worker queuing on it switches off the siren, then we may

342− -- sleep here. In that case we are guaranteed to be woken up on the

343− -- next siren. The next siren is guaranteed as we send a doorbell

344− -- before switching off the siren, and the consumer switches on the

345− -- siren after receiving the doorbell.

346− readMVar (siren sv)

335+ takeMVar (siren sv)

347− else return ()

336+ (_, n) <- readIORef (outputQueue sv)

337+ when (n <= 1500) $ void $ tryPutMVar (siren sv) ()

348

338

349{-# INLINE sendStop #-}

339{-# INLINE sendStop #-}

350sendStop :: SVar m a -> IO ()

340sendStop :: SVar m a -> IO ()

514 res <- liftIO $ tryTakeMVar (doorBell sv)

504 res <- liftIO $ tryTakeMVar (doorBell sv)

515 when (isNothing res) $ sendWorkerWait sv

505 when (isNothing res) $ sendWorkerWait sv

516

506

517− void $ liftIO $ tryPutMVar (siren sv) ()

518 (list, _) <- liftIO $ atomicModifyIORefCAS (outputQueue sv) $ \x -> (([],0), x)

507 (list, _) <- liftIO $ atomicModifyIORefCAS (outputQueue sv) $ \x -> (([],0), x)

508+ void $ liftIO $ tryPutMVar (siren sv) ()

519 -- Reversing the output is important to guarantee that we process the

509 -- Reversing the output is important to guarantee that we process the

520 -- outputs in the same order as they were generated by the constituent

510 -- outputs in the same order as they were generated by the constituent

521 -- streams.

511 -- streams.