bfec356 · streamly · git.iko.soy

git.iko.soy/streamly

Implement "interleaved" config option

For concurrent streams.

commitbfec3569512d15c4e7f08d0c16624fd7950e6e59 authorHarendra Kumar <harendra@composewell.com> dateSat 15 Oct 2022, 02:52 (4 years ago) parentb7e8907

8 files changed+66−97

compare with master

unifiedsplit

benchmark/Streamly/Benchmark/Data/Stream/AsyncCommon.hs+3−3 src/Streamly/Data/Stream/Concurrent.hs+3−7 src/Streamly/Internal/Data/Stream/Channel/Types.hs+18−1 src/Streamly/Internal/Data/Stream/Concurrent.hs+20−41 src/Streamly/Internal/Data/Stream/Concurrent/Channel.hs+16−39 src/Streamly/Internal/Data/Stream/Concurrent/Channel/Append.hs+2−2 src/Streamly/Internal/Data/Stream/Concurrent/Channel/Interleave.hs+2−2 test/Streamly/Test/Data/Stream/Concurrent.hs+2−2

benchmark/Streamly/Benchmark/Data/Stream/AsyncCommon.hs+3−3

@@ -69,7 +69,7 @@ o_n_heap_buffering value f =

6969async2 :: (Config -> Config) -> Int -> Int -> IO ()

7070async2 f count n =

7171 Stream.fold Fold.drain

72− $ Async.appendWith f

72+ $ Async.combineWith f

7373 (sourceUnfoldrM count n) (sourceUnfoldrM count (n + 1))

7474

7575{-# INLINE concatAsync2 #-}

@@ -84,7 +84,7 @@ concatAsync2 f count n =

8484interleave2 :: (Config -> Config) -> Int -> Int -> IO ()

8585interleave2 f count n =

8686 Stream.fold Fold.drain

87− $ Async.interleaveWith f

87+ $ Async.combineWith (f . Async.interleaved)

8888 (sourceUnfoldrM count n) (sourceUnfoldrM count (n + 1))

8989

9090o_1_space_joining :: Int -> (Config -> Config) -> [Benchmark]

@@ -165,7 +165,7 @@ concatMapInterleaveStreamsWith

165165 -> IO ()

166166concatMapInterleaveStreamsWith f outer inner n =

167167 Stream.fold Fold.drain

168− $ Async.concatMapInterleaveWith f

168+ $ Async.concatMapWith (f . Async.interleaved)

169169 (sourceUnfoldrM inner) (sourceUnfoldrM outer n)

170170

171171o_1_space_concatMap :: Int -> (Config -> Config) -> [Benchmark]

src/Streamly/Data/Stream/Concurrent.hs+3−7

@@ -18,6 +18,7 @@ module Streamly.Data.Stream.Concurrent

1818 , eagerEval

1919 , stopWhen

2020 , ordered

21+ , interleaved

2122

2223 , rate

2324 , avgRate

@@ -31,23 +32,18 @@ module Streamly.Data.Stream.Concurrent

3132 , sequenceWith

3233

3334 , append

34− , appendWith

35− , interleave

36− , interleaveWith

3735 , ahead

3836 , parallel

39− , parallelFst

40− , parallelMin

37+ , combineWith

4138

4239 , apply

4340 , applyWith

41+

4442 , concatList

4543 , concat

4644 , concatWith

4745 , concatMap

48− , concatMapInterleave

4946 , concatMapWith

50− , concatMapInterleaveWith

5147 )

5248where

5349

src/Streamly/Internal/Data/Stream/Channel/Types.hs+18−1

@@ -75,6 +75,7 @@ module Streamly.Internal.Data.Stream.Channel.Types

7575 , eagerEval

7676 , stopWhen

7777 , ordered

78+ , interleaved

7879

7980 , rate

8081 , avgRate

@@ -93,6 +94,7 @@ module Streamly.Internal.Data.Stream.Channel.Types

9394 , getEagerDispatch

9495 , getStopWhen

9596 , getOrdered

97+ , getInterleaved

9698

9799 -- * Cleanup

98100 , cleanupSVar

@@ -342,6 +344,7 @@ data Config = Config

342344 , _eagerDispatch :: Bool

343345 , _stopWhen :: StopWhen

344346 , _ordered :: Bool

347+ , _interleaved :: Bool

345348 }

346349

347350-------------------------------------------------------------------------------

@@ -375,6 +378,7 @@ defaultConfig = Config

375378 , _eagerDispatch = False

376379 , _stopWhen = AllStop

377380 , _ordered = False

381+ , _interleaved = False

378382 }

379383

380384-------------------------------------------------------------------------------

@@ -522,7 +526,7 @@ getStopWhen = _stopWhen

522526-- | When enabled the streams may be evaluated cocnurrently but the results are

523527-- produced in the same sequence as a serial evaluation would produce.

524528--

525−-- /Note:/ this option is not supported with interleave operations.

529+-- /Note:/ Not supported with 'interleaved'.

526530--

527531ordered :: Config -> Config

528532ordered st = st { _ordered = True }

@@ -530,6 +534,19 @@ ordered st = st { _ordered = True }

530534getOrdered :: Config -> Bool

531535getOrdered = _ordered

532536

537+-- | Interleave the streams fairly instead of prioritizing the left stream.

538+-- This schedules all streams in a round robin fashion over limited number of

539+-- threads.

540+--

541+-- /Note:/ Can only be used on finite number of streams.

542+-- /Note:/ Not supported with 'ordered'.

543+--

544+interleaved :: Config -> Config

545+interleaved st = st { _interleaved = True }

546+

547+getInterleaved :: Config -> Bool

548+getInterleaved = _interleaved

549+

533550-------------------------------------------------------------------------------

534551-- Initialization

535552-------------------------------------------------------------------------------

src/Streamly/Internal/Data/Stream/Concurrent.hs+20−41

@@ -32,6 +32,7 @@ module Streamly.Internal.Data.Stream.Concurrent

3232 , StopWhen (..)

3333 , stopWhen

3434 , ordered

35+ , interleaved

3536 -- maxYields

3637 , Rate(..)

3738 , rate

@@ -62,13 +63,12 @@ module Streamly.Internal.Data.Stream.Concurrent

6263 -- | Use a channel for each pair.

6364 -- combine/concur/conjoin

6465 , append

65− , appendWith

6666 , interleave

67− , interleaveWith

6867 , ahead

6968 , parallel

7069 , parallelFst

7170 , parallelMin

71+ , combineWith

7272

7373 -- ** Apply

7474 -- | Use a separate channel for each application.

@@ -83,7 +83,6 @@ module Streamly.Internal.Data.Stream.Concurrent

8383 , concatMap

8484 , concatMapInterleave

8585 , concatMapWith

86− , concatMapInterleaveWith

8786 )

8887where

8988

@@ -140,23 +139,14 @@ eval = evalWith id

140139-------------------------------------------------------------------------------

141140

142141-- | Like 'append' but with a Config modifier.

143−{-# INLINE appendWith #-}

144−appendWith :: MonadAsync m =>

142+{-# INLINE combineWith #-}

143+combineWith :: MonadAsync m =>

145144 (Config -> Config) -> Stream m a -> Stream m a -> Stream m a

146−appendWith modifier stream1 stream2 =

145+combineWith modifier stream1 stream2 =

147146 Stream.fromStreamK

148147 $ appendWithK

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

150149

151−-- | Like 'interleave' but with a Config modifier.

152−{-# INLINE interleaveWith #-}

153−interleaveWith :: MonadAsync m =>

154− (Config -> Config) -> Stream m a -> Stream m a -> Stream m a

155−interleaveWith modifier stream1 stream2 =

156− Stream.fromStreamK

157− $ interleaveWithK

158− modifier (Stream.toStreamK stream1) (Stream.toStreamK stream2)

159−

160150-- | Binary operation to evaluate two streams concurrently prioritizing the

161151-- left stream.

162152--

@@ -167,7 +157,7 @@ interleaveWith modifier stream1 stream2 =

167157-- all streams. However, with this operation you can precisely control the

168158-- scheduling by creating arbitrary shape expression trees.

169159--

170−-- >>> append = Async.appendWith id

160+-- >>> append = Async.combineWith id

171161--

172162-- The following code finishes in 4 seconds:

173163--

@@ -180,52 +170,52 @@ interleaveWith modifier stream1 stream2 =

180170--

181171{-# INLINE append #-}

182172append :: MonadAsync m => Stream m a -> Stream m a -> Stream m a

183−append = appendWith id

173+append = combineWith id

184174

185175-- | Like 'append' but interleaves the streams fairly instead of prioritizing

186176-- the left stream. This schedules all streams in a round robin fashion over

187177-- limited number of threads.

188178--

189−-- >>> interleave = Async.interleaveWith id

179+-- >>> interleave = Async.combineWith Async.interleaved

190180--

191181{-# INLINE interleave #-}

192182interleave :: MonadAsync m => Stream m a -> Stream m a -> Stream m a

193−interleave = interleaveWith id

183+interleave = combineWith interleaved

194184

195185-- | Like 'append' but with 'ordered' on.

196186--

197−-- >>> ahead = Async.appendWith Async.ordered

187+-- >>> ahead = Async.combineWith Async.ordered

198188--

199189{-# INLINE ahead #-}

200190ahead :: MonadAsync m => Stream m a -> Stream m a -> Stream m a

201−ahead = appendWith ordered

191+ahead = combineWith ordered

202192

203193-- | Like 'append' but with 'eagerEval' on.

204194--

205−-- >>> parallel = Async.appendWith Async.eagerEval

195+-- >>> parallel = Async.combineWith Async.eagerEval

206196--

207197{-# INLINE parallel #-}

208198parallel :: MonadAsync m => Stream m a -> Stream m a -> Stream m a

209−parallel = appendWith eagerEval

199+parallel = combineWith eagerEval

210200

211201-- | Like 'parallel' but stops the output as soon as the first stream stops.

212202--

213−-- >>> parallelFst = Async.appendWith (Async.eagerEval . Async.stopWhen Async.FirstStops)

203+-- >>> parallelFst = Async.combineWith (Async.eagerEval . Async.stopWhen Async.FirstStops)

214204--

215205-- /Pre-release/

216206{-# INLINE parallelFst #-}

217207parallelFst :: MonadAsync m => Stream m a -> Stream m a -> Stream m a

218−parallelFst = appendWith (eagerEval . stopWhen FirstStops)

208+parallelFst = combineWith (eagerEval . stopWhen FirstStops)

219209

220210-- | Like 'parallel' but stops the output as soon as any of the two streams

221211-- stops.

222212--

223−-- >>> parallelMin = Async.appendWith (Async.eagerEval . Async.stopWhen Async.AnyStops)

213+-- >>> parallelMin = Async.combineWith (Async.eagerEval . Async.stopWhen Async.AnyStops)

224214--

225215-- /Pre-release/

226216{-# INLINE parallelMin #-}

227217parallelMin :: MonadAsync m => Stream m a -> Stream m a -> Stream m a

228−parallelMin = appendWith (eagerEval . stopWhen AnyStops)

218+parallelMin = combineWith (eagerEval . stopWhen AnyStops)

229219

230220-- concatMapWith modifier f stream = concatWith modifier $ fmap f stream

231221

@@ -240,17 +230,6 @@ concatMapWith modifier f stream =

240230 $ concatMapWithK

241231 modifier (Stream.toStreamK . f) (Stream.toStreamK stream)

242232

243−-- | Like 'concatMapInterleave' but we can also specify the concurrent

244−-- channel's configuration parameters using Config modifiers.

245−--

246−{-# INLINE concatMapInterleaveWith #-}

247−concatMapInterleaveWith :: MonadAsync m =>

248− (Config -> Config) -> (a -> Stream m b) -> Stream m a -> Stream m b

249−concatMapInterleaveWith modifier f stream =

250− Stream.fromStreamK

251− $ concatMapInterleaveWithK

252− modifier (Stream.toStreamK . f) (Stream.toStreamK stream)

253−

254233-- | Map each element of the input to a stream and then concurrently evaluate

255234-- and concatenate the resulting streams. Multiple streams may be evaluated

256235-- concurrently but earlier streams are perferred. Output from the streams are

@@ -298,11 +277,11 @@ concatMap = concatMapWith id

298277-- evaluate the earlier stream first, this schedules all streams in a round

299278-- robin fashion over the available threads.

300279--

301−-- >>> concatMapInterleave = Async.concatMapInterleaveWith id

280+-- >>> concatMapInterleave = Async.concatMapWith Async.interleaved

302281--

303282-- When used with a single thread it behaves like serial interleaving:

304283--

305−-- >>> f cfg xs = Stream.fold Fold.toList $ Async.concatMapInterleaveWith cfg id $ Stream.fromList xs

284+-- >>> f cfg xs = Stream.fold Fold.toList $ Async.concatMapWith (Async.interleaved . cfg) id $ Stream.fromList xs

306285--

307286-- >>> stream1 = Stream.fromList [1,2,3]

308287-- >>> stream2 = Stream.fromList [4,5,6]

@@ -313,7 +292,7 @@ concatMap = concatMapWith id

313292--

314293{-# INLINE concatMapInterleave #-}

315294concatMapInterleave :: MonadAsync m => (a -> Stream m b) -> Stream m a -> Stream m b

316−concatMapInterleave = concatMapInterleaveWith id

295+concatMapInterleave = concatMapWith interleaved

317296

318297-- | Like 'concat' but we can also specify the concurrent channel's

319298-- configuration parameters using Config modifiers.

src/Streamly/Internal/Data/Stream/Concurrent/Channel.hs+16−39

@@ -10,6 +10,7 @@ module Streamly.Internal.Data.Stream.Concurrent.Channel

1010 (

1111 -- * Channel

1212 Channel

13+ , newChannel

1314

1415 -- * Configuration

1516 , Config

@@ -26,6 +27,7 @@ module Streamly.Internal.Data.Stream.Concurrent.Channel

2627 , StopWhen (..)

2728 , stopWhen

2829 , ordered

30+ , interleaved

2931

3032 -- * Operations

3133 , fromChannel

@@ -39,13 +41,12 @@ module Streamly.Internal.Data.Stream.Concurrent.Channel

3941 -- * Stream operations

4042 , evalWithK

4143 , appendWithK

42− , interleaveWithK

4344 , concatMapWithK -- appendMap

44− , concatMapInterleaveWithK -- interleaveMap

4545 )

4646where

4747

4848import Control.Monad.IO.Class (MonadIO(liftIO))

49+import Control.Monad.Trans.Control (MonadBaseControl)

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

5051import Streamly.Internal.Data.Stream.Concurrent.Channel.Operations

5152 (fromChannel, fromChannelK, toChannel, toChannelK)

@@ -59,6 +60,18 @@ import qualified Streamly.Internal.Data.Stream.StreamK.Type as K

5960import Streamly.Internal.Data.Stream.Concurrent.Channel.Type

6061import Streamly.Internal.Data.Stream.Channel.Types

6162

63+-- | Create a new concurrent stream evaluation channel. The monad

64+-- state used to run the stream actions is taken from the call site of

65+-- newChannel.

66+{-# INLINE newChannel #-}

67+newChannel :: (MonadIO m, MonadBaseControl IO m) =>

68+ (Config -> Config) -> m (Channel m a)

69+newChannel modifier =

70+ let cfg = modifier defaultConfig

71+ in if getInterleaved cfg

72+ then Interleave.newChannel modifier

73+ else Append.newChannel modifier

74+

6275-- | Allocate a channel and evaluate the stream using the channel and the

6376-- supplied evaluator function. The evaluator is run in a worker thread.

6477{-# INLINE withChannelK #-}

@@ -72,27 +85,10 @@ withChannelK evaluator modifier input = K.concatEffect action

7285 where

7386

7487 action = do

75− chan <- Append.newChannel modifier

88+ chan <- newChannel modifier

7689 toChannelK chan (evaluator chan input)

7790 return $ fromChannelK chan

7891

79−-- XXX Dedup with withChannelK

80−

81−{-# INLINE withInterleaveChannelK #-}

82−withInterleaveChannelK :: MonadAsync m =>

83− (Channel m b -> K.Stream m a -> K.Stream m b)

84− -> (Config -> Config)

85− -> K.Stream m a

86− -> K.Stream m b

87−withInterleaveChannelK evaluator modifier input = K.concatEffect action

88−

89− where

90−

91− action = do

92− chan <- Interleave.newChannel modifier

93− toChannelK chan (evaluator chan input)

94− return $ Stream.toStreamK $ fromChannel chan

95−

9692-------------------------------------------------------------------------------

9793-- Evaluating a stream

9894-------------------------------------------------------------------------------

@@ -172,15 +168,6 @@ appendWithK modifier stream1 stream2 =

172168-}

173169 concatMapWithK modifier id (stream1 `K.cons` K.fromPure stream2)

174170

175−-- | Create a new channel and add both the streams to it for async evaluation.

176−-- The output stream is the result of the evaluation.

177−{-# INLINE interleaveWithK #-}

178−interleaveWithK :: MonadAsync m =>

179− (Config -> Config) -> K.Stream m a -> K.Stream m a -> K.Stream m a

180−-- interleaveWithK = _appendGeneric Interleave.newChannel

181−interleaveWithK modifier stream1 stream2 =

182− concatMapInterleaveWithK modifier id (stream1 `K.cons` K.fromPure stream2)

183−

184171-- | Evaluate the first stream in the current thread and add the second stream

185172-- to the supplied channel. This is to be used by a worker thread.

186173--

@@ -290,13 +277,3 @@ concatMapWithK :: MonadAsync m =>

290277concatMapWithK modifier f =

291278 let g = concatMapWithChanKGeneric modifier

292279 in withChannelK (`g` f) modifier

293−

294−-- | Allocate a channel and use it to concurrently evaluate the streams

295−-- generated by the mapped function.

296−--

297−{-# INLINE concatMapInterleaveWithK #-}

298−concatMapInterleaveWithK :: MonadAsync m =>

299− (Config -> Config) -> (a -> K.Stream m b) -> K.Stream m a -> K.Stream m b

300−concatMapInterleaveWithK modifier f =

301− let g = concatMapWithChanKGeneric modifier

302− in withInterleaveChannelK (`g` f) modifier

src/Streamly/Internal/Data/Stream/Concurrent/Channel/Append.hs+2−2

@@ -30,7 +30,7 @@ import Data.IORef (IORef, newIORef, readIORef, atomicModifyIORef, writeIORef)

3030import Data.Kind (Type)

3131import GHC.Exts (inline)

3232import Streamly.Internal.Control.Concurrent

33− (MonadRunInIO, MonadAsync, RunInIO(..), askRunInIO)

33+ (MonadRunInIO, RunInIO(..), askRunInIO)

3434import Streamly.Internal.Data.Atomics

3535 (atomicModifyIORefCAS, atomicModifyIORefCAS_)

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

@@ -1052,7 +1052,7 @@ getLifoSVar mrun cfg = do

10521052-- newChannel.

10531053{-# INLINABLE newChannel #-}

10541054{-# SPECIALIZE newChannel :: (Config -> Config) -> IO (Channel IO a) #-}

1055−newChannel :: (MonadAsync m) =>

1055+newChannel :: (MonadIO m, MonadBaseControl IO m) =>

10561056 (Config -> Config) -> m (Channel m a)

10571057newChannel modifier = do

10581058 mrun <- askRunInIO

src/Streamly/Internal/Data/Stream/Concurrent/Channel/Interleave.hs+2−2

@@ -21,7 +21,7 @@ import Control.Monad.Trans.Control (MonadBaseControl, restoreM)

2121import Data.Concurrent.Queue.MichaelScott (LinkedQueue, newQ, nullQ, tryPopR, pushL)

2222import Data.IORef (newIORef, readIORef)

2323import Streamly.Internal.Control.Concurrent

24− (MonadRunInIO, MonadAsync, RunInIO(..), askRunInIO)

24+ (MonadRunInIO, RunInIO(..), askRunInIO)

2525import Streamly.Internal.Data.Stream.Channel.Dispatcher (delThread)

2626

2727import qualified Data.Set as Set

@@ -241,7 +241,7 @@ getFifoSVar mrun cfg = do

241241-- newChannel.

242242{-# INLINABLE newChannel #-}

243243{-# SPECIALIZE newChannel :: (Config -> Config) -> IO (Channel IO a) #-}

244−newChannel :: (MonadAsync m) =>

244+newChannel :: (MonadIO m, MonadBaseControl IO m) =>

245245 (Config -> Config) -> m (Channel m a)

246246newChannel modifier = do

247247 mrun <- askRunInIO

test/Streamly/Test/Data/Stream/Concurrent.hs+2−2

@@ -283,9 +283,9 @@ main = hspec

283283 stream = leaf11 `Async.append` leaf12

284284 in cmp stream sortEq [0, 1, 2, 3, 4, 5, 6,7]

285285

286− prop1 "appendWith (maxThreads 1)"

286+ prop1 "combineWith (maxThreads 1)"

287287 $ let stream =

288− Async.appendWith (Async.maxThreads 1)

288+ Async.combineWith (Async.maxThreads 1)

289289 (Stream.fromList [1,2,3,4,5])

290290 (Stream.fromList [6,7,8,9,10])

291291 in cmp stream (==) [1,2,3,4,5,6,7,8,9,10]