3d2c545 · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Add write barriers for doorBell

3d2c545Harendra Kumar8 years ago

src/Streamly/Core.hs+13−5

compare with master

unifiedsplit

@@ -75,7 +75,8 @@ import Control.Monad.IO.Class (MonadIO(..))

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

76import Control.Monad.Trans.Control (MonadBaseControl, control)

77import Data.Atomics (casIORef, readForCAS, peekTicket

78− ,atomicModifyIORefCAS_)

78+ ,atomicModifyIORefCAS_

79+ ,writeBarrier)

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

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

80 tryPopR, nullQ)

81 tryPopR, nullQ)

81import Data.Functor (void)

82import Data.Functor (void)

@@ -424,9 +425,9 @@ send sv msg = do

424 len <- atomicModifyIORefCAS (outputQueue sv) $ \(es, n) ->

425 len <- atomicModifyIORefCAS (outputQueue sv) $ \(es, n) ->

425 ((msg : es, n + 1), n)

426 ((msg : es, n + 1), n)

426 when (len <= 0) $ do

427 when (len <= 0) $ do

427− -- XXX need a memory barrier? The wake up must happen only after the

428+ -- The wake up must happen only after the store has finished otherwise

428− -- store has finished otherwise we can have lost wakeup problems.

429+ -- we can have lost wakeup problems.

429− --

430+ writeBarrier

430 -- Since multiple workers can try this at the same time, it is possible

431 -- Since multiple workers can try this at the same time, it is possible

431 -- that we may put a spurious MVar after the consumer has already seen

432 -- that we may put a spurious MVar after the consumer has already seen

432 -- the output. But that's harmless, at worst it may cause the consumer

433 -- the output. But that's harmless, at worst it may cause the consumer

@@ -463,6 +464,7 @@ enqueueLIFO sv q m = do

463 case ms of

464 case ms of

464 [] -> (m : ms, True)

465 [] -> (m : ms, True)

465 _ -> (m : ms, False)

466 _ -> (m : ms, False)

467+ writeBarrier

466 when v $ void $ tryPutMVar (doorBell sv) ()

468 when v $ void $ tryPutMVar (doorBell sv) ()

467 atomicModifyIORefCAS_ (waitingForWork sv) (const False)

469 atomicModifyIORefCAS_ (waitingForWork sv) (const False)

468

470

@@ -507,7 +509,9 @@ enqueueFIFO sv q m = do

507 withDoorBell = do

509 withDoorBell = do

508 emp <- nullQ q

510 emp <- nullQ q

509 pushL q m

511 pushL q m

512+ writeBarrier

510 when emp $ void $ tryPutMVar (doorBell sv) ()

513 when emp $ void $ tryPutMVar (doorBell sv) ()

514+ atomicModifyIORefCAS_ (waitingForWork sv) (const False)

511

515

512runqueueFIFO :: MonadIO m => SVar m a -> LinkedQueue (Stream m a) -> m ()

516runqueueFIFO :: MonadIO m => SVar m a -> LinkedQueue (Stream m a) -> m ()

513runqueueFIFO sv q = run

517runqueueFIFO sv q = run

@@ -612,6 +616,7 @@ enqueueAhead sv q m = do

612 ([], n) -> ([m], n + 1) -- increment sequence

616 ([], n) -> ([m], n + 1) -- increment sequence

613 _ -> error "not empty"

617 _ -> error "not empty"

614 when w $ do

618 when w $ do

619+ writeBarrier

615 void $ tryPutMVar (doorBell sv) ()

620 void $ tryPutMVar (doorBell sv) ()

616 atomicModifyIORefCAS_ (waitingForWork sv) (const False)

621 atomicModifyIORefCAS_ (waitingForWork sv) (const False)

617

622

@@ -770,7 +775,9 @@ modifyThread sv tid = do

770 then let new = (S.delete tid old) in (new, new)

775 then let new = (S.delete tid old) in (new, new)

771 else let new = (S.insert tid old) in (new, old)

776 else let new = (S.insert tid old) in (new, old)

772 if null changed

777 if null changed

778+ then liftIO $ do

779+ writeBarrier

773− then liftIO $ void $ tryPutMVar (doorBell sv) ()

780+ void $ tryPutMVar (doorBell sv) ()

774 else return ()

781 else return ()

775

782

776-- | This is safe even if we are adding more threads concurrently because if

783-- | This is safe even if we are adding more threads concurrently because if

@@ -878,6 +885,7 @@ sendWorkerWait sv = do

878 -- the enqueue to send us a doorbell.

885 -- the enqueue to send us a doorbell.

879

886

880 liftIO $ atomicModifyIORefCAS_ (waitingForWork sv) $ const True

887 liftIO $ atomicModifyIORefCAS_ (waitingForWork sv) $ const True

888+ liftIO $ writeBarrier

881

889

882 -- check again, this time we have set the waitingForWork flag so we

890 -- check again, this time we have set the waitingForWork flag so we

883 -- are guaranteed to get a doorbell in case the status changed

891 -- are guaranteed to get a doorbell in case the status changed