d7d4613 · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Prefix par to concurrent APIs

d7d4613Harendra Kumar4 years ago

benchmark/Streamly/Benchmark/Data/Stream/ConcurrentCommon.hs+12−12

compare with master

unifiedsplit

@@ -39,7 +39,7 @@ mapM ::

39 -> Int

40 -> Stream m Int

41 -> m ()

42−mapM f n = composeN n $ Async.mapMWith f return

42+mapM f n = composeN n $ Async.parMapM f return

44o_1_space_mapping :: Int -> (Config -> Config) -> [Benchmark]

45o_1_space_mapping value f =

@@ -56,7 +56,7 @@ o_n_heap_buffering :: Int -> (Config -> Config) -> [Benchmark]

56o_n_heap_buffering value f =

57 [ bgroup "buffered"

58 [ benchIOSink value "mkAsync"

59− (Stream.fold Fold.drain . Async.evalWith f)

59+ (Stream.fold Fold.drain . Async.parEval f)

60 ]

61 ]

@@ -75,7 +75,7 @@ async2 f count n =

75concatAsync2 :: (Config -> Config) -> Int -> Int -> IO ()

76concatAsync2 f count n =

77 Stream.fold Fold.drain

78− $ Async.concatWith f

78+ $ Async.parConcat f

79 $ Stream.fromList

80 [sourceUnfoldrM count n, sourceUnfoldrM count (n + 1)]

@@ -94,12 +94,12 @@ o_1_space_joining value f =

94{-# INLINE sourceFoldMapWith #-}

95sourceFoldMapWith :: (Config -> Config) -> Int -> Int -> Stream IO Int

96sourceFoldMapWith f value n =

97− Async.concatMapWith f Stream.fromPure $ Stream.fromList [n..n+value]

97+ Async.parConcatMap f Stream.fromPure $ Stream.fromList [n..n+value]

99{-# INLINE sourceFoldMapWithStream #-}

100sourceFoldMapWithStream :: (Config -> Config) -> Int -> Int -> Stream IO Int

101sourceFoldMapWithStream f value n =

102− Async.concatMapWith f Stream.fromPure

102+ Async.parConcatMap f Stream.fromPure

103 $ Stream.enumerateFromTo n (n + value)

104

105{-# INLINE concatFoldableWith #-}

@@ -110,7 +110,7 @@ concatFoldableWith f value n =

110 then Just (Stream.fromPure x, x + 1)

111 else Nothing

112 list = List.unfoldr step n

113− in Async.concatWith f (Stream.fromList list)

113+ in Async.parConcat f (Stream.fromList list)

114

115o_1_space_concatFoldable :: Int -> (Config -> Config) -> [Benchmark]

116o_1_space_concatFoldable value f =

@@ -133,7 +133,7 @@ concatMapStreamsWith

133 -> IO ()

134concatMapStreamsWith f outer inner n =

135 Stream.fold Fold.drain

136− $ Async.concatMapWith f (sourceUnfoldrM inner) (sourceUnfoldrM outer n)

136+ $ Async.parConcatMap f (sourceUnfoldrM inner) (sourceUnfoldrM outer n)

137

138{-# INLINE concatFmapStreamsWith #-}

139concatFmapStreamsWith

@@ -144,18 +144,18 @@ concatFmapStreamsWith

144 -> IO ()

145concatFmapStreamsWith f outer inner n =

146 Stream.fold Fold.drain

147− $ Async.concatWith f

147+ $ Async.parConcat f

148 $ fmap (sourceUnfoldrM inner) (sourceUnfoldrM outer n)

149

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

151o_1_space_concatMap value f =

152 value2 `seq`

153 [ bgroup "concat"

154− [ benchIO "concatMapWith (n of 1)"

154+ [ benchIO "parConcatMap (n of 1)"

155 (concatMapStreamsWith f value 1)

156− , benchIO "concatMapWith (sqrt x of sqrt x)"

156+ , benchIO "parConcatMap (sqrt x of sqrt x)"

157 (concatMapStreamsWith f value2 value2)

158− , benchIO "concatMapWith (1 of n)"

158+ , benchIO "parConcatMap (1 of n)"

159 (concatMapStreamsWith f 1 value)

160 , benchIO "concat . fmap (n of 1)"

161 (concatFmapStreamsWith f value 1)

@@ -174,7 +174,7 @@ o_1_space_concatMap value f =

174toNullAp :: (Config -> Config) -> Int -> Int -> IO ()

175toNullAp f linearCount start =

176 Stream.fold Fold.drain

177− $ Async.applyWith f

177+ $ Async.parApply f

178 (fmap (+) (sourceUnfoldrM nestedCount2 start))

179 (sourceUnfoldrM nestedCount2 start)

180

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

@@ -80,6 +80,13 @@

80-- A concurrent 'repeatM' would repeat an action using multiple concurrent

81-- executions of the action. Similarly, a concurrent 'mapM' would perform the

82-- mapped action in independent threads.

83+--

84+-- == Programming Tips

85+--

86+-- The names in this module do not conflict with other stream modules,

87+-- therefore, you can import it in the same namespace:

88+--

89+-- >>> import qualified Streamly.Data.Stream.Concurrent as Stream

84module Streamly.Data.Stream.Concurrent

91module Streamly.Data.Stream.Concurrent

85 (

92 (

@@ -119,20 +126,20 @@ module Streamly.Data.Stream.Concurrent

119 -- | Evaluates a stream concurrently using a channel.

126 -- | Evaluates a stream concurrently using a channel.

120

127

121 -- , eval

128 -- , eval

122− , evalWith

129+ , parEval

123

130

124 -- ** Generate

131 -- ** Generate

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

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

126− , repeatMWith

133+ , parRepeatM

127− , replicateMWith

134+ , parReplicateM

128

135

129 -- ** Map

136 -- ** Map

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

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

131

138

132 -- , mapM

139 -- , mapM

133− , mapMWith

140+ , parMapM

134 -- , sequence

141 -- , sequence

135− , sequenceWith

142+ , parSequence

136

143

137 -- ** List of streams

144 -- ** List of streams

138 -- | Shares a single channel across many streams.

145 -- | Shares a single channel across many streams.

@@ -140,7 +147,7 @@ module Streamly.Data.Stream.Concurrent

140 -- , append

147 -- , append

141 -- , ahead

148 -- , ahead

142 -- , parallel

149 -- , parallel

143− , concatListWith

150+ , parConcatList

144 -- , parallelFst

151 -- , parallelFst

145 -- , parallelMin

152 -- , parallelMin

146 -- , zipWithM

153 -- , zipWithM

@@ -150,15 +157,15 @@ module Streamly.Data.Stream.Concurrent

150 -- *** Apply

157 -- *** Apply

151

158

152 -- , apply

159 -- , apply

153− , applyWith

160+ , parApply

154

161

155 -- *** Concat

162 -- *** Concat

156 -- | Shares a single channel across many streams.

163 -- | Shares a single channel across many streams.

157

164

158 -- , concat

165 -- , concat

159− , concatWith

166+ , parConcat

160 -- , concatMap

167 -- , concatMap

161− , concatMapWith

168+ , parConcatMap

162 )

169 )

163where

170where

164

171

src/Streamly/Internal/Data/Stream/Concurrent.hs+182−170

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

48 -- ** Evaluate

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

50 , eval

51− , evalWith

51+ , parEval

52 -- Add unfoldrM/iterateM?

54 -- ** Generate

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

56 , repeatM

57− , repeatMWith

57+ , parRepeatM

58 , replicateM

59− , replicateMWith

59+ , parReplicateM

61 -- ** Map

62 -- | Uses a single channel to evaluate all actions.

63 , mapM

64− , mapMWith

64+ , parMapM

65 , sequence

66− , sequenceWith

66+ , parSequence

68 -- ** Combine two

69 -- | Use a channel for each pair.

@@ -78,28 +78,29 @@ module Streamly.Internal.Data.Stream.Concurrent

79 -- ** List of streams

80 -- | Shares a single channel across many streams.

81− , append

81+ , parLazy

82− , interleave

82+ , parLazyOrdered

83− , ahead

83+ , parLazyInterleaved

84− , parallel

84+ , parEager

85− , parallelFst

85+ , parEagerFst

86− , parallelMin

86+ , parEagerMin

87− , concatListWith

87+ , parConcatList

88− , zipWithM

88+

89− , zipWith

89+ , parZipWithM

90+ , parZipWith

91 -- ** Stream of streams

92 -- ** Stream of streams

92 -- *** Apply

93 -- *** Apply

93 , apply

94 , apply

94− , applyWith

95+ , parApply

96 -- *** Concat

97 -- *** Concat

97 -- | Shares a single channel across many streams.

98 -- | Shares a single channel across many streams.

98 , concat

99 , concat

99− , concatWith

100+ , parConcat

100 , concatMap

101 , concatMap

101 , concatMapInterleave

102 , concatMapInterleave

102− , concatMapWith

103+ , parConcatMap

103

104

104 -- ** Reactive

105 -- ** Reactive

105 , fromCallback

106 , fromCallback

@@ -144,7 +145,7 @@ import Streamly.Internal.Data.Stream.Concurrent.Channel

144-- >>> import qualified Streamly.Data.Fold as Fold

145-- >>> import qualified Streamly.Data.Fold as Fold

145-- >>> import qualified Streamly.Data.Parser as Parser

146-- >>> import qualified Streamly.Data.Parser as Parser

146-- >>> import qualified Streamly.Data.Stream as Stream

147-- >>> import qualified Streamly.Data.Stream as Stream

147−-- >>> import qualified Streamly.Internal.Data.Stream.Concurrent as Concur

148+-- >>> import qualified Streamly.Internal.Data.Stream.Concurrent as Stream

148-- >>> import Prelude hiding (concatMap, concat, zipWith)

149-- >>> import Prelude hiding (concatMap, concat, zipWith)

149-- >>> :{

150-- >>> :{

150-- delay n = do

151-- delay n = do

@@ -158,9 +159,9 @@ import Streamly.Internal.Data.Stream.Concurrent.Channel

158-------------------------------------------------------------------------------

159-------------------------------------------------------------------------------

159

160

160{-

161{-

161−{-# INLINE_NORMAL evalWithD #-}

162+{-# INLINE_NORMAL parEvalD #-}

162−evalWithD :: MonadAsync m => (Config -> Config) -> D.Stream m a -> D.Stream m a

163+parEvalD :: MonadAsync m => (Config -> Config) -> D.Stream m a -> D.Stream m a

163−evalWithD modifier m = D.Stream step Nothing

164+parEvalD modifier m = D.Stream step Nothing

164 where

165 where

165

166

166 step _ Nothing = do

167 step _ Nothing = do

@@ -180,18 +181,23 @@ evalWithD modifier m = D.Stream step Nothing

180-- | Evaluate a stream asynchronously using a channel and serve the consumer

181-- | Evaluate a stream asynchronously using a channel and serve the consumer

181-- from the evaluation buffer.

182-- from the evaluation buffer.

182--

183--

184+-- Note that the evaluation requires a single thread as only one stream needs

185+-- to be evaluated. Therefore, the concurrency options that are relevant to

186+-- multiple streams won't apply here e.g. maxThreads, eager, interleaved,

187+-- ordered, stopWhen options won't have any effect.

188+--

183−{-# INLINE evalWith #-}

189+{-# INLINE parEval #-}

184−evalWith :: MonadAsync m => (Config -> Config) -> Stream m a -> Stream m a

190+parEval :: MonadAsync m => (Config -> Config) -> Stream m a -> Stream m a

185−evalWith modifier input = withChannel modifier input (const id)

191+parEval modifier input = withChannel modifier input (const id)

186− -- Stream.fromStreamD $ evalWithD cfg $ Stream.toStreamD stream

192+ -- Stream.fromStreamD $ parEvalD cfg $ Stream.toStreamD stream

187

193

188-- | Definition:

194-- | Definition:

189--

195--

190−-- >>> eval = Concur.evalWith id

196+-- >>> eval = Stream.parEval id

191--

197--

192{-# INLINE eval #-}

198{-# INLINE eval #-}

193eval :: MonadAsync m => Stream m a -> Stream m a

199eval :: MonadAsync m => Stream m a -> Stream m a

194−eval = evalWith id

200+eval = parEval id

195

201

196-------------------------------------------------------------------------------

202-------------------------------------------------------------------------------

197-- combining two streams

203-- combining two streams

@@ -232,18 +238,18 @@ appendWithK :: MonadAsync m =>

232appendWithK modifier stream1 stream2 =

238appendWithK modifier stream1 stream2 =

233{-

239{-

234 if getOrdered (modifier defaultConfig)

240 if getOrdered (modifier defaultConfig)

235− then concatMapWithK modifier id (stream1 `K.cons` K.fromPure stream2)

241+ then parConcatMapK modifier id (stream1 `K.cons` K.fromPure stream2)

236 else _appendGeneric Append.newChannel modifier stream1 stream2

242 else _appendGeneric Append.newChannel modifier stream1 stream2

237-}

243-}

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

244+ parConcatMapK modifier id (stream1 `K.cons` K.fromPure stream2)

239

245

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

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

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

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

242--

248--

243−-- This can be used with concatMapWith:

249+-- This can be used with parConcatMap:

244--

250--

245-- @

251-- @

246−-- concatMap = K.concatMapWith (_appendWithChanK chan) f stream

252+-- concatMap = K.parConcatMap (_appendWithChanK chan) f stream

247-- @

253-- @

248--

254--

249{-# INLINE _appendWithChanK #-}

255{-# INLINE _appendWithChanK #-}

@@ -272,13 +278,13 @@ combineWith modifier stream1 stream2 =

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

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

273-- scheduling by creating arbitrary shape expression trees.

279-- scheduling by creating arbitrary shape expression trees.

274--

280--

275−-- >>> append2 = Concur.combineWith id

281+-- >>> append2 = Stream.combineWith id

276--

282--

277-- The following code finishes in 4 seconds:

283-- The following code finishes in 4 seconds:

278--

284--

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

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

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

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

281−-- >>> Stream.fold Fold.toList $ stream1 `Concur.append2` stream2

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

282-- 2 sec

288-- 2 sec

283-- 4 sec

289-- 4 sec

284-- [2,4]

290-- [2,4]

@@ -291,7 +297,7 @@ append2 = combineWith id

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

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

292-- limited number of threads.

298-- limited number of threads.

293--

299--

294−-- >>> interleave2 = Concur.combineWith (Concur.interleaved True)

300+-- >>> interleave2 = Stream.combineWith (Stream.interleaved True)

295--

301--

296{-# INLINE interleave2 #-}

302{-# INLINE interleave2 #-}

297interleave2 :: MonadAsync m => Stream m a -> Stream m a -> Stream m a

303interleave2 :: MonadAsync m => Stream m a -> Stream m a -> Stream m a

@@ -299,7 +305,7 @@ interleave2 = combineWith (interleaved True)

299

305

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

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

301--

307--

302−-- >>> ahead2 = Concur.combineWith (Concur.ordered True)

308+-- >>> ahead2 = Stream.combineWith (Stream.ordered True)

303--

309--

304{-# INLINE ahead2 #-}

310{-# INLINE ahead2 #-}

305ahead2 :: MonadAsync m => Stream m a -> Stream m a -> Stream m a

311ahead2 :: MonadAsync m => Stream m a -> Stream m a -> Stream m a

@@ -307,7 +313,7 @@ ahead2 = combineWith (ordered True)

307

313

308-- | Like 'append2' but with 'eager' on.

314-- | Like 'append2' but with 'eager' on.

309--

315--

310−-- >>> parallel2 = Concur.combineWith (Concur.eager True)

316+-- >>> parallel2 = Stream.combineWith (Stream.eager True)

311--

317--

312{-# INLINE parallel2 #-}

318{-# INLINE parallel2 #-}

313parallel2 :: MonadAsync m => Stream m a -> Stream m a -> Stream m a

319parallel2 :: MonadAsync m => Stream m a -> Stream m a -> Stream m a

@@ -315,7 +321,7 @@ parallel2 = combineWith (eager True)

315

321

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

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

317--

323--

318−-- >>> parallelFst2 = Concur.combineWith (Concur.eager True . Concur.stopWhen Concur.FirstStops)

324+-- >>> parallelFst2 = Stream.combineWith (Stream.eager True . Stream.stopWhen Stream.FirstStops)

319--

325--

320-- /Pre-release/

326-- /Pre-release/

321{-# INLINE parallelFst2 #-}

327{-# INLINE parallelFst2 #-}

@@ -325,7 +331,7 @@ parallelFst2 = combineWith (eager True . stopWhen FirstStops)

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

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

326-- stops.

332-- stops.

327--

333--

328−-- >>> parallelMin2 = Concur.combineWith (Concur.eager True . Concur.stopWhen Concur.AnyStops)

334+-- >>> parallelMin2 = Stream.combineWith (Stream.eager True . Stream.stopWhen Stream.AnyStops)

329--

335--

330-- /Pre-release/

336-- /Pre-release/

331{-# INLINE parallelMin2 #-}

337{-# INLINE parallelMin2 #-}

@@ -366,26 +372,26 @@ mkEnqueue chan runner = do

366--

372--

367-- This function is designed to be used by worker threads on a channel to

373-- This function is designed to be used by worker threads on a channel to

368-- concurrently map and evaluate a stream.

374-- concurrently map and evaluate a stream.

369−{-# INLINE concatMapWithChanK #-}

375+{-# INLINE parConcatMapChanK #-}

370−concatMapWithChanK :: MonadAsync m =>

376+parConcatMapChanK :: MonadAsync m =>

371 Channel m b -> (a -> K.Stream m b) -> K.Stream m a -> K.Stream m b

377 Channel m b -> (a -> K.Stream m b) -> K.Stream m a -> K.Stream m b

372−concatMapWithChanK chan f stream =

378+parConcatMapChanK chan f stream =

373 let run q = concatMapDivK q f

379 let run q = concatMapDivK q f

374 in K.concatMapEffect (`run` stream) (mkEnqueue chan run)

380 in K.concatMapEffect (`run` stream) (mkEnqueue chan run)

375− -- K.concatMapWith (_appendWithChanK chan) f stream

381+ -- K.parConcatMap (_appendWithChanK chan) f stream

376

382

377−{-# INLINE concatMapWithChanKAny #-}

383+{-# INLINE parConcatMapChanKAny #-}

378−concatMapWithChanKAny :: MonadAsync m =>

384+parConcatMapChanKAny :: MonadAsync m =>

379 Channel m b -> (a -> K.Stream m b) -> K.Stream m a -> K.Stream m b

385 Channel m b -> (a -> K.Stream m b) -> K.Stream m a -> K.Stream m b

380−concatMapWithChanKAny chan f stream =

386+parConcatMapChanKAny chan f stream =

381 let done = K.nilM (stopChannel chan)

387 let done = K.nilM (stopChannel chan)

382 run q = concatMapDivK q (\x -> K.serial (f x) done)

388 run q = concatMapDivK q (\x -> K.serial (f x) done)

383 in K.concatMapEffect (`run` stream) (mkEnqueue chan run)

389 in K.concatMapEffect (`run` stream) (mkEnqueue chan run)

384

390

385−{-# INLINE concatMapWithChanKFirst #-}

391+{-# INLINE parConcatMapChanKFirst #-}

386−concatMapWithChanKFirst :: MonadAsync m =>

392+parConcatMapChanKFirst :: MonadAsync m =>

387 Channel m b -> (a -> K.Stream m b) -> K.Stream m a -> K.Stream m b

393 Channel m b -> (a -> K.Stream m b) -> K.Stream m a -> K.Stream m b

388−concatMapWithChanKFirst chan f stream =

394+parConcatMapChanKFirst chan f stream =

389 let done = K.nilM (stopChannel chan)

395 let done = K.nilM (stopChannel chan)

390 run q = concatMapDivK q f

396 run q = concatMapDivK q f

391 in K.concatEffect $ do

397 in K.concatEffect $ do

@@ -397,19 +403,19 @@ concatMapWithChanKFirst chan f stream =

397 q t

403 q t

398 return $ K.serial (f h) done

404 return $ K.serial (f h) done

399

405

400−{-# INLINE concatMapWithChanKGeneric #-}

406+{-# INLINE parConcatMapChanKGeneric #-}

401−concatMapWithChanKGeneric :: MonadAsync m =>

407+parConcatMapChanKGeneric :: MonadAsync m =>

402 (Config -> Config)

408 (Config -> Config)

403 -> Channel m b

409 -> Channel m b

404 -> (a -> K.Stream m b)

410 -> (a -> K.Stream m b)

405 -> K.Stream m a

411 -> K.Stream m a

406 -> K.Stream m b

412 -> K.Stream m b

407−concatMapWithChanKGeneric modifier chan f stream = do

413+parConcatMapChanKGeneric modifier chan f stream = do

408 let cfg = modifier defaultConfig

414 let cfg = modifier defaultConfig

409 case getStopWhen cfg of

415 case getStopWhen cfg of

410− AllStop -> concatMapWithChanK chan f stream

416+ AllStop -> parConcatMapChanK chan f stream

411− FirstStops -> concatMapWithChanKFirst chan f stream

417+ FirstStops -> parConcatMapChanKFirst chan f stream

412− AnyStops -> concatMapWithChanKAny chan f stream

418+ AnyStops -> parConcatMapChanKAny chan f stream

413

419

414-- XXX Add a deep evaluation variant that evaluates individual elements in the

420-- XXX Add a deep evaluation variant that evaluates individual elements in the

415-- generated streams in parallel.

421-- generated streams in parallel.

@@ -417,11 +423,11 @@ concatMapWithChanKGeneric modifier chan f stream = do

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

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

418-- generated by the mapped function.

424-- generated by the mapped function.

419--

425--

420−{-# INLINE concatMapWithK #-}

426+{-# INLINE parConcatMapK #-}

421−concatMapWithK :: MonadAsync m =>

427+parConcatMapK :: MonadAsync m =>

422 (Config -> Config) -> (a -> K.Stream m b) -> K.Stream m a -> K.Stream m b

428 (Config -> Config) -> (a -> K.Stream m b) -> K.Stream m a -> K.Stream m b

423−concatMapWithK modifier f input =

429+parConcatMapK modifier f input =

424− let g = concatMapWithChanKGeneric modifier

430+ let g = parConcatMapChanKGeneric modifier

425 in withChannelK modifier input (`g` f)

431 in withChannelK modifier input (`g` f)

426

432

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

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

@@ -431,11 +437,11 @@ concatMapWithK modifier f input =

431--

437--

432-- Definition:

438-- Definition:

433--

439--

434−-- >>> concatMapWith modifier f stream = Concur.concatWith modifier $ fmap f stream

440+-- >>> parConcatMap modifier f stream = Stream.parConcat modifier $ fmap f stream

435--

441--

436-- Examples:

442-- Examples:

437--

443--

438−-- >>> f cfg xs = Stream.fold Fold.toList $ Concur.concatMapWith cfg id $ Stream.fromList xs

444+-- >>> f cfg xs = Stream.fold Fold.toList $ Stream.parConcatMap cfg id $ Stream.fromList xs

439--

445--

440-- The following streams finish in 4 seconds:

446-- The following streams finish in 4 seconds:

441--

447--

@@ -451,140 +457,140 @@ concatMapWithK modifier f input =

451-- Limiting threads to 2 schedules the third stream only after one of the first

457-- Limiting threads to 2 schedules the third stream only after one of the first

452-- two has finished, releasing a thread:

458-- two has finished, releasing a thread:

453--

459--

454−-- >>> f (Concur.maxThreads 2) [stream1, stream2, stream3]

460+-- >>> f (Stream.maxThreads 2) [stream1, stream2, stream3]

455-- ...

461-- ...

456-- [2,1,4]

462-- [2,1,4]

457--

463--

458-- When used with a Single thread it behaves like serial concatMap:

464-- When used with a Single thread it behaves like serial concatMap:

459--

465--

460−-- >>> f (Concur.maxThreads 1) [stream1, stream2, stream3]

466+-- >>> f (Stream.maxThreads 1) [stream1, stream2, stream3]

461-- ...

467-- ...

462-- [4,2,1]

468-- [4,2,1]

463--

469--

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

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

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

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

466−-- >>> f (Concur.maxThreads 1) [stream1, stream2]

472+-- >>> f (Stream.maxThreads 1) [stream1, stream2]

467-- [1,2,3,4,5,6]

473-- [1,2,3,4,5,6]

468--

474--

469−{-# INLINE concatMapWith #-}

475+{-# INLINE parConcatMap #-}

470−concatMapWith :: MonadAsync m =>

476+parConcatMap :: MonadAsync m =>

471 (Config -> Config) -> (a -> Stream m b) -> Stream m a -> Stream m b

477 (Config -> Config) -> (a -> Stream m b) -> Stream m a -> Stream m b

472−concatMapWith modifier f stream =

478+parConcatMap modifier f stream =

473 Stream.fromStreamK

479 Stream.fromStreamK

474− $ concatMapWithK

480+ $ parConcatMapK

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

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

476

482

477-- | Definition:

483-- | Definition:

478--

484--

479−-- >>> concatMap = Concur.concatMapWith id

485+-- >>> concatMap = Stream.parConcatMap id

480--

486--

481{-# INLINE concatMap #-}

487{-# INLINE concatMap #-}

482concatMap :: MonadAsync m => (a -> Stream m b) -> Stream m a -> Stream m b

488concatMap :: MonadAsync m => (a -> Stream m b) -> Stream m a -> Stream m b

483−concatMap = concatMapWith id

489+concatMap = parConcatMap id

484

490

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

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

486-- and interleave the resulting streams. Unlike 'concatMap' which prefers to

492-- and interleave the resulting streams. Unlike 'concatMap' which prefers to

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

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

488-- robin fashion over the available threads.

494-- robin fashion over the available threads.

489--

495--

490−-- >>> concatMapInterleave = Concur.concatMapWith (Concur.interleaved True)

496+-- >>> concatMapInterleave = Stream.parConcatMap (Stream.interleaved True)

491--

497--

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

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

493--

499--

494−-- >>> f cfg xs = Stream.fold Fold.toList $ Concur.concatMapWith (Concur.interleaved True . cfg) id $ Stream.fromList xs

500+-- >>> f cfg xs = Stream.fold Fold.toList $ Stream.parConcatMap (Stream.interleaved True . cfg) id $ Stream.fromList xs

495--

501--

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

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

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

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

498−-- >>> f (Concur.maxThreads 1) [stream1, stream2]

504+-- >>> f (Stream.maxThreads 1) [stream1, stream2]

499-- [1,4,2,5,3,6]

505-- [1,4,2,5,3,6]

500--

506--

501-- /Works only on finite number of streams/

507-- /Works only on finite number of streams/

502--

508--

503{-# INLINE concatMapInterleave #-}

509{-# INLINE concatMapInterleave #-}

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

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

505−concatMapInterleave = concatMapWith (interleaved True)

511+concatMapInterleave = parConcatMap (interleaved True)

506

512

507-- | Evaluate the streams in the input stream concurrently and combine them.

513-- | Evaluate the streams in the input stream concurrently and combine them.

508--

514--

509−-- >>> concatWith modifier = Concur.concatMapWith modifier id

515+-- >>> parConcat modifier = Stream.parConcatMap modifier id

510--

516--

511−{-# INLINE concatWith #-}

517+{-# INLINE parConcat #-}

512−concatWith :: MonadAsync m =>

518+parConcat :: MonadAsync m =>

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

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

514−concatWith modifier = concatMapWith modifier id

520+parConcat modifier = parConcatMap modifier id

515

521

516-- | Definition:

522-- | Definition:

517--

523--

518−-- >>> concat = Concur.concatWith id

524+-- >>> concat = Stream.parConcat id

519--

525--

520{-# INLINE concat #-}

526{-# INLINE concat #-}

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

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

522−concat = concatWith id

528+concat = parConcat id

523

529

524-------------------------------------------------------------------------------

530-------------------------------------------------------------------------------

525-- concat Lists

531-- concat Lists

526-------------------------------------------------------------------------------

532-------------------------------------------------------------------------------

527

533

528−-- | Like 'concatWith' but works on a list of streams.

534+-- | Like 'parConcat' but works on a list of streams.

529--

535--

530−-- >>> concatListWith modifier = Concur.concatWith modifier . Stream.fromList

536+-- >>> parConcatList modifier = Stream.parConcat modifier . Stream.fromList

531--

537--

532−{-# INLINE concatListWith #-}

538+{-# INLINE parConcatList #-}

533−concatListWith :: MonadAsync m => (Config -> Config) -> [Stream m a] -> Stream m a

539+parConcatList :: MonadAsync m => (Config -> Config) -> [Stream m a] -> Stream m a

534−concatListWith modifier = concatWith modifier . Stream.fromList

540+parConcatList modifier = parConcat modifier . Stream.fromList

535

541

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

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

537--

543--

538−-- >>> append = Concur.concatListWith id

544+-- >>> parLazy = Stream.parConcatList id

539--

545--

540−{-# INLINE append #-}

546+{-# INLINE parLazy #-}

541−append :: MonadAsync m => [Stream m a] -> Stream m a

547+parLazy :: MonadAsync m => [Stream m a] -> Stream m a

542−append = concatListWith id

548+parLazy = parConcatList id

543

549

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

550+-- | Like 'parLazy' but interleaves the streams fairly instead of prioritizing

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

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

546-- limited number of threads.

552-- limited number of threads.

547--

553--

548−-- >>> interleave = Concur.concatListWith (Concur.interleaved True)

554+-- >>> parLazyInterleaved = Stream.parConcatList (Stream.interleaved True)

549--

555--

550−{-# INLINE interleave #-}

556+{-# INLINE parLazyInterleaved #-}

551−interleave :: MonadAsync m => [Stream m a] -> Stream m a

557+parLazyInterleaved :: MonadAsync m => [Stream m a] -> Stream m a

552−interleave = concatListWith (interleaved True)

558+parLazyInterleaved = parConcatList (interleaved True)

553

559

554−-- | Like 'append' but with 'ordered' on.

560+-- | Like 'parLazy' but with 'ordered' on.

555--

561--

556−-- >>> ahead = Concur.concatListWith (Concur.ordered True)

562+-- >>> parLazyOrdered = Stream.parConcatList (Stream.ordered True)

557--

563--

558−{-# INLINE ahead #-}

564+{-# INLINE parLazyOrdered #-}

559−ahead :: MonadAsync m => [Stream m a] -> Stream m a

565+parLazyOrdered :: MonadAsync m => [Stream m a] -> Stream m a

560−ahead = concatListWith (ordered True)

566+parLazyOrdered = parConcatList (ordered True)

561

567

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

568+-- | Like 'parLazy' but with 'eager' on.

563--

569--

564−-- >>> parallel = Concur.concatListWith (Concur.eager True)

570+-- >>> parEager = Stream.parConcatList (Stream.eager True)

565--

571--

566−{-# INLINE parallel #-}

572+{-# INLINE parEager #-}

567−parallel :: MonadAsync m => [Stream m a] -> Stream m a

573+parEager :: MonadAsync m => [Stream m a] -> Stream m a

568−parallel = concatListWith (eager True)

574+parEager = parConcatList (eager True)

569

575

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

576+-- | Like 'parEager' but stops the output as soon as the first stream stops.

571--

577--

572−-- >>> parallelFst = Concur.concatListWith (Concur.eager True . Concur.stopWhen Concur.FirstStops)

578+-- >>> parEagerFst = Stream.parConcatList (Stream.eager True . Stream.stopWhen Stream.FirstStops)

573--

579--

574−{-# INLINE parallelFst #-}

580+{-# INLINE parEagerFst #-}

575−parallelFst :: MonadAsync m => [Stream m a] -> Stream m a

581+parEagerFst :: MonadAsync m => [Stream m a] -> Stream m a

576−parallelFst = concatListWith (eager True . stopWhen FirstStops)

582+parEagerFst = parConcatList (eager True . stopWhen FirstStops)

577

583

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

584+-- | Like 'parEager' but stops the output as soon as any of the two streams

579-- stops.

585-- stops.

580--

586--

581-- Definition:

587-- Definition:

582--

588--

583−-- >>> parallelMin = Concur.concatListWith (Concur.eager True . Concur.stopWhen Concur.AnyStops)

589+-- >>> parEagerMin = Stream.parConcatList (Stream.eager True . Stream.stopWhen Stream.AnyStops)

584--

590--

585−{-# INLINE parallelMin #-}

591+{-# INLINE parEagerMin #-}

586−parallelMin :: MonadAsync m => [Stream m a] -> Stream m a

592+parEagerMin :: MonadAsync m => [Stream m a] -> Stream m a

587−parallelMin = concatListWith (eager True . stopWhen AnyStops)

593+parEagerMin = parConcatList (eager True . stopWhen AnyStops)

588

594

589-------------------------------------------------------------------------------

595-------------------------------------------------------------------------------

590-- Applicative

596-- Applicative

@@ -592,21 +598,21 @@ parallelMin = concatListWith (eager True . stopWhen AnyStops)

592

598

593-- | Apply an argument stream to a function stream concurrently. Uses a

599-- | Apply an argument stream to a function stream concurrently. Uses a

594-- shared channel for all individual applications within a stream application.

600-- shared channel for all individual applications within a stream application.

595−{-# INLINE applyWith #-}

601+{-# INLINE parApply #-}

596−{-# SPECIALIZE applyWith ::

602+{-# SPECIALIZE parApply ::

597 (Config -> Config) -> Stream IO (a -> b) -> Stream IO a -> Stream IO b #-}

603 (Config -> Config) -> Stream IO (a -> b) -> Stream IO a -> Stream IO b #-}

598−applyWith :: MonadAsync m =>

604+parApply :: MonadAsync m =>

599 (Config -> Config) -> Stream m (a -> b) -> Stream m a -> Stream m b

605 (Config -> Config) -> Stream m (a -> b) -> Stream m a -> Stream m b

600−applyWith modifier stream1 stream2 =

606+parApply modifier stream1 stream2 =

601− concatMapWith

607+ parConcatMap

602 modifier

608 modifier

603− (\g -> concatMapWith modifier (pure . g) stream2)

609+ (\g -> parConcatMap modifier (pure . g) stream2)

604 stream1

610 stream1

605

611

606{-# INLINE apply #-}

612{-# INLINE apply #-}

607{-# SPECIALIZE apply :: Stream IO (a -> b) -> Stream IO a -> Stream IO b #-}

613{-# SPECIALIZE apply :: Stream IO (a -> b) -> Stream IO a -> Stream IO b #-}

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

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

609−apply = applyWith id

615+apply = parApply id

610

616

611-------------------------------------------------------------------------------

617-------------------------------------------------------------------------------

612-- Map

618-- Map

@@ -615,79 +621,85 @@ apply = applyWith id

615-- |

621-- |

616-- Definition:

622-- Definition:

617--

623--

618−-- >>> mapMWith modifier f = Concur.concatMapWith modifier (Stream.fromEffect . f)

624+-- >>> parMapM modifier f = Stream.parConcatMap modifier (Stream.fromEffect . f)

619--

625--

620-- Example, the following example finishes in 1 second as all actions run in

626-- Example, the following example finishes in 1 second as all actions run in

621-- parallel. Even though results are available out of order they are ordered

627-- parallel. Even though results are available out of order they are ordered

622-- due to the config option::

628-- due to the config option::

623--

629--

624-- >>> f x = delay x >> return x

630-- >>> f x = delay x >> return x

625−-- >>> Stream.fold Fold.toList $ Concur.mapMWith (Concur.ordered True) f $ Stream.fromList [3,2,1]

631+-- >>> Stream.fold Fold.toList $ Stream.parMapM (Stream.ordered True) f $ Stream.fromList [3,2,1]

626-- 1 sec

632-- 1 sec

627-- 2 sec

633-- 2 sec

628-- 3 sec

634-- 3 sec

629-- [3,2,1]

635-- [3,2,1]

630--

636--

631−{-# INLINE mapMWith #-}

637+{-# INLINE parMapM #-}

632−mapMWith :: MonadAsync m =>

638+parMapM :: MonadAsync m =>

633 (Config -> Config) -> (a -> m b) -> Stream m a -> Stream m b

639 (Config -> Config) -> (a -> m b) -> Stream m a -> Stream m b

634−mapMWith modifier f = concatMapWith modifier (Stream.fromEffect . f)

640+parMapM modifier f = parConcatMap modifier (Stream.fromEffect . f)

635

641

636-- |

642-- |

637-- Definition:

643-- Definition:

638--

644--

639−-- >>> mapM = Concur.mapMWith id

645+-- >>> mapM = Stream.parMapM id

640−--

641−-- Example, the following example finishes in 1 second as all actions run in

642−-- parallel, by default, the results are output as soon as they are available::

643−--

644−-- >>> f x = delay x >> return x

645−-- >>> Stream.fold Fold.toList $ Concur.mapM f $ Stream.fromList [3,2,1]

646−-- 1 sec

647−-- 2 sec

648−-- 3 sec

649−-- [1,2,3]

650--

646--

651{-# INLINE mapM #-}

647{-# INLINE mapM #-}

652mapM :: MonadAsync m => (a -> m b) -> Stream m a -> Stream m b

648mapM :: MonadAsync m => (a -> m b) -> Stream m a -> Stream m b

653−mapM = mapMWith id

649+mapM = parMapM id

654

650

655-- |

651-- |

656−-- >>> sequenceWith modifier = Concur.mapMWith modifier id

652+-- >>> parSequence modifier = Stream.parMapM modifier id

657--

653--

658−{-# INLINE sequenceWith #-}

654+{-# INLINE parSequence #-}

659−sequenceWith :: MonadAsync m =>

655+parSequence :: MonadAsync m =>

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

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

661−sequenceWith modifier = mapMWith modifier id

657+parSequence modifier = parMapM modifier id

662

658

663-- |

659-- |

664−-- >>> sequence = Concur.sequenceWith id

660+-- >>> sequence = Stream.parSequence id

665--

661--

666{-# INLINE sequence #-}

662{-# INLINE sequence #-}

667sequence :: MonadAsync m =>

663sequence :: MonadAsync m =>

668 Stream m (m a) -> Stream m a

664 Stream m (m a) -> Stream m a

669−sequence = sequenceWith id

665+sequence = parSequence id

670

666

671−-- |

667+-- | Evaluates the streams being zipped in separate threads than the consumer.

672−-- >>> zipWithM f m1 m2 = Stream.zipWithM f (Concur.eval m1) (Concur.eval m2)

668+-- The zip function is evaluated in the consumer thread.

669+--

670+-- >>> parZipWithM cfg f m1 m2 = Stream.zipWithM f (Stream.parEval cfg m1) (Stream.parEval cfg m2)

673--

671--

674−{-# INLINE zipWithM #-}

672+-- Multi-stream concurrency options won't apply here, see the notes in

673+-- 'parEval'.

674+--

675+-- If you want to evaluate the zip function as well in a separate thread, you

676+-- can use a 'parEval' on 'parZipWithM'.

677+--

678+{-# INLINE parZipWithM #-}

675−zipWithM :: MonadAsync m

679+parZipWithM :: MonadAsync m

680+ => (Config -> Config)

681+ -> (a -> b -> m c)

676− => (a -> b -> m c) -> Stream m a -> Stream m b -> Stream m c

682+ -> Stream m a

683+ -> Stream m b

684+ -> Stream m c

677−zipWithM f m1 m2 = Stream.zipWithM f (eval m1) (eval m2)

685+parZipWithM cfg f m1 m2 = Stream.zipWithM f (parEval cfg m1) (parEval cfg m2)

678

686

679-- |

687-- |

680−-- >>> zipWith f = Stream.zipWithM (\a b -> return $ f a b)

688+-- >>> parZipWith cfg f = Stream.parZipWithM cfg (\a b -> return $ f a b)

681--

689--

682-- >>> m1 = Stream.fromList [1,2,3]

690-- >>> m1 = Stream.fromList [1,2,3]

683-- >>> m2 = Stream.fromList [4,5,6]

691-- >>> m2 = Stream.fromList [4,5,6]

684−-- >>> Stream.fold Fold.toList $ Concur.zipWith (,) m1 m2

692+-- >>> Stream.fold Fold.toList $ Stream.parZipWith id (,) m1 m2

685-- [(1,4),(2,5),(3,6)]

693-- [(1,4),(2,5),(3,6)]

686--

694--

687−{-# INLINE zipWith #-}

695+{-# INLINE parZipWith #-}

688−zipWith :: MonadAsync m

696+parZipWith :: MonadAsync m

697+ => (Config -> Config)

698+ -> (a -> b -> c)

689− => (a -> b -> c) -> Stream m a -> Stream m b -> Stream m c

699+ -> Stream m a

690−zipWith f = zipWithM (\a b -> return $ f a b)

700+ -> Stream m b

701+ -> Stream m c

702+parZipWith cfg f = parZipWithM cfg (\a b -> return $ f a b)

691

703

692-------------------------------------------------------------------------------

704-------------------------------------------------------------------------------

693-- Generate

705-- Generate

@@ -696,46 +708,46 @@ zipWith f = zipWithM (\a b -> return $ f a b)

696-- |

708-- |

697-- Definition:

709-- Definition:

698--

710--

699−-- >>> repeatMWith cfg = Concur.sequenceWith cfg . Stream.repeat

711+-- >>> parRepeatM cfg = Stream.parSequence cfg . Stream.repeat

700--

712--

701-- Generate a stream by repeatedly executing a monadic action forever.

713-- Generate a stream by repeatedly executing a monadic action forever.

702−{-# INLINE repeatMWith #-}

714+{-# INLINE parRepeatM #-}

703−repeatMWith :: MonadAsync m => (Config -> Config) -> m a -> Stream m a

715+parRepeatM :: MonadAsync m => (Config -> Config) -> m a -> Stream m a

704−repeatMWith cfg = sequenceWith cfg . Stream.repeat

716+parRepeatM cfg = parSequence cfg . Stream.repeat

705

717

706-- |

718-- |

707-- Definition:

719-- Definition:

708--

720--

709−-- >>> repeatM = Concur.repeatMWith id

721+-- >>> repeatM = Stream.parRepeatM id

710--

722--

711{-# INLINE repeatM #-}

723{-# INLINE repeatM #-}

712repeatM :: MonadAsync m => m a -> Stream m a

724repeatM :: MonadAsync m => m a -> Stream m a

713−repeatM = repeatMWith id

725+repeatM = parRepeatM id

714

726

715-- | Generate a stream by concurrently performing a monadic action @n@ times.

727-- | Generate a stream by concurrently performing a monadic action @n@ times.

716--

728--

717-- Definition:

729-- Definition:

718--

730--

719−-- >>> replicateMWith cfg n = Concur.sequenceWith cfg . Stream.replicate n

731+-- >>> parReplicateM cfg n = Stream.parSequence cfg . Stream.replicate n

720--

732--

721−-- Example, 'replicateMWith' in the following example executes all the

733+-- Example, 'parReplicateM' in the following example executes all the

722-- replicated actions concurrently, thus taking only 1 second:

734-- replicated actions concurrently, thus taking only 1 second:

723--

735--

724−-- >>> Stream.fold Fold.drain $ Concur.replicateMWith id 10 $ delay 1

736+-- >>> Stream.fold Fold.drain $ Stream.parReplicateM id 10 $ delay 1

725-- ...

737-- ...

726--

738--

727−{-# INLINE replicateMWith #-}

739+{-# INLINE parReplicateM #-}

728−replicateMWith :: MonadAsync m => (Config -> Config) -> Int -> m a -> Stream m a

740+parReplicateM :: MonadAsync m => (Config -> Config) -> Int -> m a -> Stream m a

729−replicateMWith cfg n = sequenceWith cfg . Stream.replicate n

741+parReplicateM cfg n = parSequence cfg . Stream.replicate n

730

742

731-- |

743-- |

732-- Definition:

744-- Definition:

733--

745--

734−-- >>> replicateM n = Concur.replicateMWith id

746+-- >>> replicateM n = Stream.parReplicateM id

735--

747--

736{-# INLINE replicateM #-}

748{-# INLINE replicateM #-}

737replicateM :: MonadAsync m => Int -> m a -> Stream m a

749replicateM :: MonadAsync m => Int -> m a -> Stream m a

738−replicateM = replicateMWith id

750+replicateM = parReplicateM id

739

751

740-------------------------------------------------------------------------------

752-------------------------------------------------------------------------------

741-- Reactive

753-- Reactive

@@ -824,7 +836,7 @@ pollCountsD predicate fld (D.Stream step state) = D.Stream step' Nothing

824-- For example, to print the count of elements processed every second:

836-- For example, to print the count of elements processed every second:

825--

837--

826-- @

838-- @

827−-- > Stream.drain $ Concur.pollCounts (const True) (Stream.rollingMap (-) . Stream.delayPost 1) (Fold.drainBy print)

839+-- > Stream.drain $ Stream.pollCounts (const True) (Stream.rollingMap (-) . Stream.delayPost 1) (Fold.drainBy print)

828-- $ Stream.enumerateFrom 0

840-- $ Stream.enumerateFrom 0

829-- @

841-- @

830--

842--

src/Streamly/Internal/Data/Stream/Time.hs+7−6

@@ -97,7 +97,8 @@ import Streamly.Internal.Data.Stream.Concurrent

97-- >>> import qualified Streamly.Data.Parser as Parser

98-- >>> import qualified Streamly.Data.Stream as Stream

99-- >>> import qualified Streamly.Internal.Data.Stream as Stream (delayPost, timestamped)

100+-- >>> import qualified Streamly.Data.Stream.Concurrent as Stream

100−-- >>> import qualified Streamly.Internal.Data.Stream.Concurrent as Concur

101+-- >>> import qualified Streamly.Internal.Data.Stream.Concurrent as Stream (parEagerFst)

101-- >>> import qualified Streamly.Internal.Data.Stream.Time as Stream

102-- >>> import qualified Streamly.Internal.Data.Stream.Time as Stream

102-- >>> import Prelude hiding (concatMap, concat)

103-- >>> import Prelude hiding (concatMap, concat)

103-- >>> :{

104-- >>> :{

@@ -139,11 +140,11 @@ ticks = periodic (return ())

139-- speed.

140-- speed.

140--

141--

141-- >>> tickStream = Stream.repeatM (return ())

142-- >>> tickStream = Stream.repeatM (return ())

142−-- >>> ticksRate r = Concur.evalWith (Concur.rate (Just r)) tickStream

143+-- >>> ticksRate r = Stream.parEval (Stream.rate (Just r)) tickStream

143--

144--

144{-# INLINE ticksRate #-}

145{-# INLINE ticksRate #-}

145ticksRate :: MonadAsync m => Rate -> Stream m ()

146ticksRate :: MonadAsync m => Rate -> Stream m ()

146−ticksRate r = evalWith (rate (Just r)) $ Stream.repeatM (return ())

147+ticksRate r = parEval (rate (Just r)) $ Stream.repeatM (return ())

147

148

148-- XXX The case when the interval is 0, we should run only the stream being

149-- XXX The case when the interval is 0, we should run only the stream being

149-- interjected.

150-- interjected.

@@ -153,7 +154,7 @@ ticksRate r = evalWith (rate (Just r)) $ Stream.repeatM (return ())

153--

154--

154-- Definition:

155-- Definition:

155--

156--

156−-- >>> interject n f xs = Concur.parallelFst [xs, Stream.periodic f n]

157+-- >>> interject n f xs = Stream.parEagerFst [xs, Stream.periodic f n]

157--

158--

158-- Example:

159-- Example:

159--

160--

@@ -164,7 +165,7 @@ ticksRate r = evalWith (rate (Just r)) $ Stream.repeatM (return ())

164--

165--

165{-# INLINE interject #-}

166{-# INLINE interject #-}

166interject :: MonadAsync m => m a -> Double -> Stream m a -> Stream m a

167interject :: MonadAsync m => m a -> Double -> Stream m a -> Stream m a

167−interject f n xs = parallelFst [xs, periodic f n]

168+interject f n xs = parEagerFst [xs, periodic f n]

168

169

169-- XXX No element should be yielded if the duration is zero.

170-- XXX No element should be yielded if the duration is zero.

170

171

@@ -217,7 +218,7 @@ dropLastInterval = undefined

217-- | Group the input stream into windows of @n@ second each and then fold each

218-- | Group the input stream into windows of @n@ second each and then fold each

218-- group using the provided fold function.

219-- group using the provided fold function.

219--

220--

220−-- >>> twoPerSec = Concur.evalWith (Concur.constRate 2) $ Stream.enumerateFrom 1

221+-- >>> twoPerSec = Stream.parEval (Stream.constRate 2) $ Stream.enumerateFrom 1

221-- >>> intervals = Stream.intervalsOf 1 Fold.toList twoPerSec

222-- >>> intervals = Stream.intervalsOf 1 Fold.toList twoPerSec

222-- >>> Stream.fold Fold.toList $ Stream.take 2 intervals

223-- >>> Stream.fold Fold.toList $ Stream.take 2 intervals

223-- [...,...]

224-- [...,...]

src/Streamly/Internal/Data/Stream/Zip/Concurrent.hs+5−3

@@ -19,7 +19,7 @@ module Streamly.Internal.Data.Stream.Zip.Concurrent

19where

21import Streamly.Internal.Data.Stream (Stream)

22−import Streamly.Internal.Data.Stream.Concurrent (MonadAsync, zipWith)

22+import Streamly.Internal.Data.Stream.Concurrent (MonadAsync, parZipWith)

24import qualified Streamly.Internal.Data.Stream as Stream (repeat)

25import Prelude hiding (map, repeat, zipWith)

@@ -32,7 +32,8 @@ import Prelude hiding (map, repeat, zipWith)

32newtype ZipConcurrent m a = ZipConcurrent {getZipConcurrent :: Stream m a}

33 deriving (Functor)

35−-- | An IO stream whose applicative instance zips streams concurrently.

35+-- | An IO stream whose applicative instance zips streams concurrently. Note

36+-- that it uses the default concurrency options.

36--

37--

37-- >>> s = ZipConcurrent $ Stream.fromList [1, 2, 3]

38-- >>> s = ZipConcurrent $ Stream.fromList [1, 2, 3]

38-- >>> x = (,,) <$> s <*> s <*> s

39-- >>> x = (,,) <$> s <*> s <*> s

@@ -45,4 +46,5 @@ instance MonadAsync m => Applicative (ZipConcurrent m) where

45 pure = ZipConcurrent . Stream.repeat

46 pure = ZipConcurrent . Stream.repeat

47 {-# INLINE (<*>) #-}

48 {-# INLINE (<*>) #-}

48− ZipConcurrent m1 <*> ZipConcurrent m2 = ZipConcurrent $ zipWith id m1 m2

49+ ZipConcurrent m1 <*> ZipConcurrent m2 =

50+ ZipConcurrent $ parZipWith id id m1 m2

src/Streamly/Internal/FileSystem/Event/Windows.hs+2−2

@@ -121,10 +121,10 @@ import System.Win32.Types (BOOL, DWORD, HANDLE, LPVOID, LPDWORD, failIfFalse_)

121import qualified Data.List.NonEmpty as NonEmpty

122import qualified Streamly.Data.Fold as Fold

123import qualified Streamly.Internal.Data.Stream as S

124+import qualified Streamly.Internal.Data.Stream.Concurrent as S (parConcatMap)

124import qualified Streamly.Internal.Unicode.Stream as U

125import qualified Streamly.Internal.Unicode.Stream as U

125import qualified Streamly.Internal.Unicode.Utf8 as UTF8

126import qualified Streamly.Internal.Unicode.Utf8 as UTF8

126import qualified Streamly.Internal.Data.Array as A

127import qualified Streamly.Internal.Data.Array as A

127−import qualified Streamly.Internal.Data.Stream.Concurrent as Concur

128import Streamly.Internal.Data.Array (Array)

129

130-- | Watch configuration, used to specify the events of interest and the

@@ -471,7 +471,7 @@ closePathHandleStream =

471--

472watchWith :: (Config -> Config) -> NonEmpty (Array Word8) -> Stream IO Event

473watchWith f paths =

474− S.bracketIO before after (Concur.concatMapWith (eager True) eventStreamAggr)

474+ S.bracketIO before after (S.parConcatMap (eager True) eventStreamAggr)

475

476 where

477

src/Streamly/Network/Socket.hs+2−2

@@ -34,7 +34,7 @@

34-- >>>

35-- >>> import qualified Streamly.Data.Fold as Fold

36-- >>> import qualified Streamly.Data.Stream as Stream

37−-- >>> import qualified Streamly.Data.Stream.Concurrent as Concur

37+-- >>> import qualified Streamly.Data.Stream.Concurrent as Stream

38-- >>> import qualified Streamly.Network.Socket as Socket

39-- >>>

40-- >>> :{

@@ -51,7 +51,7 @@

51-- where

52-- server spec addr =

53-- Stream.unfold Socket.acceptor (maxListenQueue, spec, addr)

54−-- & Concur.mapMWith (Concur.eager True) (Socket.forSocketM echo)

54+-- & Stream.parMapM (Stream.eager True) (Socket.forSocketM echo)

55-- & Stream.fold Fold.drain

56-- echo sk =

57-- Stream.unfold Socket.chunkReader sk -- Stream IO (Array Word8)

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

@@ -68,8 +68,8 @@ 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])

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

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

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

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

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

75 listEquals eq stream list

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

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

146

147 it "concatMap throwM" $ do

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

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

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

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

150 try $ tl (

151 let bind = flip Async.concatMap

152 in bind s1 $ \x ->

@@ -187,7 +187,7 @@ timeOrdering f = do

187takeCombined :: Int -> IO ()

188takeCombined n = do

189 let constr = Stream.fromFoldable

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

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

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

192 r `shouldBe` []

193

@@ -219,7 +219,7 @@ sequenceReplicate cfg = constructWithLenM stream list

219 where

220

221 list = flip replicateM (return 1 :: IO Int)

222− stream = Async.sequenceWith cfg . flip Stream.replicate (return 1 :: IO Int)

222+ stream = Async.parSequence cfg . flip Stream.replicate (return 1 :: IO Int)

223

224main :: IO ()

225main = hspec

@@ -235,7 +235,7 @@ main = hspec

235 (fmap (+2))

236 (fmap (+1) . Async.eval . fmap (+1))

237

238− asyncSpec $ prop "sequenceWith" . sequenceReplicate

238+ asyncSpec $ prop "parSequence" . sequenceReplicate

239 -- XXX Need to use asyncSpec in all tests

240 prop "mapM (+1)" $

241 transform (fmap (+1)) (Async.mapM (\x -> return (x + 1)))

@@ -243,18 +243,18 @@ main = hspec

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

244 -- Binary append

245 prop1 "append [] []"

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

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

247 prop1 "append [] [1]"

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

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

249 prop1 "append [1] []"

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

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

251 prop1 "append [0] [1]"

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

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

253 in cmp stream sortEq [0, 1]

254

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

256 $ let stream =

257− Async.append

257+ Async.parLazy

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

259 in cmp stream sortEq [0, 1]

260

@@ -315,7 +315,7 @@ main = hspec

315 (Async.concatMap (const (Stream.fromList [1..n])))

316

317#ifdef DEVBUILD

318− describe "Time ordering" $ timeOrdering Async.append

318+ describe "Time ordering" $ timeOrdering Async.parLazy

319#endif

320 describe "Exception propagation" $ exceptionPropagation Async.append2

321 -- Ad-hoc tests