git.iko.soy/streamly
on fa417eb help sign in
streamly/src/Streamly/Internal/Data/Stream/Async.hs
320 lines 10 kB history raw
1-- |
2-- Module : Streamly.Internal.Data.Stream.Async
3-- Copyright : (c) 2017 Composewell Technologies
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 -- maxYields
32 , Rate(..)
33 , rate
34 , avgRate
35 , minRate
36 , maxRate
37 , constRate
39 -- * Combinators
40 -- | Stream combinators using Async channel
42 -- XXX Move to parallel module
43 , eval
44 , evalWith
46 , mapM
47 , mapMWith
48 , sequence
49 , sequenceWith
51 -- XXX experimental binary ops, move to parallel?
52 , append
53 , appendWith
54 , interleave
55 , interleaveWith
57 , apply
58 , concatList
59 , concat
60 , concatWith
61 , concatMap
62 , concatMapInterleave
63 , concatMapWith
64 , concatMapInterleaveWith
65 )
66where
68import Streamly.Internal.Control.Concurrent (MonadAsync)
69import Streamly.Internal.Data.Stream.Type (Stream)
71import qualified Streamly.Internal.Data.Stream as Stream
73import Streamly.Internal.Data.Stream.Async.Channel
74import Prelude hiding (mapM, sequence, concat, concatMap)
76-- $setup
77--
78-- Imports for example snippets in this module.
79--
80-- >>> :m
81-- >>> import Control.Concurrent (threadDelay)
82-- >>> import qualified Streamly.Data.Array.Unboxed as Array
83-- >>> import qualified Streamly.Data.Fold as Fold
84-- >>> import qualified Streamly.Data.Parser as Parser
85-- >>> import qualified Streamly.Data.Stream as Stream
86-- >>> import qualified Streamly.Internal.Data.Stream.Async as Async
87-- >>> import Prelude hiding (concatMap, concat)
88-- >>> :{
89-- delay n = do
90-- threadDelay (n * 1000000) -- sleep for n seconds
91-- putStrLn (show n ++ " sec") -- print "n sec"
92-- return n -- IO Int
93-- :}
95-------------------------------------------------------------------------------
96-- Useful operations
97-------------------------------------------------------------------------------
99-- | Like 'eval' but can specify a config modifier to change the concurrent
100-- channel parameters.
101{-# INLINE evalWith #-}
102evalWith :: MonadAsync m => (Config -> Config) -> Stream m a -> Stream m a
103evalWith cfg stream =
104 -- Stream.fromStreamD $ evalWithD cfg $ Stream.toStreamD stream
105 Stream.fromStreamK $ evalWithK cfg $ Stream.toStreamK stream
107-- | Evaluate a stream asynchronously using a channel and serve the consumer
108-- from the evaluation buffer.
109--
110-- >>> eval = Async.evalWith id
111--
112{-# INLINE eval #-}
113eval :: MonadAsync m => Stream m a -> Stream m a
114eval = evalWith id
116-------------------------------------------------------------------------------
117-- Concurrent combinators for "Stream" type
118-------------------------------------------------------------------------------
120-- | Like 'append' but with a Config modifier.
121{-# INLINE appendWith #-}
122appendWith :: MonadAsync m =>
123 (Config -> Config) -> Stream m a -> Stream m a -> Stream m a
124appendWith modifier stream1 stream2 =
125 Stream.fromStreamK
126 $ appendWithK
127 modifier (Stream.toStreamK stream1) (Stream.toStreamK stream2)
129-- | Like 'interleave' but with a Config modifier.
130{-# INLINE interleaveWith #-}
131interleaveWith :: MonadAsync m =>
132 (Config -> Config) -> Stream m a -> Stream m a -> Stream m a
133interleaveWith modifier stream1 stream2 =
134 Stream.fromStreamK
135 $ interleaveWithK
136 modifier (Stream.toStreamK stream1) (Stream.toStreamK stream2)
138-- | Binary operation to evaluate two streams concurrently prioritizing the
139-- left stream.
140--
141-- If you want to combine more than two streams you almost always want the
142-- 'concat' operation instead. The performance of this operation degrades
143-- rapidly when more streams are combined as each operation adds one more
144-- concurrent channel. On the other hand, 'concat' uses a single channel for
145-- all streams. However, with this operation you can precisely control the
146-- scheduling by creating arbitrary shape expression trees.
147--
148-- >>> append = Async.appendWith id
149--
150-- The following code finishes in 4 seconds:
151--
152-- >>> stream1 = Stream.fromEffect (delay 4)
153-- >>> stream2 = Stream.fromEffect (delay 2)
154-- >>> Stream.fold Fold.toList $ stream1 `Async.append` stream2
155-- 2 sec
156-- 4 sec
157-- [2,4]
158--
159{-# INLINE append #-}
160append :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
161append = appendWith id
163-- | Like 'append' but interleaves the streams fairly instead of prioritizing
164-- the left stream. This schedules all streams in a round robin fashion over
165-- limited number of threads.
166--
167-- >>> interleave = Async.interleaveWith id
168--
169{-# INLINE interleave #-}
170interleave :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
171interleave = interleaveWith id
173-- concatMapWith modifier f stream = concatWith modifier $ fmap f stream
175-- | Like 'concatMap' but we can also specify the concurrent channel's
176-- configuration parameters using Config modifiers.
177--
178{-# INLINE concatMapWith #-}
179concatMapWith :: MonadAsync m =>
180 (Config -> Config) -> (a -> Stream m b) -> Stream m a -> Stream m b
181concatMapWith modifier f stream =
182 Stream.fromStreamK
183 $ concatMapWithK
184 modifier (Stream.toStreamK . f) (Stream.toStreamK stream)
186-- | Like 'concatMapInterleave' but we can also specify the concurrent
187-- channel's configuration parameters using Config modifiers.
188--
189{-# INLINE concatMapInterleaveWith #-}
190concatMapInterleaveWith :: MonadAsync m =>
191 (Config -> Config) -> (a -> Stream m b) -> Stream m a -> Stream m b
192concatMapInterleaveWith modifier f stream =
193 Stream.fromStreamK
194 $ concatMapInterleaveWithK
195 modifier (Stream.toStreamK . f) (Stream.toStreamK stream)
197-- | Map each element of the input to a stream and then concurrently evaluate
198-- and concatenate the resulting streams. Multiple streams may be evaluated
199-- concurrently but earlier streams are perferred. Output from the streams are
200-- used as they arrive.
201--
202-- >>> concatMap = Async.concatMapWith id
203--
204-- >>> f cfg xs = Stream.fold Fold.toList $ Async.concatMapWith cfg id $ Stream.fromList xs
205--
206-- The following streams finish in 4 seconds:
207--
208-- >>> stream1 = Stream.fromEffect (delay 4)
209-- >>> stream2 = Stream.fromEffect (delay 2)
210-- >>> stream3 = Stream.fromEffect (delay 1)
211-- >>> f id [stream1, stream2, stream3]
212-- 1 sec
213-- 2 sec
214-- 4 sec
215-- [1,2,4]
216--
217-- Limiting threads to 2 schedules the third stream only after one of the first
218-- two has finished, releasing a thread:
219--
220-- >>> f (Async.maxThreads 2) [stream1, stream2, stream3]
221-- ...
222-- [2,1,4]
223--
224-- When used with a Single thread it behaves like serial concatMap:
225--
226-- >>> f (Async.maxThreads 1) [stream1, stream2, stream3]
227-- ...
228-- [4,2,1]
229--
230-- >>> stream1 = Stream.fromList [1,2,3]
231-- >>> stream2 = Stream.fromList [4,5,6]
232-- >>> f (Async.maxThreads 1) [stream1, stream2]
233-- [1,2,3,4,5,6]
234--
235{-# INLINE concatMap #-}
236concatMap :: MonadAsync m => (a -> Stream m b) -> Stream m a -> Stream m b
237concatMap = concatMapWith id
239-- | Map each element of the input to a stream and then concurrently evaluate
240-- and interleave the resulting streams. Unlike 'concatMap' which prefers to
241-- evaluate the earlier stream first, this schedules all streams in a round
242-- robin fashion over the available threads.
243--
244-- >>> concatMapInterleave = Async.concatMapInterleaveWith id
245--
246-- When used with a single thread it behaves like serial interleaving:
247--
248-- >>> f cfg xs = Stream.fold Fold.toList $ Async.concatMapInterleaveWith cfg id $ Stream.fromList xs
249--
250-- >>> stream1 = Stream.fromList [1,2,3]
251-- >>> stream2 = Stream.fromList [4,5,6]
252-- >>> f (Async.maxThreads 1) [stream1, stream2]
253-- [1,4,2,5,3,6]
254--
255-- /Works only on finite number of streams/
256--
257{-# INLINE concatMapInterleave #-}
258concatMapInterleave :: MonadAsync m => (a -> Stream m b) -> Stream m a -> Stream m b
259concatMapInterleave = concatMapInterleaveWith id
261-- | Like 'concat' but we can also specify the concurrent channel's
262-- configuration parameters using Config modifiers.
263--
264-- >>> concatWith modifier = Async.concatMapWith modifier id
265--
266{-# INLINE concatWith #-}
267concatWith :: MonadAsync m =>
268 (Config -> Config) -> Stream m (Stream m a) -> Stream m a
269concatWith modifier = concatMapWith modifier id
271-- | Evaluate the streams in the input stream concurrently and combine them.
272--
273-- >>> concat = Async.concatWith id
274--
275{-# INLINE concat #-}
276concat :: MonadAsync m => Stream m (Stream m a) -> Stream m a
277concat = concatWith id
279-- | Like 'concat' but works on a list of streams.
280--
281-- >>> concatList = Async.concat . Stream.fromList
282--
283{-# INLINE concatList #-}
284concatList :: MonadAsync m => [Stream m a] -> Stream m a
285concatList = concat . Stream.fromList
287{-# INLINE apply #-}
288{-# SPECIALIZE apply :: Stream IO (a -> b) -> Stream IO a -> Stream IO b #-}
289apply :: MonadAsync m => Stream m (a -> b) -> Stream m a -> Stream m b
290apply stream1 stream2 = concatMap (\g -> concatMap (pure . g) stream2) stream1
292-- |
293-- >>> mapMWith modifier f = Async.concatMapWith modifier (Stream.fromEffect . f)
294--
295{-# INLINE mapMWith #-}
296mapMWith :: MonadAsync m =>
297 (Config -> Config) -> (a -> m b) -> Stream m a -> Stream m b
298mapMWith modifier f = concatMapWith modifier (Stream.fromEffect . f)
300-- |
301-- >>> mapM = Async.mapMWith id
302{-# INLINE mapM #-}
303mapM :: MonadAsync m => (a -> m b) -> Stream m a -> Stream m b
304mapM = mapMWith id
306-- |
307-- >>> sequenceWith modifier = Async.mapMWith modifier id
308--
309{-# INLINE sequenceWith #-}
310sequenceWith :: MonadAsync m =>
311 (Config -> Config) -> Stream m (m a) -> Stream m a
312sequenceWith modifier = mapMWith modifier id
314-- |
315-- >>> sequence = Async.sequenceWith id
316--
317{-# INLINE sequence #-}
318sequence :: MonadAsync m =>
319 Stream m (m a) -> Stream m a
320sequence = sequenceWith id