1e51921 · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

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

1e51921pranaysashank6 years ago

.hlint.ignore+1−1

compare with master

unifiedsplit

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

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

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

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

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

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

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

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

20src/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

66 , fromListM

67 , K.fromFoldable

68 , fromFoldableM

69− , fromPrimVar

69+ , fromPrimIORef

70 , fromCallback

72 -- * Elimination

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

553import Streamly.Internal.Data.Time.Units

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

555 , toAbsTime, TimeUnit64, RelTime64, addToAbsTime64)

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

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

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

558

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

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

978 str <- liftIO $ IO.hGetLine h

979 yld str go

980

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

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

982--

983-- /Internal/

984--

985−{-# INLINE fromPrimVar #-}

985+{-# INLINE fromPrimIORef #-}

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

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

987−fromPrimVar = fromStreamD . D.fromPrimVar

987+fromPrimIORef = fromStreamD . D.fromPrimIORef

988

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

990-- 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

81 , fromListM

82 , fromStreamK

83 , fromStreamD

84− , fromPrimVar

84+ , fromPrimIORef

85 , fromSVar

87 -- * Elimination

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

325import GHC.Types (SPEC(..))

326import System.Mem (performMajorGC)

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

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

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

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

330import Streamly.Internal.Data.Time.Units

329import Streamly.Internal.Data.Time.Units

331 (TimeUnit64, toRelTime64, diffAbsTime64, RelTime64)

330 (TimeUnit64, toRelTime64, diffAbsTime64, RelTime64)

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

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

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

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

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

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

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

343

342

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

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

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

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

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

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

678toStreamD = fromStreamK . K.toStream

679

680−{-# INLINE_NORMAL fromPrimVar #-}

680+{-# INLINE_NORMAL fromPrimIORef #-}

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

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

682−fromPrimVar var = Stream step ()

682+fromPrimIORef var = Stream step ()

683 where

684 {-# INLINE_LATE step #-}

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

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

686

687-------------------------------------------------------------------------------

688-- Generation from SVar

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

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

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

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

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

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

3815 tid <- forkManaged

3816 $ void $ runFold fld

3817− $ transf $ fromPrimVar countVar

3817+ $ transf $ fromPrimIORef countVar

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

3819

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

3821 r <- step gst st

3822 case r of

3823 Yield x s -> do

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

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

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

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

3827 Stop -> do

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

3842 i <-

3843 MC.catch

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

3845− i <- liftIO $ readVar countVar

3845+ i <- liftIO $ Prim.readIORef countVar

3846 let !diff = i - prev

3847 void $ action diff

3848 return i)

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

3850− i <- liftIO $ readVar countVar

3850+ i <- liftIO $ Prim.readIORef countVar

3851 let !diff = i - prev

3852 void $ action diff

3853 throwM (MC.toException e))

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

3855

3856 {-# INLINE_LATE step' #-}

3857 step' _ Nothing = do

3858− countVar <- liftIO $ newVar 0

3858+ countVar <- liftIO $ Prim.newIORef 0

3859 tid <- fork $ loop countVar 0

3860 ref <- liftIO $ newIORef ()

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

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

3865 r <- step gst st

3866 case r of

3867 Yield x s -> do

3868− liftIO $ modifyVar' countVar (+ 1)

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

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

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

3871 Stop -> do

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

4628 Stop -> Stop

4629

4630{-# INLINE updateTimeVar #-}

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

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

4632updateTimeVar timeVar = do

4633 MicroSecond64 t <- fromAbsTime <$> getTime Monotonic

4634− modifyVar' timeVar (const t)

4634+ Prim.modifyIORef' timeVar (const t)

4635

4636{-# INLINE updateWithDelay #-}

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

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

4638updateWithDelay precision timeVar = do

4639 threadDelay (delayTime precision)

4640 updateTimeVar timeVar

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

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

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

4668 -- while another thread is writing to it.

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

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

4670 liftIO $ updateTimeVar timeVar

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

4672− a <- liftIO $ readVar timeVar

4672+ a <- liftIO $ Prim.readIORef timeVar

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

4674

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

4676− a <- liftIO $ readVar timeVar

4676+ a <- liftIO $ Prim.readIORef timeVar

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

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

4679 -- 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

386 , Streamly.Internal.Data.SVar

387

388 -- Mutable data

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

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

390

391 -- Arrays

392 , Streamly.Internal.Data.Array