@@ -75,7 +75,8 @@ import Control.Monad.IO.Class (MonadIO(..))
7575import Control.Monad.Trans.Class (MonadTrans (lift))
7676import Control.Monad.Trans.Control (MonadBaseControl, control)
7777import Data.Atomics (casIORef, readForCAS, peekTicket
78− ,atomicModifyIORefCAS_)
78+ ,atomicModifyIORefCAS_
79+ ,writeBarrier)
7980import Data.Concurrent.Queue.MichaelScott (LinkedQueue, newQ, pushL,
8081 tryPopR, nullQ)
8182import Data.Functor (void)
@@ -424,9 +425,9 @@ send sv msg = do
424425 len <- atomicModifyIORefCAS (outputQueue sv) $ \(es, n) ->
425426 ((msg : es, n + 1), n)
426427 when (len <= 0) $ do
427− -- XXX need a memory barrier? The wake up must happen only after the
428− -- store has finished otherwise we can have lost wakeup problems.
429− --
428+ -- The wake up must happen only after the store has finished otherwise
429+ -- we can have lost wakeup problems.
430+ writeBarrier
430431 -- Since multiple workers can try this at the same time, it is possible
431432 -- that we may put a spurious MVar after the consumer has already seen
432433 -- the output. But that's harmless, at worst it may cause the consumer
@@ -463,6 +464,7 @@ enqueueLIFO sv q m = do
463464 case ms of
464465 [] -> (m : ms, True)
465466 _ -> (m : ms, False)
467+ writeBarrier
466468 when v $ void $ tryPutMVar (doorBell sv) ()
467469 atomicModifyIORefCAS_ (waitingForWork sv) (const False)
468470
@@ -507,7 +509,9 @@ enqueueFIFO sv q m = do
507509 withDoorBell = do
508510 emp <- nullQ q
509511 pushL q m
512+ writeBarrier
510513 when emp $ void $ tryPutMVar (doorBell sv) ()
514+ atomicModifyIORefCAS_ (waitingForWork sv) (const False)
511515
512516runqueueFIFO :: MonadIO m => SVar m a -> LinkedQueue (Stream m a) -> m ()
513517runqueueFIFO sv q = run
@@ -612,6 +616,7 @@ enqueueAhead sv q m = do
612616 ([], n) -> ([m], n + 1) -- increment sequence
613617 _ -> error "not empty"
614618 when w $ do
619+ writeBarrier
615620 void $ tryPutMVar (doorBell sv) ()
616621 atomicModifyIORefCAS_ (waitingForWork sv) (const False)
617622
@@ -770,7 +775,9 @@ modifyThread sv tid = do
770775 then let new = (S.delete tid old) in (new, new)
771776 else let new = (S.insert tid old) in (new, old)
772777 if null changed
773− then liftIO $ void $ tryPutMVar (doorBell sv) ()
778+ then liftIO $ do
779+ writeBarrier
780+ void $ tryPutMVar (doorBell sv) ()
774781 else return ()
775782
776783-- | This is safe even if we are adding more threads concurrently because if
@@ -878,6 +885,7 @@ sendWorkerWait sv = do
878885 -- the enqueue to send us a doorbell.
879886
880887 liftIO $ atomicModifyIORefCAS_ (waitingForWork sv) $ const True
888+ liftIO $ writeBarrier
881889
882890 -- check again, this time we have set the waitingForWork flag so we
883891 -- are guaranteed to get a doorbell in case the status changed