@@ -66,7 +66,7 @@ where
6666
6767import Control.Concurrent (ThreadId, myThreadId, threadDelay)
6868import Control.Concurrent.MVar (MVar, newEmptyMVar, tryTakeMVar,
69− tryPutMVar, takeMVar, readMVar)
69+ tryPutMVar, takeMVar)
7070import Control.Exception (SomeException (..))
7171import qualified Control.Exception.Lifted as EL
7272import Control.Monad (when)
@@ -324,27 +324,17 @@ send sv msg = do
324324 -- The important point is that the consumer is guaranteed to receive a
325325 -- doorbell if something was added to the queue after it empties it.
326326 void $ tryPutMVar (doorBell sv) ()
327− -- The first worker who notices the output queue was emptied puts the
328− -- siren off.
329− void $ tryTakeMVar (siren sv)
330− else if (len + 1 >= 1500) then do
331− -- We are guaranteed to receive the siren if the consumer reads the
332− -- queue because the consumer puts the siren on before reading the
333− -- queue.
334− --
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)
347− else return ()
327+ -- those who were waiting because of buffer limit should now go ahead
328+ void $ tryPutMVar (siren sv) ()
329+ -- The only thread that reduces the length of the queue is the consumer
330+ -- thread. It is possible that the queue may have been emptied in the
331+ -- window when we acquired "len" and now when we are using it. In that case
332+ -- we will be woken up by the consumer blowing the siren.
333+ -- Note: current length after adding our output is len + 1
334+ else when (len + 1 >= 1500) $ do
335+ takeMVar (siren sv)
336+ (_, n) <- readIORef (outputQueue sv)
337+ when (n <= 1500) $ void $ tryPutMVar (siren sv) ()
348338
349339{-# INLINE sendStop #-}
350340sendStop :: SVar m a -> IO ()
@@ -514,8 +504,8 @@ fromStreamVar sv = Stream $ \_ stp sng yld -> do
514504 res <- liftIO $ tryTakeMVar (doorBell sv)
515505 when (isNothing res) $ sendWorkerWait sv
516506
517− void $ liftIO $ tryPutMVar (siren sv) ()
518507 (list, _) <- liftIO $ atomicModifyIORefCAS (outputQueue sv) $ \x -> (([],0), x)
508+ void $ liftIO $ tryPutMVar (siren sv) ()
519509 -- Reversing the output is important to guarantee that we process the
520510 -- outputs in the same order as they were generated by the constituent
521511 -- streams.