f7415e9 · streamly · git.iko.soy

git.iko.soy/streamly

Use list based concat functions in place of binary ops

commitf7415e9b443d33bb185db8a688613885da1738e8 authorHarendra Kumar <harendra@composewell.com> dateSun 16 Oct 2022, 16:18 (4 years ago) parent0cd6d62

3 files changed+155−90

compare with master

unifiedsplit

src/Streamly/Data/Stream/Concurrent.hs+11−16 src/Streamly/Internal/Data/Stream/Concurrent.hs+117−45 test/Streamly/Test/Data/Stream/Concurrent.hs+27−29

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

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

4141 -- | Stream combinators using a concurrent channel.

4242

4343 -- ** Evaluate

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

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

4545 , eval

4646 , evalWith

4747

4848 -- ** Map

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

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

5050 , mapM

5151 , mapMWith

5252 , sequence

5353 , sequenceWith

5454

55− -- ** Combine two

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.

59− --

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

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

55+ -- ** List of streams

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

6257 , append

6358 , ahead

6459 , parallel

65− , combineWith

60+ , concatListWith

6661

67− -- ** Apply

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

69− -- for each application.

62+ -- ** Stream of streams

63+ -- *** Apply

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

65+ -- channel for each application.

7066 , apply

7167 , applyWith

7268

73− -- ** Combine many

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

75− , concatList

69+ -- *** Concat

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

7671 , concat

7772 , concatWith

7873 , concatMap

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

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

4646 -- | Stream combinators using a concurrent channel

4747

4848 -- ** Evaluate

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

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

5050 , eval

5151 , evalWith

5252 -- Add unfoldrM/iterateM?

5353

5454 -- ** Map

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

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

5656 , mapM

5757 , mapMWith

5858 , sequence

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

6262 -- ** Combine two

6363 -- | Use a channel for each pair.

6464 -- 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.

6575 , append

6676 , interleave

6777 , ahead

6878 , parallel

6979 , parallelFst

7080 , parallelMin

71− , combineWith

81+ , concatListWith

7282

73− -- ** Apply

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

83+ -- ** Stream of streams

84+ -- *** Apply

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

7586 , apply

7687 , applyWith

7788

78− -- ** Combine many

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

80− , concatList

89+ -- *** Concat

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

8191 , concat

8292 , concatWith

8393 , concatMap

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

161171eval = evalWith id

162172

163173-------------------------------------------------------------------------------

164−-- appending two streams

174+-- combining two streams

165175-------------------------------------------------------------------------------

166176

167177{-# INLINE _appendGeneric #-}

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

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

240250-- scheduling by creating arbitrary shape expression trees.

241251--

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

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

243253--

244254-- The following code finishes in 4 seconds:

245255--

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

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

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

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

249259-- 2 sec

250260-- 4 sec

251261-- [2,4]

252262--

253−{-# INLINE append #-}

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

255−append = combineWith id

263+{-# INLINE append2 #-}

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

265+append2 = combineWith id

256266

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

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

259269-- limited number of threads.

260270--

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

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

262272--

263−{-# INLINE interleave #-}

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

265−interleave = combineWith interleaved

273+{-# INLINE interleave2 #-}

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

275+interleave2 = combineWith interleaved

266276

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

268278--

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

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

270280--

271−{-# INLINE ahead #-}

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

273−ahead = combineWith ordered

281+{-# INLINE ahead2 #-}

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

283+ahead2 = combineWith ordered

274284

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

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

276286--

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

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

278288--

279−{-# INLINE parallel #-}

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

281−parallel = combineWith eager

289+{-# INLINE parallel2 #-}

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

291+parallel2 = combineWith eager

282292

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.

284294--

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

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

286296--

287297-- /Pre-release/

288−{-# INLINE parallelFst #-}

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

290−parallelFst = combineWith (eager . stopWhen FirstStops)

298+{-# INLINE parallelFst2 #-}

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

300+parallelFst2 = combineWith (eager . stopWhen FirstStops)

291301

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

293303-- stops.

294304--

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

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

296306--

297307-- /Pre-release/

298−{-# INLINE parallelMin #-}

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

300−parallelMin = combineWith (eager . stopWhen AnyStops)

308+{-# INLINE parallelMin2 #-}

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

310+parallelMin2 = combineWith (eager . stopWhen AnyStops)

301311

302312-------------------------------------------------------------------------------

303−-- concat

313+-- concat streams

304314-------------------------------------------------------------------------------

305315

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

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

327337 eagerDispatch chan

328338 in q

329339

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−

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

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

335342-- evaluates the resulting stream.

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

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

490497concat = concatWith id

491498

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+

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

493512--

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)

495558--

496−{-# INLINE concatList #-}

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

498−concatList = concat . Stream.fromList

559+-- /Pre-release/

560+{-# INLINE parallelMin #-}

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

562+parallelMin = concatListWith (eager . stopWhen AnyStops)

563+

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

565+-- Applicative

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

499567

500568{-# INLINE applyWith #-}

501569{-# SPECIALIZE applyWith ::

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

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

514582apply = applyWith id

515583

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

585+-- Map

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

587+

516588-- |

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

518590--

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

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

6868transformCombineFromList constr eq listOp op a b c =

6969 withMaxSuccess maxTestCount $

7070 monadicIO $ do

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

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

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

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

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

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

7475 listEquals eq stream list

7576

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

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

145146

146147 it "concatMap throwM" $ do

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

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

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

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

149150 try $ tl (

150151 let bind = flip Async.concatMap

151152 in bind s1 $ \x ->

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

186187takeCombined :: Int -> IO ()

187188takeCombined n = do

188189 let constr = Stream.fromFoldable

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

190− Stream.take n

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

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

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

192192 r `shouldBe` []

193193

194194---------------------------------------------------------------------------

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

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

244244 -- Binary append

245245 prop1 "append [] []"

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

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

247247 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]

249249 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]

251251 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]

254253 in cmp stream sortEq [0, 1]

255254

256255 prop1 "append [0] [] [1]"

257256 $ let stream =

258− Stream.fromPure 0

259− `Async.append` Stream.nil

260− `Async.append` Stream.fromPure 1

257+ Async.append

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

261259 in cmp stream sortEq [0, 1]

262260

263− prop1 "append left associated"

261+ prop1 "append2 left associated"

264262 $ let stream =

265263 Stream.fromPure 0

266− `Async.append` Stream.fromPure 1

267− `Async.append` Stream.fromPure 2

268− `Async.append` Stream.fromPure 3

264+ `Async.append2` Stream.fromPure 1

265+ `Async.append2` Stream.fromPure 2

266+ `Async.append2` Stream.fromPure 3

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

270268

271269 prop1 "append right associated"

272270 $ let stream =

273271 Stream.fromPure 0

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

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

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

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

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

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

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

278276

279277 prop1 "append balanced"

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

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

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

283− stream = leaf11 `Async.append` leaf12

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

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

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

281+ stream = leaf11 `Async.append2` leaf12

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

285283

286284 prop1 "combineWith (maxThreads 1)"

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

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

292290

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

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

296294 xs = Stream.fold Fold.toList s2

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

298296

299297 prop1 "apply (async arg2)"

300298 $ 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)

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

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

304302

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

319317#ifdef DEVBUILD

320318 describe "Time ordering" $ timeOrdering Async.append

321319#endif

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

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

323321 -- Ad-hoc tests

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