git.iko.soy/streamly
on master help sign in
Move concatMapDivK to higher level module
commit9251eae2d01a385f9ec2ac22d5c757317a3e16f0 authorHarendra Kumar <harendra@composewell.com> dateMon 27 Nov 2023, 01:26 (3 years ago) parenta1e1468
2 files changed+24−30
compare with master
unifiedsplit
src/Streamly/Internal/Data/Stream/Channel/Types.hs−26 src/Streamly/Internal/Data/Stream/Concurrent.hs+24−4
src/Streamly/Internal/Data/Stream/Channel/Types.hs−26
@@ -111,8 +111,6 @@ module Streamly.Internal.Data.Stream.Channel.Types
111111
112112 , withDiagMVar
113113 , printSVar
114−
115− , concatMapDivK
116114 )
117115where
118116
@@ -127,13 +125,11 @@ import Data.IORef (IORef, newIORef, readIORef, writeIORef)
127125import Data.Set (Set)
128126import Streamly.Internal.Data.Atomics
129127 (atomicModifyIORefCAS, atomicModifyIORefCAS_, storeLoadBarrier)
130−import Streamly.Internal.Data.SVar.Type (adaptState)
131128import Streamly.Internal.Data.Time.Clock (Clock(Monotonic), getTime)
132129import Streamly.Internal.Data.Time.Units (AbsTime, NanoSecond64(..))
133130import System.IO (hPutStrLn, stderr)
134131
135132import qualified Data.Set as Set
136−import qualified Streamly.Internal.Data.StreamK as K
137133
138134------------------------------------------------------------------------------
139135-- Common types
@@ -819,25 +815,3 @@ cleanupSVar workerSet = do
819815 Prelude.mapM_ (`throwTo` ThreadAbort)
820816 -- (Prelude.filter (/= self) $ Set.toList workers)
821817 (Set.toList workers)
822−
823−-------------------------------------------------------------------------------
824−-- Evaluator
825−-------------------------------------------------------------------------------
826−
827−-- | @concatMapDivK useTail useHead stream@, divides the stream in head and
828−-- tail, maps a stream generator on the head and maps an action on the tail of
829−-- a stream. Returns the stream generated by the head.
830−--
831−-- Used for concurrent evaluation of streams using a Channel.
832−{-# INLINE concatMapDivK #-}
833−concatMapDivK :: Monad m =>
834− (K.StreamK m a -> m ())
835− -> (a -> K.StreamK m b)
836− -> K.StreamK m a
837− -> K.StreamK m b
838−concatMapDivK useTail useHead stream =
839− K.mkStream $ \st yld sng stp -> do
840− let foldShared = K.foldStreamShared st yld sng stp
841− single a = foldShared $ useHead a
842− yieldk a r = useTail r >> single a
843− in K.foldStreamShared (adaptState st) yieldk single stp stream
src/Streamly/Internal/Data/Stream/Concurrent.hs+24−4
@@ -105,12 +105,10 @@ import Control.Monad.IO.Class (MonadIO(liftIO))
105105import Streamly.Internal.Control.Concurrent (MonadAsync, askRunInIO)
106106import Streamly.Internal.Control.ForkLifted (forkManaged)
107107import Streamly.Internal.Data.Stream.Channel.Dispatcher (modifyThread)
108−import Streamly.Internal.Data.Stream.Channel.Types
109− ( ChildEvent(..)
110− , concatMapDivK
111− )
108+import Streamly.Internal.Data.Stream.Channel.Types (ChildEvent(..))
112109import Streamly.Internal.Data.Stream.Channel.Worker (sendWithDoorBell)
113110import Streamly.Internal.Data.Stream (Stream, Step(..))
111+import Streamly.Internal.Data.SVar.Type (adaptState)
114112
115113import qualified Streamly.Internal.Data.MutArray as Unboxed
116114import qualified Streamly.Internal.Data.Stream as Stream
@@ -275,6 +273,28 @@ parTwo modifier stream1 stream2 =
275273 $ appendWithK
276274 modifier (Stream.toStreamK stream1) (Stream.toStreamK stream2)
277275
276+-------------------------------------------------------------------------------
277+-- Evaluator
278+-------------------------------------------------------------------------------
279+
280+-- | @concatMapDivK useTail useHead stream@, divides the stream in head and
281+-- tail, maps a stream generator on the head and maps an action on the tail of
282+-- a stream. Returns the stream generated by the head.
283+--
284+-- Used for concurrent evaluation of streams using a Channel.
285+{-# INLINE concatMapDivK #-}
286+concatMapDivK :: Monad m =>
287+ (K.StreamK m a -> m ())
288+ -> (a -> K.StreamK m b)
289+ -> K.StreamK m a
290+ -> K.StreamK m b
291+concatMapDivK useTail useHead stream =
292+ K.mkStream $ \st yld sng stp -> do
293+ let foldShared = K.foldStreamShared st yld sng stp
294+ single a = foldShared $ useHead a
295+ yieldk a r = useTail r >> single a
296+ in K.foldStreamShared (adaptState st) yieldk single stp stream
297+
278298-------------------------------------------------------------------------------
279299-- concat streams
280300-------------------------------------------------------------------------------