1e51921 · streamly · git.iko.soy

git.iko.soy/streamly

Rename Internal.Mutable.Prim.Var to Internal.Data.IORef.Prim.

- Rename function fromPrimVar to fromPrimIORef.

commit1e5192145bd4ec472596eef3e31ce1b55df6f9df authorpranaysashank <pranaysashank@composewell.com> dateMon 24 Aug 2020, 19:23 (6 years ago) parent971da77

6 files changed+112−111

compare with master

unifiedsplit

.hlint.ignore+1−1 src/Streamly/Internal/Data/IORef/Prim.hs+84 src/Streamly/Internal/Data/Stream/IsStream.hs+6−6 src/Streamly/Internal/Data/Stream/StreamD.hs+20−20 src/Streamly/Internal/Mutable/Prim/Var.hs−83 streamly.cabal+1−1

.hlint.ignore+1−1

@@ -14,7 +14,7 @@ src/Streamly/Internal/Data/Stream/StreamD.hs

1414src/Streamly/Internal/Data/Pipe/Types.hs

1515src/Streamly/Internal/Data/SmallArray/Types.hs

1616src/Streamly/Internal/Data/Unicode/Stream.hs

17−src/Streamly/Internal/Mutable/Prim/Var.hs

17+src/Streamly/Internal/Data/IORef/Prim.hs

1818src/Streamly/Internal/Data/Array/Prim/Types.hs

1919src/Streamly/Internal/Data/Array/Prim/MutTypesInclude.hs

2020src/Streamly/Internal/Data/Array/Prim/TypesInclude.hs

src/Streamly/Internal/Data/IORef/Prim.hsnew file+84

@@ -0,0 +1,84 @@

1+{-# LANGUAGE UnboxedTuples #-}

3+#include "inline.hs"

5+-- |

6+-- Module : Streamly.Internal.Data.IORef.Prim

8+--

9+-- License : BSD3

10+-- Maintainer : streamly@composewell.com

11+-- Stability : experimental

12+-- Portability : GHC

13+--

14+-- A mutable variable in a mutation capable monad (IO) holding a 'Prim'

15+-- value. This allows fast modification because of unboxed storage.

16+--

17+-- = Multithread Consistency Notes

18+--

19+-- In general, any value that straddles a machine word cannot be guaranteed to

20+-- be consistently read from another thread without a lock. GHC heap objects

21+-- are always machine word aligned, therefore, a 'IORef' is also word aligned.

22+-- On a 64-bit platform, writing a 64-bit aligned type from one thread and

23+-- reading it from another thread should give consistent old or new value. The

24+-- same holds true for 32-bit values on a 32-bit platform.

25+

26+module Streamly.Internal.Data.IORef.Prim

27+ (

28+ IORef

29+ , Prim

30+

31+ -- * Construction

32+ , newIORef

33+

34+ -- * Write

35+ , writeIORef

36+ , modifyIORef'

37+

38+ -- * Read

39+ , readIORef

40+ )

41+where

42+

43+import Control.Monad.Primitive (primitive_)

44+import Data.Primitive.Types (Prim, sizeOf#, readByteArray#, writeByteArray#)

45+import GHC.Exts (MutableByteArray#, newByteArray#, RealWorld)

46+import GHC.IO (IO(..))

47+

48+-- | An 'IORef' holds a single 'Prim' value.

49+data IORef a = IORef (MutableByteArray# RealWorld)

50+

51+-- | Create a new 'IORef'.

52+--

53+-- /Internal/

54+{-# INLINE newIORef #-}

55+newIORef :: forall a. Prim a => a -> IO (IORef a)

56+newIORef x = IO (\s# ->

57+ case newByteArray# (sizeOf# (undefined :: a)) s# of

58+ (# s1#, arr# #) ->

59+ case writeByteArray# arr# 0# x s1# of

60+ s2# -> (# s2#, IORef arr# #)

61+ )

62+

63+-- | Write a value to an 'IORef'.

64+--

65+-- /Internal/

66+{-# INLINE writeIORef #-}

67+writeIORef :: Prim a => IORef a -> a -> IO ()

68+writeIORef (IORef arr#) x = primitive_ (writeByteArray# arr# 0# x)

69+

70+-- | Read a value from an 'IORef'.

71+--

72+-- /Internal/

73+{-# INLINE readIORef #-}

74+readIORef :: Prim a => IORef a -> IO a

75+readIORef (IORef arr#) = IO (readByteArray# arr# 0#)

76+

77+-- | Modify the value of an 'IORef' using a function with strict application.

78+--

79+-- /Internal/

80+{-# INLINE modifyIORef' #-}

81+modifyIORef' :: Prim a => IORef a -> (a -> a) -> IO ()

82+modifyIORef' (IORef arr#) g = primitive_ $ \s# ->

83+ case readByteArray# arr# 0# s# of

84+ (# s'#, a #) -> let a' = g a in a' `seq` writeByteArray# arr# 0# a' s'#

src/Streamly/Internal/Data/Stream/IsStream.hs+6−6

@@ -66,7 +66,7 @@ module Streamly.Internal.Data.Stream.IsStream

6666 , fromListM

6767 , K.fromFoldable

6868 , fromFoldableM

69− , fromPrimVar

69+ , fromPrimIORef

7070 , fromCallback

7171

7272 -- * Elimination

@@ -553,7 +553,7 @@ import Streamly.Internal.Data.Pipe.Types (Pipe (..))

553553import Streamly.Internal.Data.Time.Units

554554 ( AbsTime, MilliSecond64(..), addToAbsTime, toRelTime

555555 , toAbsTime, TimeUnit64, RelTime64, addToAbsTime64)

556−import Streamly.Internal.Mutable.Prim.Var (Prim, Var)

556+import Streamly.Internal.Data.IORef.Prim (Prim, IORef)

557557import Streamly.Internal.Data.Tuple.Strict (Tuple'(..))

558558

559559import qualified Streamly.Internal.Memory.Array as A

@@ -978,13 +978,13 @@ fromHandle h = go

978978 str <- liftIO $ IO.hGetLine h

979979 yld str go

980980

981−-- | Construct a stream by reading a 'Prim' 'Var' repeatedly.

981+-- | Construct a stream by reading a 'Prim' 'IORef' repeatedly.

982982--

983983-- /Internal/

984984--

985−{-# INLINE fromPrimVar #-}

986−fromPrimVar :: (IsStream t, MonadIO m, Prim a) => Var IO a -> t m a

987−fromPrimVar = fromStreamD . D.fromPrimVar

985+{-# INLINE fromPrimIORef #-}

986+fromPrimIORef :: (IsStream t, MonadIO m, Prim a) => IORef a -> t m a

987+fromPrimIORef = fromStreamD . D.fromPrimIORef

988988

989989-- | Takes a callback setter function and provides it with a callback. The

990990-- callback when invoked adds a value at the tail of the stream. Returns a

src/Streamly/Internal/Data/Stream/StreamD.hs+20−20

@@ -81,7 +81,7 @@ module Streamly.Internal.Data.Stream.StreamD

8181 , fromListM

8282 , fromStreamK

8383 , fromStreamD

84− , fromPrimVar

84+ , fromPrimIORef

8585 , fromSVar

8686

8787 -- * Elimination

@@ -325,8 +325,7 @@ import Foreign.Storable (Storable(..))

325325import GHC.Types (SPEC(..))

326326import System.Mem (performMajorGC)

327327import Fusion.Plugin.Types (Fuse(..))

328−import Streamly.Internal.Mutable.Prim.Var

329− (Prim, Var, readVar, newVar, modifyVar')

328+import Streamly.Internal.Data.IORef.Prim (Prim)

330329import Streamly.Internal.Data.Time.Units

331330 (TimeUnit64, toRelTime64, diffAbsTime64, RelTime64)

332331import Streamly.Internal.Data.Atomics (atomicModifyIORefCAS_)

@@ -341,6 +340,7 @@ import Streamly.Internal.Data.Unfold.Types (Unfold(..))

341340import Streamly.Internal.Data.Tuple.Strict (Tuple3'(..))

342341import Streamly.Internal.Data.Stream.SVar (fromConsumer, pushToFold)

343342

343+import qualified Streamly.Internal.Data.IORef.Prim as Prim

344344import qualified Streamly.Internal.Data.Pipe.Types as Pipe

345345import qualified Streamly.Internal.Memory.Array.Types as A

346346import qualified Streamly.Internal.Memory.Mutable.Array.Types as MA

@@ -677,12 +677,12 @@ fromListM = Stream step

677677toStreamD :: (K.IsStream t, Monad m) => t m a -> Stream m a

678678toStreamD = fromStreamK . K.toStream

679679

680−{-# INLINE_NORMAL fromPrimVar #-}

681−fromPrimVar :: (MonadIO m, Prim a) => Var IO a -> Stream m a

682−fromPrimVar var = Stream step ()

680+{-# INLINE_NORMAL fromPrimIORef #-}

681+fromPrimIORef :: (MonadIO m, Prim a) => Prim.IORef a -> Stream m a

682+fromPrimIORef var = Stream step ()

683683 where

684684 {-# INLINE_LATE step #-}

685− step _ () = liftIO (readVar var) >>= \x -> return $ Yield x ()

685+ step _ () = liftIO (Prim.readIORef var) >>= \x -> return $ Yield x ()

686686

687687-------------------------------------------------------------------------------

688688-- Generation from SVar

@@ -3811,17 +3811,17 @@ pollCounts predicate transf fld (Stream step state) = Stream step' Nothing

38113811 -- As long as we are using an "Int" for counts lockfree reads from

38123812 -- Var should work correctly on both 32-bit and 64-bit machines.

38133813 -- However, an Int on a 32-bit machine may overflow quickly.

3814− countVar <- liftIO $ newVar (0 :: Int)

3814+ countVar <- liftIO $ Prim.newIORef (0 :: Int)

38153815 tid <- forkManaged

38163816 $ void $ runFold fld

3817− $ transf $ fromPrimVar countVar

3817+ $ transf $ fromPrimIORef countVar

38183818 return $ Skip (Just (countVar, tid, state))

38193819

38203820 step' gst (Just (countVar, tid, st)) = do

38213821 r <- step gst st

38223822 case r of

38233823 Yield x s -> do

3824− when (predicate x) $ liftIO $ modifyVar' countVar (+ 1)

3824+ when (predicate x) $ liftIO $ Prim.modifyIORef' countVar (+ 1)

38253825 return $ Yield x (Just (countVar, tid, s))

38263826 Skip s -> return $ Skip (Just (countVar, tid, s))

38273827 Stop -> do

@@ -3842,12 +3842,12 @@ tapRate samplingRate action (Stream step state) = Stream step' Nothing

38423842 i <-

38433843 MC.catch

38443844 (do liftIO $ threadDelay (round $ samplingRate * 1000000)

3845− i <- liftIO $ readVar countVar

3845+ i <- liftIO $ Prim.readIORef countVar

38463846 let !diff = i - prev

38473847 void $ action diff

38483848 return i)

38493849 (\(e :: AsyncException) -> do

3850− i <- liftIO $ readVar countVar

3850+ i <- liftIO $ Prim.readIORef countVar

38513851 let !diff = i - prev

38523852 void $ action diff

38533853 throwM (MC.toException e))

@@ -3855,7 +3855,7 @@ tapRate samplingRate action (Stream step state) = Stream step' Nothing

38553855

38563856 {-# INLINE_LATE step' #-}

38573857 step' _ Nothing = do

3858− countVar <- liftIO $ newVar 0

3858+ countVar <- liftIO $ Prim.newIORef 0

38593859 tid <- fork $ loop countVar 0

38603860 ref <- liftIO $ newIORef ()

38613861 _ <- liftIO $ mkWeakIORef ref (killThread tid)

@@ -3865,7 +3865,7 @@ tapRate samplingRate action (Stream step state) = Stream step' Nothing

38653865 r <- step gst st

38663866 case r of

38673867 Yield x s -> do

3868− liftIO $ modifyVar' countVar (+ 1)

3868+ liftIO $ Prim.modifyIORef' countVar (+ 1)

38693869 return $ Yield x (Just (countVar, tid, s, ref))

38703870 Skip s -> return $ Skip (Just (countVar, tid, s, ref))

38713871 Stop -> do

@@ -4628,13 +4628,13 @@ dropByTime duration (Stream step1 state1) = Stream step (DropByTimeInit state1)

46284628 Stop -> Stop

46294629

46304630{-# INLINE updateTimeVar #-}

4631−updateTimeVar :: Var IO Int64 -> IO ()

4631+updateTimeVar :: Prim.IORef Int64 -> IO ()

46324632updateTimeVar timeVar = do

46334633 MicroSecond64 t <- fromAbsTime <$> getTime Monotonic

4634− modifyVar' timeVar (const t)

4634+ Prim.modifyIORef' timeVar (const t)

46354635

46364636{-# INLINE updateWithDelay #-}

4637−updateWithDelay :: RealFrac a => a -> Var IO Int64 -> IO ()

4637+updateWithDelay :: RealFrac a => a -> Prim.IORef Int64 -> IO ()

46384638updateWithDelay precision timeVar = do

46394639 threadDelay (delayTime precision)

46404640 updateTimeVar timeVar

@@ -4666,14 +4666,14 @@ times g = Stream step Nothing

46664666 -- XXX note that this is safe only on a 64-bit machine. On a 32-bit

46674667 -- machine a 64-bit 'Var' cannot be read consistently without a lock

46684668 -- while another thread is writing to it.

4669− timeVar <- liftIO $ newVar (0 :: Int64)

4669+ timeVar <- liftIO $ Prim.newIORef (0 :: Int64)

46704670 liftIO $ updateTimeVar timeVar

46714671 tid <- forkManaged $ liftIO $ forever (updateWithDelay g timeVar)

4672− a <- liftIO $ readVar timeVar

4672+ a <- liftIO $ Prim.readIORef timeVar

46734673 return $ Skip $ Just (timeVar, tid, a)

46744674

46754675 step _ s@(Just (timeVar, _, t0)) = do

4676− a <- liftIO $ readVar timeVar

4676+ a <- liftIO $ Prim.readIORef timeVar

46774677 -- XXX we can perhaps use an AbsTime64 using a 64 bit Int for

46784678 -- efficiency. or maybe we can use a representation using Double for

46794679 -- floating precision time

src/Streamly/Internal/Mutable/Prim/Var.hsdeleted−83

@@ -1,83 +0,0 @@

1−{-# LANGUAGE UnboxedTuples #-}

2−

3−#include "inline.hs"

4−

5−-- |

6−-- Module : Streamly.Internal.Mutable.Prim.Var

8−--

9−-- License : BSD3

10−-- Maintainer : streamly@composewell.com

11−-- Stability : experimental

12−-- Portability : GHC

13−--

14−-- A mutable variable in a mutation capable monad (IO/ST) holding a 'Prim'

15−-- value. This allows fast modification because of unboxed storage.

16−--

17−-- = Multithread Consistency Notes

18−--

19−-- In general, any value that straddles a machine word cannot be guaranteed to

20−-- be consistently read from another thread without a lock. GHC heap objects

21−-- are always machine word aligned, therefore, a 'Var' is also word aligned. On

22−-- a 64-bit platform, writing a 64-bit aligned type from one thread and reading

23−-- it from another thread should give consistent old or new value. The same

24−-- holds true for 32-bit values on a 32-bit platform.

25−

26−module Streamly.Internal.Mutable.Prim.Var

27− (

28− Var

29− , MonadMut

30− , Prim

31−

32− -- * Construction

33− , newVar

34−

35− -- * Write

36− , writeVar

37− , modifyVar'

38−

39− -- * Read

40− , readVar

41− )

42−where

43−

44−import Control.Monad.Primitive (PrimMonad(..), primitive_)

45−import Data.Primitive.Types (Prim, sizeOf#, readByteArray#, writeByteArray#)

46−import GHC.Exts (MutableByteArray#, newByteArray#)

47−

48−-- | A 'Var' holds a single 'Prim' value.

49−data Var m a = Var (MutableByteArray# (PrimState m))

50−

51−-- The name PrimMonad does not give a clue what it means, an explicit "Mut"

52−-- suffix provides a better hint. MonadMut is just a generalization of MonadIO.

53−--

54−-- | A monad that allows mutable operations using a state token.

55−type MonadMut = PrimMonad

56−

57−-- | Create a new mutable variable.

58−{-# INLINE newVar #-}

59−newVar :: forall m a. (MonadMut m, Prim a) => a -> m (Var m a)

60−newVar x = primitive (\s# ->

61− case newByteArray# (sizeOf# (undefined :: a)) s# of

62− (# s1#, arr# #) ->

63− case writeByteArray# arr# 0# x s1# of

64− s2# -> (# s2#, Var arr# #)

65− )

66−

67−-- | Write a value to a mutable variable.

68−{-# INLINE writeVar #-}

69−writeVar :: (MonadMut m, Prim a) => Var m a -> a -> m ()

70−writeVar (Var arr#) x = primitive_ (writeByteArray# arr# 0# x)

71−

72−-- | Read a value from a variable.

73−{-# INLINE readVar #-}

74−readVar :: (MonadMut m, Prim a) => Var m a -> m a

75−readVar (Var arr#) = primitive (readByteArray# arr# 0#)

76−

77−-- | Modify the value of a mutable variable using a function with strict

78−-- application.

79−{-# INLINE modifyVar' #-}

80−modifyVar' :: (MonadMut m, Prim a) => Var m a -> (a -> a) -> m ()

81−modifyVar' (Var arr#) g = primitive_ $ \s# ->

82− case readByteArray# arr# 0# s# of

83− (# s'#, a #) -> let a' = g a in a' `seq` writeByteArray# arr# 0# a' s'#

streamly.cabal+1−1

@@ -386,7 +386,7 @@ library

386386 , Streamly.Internal.Data.SVar

387387

388388 -- Mutable data

389− , Streamly.Internal.Mutable.Prim.Var

389+ , Streamly.Internal.Data.IORef.Prim

390390

391391 -- Arrays

392392 , Streamly.Internal.Data.Array