@@ -68,7 +68,7 @@ where
6868
6969import Control.Concurrent (ThreadId, myThreadId, threadDelay)
7070import Control.Concurrent.MVar (MVar, newEmptyMVar, tryTakeMVar,
71− tryPutMVar, takeMVar)
71+ tryPutMVar, takeMVar, readMVar)
7272import Control.Exception (SomeException (..))
7373import qualified Control.Exception.Lifted as EL
7474import Control.Monad (when)
@@ -148,8 +148,9 @@ data SVarStyle = SVarStyle SVarTag SVarSched deriving Eq
148148-- already enqueued computations get evaluated. See 'joinStreamVar2'.
149149--
150150data SVar m a =
151− SVar { outputQueue :: IORef [ChildEvent a]
152− , doorBell :: MVar () -- wakeup mechanism for outQ
151+ SVar { outputQueue :: IORef ([ChildEvent a], Int)
152+ , doorBell :: MVar () -- signal the consumer about output
153+ , siren :: MVar () -- hooter for workers to begin work
153154 , enqueue :: Stream m a -> IO ()
154155 , runqueue :: m ()
155156 , runningThreads :: IORef (Set ThreadId)
@@ -304,15 +305,45 @@ doFork action exHandler =
304305
305306-- XXX exception safety of all atomic/MVar operations
306307
308+-- TBD Each worker can have their own queue and the consumer can empty one
309+-- queue at a time, that way contention can be reduced.
307310{-# INLINE send #-}
308311send :: MonadIO m => SVar m a -> ChildEvent a -> m ()
309312send sv msg = liftIO $ do
310− old <- atomicModifyIORefCAS (outputQueue sv) $ \es -> (msg : es, es)
311− if null old
313+ len <- atomicModifyIORefCAS (outputQueue sv) $ \(es, n) ->
314+ ((msg : es, n + 1), n)
315+ if (len <= 0) then do
312316 -- XXX need a memory barrier? The wake up must happen only after the
313317 -- store has finished otherwise we can have lost wakeup problems.
314− then void $ tryPutMVar (doorBell sv) ()
315− else return ()
318+ --
319+ -- Since multiple workers can try this at the same time, it is possible
320+ -- that we may put a spurious MVar after the consumer has already seen
321+ -- the output. But that's harmless, at worst it may cause the consumer
322+ -- to read the queue again and find it empty.
323+ -- The important point is that the consumer is guaranteed to receive a
324+ -- doorbell if something was added to the queue after it empties it.
325+ void $ tryPutMVar (doorBell sv) ()
326+ -- The first worker who notices the output queue was emptied puts the
327+ -- siren off.
328+ void $ tryTakeMVar (siren sv)
329+ else if (len + 1 >= 1500) then do
330+ -- We are guaranteed to receive the siren if the consumer reads the
331+ -- queue because the consumer puts the siren on before reading the
332+ -- queue.
333+ --
334+ -- We may get the siren between the siren being swicthed on and the
335+ -- queue getting read but that's harmless, at amost everyone will go
336+ -- back to work and will have to sleep again if queue was still not
337+ -- emptied.
338+ --
339+ -- If even before a worker could read the MVar, the queue gets emptied
340+ -- and another worker queuing on it switches off the siren, then we may
341+ -- sleep here. In that case we are guaranteed to be woken up on the
342+ -- next siren. The next siren is guaranteed as we send a doorbell
343+ -- before switching off the siren, and the consumer switches on the
344+ -- siren after receiving the doorbell.
345+ readMVar (siren sv)
346+ else return ()
316347
317348{-# INLINE sendStop #-}
318349sendStop :: MonadIO m => SVar m a -> m ()
@@ -451,8 +482,8 @@ sendWorkerWait sv = do
451482 -- sent then we block, so that we do not keep looping fruitlessly.
452483
453484 liftIO $ threadDelay 200
454− output <- liftIO $ readIORef (outputQueue sv)
455− when (null output) $ do
485+ (_, n) <- liftIO $ readIORef (outputQueue sv)
486+ when (n <= 0) $ do
456487 done <- queueEmpty sv
457488 if not done
458489 then pushWorker sv >> sendWorkerWait sv
@@ -469,7 +500,8 @@ fromStreamVar sv = Stream $ \_ stp sng yld -> do
469500 res <- liftIO $ tryTakeMVar (doorBell sv)
470501 when (isNothing res) $ sendWorkerWait sv
471502
472− list <- liftIO $ atomicModifyIORefCAS (outputQueue sv) $ \x -> ([], x)
503+ void $ liftIO $ tryPutMVar (siren sv) ()
504+ (list, _) <- liftIO $ atomicModifyIORefCAS (outputQueue sv) $ \x -> (([],0), x)
473505 -- Reversing the output is important to guarantee that we process the
474506 -- outputs in the same order as they were generated by the constituent
475507 -- streams.
@@ -507,14 +539,16 @@ fromStreamVar sv = Stream $ \_ stp sng yld -> do
507539
508540getFifoSVar :: MonadIO m => SVarStyle -> IO (SVar m a)
509541getFifoSVar ctype = do
510− outQ <- newIORef []
542+ outQ <- newIORef ([], 0)
511543 outQMv <- newEmptyMVar
544+ hooter <- newEmptyMVar
512545 active <- newIORef 0
513546 running <- newIORef S.empty
514547 q <- newQ
515548 let sv =
516549 SVar { outputQueue = outQ
517550 , doorBell = outQMv
551+ , siren = hooter
518552 , runningThreads = running
519553 , runqueue = runqueueFIFO sv q
520554 , enqueue = pushL q
@@ -526,8 +560,9 @@ getFifoSVar ctype = do
526560
527561getLifoSVar :: MonadIO m => SVarStyle -> IO (SVar m a)
528562getLifoSVar ctype = do
529− outQ <- newIORef []
563+ outQ <- newIORef ([], 0)
530564 outQMv <- newEmptyMVar
565+ hooter <- newEmptyMVar
531566 active <- newIORef 0
532567 running <- newIORef S.empty
533568 q <- newIORef []
@@ -535,6 +570,7 @@ getLifoSVar ctype = do
535570 let sv =
536571 SVar { outputQueue = outQ
537572 , doorBell = outQMv
573+ , siren = hooter
538574 , runningThreads = running
539575 , runqueue = runqueueLIFO sv q
540576 , enqueue = enqueueLIFO q
@@ -546,13 +582,15 @@ getLifoSVar ctype = do
546582
547583getParSVar :: SVarStyle -> IO (SVar m a)
548584getParSVar style = do
549− outQ <- newIORef []
585+ outQ <- newIORef ([], 0)
550586 outQMv <- newEmptyMVar
587+ hooter <- newEmptyMVar
551588 active <- newIORef 0
552589 running <- newIORef S.empty
553590 let sv =
554591 SVar { outputQueue = outQ
555592 , doorBell = outQMv
593+ , siren = hooter
556594 , runningThreads = running
557595 , runqueue = undefined
558596 , enqueue = undefined