9b9b0d8 · streamly · git.iko.soy

core/src/Streamly/Internal/Data/IORef/Unboxed.hs+10−4

compare with master

unifiedsplit

@@ -32,7 +32,7 @@ module Streamly.Internal.Data.IORef.Unboxed

32

33 -- Read

34 , readIORef

35− , toStreamD

35+ , pollIntIORef

36 )

37where

38

@@ -87,10 +87,16 @@ modifyIORef' var g = do

87

88-- | Generate a stream by continuously reading the IORef.

89--

90+-- This operation reads the IORef without any synchronization. It can be

91+-- assumed to be atomic because the IORef (MutableByteArray) is always aligned

92+-- to Int boundaries, we are assuming that compiler uses single instructions to

93+-- access the memory. It may read stale values though until caches are

94+-- synchronised in a multiprocessor architecture.

95+--

90-- /Pre-release/

96-- /Pre-release/

91−{-# INLINE_NORMAL toStreamD #-}

97+{-# INLINE_NORMAL pollIntIORef #-}

92−toStreamD :: (MonadIO m, Unbox a) => IORef a -> D.Stream m a

98+pollIntIORef :: (MonadIO m, Unbox a) => IORef a -> D.Stream m a

93−toStreamD var = D.Stream step ()

99+pollIntIORef var = D.Stream step ()

94

100

95 where

101 where

96

102

src/Streamly/Internal/Data/Stream/Concurrent.hs+1−1

@@ -780,7 +780,7 @@ parTapCount predicate fld (D.Stream step state) = D.Stream step' Nothing

780 countVar <- liftIO $ Unboxed.newIORef (0 :: Int)

781 tid <- forkManaged

782 $ void $ fld

783− $ Unboxed.toStreamD countVar

783+ $ Unboxed.pollIntIORef countVar

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

785

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

src/Streamly/Internal/Data/Stream/IsStream/Generate.hs+2−2

@@ -106,7 +106,7 @@ import Streamly.Internal.Data.Time.Units (AbsTime , RelTime64, addToAbsTime64)

106import Streamly.Data.Serialize (Unbox)

107

108import qualified Streamly.Internal.Data.MutArray as Unboxed

109− (toStreamD, IORef)

109+ (pollIntIORef, IORef)

110import qualified Streamly.Internal.Data.Stream.IsStream.Type as IsStream

111import qualified Streamly.Internal.Data.Stream.Parallel as Par

112import qualified Streamly.Internal.Data.Stream.Serial as Serial

@@ -637,4 +637,4 @@ fromCallback setCallback = concatM $ do

637--

638{-# INLINE fromPrimIORef #-}

639fromPrimIORef :: (IsStream t, MonadIO m, Unbox a) => Unboxed.IORef a -> t m a

640−fromPrimIORef = fromStreamD . Unboxed.toStreamD

640+fromPrimIORef = fromStreamD . Unboxed.pollIntIORef