src/Streamly/Internal/Data/Stream/Async.hs · streamly

git.iko.soy/streamly

on 0945bc3 help sign in

streamly/src/Streamly/Internal/Data/Stream/Async.hs

334 lines 11 kB history raw

1-- |

2-- Module : Streamly.Internal.Data.Stream.Async

4-- License : BSD-3-Clause

5-- Maintainer : streamly@composewell.com

6-- Stability : experimental

7-- Portability : GHC

8--

9-- Non-parallelizable stream combinators like unfoldrM, iterateM etc. can be

10-- evaluated concurrently with the stream consumer by using `eval`.

11-- Parallelizable combinators like repeatM, replicateM can generate the stream

12-- concurrently using 'concatMap'.

14-- Single effects related functionality can be moved to

15-- Data.Async/Control.Async.

16-- Common Channel functionality to Data.Channel.

17-- Stream channel to Data.Stream.Channel.

19module Streamly.Internal.Data.Stream.Async

20 (

21 -- * Imports

22 -- $setup

24 -- * Types

25 MonadAsync

27 -- * Configuration

28 , Config

29 , maxThreads

30 , maxBuffer

31 , eagerEval

32 -- maxYields

33 , Rate(..)

34 , rate

35 , avgRate

36 , minRate

37 , maxRate

38 , constRate

39 , inspectMode

41 -- * Combinators

42 -- | Stream combinators using Async channel

44 -- XXX Move to parallel module

45 , eval

46 , evalWith

48 , mapM

49 , mapMWith

50 , sequence

51 , sequenceWith

53 -- XXX experimental binary ops, move to parallel?

54 , append

55 , appendWith

56 , interleave

57 , interleaveWith

59 , apply

60 , applyWith

61 , concatList

62 , concat

63 , concatWith

64 , concatMap

65 , concatMapInterleave

66 , concatMapWith

67 , concatMapInterleaveWith

68 )

69where

71import Streamly.Internal.Control.Concurrent (MonadAsync)

72import Streamly.Internal.Data.Stream.Type (Stream)

74import qualified Streamly.Internal.Data.Stream as Stream

76import Streamly.Internal.Data.Stream.Async.Channel

77import Prelude hiding (mapM, sequence, concat, concatMap)

79-- $setup

80--

81-- Imports for example snippets in this module.

82--

83-- >>> :m

84-- >>> import Control.Concurrent (threadDelay)

85-- >>> import qualified Streamly.Data.Array.Unboxed as Array

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

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

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

89-- >>> import qualified Streamly.Internal.Data.Stream.Async as Async

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

91-- >>> :{

92-- delay n = do

93-- threadDelay (n * 1000000) -- sleep for n seconds

94-- putStrLn (show n ++ " sec") -- print "n sec"

95-- return n -- IO Int

96-- :}

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

99-- Useful operations

100-------------------------------------------------------------------------------

101

102-- | Like 'eval' but can specify a config modifier to change the concurrent

103-- channel parameters.

104{-# INLINE evalWith #-}

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

106evalWith cfg stream =

107 -- Stream.fromStreamD $ evalWithD cfg $ Stream.toStreamD stream

108 Stream.fromStreamK $ evalWithK cfg $ Stream.toStreamK stream

109

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

111-- from the evaluation buffer.

112--

113-- >>> eval = Async.evalWith id

114--

115{-# INLINE eval #-}

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

117eval = evalWith id

118

119-------------------------------------------------------------------------------

120-- Concurrent combinators for "Stream" type

121-------------------------------------------------------------------------------

122

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

124{-# INLINE appendWith #-}

125appendWith :: MonadAsync m =>

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

127appendWith modifier stream1 stream2 =

128 Stream.fromStreamK

129 $ appendWithK

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

131

132-- | Like 'interleave' but with a Config modifier.

133{-# INLINE interleaveWith #-}

134interleaveWith :: MonadAsync m =>

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

136interleaveWith modifier stream1 stream2 =

137 Stream.fromStreamK

138 $ interleaveWithK

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

140

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

142-- left stream.

143--

144-- If you want to combine more than two streams you almost always want the

145-- 'concat' operation instead. The performance of this operation degrades

146-- rapidly when more streams are combined as each operation adds one more

147-- concurrent channel. On the other hand, 'concat' uses a single channel for

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

149-- scheduling by creating arbitrary shape expression trees.

150--

151-- >>> append = Async.appendWith id

152--

153-- The following code finishes in 4 seconds:

154--

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

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

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

158-- 2 sec

159-- 4 sec

160-- [2,4]

161--

162{-# INLINE append #-}

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

164append = appendWith id

165

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

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

168-- limited number of threads.

169--

170-- >>> interleave = Async.interleaveWith id

171--

172{-# INLINE interleave #-}

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

174interleave = interleaveWith id

175

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

177

178-- | Like 'concatMap' but we can also specify the concurrent channel's

179-- configuration parameters using Config modifiers.

180--

181{-# INLINE concatMapWith #-}

182concatMapWith :: MonadAsync m =>

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

184concatMapWith modifier f stream =

185 Stream.fromStreamK

186 $ concatMapWithK

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

188

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

190-- channel's configuration parameters using Config modifiers.

191--

192{-# INLINE concatMapInterleaveWith #-}

193concatMapInterleaveWith :: MonadAsync m =>

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

195concatMapInterleaveWith modifier f stream =

196 Stream.fromStreamK

197 $ concatMapInterleaveWithK

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

199

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

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

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

203-- used as they arrive.

204--

205-- >>> concatMap = Async.concatMapWith id

206--

207-- >>> f cfg xs = Stream.fold Fold.toList $ Async.concatMapWith cfg id $ Stream.fromList xs

208--

209-- The following streams finish in 4 seconds:

210--

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

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

213-- >>> stream3 = Stream.fromEffect (delay 1)

214-- >>> f id [stream1, stream2, stream3]

215-- 1 sec

216-- 2 sec

217-- 4 sec

218-- [1,2,4]

219--

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

221-- two has finished, releasing a thread:

222--

223-- >>> f (Async.maxThreads 2) [stream1, stream2, stream3]

224-- ...

225-- [2,1,4]

226--

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

228--

229-- >>> f (Async.maxThreads 1) [stream1, stream2, stream3]

230-- ...

231-- [4,2,1]

232--

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

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

235-- >>> f (Async.maxThreads 1) [stream1, stream2]

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

237--

238{-# INLINE concatMap #-}

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

240concatMap = concatMapWith id

241

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

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

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

245-- robin fashion over the available threads.

246--

247-- >>> concatMapInterleave = Async.concatMapInterleaveWith id

248--

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

250--

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

252--

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

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

255-- >>> f (Async.maxThreads 1) [stream1, stream2]

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

257--

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

259--

260{-# INLINE concatMapInterleave #-}

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

262concatMapInterleave = concatMapInterleaveWith id

263

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

265-- configuration parameters using Config modifiers.

266--

267-- >>> concatWith modifier = Async.concatMapWith modifier id

268--

269{-# INLINE concatWith #-}

270concatWith :: MonadAsync m =>

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

272concatWith modifier = concatMapWith modifier id

273

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

275--

276-- >>> concat = Async.concatWith id

277--

278{-# INLINE concat #-}

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

280concat = concatWith id

281

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

283--

284-- >>> concatList = Async.concat . Stream.fromList

285--

286{-# INLINE concatList #-}

287concatList :: MonadAsync m => [Stream m a] -> Stream m a

288concatList = concat . Stream.fromList

289

290{-# INLINE applyWith #-}

291{-# SPECIALIZE applyWith ::

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

293applyWith :: MonadAsync m =>

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

295applyWith modifier stream1 stream2 =

296 concatMapWith

297 modifier

298 (\g -> concatMapWith modifier (pure . g) stream2)

299 stream1

300

301{-# INLINE apply #-}

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

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

304apply = applyWith id

305

306-- |

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

308--

309{-# INLINE mapMWith #-}

310mapMWith :: MonadAsync m =>

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

312mapMWith modifier f = concatMapWith modifier (Stream.fromEffect . f)

313

314-- |

315-- >>> mapM = Async.mapMWith id

316{-# INLINE mapM #-}

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

318mapM = mapMWith id

319

320-- |

321-- >>> sequenceWith modifier = Async.mapMWith modifier id

322--

323{-# INLINE sequenceWith #-}

324sequenceWith :: MonadAsync m =>

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

326sequenceWith modifier = mapMWith modifier id

327

328-- |

329-- >>> sequence = Async.sequenceWith id

330--

331{-# INLINE sequence #-}

332sequence :: MonadAsync m =>

333 Stream m (m a) -> Stream m a

334sequence = sequenceWith id