cbff63f · streamly · git.iko.soy

src/Streamly/Atomics.hsnew file+82

compare with master

unifiedsplit

@@ -0,0 +1,82 @@

1+{-# LANGUAGE CPP #-}

2+

3+-- |

4+-- Module : Streamly.Atomics

6+--

7+-- License : BSD3

8+-- Maintainer : harendra.kumar@gmail.com

9+-- Stability : experimental

10+-- Portability : GHC

11+

12+module Streamly.Atomics

13+ (

14+ atomicModifyIORefCAS

15+ , atomicModifyIORefCAS_

16+ , writeBarrier

17+ , storeLoadBarrier

18+ )

19+where

20+

21+import Data.IORef (IORef, atomicModifyIORef)

22+#ifdef ghcjs_HOST_OS

23+import Data.IORef (modifyIORef)

24+#else

25+import qualified Data.Atomics as A

26+#endif

27+

28+#ifndef ghcjs_HOST_OS

29+

30+-- XXX Does it make sense to have replacements for atomicModifyIORef etc. on a

31+-- single threaded system.

32+--

33+-- Slightly faster version of CAS. Gained some improvement by avoiding the use

34+-- of "evaluate" because we know we do not have exceptions in fn.

35+{-# INLINE atomicModifyIORefCAS #-}

36+atomicModifyIORefCAS :: IORef a -> (a -> (a,b)) -> IO b

37+atomicModifyIORefCAS ref fn = do

38+ tkt <- A.readForCAS ref

39+ loop tkt retries

40+

41+ where

42+

43+ retries = 25 :: Int

44+ loop _ 0 = atomicModifyIORef ref fn

45+ loop old tries = do

46+ let (new, result) = fn $ A.peekTicket old

47+ (success, tkt) <- A.casIORef ref old new

48+ if success

49+ then return result

50+ else loop tkt (tries - 1)

51+

52+{-# INLINE atomicModifyIORefCAS_ #-}

53+atomicModifyIORefCAS_ :: IORef t -> (t -> t) -> IO ()

54+atomicModifyIORefCAS_ = A.atomicModifyIORefCAS_

55+

56+{-# INLINE writeBarrier #-}

57+writeBarrier :: IO ()

58+writeBarrier = A.writeBarrier

59+

60+{-# INLINE storeLoadBarrier #-}

61+storeLoadBarrier :: IO ()

62+storeLoadBarrier = A.storeLoadBarrier

63+

64+#else

65+

66+{-# INLINE atomicModifyIORefCAS #-}

67+atomicModifyIORefCAS :: IORef a -> (a -> (a,b)) -> IO b

68+atomicModifyIORefCAS = atomicModifyIORef

69+

70+{-# INLINE atomicModifyIORefCAS_ #-}

71+atomicModifyIORefCAS_ :: IORef a -> (a -> a) -> IO ()

72+atomicModifyIORefCAS_ = modifyIORef

73+

74+{-# INLINE writeBarrier #-}

75+writeBarrier :: IO ()

76+writeBarrier = return ()

77+

78+{-# INLINE storeLoadBarrier #-}

79+storeLoadBarrier :: IO ()

80+storeLoadBarrier = return ()

81+

82+#endif

src/Streamly/SVar.hs+3−23

@@ -53,7 +53,6 @@ module Streamly.SVar

53 , captureMonadState

54 , RunInIO (..)

55

56− , atomicModifyIORefCAS

57 , WorkerInfo (..)

56 , WorkerInfo (..)

58 , YieldRateInfo (..)

57 , YieldRateInfo (..)

59 , ThreadAbort (..)

58 , ThreadAbort (..)

@@ -116,9 +115,9 @@ import Control.Monad (when)

116import Control.Monad.Catch (MonadThrow)

115import Control.Monad.Catch (MonadThrow)

117import Control.Monad.IO.Class (MonadIO(..))

116import Control.Monad.IO.Class (MonadIO(..))

118import Control.Monad.Trans.Control (MonadBaseControl, control, StM)

117import Control.Monad.Trans.Control (MonadBaseControl, control, StM)

119−import Data.Atomics

118+import Streamly.Atomics

120− (casIORef, readForCAS, peekTicket, atomicModifyIORefCAS_,

119+ (atomicModifyIORefCAS, atomicModifyIORefCAS_, writeBarrier,

121− writeBarrier, storeLoadBarrier)

120+ storeLoadBarrier)

122import Data.Concurrent.Queue.MichaelScott (LinkedQueue, pushL)

121import Data.Concurrent.Queue.MichaelScott (LinkedQueue, pushL)

123import Data.Functor (void)

122import Data.Functor (void)

124import Data.Heap (Heap, Entry(..))

123import Data.Heap (Heap, Entry(..))

@@ -730,25 +729,6 @@ withDiagMVar sv label action =

730-- CAS

729-- CAS

731-------------------------------------------------------------------------------

730-------------------------------------------------------------------------------

732

731

733−-- Slightly faster version of CAS. Gained some improvement by avoiding the use

734−-- of "evaluate" because we know we do not have exceptions in fn.

735−{-# INLINE atomicModifyIORefCAS #-}

736−atomicModifyIORefCAS :: IORef a -> (a -> (a,b)) -> IO b

737−atomicModifyIORefCAS ref fn = do

738− tkt <- readForCAS ref

739− loop tkt retries

740−

741− where

742−

743− retries = 25 :: Int

744− loop _ 0 = atomicModifyIORef ref fn

745− loop old tries = do

746− let (new, result) = fn $ peekTicket old

747− (success, tkt) <- casIORef ref old new

748− if success

749− then return result

750− else loop tkt (tries - 1)

751−

752{-# INLINE ringDoorBell #-}

732{-# INLINE ringDoorBell #-}

753ringDoorBell :: SVar t m a -> IO ()

733ringDoorBell :: SVar t m a -> IO ()

754ringDoorBell sv = do

734ringDoorBell sv = do

src/Streamly/Streams/Async.hs+1

@@ -54,6 +54,7 @@ import Data.Semigroup (Semigroup(..))

54import Prelude hiding (map)

55import qualified Data.Set as S

56

57+import Streamly.Atomics (atomicModifyIORefCAS)

57import Streamly.Streams.SVar (fromSVar)

58import Streamly.Streams.SVar (fromSVar)

58import Streamly.Streams.Serial (map)

59import Streamly.Streams.Serial (map)

59import Streamly.SVar

60import Streamly.SVar

streamly.cabal+6−3

@@ -127,7 +127,8 @@ flag examples-sdl

127

128library

129 hs-source-dirs: src

130− other-modules: Streamly.SVar

130+ other-modules: Streamly.Atomics

131+ , Streamly.SVar

131 , Streamly.Clock

132 , Streamly.Clock

132

133

133 -- Base streams

134 -- Base streams

@@ -484,7 +485,8 @@ benchmark base

484 type: exitcode-stdio-1.0

485 type: exitcode-stdio-1.0

485 hs-source-dirs: benchmark, src

486 hs-source-dirs: benchmark, src

486 main-is: BaseStreams.hs

487 main-is: BaseStreams.hs

487− other-modules: Streamly.SVar

488+ other-modules: Streamly.Atomics

489+ , Streamly.SVar

488 , Streamly.Streams.StreamK.Type

490 , Streamly.Streams.StreamK.Type

489 , Streamly.Streams.StreamK

491 , Streamly.Streams.StreamK

490 , Streamly.Streams.StreamD.Type

492 , Streamly.Streams.StreamD.Type

@@ -542,7 +544,8 @@ benchmark base

542executable nano-bench

544executable nano-bench

543 hs-source-dirs: benchmark, src

545 hs-source-dirs: benchmark, src

544 main-is: NanoBenchmarks.hs

546 main-is: NanoBenchmarks.hs

545− other-modules: Streamly.SVar

547+ other-modules: Streamly.Atomics

548+ , Streamly.SVar

546 , Streamly.Clock

549 , Streamly.Clock

547 , Streamly.Streams.StreamK.Type

550 , Streamly.Streams.StreamK.Type

548 , Streamly.Streams.StreamK

551 , Streamly.Streams.StreamK