bfec356 · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Implement "interleaved" config option

bfec356Harendra Kumar4 years ago

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

compare with master

unifiedsplit

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

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

70async2 f count n =

71 Stream.fold Fold.drain

72− $ Async.appendWith f

72+ $ Async.combineWith f

73 (sourceUnfoldrM count n) (sourceUnfoldrM count (n + 1))

75{-# INLINE concatAsync2 #-}

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

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

85interleave2 f count n =

86 Stream.fold Fold.drain

87− $ Async.interleaveWith f

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

88 (sourceUnfoldrM count n) (sourceUnfoldrM count (n + 1))

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

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

165 -> IO ()

166concatMapInterleaveStreamsWith f outer inner n =

167 Stream.fold Fold.drain

168− $ Async.concatMapInterleaveWith f

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

169 (sourceUnfoldrM inner) (sourceUnfoldrM outer n)

170

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

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

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

18 , eagerEval

19 , stopWhen

20 , ordered

21+ , interleaved

22 , rate

23 , rate

23 , avgRate

24 , avgRate

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

31 , sequenceWith

32 , sequenceWith

33 , append

34 , append

34− , appendWith

35− , interleave

36− , interleaveWith

37 , ahead

35 , ahead

38 , parallel

36 , parallel

39− , parallelFst

37+ , combineWith

40− , parallelMin

42 , apply

39 , apply

43 , applyWith

40 , applyWith

41+

44 , concatList

42 , concatList

45 , concat

43 , concat

46 , concatWith

44 , concatWith

47 , concatMap

45 , concatMap

48− , concatMapInterleave

49 , concatMapWith

46 , concatMapWith

50− , concatMapInterleaveWith

51 )

47 )

52where

48where

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

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

75 , eagerEval

76 , stopWhen

77 , ordered

78+ , interleaved

79 , rate

80 , rate

80 , avgRate

81 , avgRate

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

93 , getEagerDispatch

94 , getEagerDispatch

94 , getStopWhen

95 , getStopWhen

95 , getOrdered

96 , getOrdered

97+ , getInterleaved

97 -- * Cleanup

99 -- * Cleanup

98 , cleanupSVar

100 , cleanupSVar

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

342 , _eagerDispatch :: Bool

344 , _eagerDispatch :: Bool

343 , _stopWhen :: StopWhen

345 , _stopWhen :: StopWhen

344 , _ordered :: Bool

346 , _ordered :: Bool

347+ , _interleaved :: Bool

345 }

348 }

346

349

347-------------------------------------------------------------------------------

350-------------------------------------------------------------------------------

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

375 , _eagerDispatch = False

378 , _eagerDispatch = False

376 , _stopWhen = AllStop

379 , _stopWhen = AllStop

377 , _ordered = False

380 , _ordered = False

381+ , _interleaved = False

378 }

382 }

379

383

380-------------------------------------------------------------------------------

384-------------------------------------------------------------------------------

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

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

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

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

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

524--

528--

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

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

526--

530--

527ordered :: Config -> Config

531ordered :: Config -> Config

528ordered st = st { _ordered = True }

532ordered st = st { _ordered = True }

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

530getOrdered :: Config -> Bool

534getOrdered :: Config -> Bool

531getOrdered = _ordered

535getOrdered = _ordered

532

536

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+

533-------------------------------------------------------------------------------

550-------------------------------------------------------------------------------

534-- Initialization

551-- Initialization

535-------------------------------------------------------------------------------

552-------------------------------------------------------------------------------

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

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

32 , StopWhen (..)

33 , stopWhen

34 , ordered

35+ , interleaved

35 -- maxYields

36 -- maxYields

36 , Rate(..)

37 , Rate(..)

37 , rate

38 , rate

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

62 -- | Use a channel for each pair.

63 -- | Use a channel for each pair.

63 -- combine/concur/conjoin

64 -- combine/concur/conjoin

64 , append

65 , append

65− , appendWith

66 , interleave

67− , interleaveWith

68 , ahead

67 , ahead

69 , parallel

68 , parallel

70 , parallelFst

69 , parallelFst

71 , parallelMin

70 , parallelMin

71+ , combineWith

73 -- ** Apply

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

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

83 , concatMap

84 , concatMapInterleave

85 , concatMapWith

86− , concatMapInterleaveWith

87 )

86 )

88where

87where

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

140-------------------------------------------------------------------------------

139-------------------------------------------------------------------------------

141

140

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

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

143−{-# INLINE appendWith #-}

142+{-# INLINE combineWith #-}

144−appendWith :: MonadAsync m =>

143+combineWith :: MonadAsync m =>

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

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

146−appendWith modifier stream1 stream2 =

145+combineWith modifier stream1 stream2 =

147 Stream.fromStreamK

146 Stream.fromStreamK

148 $ appendWithK

147 $ appendWithK

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

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

150

149

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−

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

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

161-- left stream.

151-- left stream.

162--

152--

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

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

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

168-- scheduling by creating arbitrary shape expression trees.

158-- scheduling by creating arbitrary shape expression trees.

169--

159--

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

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

171--

161--

172-- The following code finishes in 4 seconds:

162-- The following code finishes in 4 seconds:

173--

163--

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

180--

170--

181{-# INLINE append #-}

171{-# INLINE append #-}

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

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

183−append = appendWith id

173+append = combineWith id

184

174

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

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

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

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

187-- limited number of threads.

177-- limited number of threads.

188--

178--

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

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

190--

180--

191{-# INLINE interleave #-}

181{-# INLINE interleave #-}

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

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

193−interleave = interleaveWith id

183+interleave = combineWith interleaved

194

184

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

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

196--

186--

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

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

198--

188--

199{-# INLINE ahead #-}

189{-# INLINE ahead #-}

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

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

201−ahead = appendWith ordered

191+ahead = combineWith ordered

202

192

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

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

204--

194--

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

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

206--

196--

207{-# INLINE parallel #-}

197{-# INLINE parallel #-}

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

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

209−parallel = appendWith eagerEval

199+parallel = combineWith eagerEval

210

200

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

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

212--

202--

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

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

214--

204--

215-- /Pre-release/

205-- /Pre-release/

216{-# INLINE parallelFst #-}

206{-# INLINE parallelFst #-}

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

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

218−parallelFst = appendWith (eagerEval . stopWhen FirstStops)

208+parallelFst = combineWith (eagerEval . stopWhen FirstStops)

219

209

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

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

221-- stops.

211-- stops.

222--

212--

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

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

224--

214--

225-- /Pre-release/

215-- /Pre-release/

226{-# INLINE parallelMin #-}

216{-# INLINE parallelMin #-}

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

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

228−parallelMin = appendWith (eagerEval . stopWhen AnyStops)

218+parallelMin = combineWith (eagerEval . stopWhen AnyStops)

229

219

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

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

231

221

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

240 $ concatMapWithK

230 $ concatMapWithK

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

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

242

232

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−

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

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

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

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

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

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

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

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

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

299-- robin fashion over the available threads.

278-- robin fashion over the available threads.

300--

279--

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

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

302--

281--

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

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

304--

283--

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

306--

285--

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

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

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

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

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

313--

292--

314{-# INLINE concatMapInterleave #-}

293{-# INLINE concatMapInterleave #-}

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

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

316−concatMapInterleave = concatMapInterleaveWith id

295+concatMapInterleave = concatMapWith interleaved

317

296

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

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

319-- configuration parameters using Config modifiers.

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

10 (

11 -- * Channel

12 Channel

13+ , newChannel

14 -- * Configuration

15 -- * Configuration

15 , Config

16 , Config

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

26 , StopWhen (..)

27 , StopWhen (..)

27 , stopWhen

28 , stopWhen

28 , ordered

29 , ordered

30+ , interleaved

30 -- * Operations

32 -- * Operations

31 , fromChannel

33 , fromChannel

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

39 -- * Stream operations

41 -- * Stream operations

40 , evalWithK

42 , evalWithK

41 , appendWithK

43 , appendWithK

42− , interleaveWithK

43 , concatMapWithK -- appendMap

44 , concatMapWithK -- appendMap

44− , concatMapInterleaveWithK -- interleaveMap

45 )

46where

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

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

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

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

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

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

51 (fromChannel, fromChannelK, toChannel, toChannelK)

52 (fromChannel, fromChannelK, toChannel, toChannelK)

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

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

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

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

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

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+

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

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

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

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

64{-# INLINE withChannelK #-}

77{-# INLINE withChannelK #-}

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

72 where

85 where

74 action = do

87 action = do

75− chan <- Append.newChannel modifier

88+ chan <- newChannel modifier

76 toChannelK chan (evaluator chan input)

89 toChannelK chan (evaluator chan input)

77 return $ fromChannelK chan

90 return $ fromChannelK chan

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−

96-------------------------------------------------------------------------------

92-------------------------------------------------------------------------------

97-- Evaluating a stream

93-- Evaluating a stream

98-------------------------------------------------------------------------------

94-------------------------------------------------------------------------------

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

172-}

168-}

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

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

174

170

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−

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

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

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

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

186--

173--

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

290concatMapWithK modifier f =

277concatMapWithK modifier f =

291 let g = concatMapWithChanKGeneric modifier

278 let g = concatMapWithChanKGeneric modifier

292 in withChannelK (`g` f) modifier

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

30import Data.Kind (Type)

31import GHC.Exts (inline)

32import Streamly.Internal.Control.Concurrent

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

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

34import Streamly.Internal.Data.Atomics

35 (atomicModifyIORefCAS, atomicModifyIORefCAS_)

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

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

1052-- newChannel.

1053{-# INLINABLE newChannel #-}

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

1055−newChannel :: (MonadAsync m) =>

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

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

1057newChannel modifier = do

1058 mrun <- askRunInIO

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

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

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

22import Data.IORef (newIORef, readIORef)

23import Streamly.Internal.Control.Concurrent

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

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

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

27import qualified Data.Set as Set

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

241-- newChannel.

242{-# INLINABLE newChannel #-}

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

244−newChannel :: (MonadAsync m) =>

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

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

246newChannel modifier = do

247 mrun <- askRunInIO

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

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

283 stream = leaf11 `Async.append` leaf12

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

285

286− prop1 "appendWith (maxThreads 1)"

286+ prop1 "combineWith (maxThreads 1)"

287 $ let stream =

288− Async.appendWith (Async.maxThreads 1)

288+ Async.combineWith (Async.maxThreads 1)

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

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

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