9b9b0d8 · streamly · git.iko.soy

git.iko.soy/streamly

Rename toStreamD to pollIntIORef

commit9b9b0d8548a9d65189b03f2b5c026c99f4af4183 authorHarendra Kumar <harendra@composewell.com> dateFri 24 Nov 2023, 04:02 (3 years ago) parenteed4308

3 files changed+13−7

compare with master

unifiedsplit

core/src/Streamly/Internal/Data/IORef/Unboxed.hs+10−4 src/Streamly/Internal/Data/Stream/Concurrent.hs+1−1 src/Streamly/Internal/Data/Stream/IsStream/Generate.hs+2−2

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

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

3232

3333 -- Read

3434 , readIORef

35− , toStreamD

35+ , pollIntIORef

3636 )

3737where

3838

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

8787

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

8989--

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

9096-- /Pre-release/

91−{-# INLINE_NORMAL toStreamD #-}

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

93−toStreamD var = D.Stream step ()

97+{-# INLINE_NORMAL pollIntIORef #-}

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

99+pollIntIORef var = D.Stream step ()

94100

95101 where

96102

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

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

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

781781 tid <- forkManaged

782782 $ void $ fld

783− $ Unboxed.toStreamD countVar

783+ $ Unboxed.pollIntIORef countVar

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

785785

786786 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)

106106import Streamly.Data.Serialize (Unbox)

107107

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

109− (toStreamD, IORef)

109+ (pollIntIORef, IORef)

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

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

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

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

637637--

638638{-# INLINE fromPrimIORef #-}

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

640−fromPrimIORef = fromStreamD . Unboxed.toStreamD

640+fromPrimIORef = fromStreamD . Unboxed.pollIntIORef