9251eae · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Move concatMapDivK to higher level module

9251eaeHarendra Kumar3 years ago

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

compare with master

unifiedsplit

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

111

112 , withDiagMVar

113 , printSVar

114−

115− , concatMapDivK

116 )

114 )

117where

115where

118

116

@@ -127,13 +125,11 @@ import Data.IORef (IORef, newIORef, readIORef, writeIORef)

127import Data.Set (Set)

125import Data.Set (Set)

128import Streamly.Internal.Data.Atomics

126import Streamly.Internal.Data.Atomics

129 (atomicModifyIORefCAS, atomicModifyIORefCAS_, storeLoadBarrier)

127 (atomicModifyIORefCAS, atomicModifyIORefCAS_, storeLoadBarrier)

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

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

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

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

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

133import System.IO (hPutStrLn, stderr)

130import System.IO (hPutStrLn, stderr)

134

131

135import qualified Data.Set as Set

132import qualified Data.Set as Set

136−import qualified Streamly.Internal.Data.StreamK as K

137

133

138------------------------------------------------------------------------------

134------------------------------------------------------------------------------

139-- Common types

135-- Common types

@@ -819,25 +815,3 @@ cleanupSVar workerSet = do

819 Prelude.mapM_ (`throwTo` ThreadAbort)

815 Prelude.mapM_ (`throwTo` ThreadAbort)

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

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

821 (Set.toList workers)

817 (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))

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

106import Streamly.Internal.Control.ForkLifted (forkManaged)

107import Streamly.Internal.Data.Stream.Channel.Dispatcher (modifyThread)

108−import Streamly.Internal.Data.Stream.Channel.Types

108+import Streamly.Internal.Data.Stream.Channel.Types (ChildEvent(..))

109− ( ChildEvent(..)

110− , concatMapDivK

111− )

112import Streamly.Internal.Data.Stream.Channel.Worker (sendWithDoorBell)

109import Streamly.Internal.Data.Stream.Channel.Worker (sendWithDoorBell)

113import Streamly.Internal.Data.Stream (Stream, Step(..))

110import Streamly.Internal.Data.Stream (Stream, Step(..))

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

114

112

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

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

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

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

@@ -275,6 +273,28 @@ parTwo modifier stream1 stream2 =

275 $ appendWithK

273 $ appendWithK

276 modifier (Stream.toStreamK stream1) (Stream.toStreamK stream2)

274 modifier (Stream.toStreamK stream1) (Stream.toStreamK stream2)

277

275

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+

278-------------------------------------------------------------------------------

298-------------------------------------------------------------------------------

279-- concat streams

299-- concat streams

280-------------------------------------------------------------------------------

300-------------------------------------------------------------------------------