770887d · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Move concatMapDivK to common channel module

770887dHarendra Kumar4 years ago

src/Streamly/Internal/Data/Stream/Async/Channel.hs+2−17

compare with master

unifiedsplit

@@ -44,7 +44,6 @@ import Control.Monad.IO.Class (MonadIO(liftIO))

44import Streamly.Internal.Control.Concurrent (MonadAsync, askRunInIO)

45import Streamly.Internal.Data.Stream.Async.Channel.Operations

46 (fromChannel, fromChannelK, toChannel, toChannelK)

47−import Streamly.Internal.Data.SVar.Type (adaptState)

49import qualified Streamly.Internal.Data.Stream as Stream

48import qualified Streamly.Internal.Data.Stream as Stream

50import qualified Streamly.Internal.Data.Stream.Async.Channel.Append as Append

49import qualified Streamly.Internal.Data.Stream.Async.Channel.Append as Append

@@ -72,6 +71,8 @@ withChannelK evaluator modifier input = K.concatEffect action

72 toChannelK chan (evaluator chan input)

71 toChannelK chan (evaluator chan input)

73 return $ fromChannelK chan

72 return $ fromChannelK chan

74+-- XXX Dedup with withChannelK

75+

75{-# INLINE withInterleaveChannelK #-}

76{-# INLINE withInterleaveChannelK #-}

76withInterleaveChannelK :: MonadAsync m =>

77withInterleaveChannelK :: MonadAsync m =>

77 (Channel m b -> K.Stream m a -> K.Stream m b)

78 (Channel m b -> K.Stream m a -> K.Stream m b)

@@ -175,22 +176,6 @@ _appendWithChanK chan stream1 stream2 =

175-- concat

176-- concat

176-------------------------------------------------------------------------------

177-------------------------------------------------------------------------------

177

178

178−-- | @concatMapDivK useTail useHead stream@, divides the stream in head and

179−-- tail, maps a stream generator on the head and maps an action on the tail of

180−-- a stream. Returns the stream generated by the head.

181−{-# INLINE concatMapDivK #-}

182−concatMapDivK :: Monad m =>

183− (K.Stream m a -> m ())

184− -> (a -> K.Stream m b)

185− -> K.Stream m a

186− -> K.Stream m b

187−concatMapDivK useTail useHead stream =

188− K.mkStream $ \st yld sng stp -> do

189− let foldShared = K.foldStreamShared st yld sng stp

190− single a = foldShared $ useHead a

191− yieldk a r = useTail r >> single a

192− in K.foldStreamShared (adaptState st) yieldk single stp stream

193−

194-- | A runner function takes a queuing function @q@ and a stream, it splits the

179-- | A runner function takes a queuing function @q@ and a stream, it splits the

195-- input stream, queuing the tail and using the head to generate a stream.

180-- input stream, queuing the tail and using the head to generate a stream.

196-- 'mkEnqueue' takes a runner function and generates the queuing function @q@.

181-- 'mkEnqueue' takes a runner function and generates the queuing function @q@.

src/Streamly/Internal/Data/Stream/Channel/Types.hs+26

@@ -100,6 +100,8 @@ module Streamly.Internal.Data.Stream.Channel.Types

100

101 , withDiagMVar

102 , printSVar

103+

104+ , concatMapDivK

103 )

105 )

104where

106where

105

107

@@ -114,11 +116,13 @@ import Data.IORef (IORef, newIORef, readIORef, writeIORef)

114import Data.Set (Set)

116import Data.Set (Set)

115import Streamly.Internal.Data.Atomics

117import Streamly.Internal.Data.Atomics

116 (atomicModifyIORefCAS, atomicModifyIORefCAS_, storeLoadBarrier)

118 (atomicModifyIORefCAS, atomicModifyIORefCAS_, storeLoadBarrier)

119+import Streamly.Internal.Data.SVar.Type (adaptState)

117import Streamly.Internal.Data.Time.Clock (Clock(Monotonic), getTime)

120import Streamly.Internal.Data.Time.Clock (Clock(Monotonic), getTime)

118import Streamly.Internal.Data.Time.Units (AbsTime, NanoSecond64(..))

121import Streamly.Internal.Data.Time.Units (AbsTime, NanoSecond64(..))

119import System.IO (hPutStrLn, stderr)

122import System.IO (hPutStrLn, stderr)

120

123

121import qualified Data.Set as Set

124import qualified Data.Set as Set

125+import qualified Streamly.Internal.Data.Stream.StreamK.Type as K

122

126

123------------------------------------------------------------------------------

127------------------------------------------------------------------------------

124-- Common types

128-- Common types

@@ -708,3 +712,25 @@ cleanupSVar workerSet = do

708 self <- myThreadId

712 self <- myThreadId

709 Prelude.mapM_ (`throwTo` ThreadAbort)

713 Prelude.mapM_ (`throwTo` ThreadAbort)

710 (Prelude.filter (/= self) $ Set.toList workers)

714 (Prelude.filter (/= self) $ Set.toList workers)

715+

716+-------------------------------------------------------------------------------

717+-- Evaluator

718+-------------------------------------------------------------------------------

719+

720+-- | @concatMapDivK useTail useHead stream@, divides the stream in head and

721+-- tail, maps a stream generator on the head and maps an action on the tail of

722+-- a stream. Returns the stream generated by the head.

723+--

724+-- Used for concurrent evaluation of streams using a Channel.

725+{-# INLINE concatMapDivK #-}

726+concatMapDivK :: Monad m =>

727+ (K.Stream m a -> m ())

728+ -> (a -> K.Stream m b)

729+ -> K.Stream m a

730+ -> K.Stream m b

731+concatMapDivK useTail useHead stream =

732+ K.mkStream $ \st yld sng stp -> do

733+ let foldShared = K.foldStreamShared st yld sng stp

734+ single a = foldShared $ useHead a

735+ yieldk a r = useTail r >> single a

736+ in K.foldStreamShared (adaptState st) yieldk single stp stream