f7415e9 · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Use list based concat functions in place of binary ops

f7415e9Harendra Kumar4 years ago

src/Streamly/Data/Stream/Concurrent.hs+11−16

compare with master

unifiedsplit

@@ -41,38 +41,33 @@ module Streamly.Data.Stream.Concurrent

41 -- | Stream combinators using a concurrent channel.

43 -- ** Evaluate

44− -- | Evaluate a stream concurrently using a channel.

44+ -- | Evaluates a stream concurrently using a channel.

45 , eval

46 , evalWith

48 -- ** Map

49− -- | Use a single channel to evaluate mapped actions concurrently.

49+ -- | Uses a single channel to evaluate mapped actions concurrently.

50 , mapM

51 , mapMWith

52 , sequence

53 , sequenceWith

55− -- ** Combine two

55+ -- ** List of streams

56− -- | Use one channel for each pair. When you have to chain more than two

57− -- actions concat family of operations are much more efficient because they

58− -- use a single channel for all streams.

56+ -- | Shares a single channel across many streams.

59− --

60− -- XXX Do not expose binary operations, instead use these to concatenate n

61− -- streams from a list in the given style?

62 , append

57 , append

63 , ahead

58 , ahead

64 , parallel

59 , parallel

65− , combineWith

60+ , concatListWith

62+ -- ** Stream of streams

67− -- ** Apply

63+ -- *** Apply

68− -- | Apply arguments to a function concurrently. Uses a separate channel

64+ -- | Apply argument streams to a function concurrently. Uses a separate

69− -- for each application.

65+ -- channel for each application.

70 , apply

66 , apply

71 , applyWith

67 , applyWith

73− -- ** Combine many

69+ -- *** Concat

74− -- | Share a single channel across many streams.

70+ -- | Shares a single channel across many streams.

75− , concatList

76 , concat

71 , concat

77 , concatWith

72 , concatWith

78 , concatMap

73 , concatMap

src/Streamly/Internal/Data/Stream/Concurrent.hs+117−45

@@ -46,13 +46,13 @@ module Streamly.Internal.Data.Stream.Concurrent

46 -- | Stream combinators using a concurrent channel

48 -- ** Evaluate

49− -- | Evaluate a stream concurrently using a channel.

49+ -- | Evaluates a stream concurrently using a channel.

50 , eval

51 , evalWith

52 -- Add unfoldrM/iterateM?

54 -- ** Map

55− -- | Use a single channel to evaluate all actions.

55+ -- | Uses a single channel to evaluate all actions.

56 , mapM

57 , mapMWith

58 , sequence

@@ -62,22 +62,32 @@ module Streamly.Internal.Data.Stream.Concurrent

62 -- ** Combine two

63 -- | Use a channel for each pair.

64 -- combine/concur/conjoin

65+ , append2

66+ , interleave2

67+ , ahead2

68+ , parallel2

69+ , parallelFst2

70+ , parallelMin2

71+ , combineWith

72+

73+ -- ** List of streams

74+ -- | Shares a single channel across many streams.

65 , append

75 , append

66 , interleave

76 , interleave

67 , ahead

77 , ahead

68 , parallel

78 , parallel

69 , parallelFst

79 , parallelFst

70 , parallelMin

80 , parallelMin

71− , combineWith

81+ , concatListWith

83+ -- ** Stream of streams

73− -- ** Apply

84+ -- *** Apply

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

85+ -- | Uses a separate channel for each application.

75 , apply

86 , apply

76 , applyWith

87 , applyWith

78− -- ** Combine many

89+ -- *** Concat

79− -- | Share a single channel across many streams.

90+ -- | Shares a single channel across many streams.

80− , concatList

81 , concat

91 , concat

82 , concatWith

92 , concatWith

83 , concatMap

93 , concatMap

@@ -161,7 +171,7 @@ eval :: MonadAsync m => Stream m a -> Stream m a

161eval = evalWith id

171eval = evalWith id

162

172

163-------------------------------------------------------------------------------

173-------------------------------------------------------------------------------

164−-- appending two streams

174+-- combining two streams

165-------------------------------------------------------------------------------

175-------------------------------------------------------------------------------

166

176

167{-# INLINE _appendGeneric #-}

177{-# INLINE _appendGeneric #-}

@@ -239,68 +249,68 @@ combineWith modifier stream1 stream2 =

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

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

240-- scheduling by creating arbitrary shape expression trees.

250-- scheduling by creating arbitrary shape expression trees.

241--

251--

242−-- >>> append = Async.combineWith id

252+-- >>> append2 = Async.combineWith id

243--

253--

244-- The following code finishes in 4 seconds:

254-- The following code finishes in 4 seconds:

245--

255--

246-- >>> stream1 = Stream.fromEffect (delay 4)

256-- >>> stream1 = Stream.fromEffect (delay 4)

247-- >>> stream2 = Stream.fromEffect (delay 2)

257-- >>> stream2 = Stream.fromEffect (delay 2)

248−-- >>> Stream.fold Fold.toList $ stream1 `Async.append` stream2

258+-- >>> Stream.fold Fold.toList $ stream1 `Async.append2` stream2

249-- 2 sec

259-- 2 sec

250-- 4 sec

260-- 4 sec

251-- [2,4]

261-- [2,4]

252--

262--

253−{-# INLINE append #-}

263+{-# INLINE append2 #-}

254−append :: MonadAsync m => Stream m a -> Stream m a -> Stream m a

264+append2 :: MonadAsync m => Stream m a -> Stream m a -> Stream m a

255−append = combineWith id

265+append2 = combineWith id

256

266

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

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

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

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

259-- limited number of threads.

269-- limited number of threads.

260--

270--

261−-- >>> interleave = Async.combineWith Async.interleaved

271+-- >>> interleave2 = Async.combineWith Async.interleaved

262--

272--

263−{-# INLINE interleave #-}

273+{-# INLINE interleave2 #-}

264−interleave :: MonadAsync m => Stream m a -> Stream m a -> Stream m a

274+interleave2 :: MonadAsync m => Stream m a -> Stream m a -> Stream m a

265−interleave = combineWith interleaved

275+interleave2 = combineWith interleaved

266

276

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

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

268--

278--

269−-- >>> ahead = Async.combineWith Async.ordered

279+-- >>> ahead2 = Async.combineWith Async.ordered

270--

280--

271−{-# INLINE ahead #-}

281+{-# INLINE ahead2 #-}

272−ahead :: MonadAsync m => Stream m a -> Stream m a -> Stream m a

282+ahead2 :: MonadAsync m => Stream m a -> Stream m a -> Stream m a

273−ahead = combineWith ordered

283+ahead2 = combineWith ordered

274

284

275−-- | Like 'append' but with 'eager' on.

285+-- | Like 'append2' but with 'eager' on.

276--

286--

277−-- >>> parallel = Async.combineWith Async.eager

287+-- >>> parallel2 = Async.combineWith Async.eager

278--

288--

279−{-# INLINE parallel #-}

289+{-# INLINE parallel2 #-}

280−parallel :: MonadAsync m => Stream m a -> Stream m a -> Stream m a

290+parallel2 :: MonadAsync m => Stream m a -> Stream m a -> Stream m a

281−parallel = combineWith eager

291+parallel2 = combineWith eager

282

292

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

293+-- | Like 'parallel2' but stops the output as soon as the first stream stops.

284--

294--

285−-- >>> parallelFst = Async.combineWith (Async.eager . Async.stopWhen Async.FirstStops)

295+-- >>> parallelFst2 = Async.combineWith (Async.eager . Async.stopWhen Async.FirstStops)

286--

296--

287-- /Pre-release/

297-- /Pre-release/

288−{-# INLINE parallelFst #-}

298+{-# INLINE parallelFst2 #-}

289−parallelFst :: MonadAsync m => Stream m a -> Stream m a -> Stream m a

299+parallelFst2 :: MonadAsync m => Stream m a -> Stream m a -> Stream m a

290−parallelFst = combineWith (eager . stopWhen FirstStops)

300+parallelFst2 = combineWith (eager . stopWhen FirstStops)

291

301

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

302+-- | Like 'parallel2' but stops the output as soon as any of the two streams

293-- stops.

303-- stops.

294--

304--

295−-- >>> parallelMin = Async.combineWith (Async.eager . Async.stopWhen Async.AnyStops)

305+-- >>> parallelMin2 = Async.combineWith (Async.eager . Async.stopWhen Async.AnyStops)

296--

306--

297-- /Pre-release/

307-- /Pre-release/

298−{-# INLINE parallelMin #-}

308+{-# INLINE parallelMin2 #-}

299−parallelMin :: MonadAsync m => Stream m a -> Stream m a -> Stream m a

309+parallelMin2 :: MonadAsync m => Stream m a -> Stream m a -> Stream m a

300−parallelMin = combineWith (eager . stopWhen AnyStops)

310+parallelMin2 = combineWith (eager . stopWhen AnyStops)

301

311

302-------------------------------------------------------------------------------

312-------------------------------------------------------------------------------

303−-- concat

313+-- concat streams

304-------------------------------------------------------------------------------

314-------------------------------------------------------------------------------

305

315

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

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

@@ -327,9 +337,6 @@ mkEnqueue chan runner = do

327 eagerDispatch chan

337 eagerDispatch chan

328 in q

338 in q

329

339

330−-- XXX Can be renamed to concatMapWithK if we move concatMapWithK to higher

331−-- level module. We can keep only Channel based ops in this module.

332−

333-- | Takes the head element of the input stream and queues the tail of the

340-- | Takes the head element of the input stream and queues the tail of the

334-- stream to the channel, then maps the supplied function on the head and

341-- stream to the channel, then maps the supplied function on the head and

335-- evaluates the resulting stream.

342-- evaluates the resulting stream.

@@ -489,13 +496,74 @@ concatWith modifier = concatMapWith modifier id

489concat :: MonadAsync m => Stream m (Stream m a) -> Stream m a

496concat :: MonadAsync m => Stream m (Stream m a) -> Stream m a

490concat = concatWith id

497concat = concatWith id

491

498

499+-------------------------------------------------------------------------------

500+-- concat Lists

501+-------------------------------------------------------------------------------

502+

503+-- | Like 'concatWith' but works on a list of streams.

504+--

505+-- >>> concatListWith modifier = Async.concatWith modifier . Stream.fromList

506+--

507+{-# INLINE concatListWith #-}

508+concatListWith :: MonadAsync m => (Config -> Config) -> [Stream m a] -> Stream m a

509+concatListWith modifier = concatWith modifier . Stream.fromList

510+

492-- | Like 'concat' but works on a list of streams.

511-- | Like 'concat' but works on a list of streams.

493--

512--

494−-- >>> concatList = Async.concat . Stream.fromList

513+-- >>> append = Async.concatListWith id

514+--

515+{-# INLINE append #-}

516+append :: MonadAsync m => [Stream m a] -> Stream m a

517+append = concatListWith id

518+

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

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

521+-- limited number of threads.

522+--

523+-- >>> interleave = Async.concatListWith Async.interleaved

524+--

525+{-# INLINE interleave #-}

526+interleave :: MonadAsync m => [Stream m a] -> Stream m a

527+interleave = concatListWith interleaved

528+

529+-- | Like 'append' but with 'ordered' on.

530+--

531+-- >>> ahead = Async.concatListWith Async.ordered

532+--

533+{-# INLINE ahead #-}

534+ahead :: MonadAsync m => [Stream m a] -> Stream m a

535+ahead = concatListWith ordered

536+

537+-- | Like 'append' but with 'eager' on.

538+--

539+-- >>> parallel = Async.concatListWith Async.eager

540+--

541+{-# INLINE parallel #-}

542+parallel :: MonadAsync m => [Stream m a] -> Stream m a

543+parallel = concatListWith eager

544+

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

546+--

547+-- >>> parallelFst = Async.concatListWith (Async.eager . Async.stopWhen Async.FirstStops)

548+--

549+-- /Pre-release/

550+{-# INLINE parallelFst #-}

551+parallelFst :: MonadAsync m => [Stream m a] -> Stream m a

552+parallelFst = concatListWith (eager . stopWhen FirstStops)

553+

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

555+-- stops.

556+--

557+-- >>> parallelMin = Async.concatListWith (Async.eager . Async.stopWhen Async.AnyStops)

495--

558--

496−{-# INLINE concatList #-}

559+-- /Pre-release/

560+{-# INLINE parallelMin #-}

497−concatList :: MonadAsync m => [Stream m a] -> Stream m a

561+parallelMin :: MonadAsync m => [Stream m a] -> Stream m a

498−concatList = concat . Stream.fromList

562+parallelMin = concatListWith (eager . stopWhen AnyStops)

563+

564+-------------------------------------------------------------------------------

565+-- Applicative

566+-------------------------------------------------------------------------------

499

567

500{-# INLINE applyWith #-}

568{-# INLINE applyWith #-}

501{-# SPECIALIZE applyWith ::

569{-# SPECIALIZE applyWith ::

@@ -513,6 +581,10 @@ applyWith modifier stream1 stream2 =

513apply :: MonadAsync m => Stream m (a -> b) -> Stream m a -> Stream m b

581apply :: MonadAsync m => Stream m (a -> b) -> Stream m a -> Stream m b

514apply = applyWith id

582apply = applyWith id

515

583

584+-------------------------------------------------------------------------------

585+-- Map

586+-------------------------------------------------------------------------------

587+

516-- |

588-- |

517-- >>> mapMWith modifier f = Async.concatMapWith modifier (Stream.fromEffect . f)

589-- >>> mapMWith modifier f = Async.concatMapWith modifier (Stream.fromEffect . f)

518--

590--

test/Streamly/Test/Data/Stream/Concurrent.hs+27−29

@@ -68,8 +68,9 @@ transformCombineFromList ::

68transformCombineFromList constr eq listOp op a b c =

69 withMaxSuccess maxTestCount $

70 monadicIO $ do

71+ let s1 = op (Async.append [constr b, constr c])

72+ let s2 = Async.append [constr a, s1]

71− stream <- run (Stream.fold Fold.toList $

73+ stream <- run (Stream.fold Fold.toList s2)

72− constr a `Async.append` op (constr b `Async.append` constr c))

73 let list = a <> listOp (b <> c)

74 let list = a <> listOp (b <> c)

74 listEquals eq stream list

75 listEquals eq stream list

@@ -144,8 +145,8 @@ exceptionPropagation f = do

144 (Left (ExampleException "E") :: Either ExampleException [Int])

145 (Left (ExampleException "E") :: Either ExampleException [Int])

145

146

146 it "concatMap throwM" $ do

147 it "concatMap throwM" $ do

147− let s1 = Async.concatList $ fmap Stream.fromPure [1..4]

148+ let s1 = Async.concatListWith id $ fmap Stream.fromPure [1..4]

148− s2 = Async.concatList $ fmap Stream.fromPure [5..8]

149+ s2 = Async.concatListWith id $ fmap Stream.fromPure [5..8]

149 try $ tl (

150 try $ tl (

150 let bind = flip Async.concatMap

151 let bind = flip Async.concatMap

151 in bind s1 $ \x ->

152 in bind s1 $ \x ->

@@ -186,9 +187,8 @@ timeOrdering f = do

186takeCombined :: Int -> IO ()

187takeCombined :: Int -> IO ()

187takeCombined n = do

188takeCombined n = do

188 let constr = Stream.fromFoldable

189 let constr = Stream.fromFoldable

190+ let s = Async.append [constr ([] :: [Int]), constr ([] :: [Int])]

189− r <- Stream.fold Fold.toList $

191+ r <- Stream.fold Fold.toList $ Stream.take n s

190− Stream.take n

191− (constr ([] :: [Int]) `Async.append` constr ([] :: [Int]))

192 r `shouldBe` []

193

194---------------------------------------------------------------------------

@@ -243,44 +243,42 @@ main = hspec

243 -- XXX Need to use eq instead of sortEq for ahead oeprations

244 -- Binary append

245 prop1 "append [] []"

246− $ cmp (Stream.nil `Async.append` Stream.nil) sortEq []

246+ $ cmp (Async.append [Stream.nil, Stream.nil]) sortEq []

247 prop1 "append [] [1]"

248− $ cmp (Stream.nil `Async.append` Stream.fromPure 1) sortEq [1]

248+ $ cmp (Async.append [Stream.nil, Stream.fromPure 1]) sortEq [1]

249 prop1 "append [1] []"

250− $ cmp (Stream.fromPure 1 `Async.append` Stream.nil) sortEq [1]

250+ $ cmp (Async.append [Stream.fromPure 1, Stream.nil]) sortEq [1]

251 prop1 "append [0] [1]"

252− $ let stream =

253− Stream.fromPure 0 `Async.append` Stream.fromPure 1

252+ $ let stream = Async.append [Stream.fromPure 0, Stream.fromPure 1]

254 in cmp stream sortEq [0, 1]

253 in cmp stream sortEq [0, 1]

255

254

256 prop1 "append [0] [] [1]"

255 prop1 "append [0] [] [1]"

257 $ let stream =

256 $ let stream =

257+ Async.append

258− Stream.fromPure 0

258+ [Stream.fromPure 0, Stream.nil, Stream.fromPure 1]

259− `Async.append` Stream.nil

260− `Async.append` Stream.fromPure 1

261 in cmp stream sortEq [0, 1]

259 in cmp stream sortEq [0, 1]

262

260

263− prop1 "append left associated"

261+ prop1 "append2 left associated"

264 $ let stream =

262 $ let stream =

265 Stream.fromPure 0

263 Stream.fromPure 0

266− `Async.append` Stream.fromPure 1

264+ `Async.append2` Stream.fromPure 1

267− `Async.append` Stream.fromPure 2

265+ `Async.append2` Stream.fromPure 2

268− `Async.append` Stream.fromPure 3

266+ `Async.append2` Stream.fromPure 3

269 in cmp stream sortEq [0, 1, 2, 3]

267 in cmp stream sortEq [0, 1, 2, 3]

270

268

271 prop1 "append right associated"

269 prop1 "append right associated"

272 $ let stream =

270 $ let stream =

273 Stream.fromPure 0

271 Stream.fromPure 0

274− `Async.append` (Stream.fromPure 1

272+ `Async.append2` (Stream.fromPure 1

275− `Async.append` (Stream.fromPure 2

273+ `Async.append2` (Stream.fromPure 2

276− `Async.append` Stream.fromPure 3))

274+ `Async.append2` Stream.fromPure 3))

277 in cmp stream sortEq [0, 1, 2, 3]

275 in cmp stream sortEq [0, 1, 2, 3]

278

276

279 prop1 "append balanced"

277 prop1 "append balanced"

280− $ let leaf x y = Stream.fromPure x `Async.append` Stream.fromPure y

278+ $ let leaf x y = Stream.fromPure x `Async.append2` Stream.fromPure y

281− leaf11 = leaf 0 1 `Async.append` leaf 2 (3 :: Int)

279+ leaf11 = leaf 0 1 `Async.append2` leaf 2 (3 :: Int)

282− leaf12 = leaf 4 5 `Async.append` leaf 6 7

280+ leaf12 = leaf 4 5 `Async.append2` leaf 6 7

283− stream = leaf11 `Async.append` leaf12

281+ stream = leaf11 `Async.append2` leaf12

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

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

285

283

286 prop1 "combineWith (maxThreads 1)"

284 prop1 "combineWith (maxThreads 1)"

@@ -291,14 +289,14 @@ main = hspec

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

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

292

290

293 prop1 "apply (async arg1)"

291 prop1 "apply (async arg1)"

294− $ let s1 = Async.apply (pure (,)) (pure 1 `Async.append` pure 2)

292+ $ let s1 = Async.apply (pure (,)) (pure 1 `Async.append2` pure 2)

295 s2 = Async.apply s1 (pure 3) :: Stream IO (Int, Int)

293 s2 = Async.apply s1 (pure 3) :: Stream IO (Int, Int)

296 xs = Stream.fold Fold.toList s2

294 xs = Stream.fold Fold.toList s2

297 in sort <$> xs `shouldReturn` [(1, 3), (2, 3)]

295 in sort <$> xs `shouldReturn` [(1, 3), (2, 3)]

298

296

299 prop1 "apply (async arg2)"

297 prop1 "apply (async arg2)"

300 $ let s1 = pure (1,)

298 $ let s1 = pure (1,)

301− s2 = Async.apply s1 (pure 2 `Async.append` pure 3)

299+ s2 = Async.apply s1 (pure 2 `Async.append2` pure 3)

302 xs = Stream.fold Fold.toList s2 :: IO [(Int, Int)]

300 xs = Stream.fold Fold.toList s2 :: IO [(Int, Int)]

303 in sort <$> xs `shouldReturn` [(1, 2), (1, 3)]

301 in sort <$> xs `shouldReturn` [(1, 2), (1, 3)]

304

302

@@ -319,6 +317,6 @@ main = hspec

319#ifdef DEVBUILD

317#ifdef DEVBUILD

320 describe "Time ordering" $ timeOrdering Async.append

318 describe "Time ordering" $ timeOrdering Async.append

321#endif

319#endif

322− describe "Exception propagation" $ exceptionPropagation Async.append

320+ describe "Exception propagation" $ exceptionPropagation Async.append2

323 -- Ad-hoc tests

321 -- Ad-hoc tests

324 it "takes n from stream of streams" $ takeCombined 2

322 it "takes n from stream of streams" $ takeCombined 2