git.iko.soy/streamly
on f3ac108 help sign in
streamly/src/Streamly/Internal/Data/Stream/Async.hs
204 lines 6.4 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 MonadAsync
23 -- * Configuration
24 , Config
25 , maxThreads
26 , maxBuffer
27 -- maxYields
28 , Rate(..)
29 , rate
30 , avgRate
31 , minRate
32 , maxRate
33 , constRate
35 -- * Combinators
36 -- | Stream combinators using Async channel
38 -- XXX Move to parallel module
39 , eval
40 , evalWith
42 , mapM
43 , mapMWith
44 , sequence
45 , sequenceWith
47 -- XXX experimental binary ops, move to parallel?
48 , append
49 , appendWith
51 , apply
52 , concatList
53 , concat
54 , concatWith
55 , concatMap
56 , concatMapWith
57 )
58where
60import Streamly.Internal.Control.Concurrent (MonadAsync)
61import Streamly.Internal.Data.Stream.Type (Stream)
63import qualified Streamly.Internal.Data.Stream as Stream
65import Streamly.Internal.Data.Stream.Async.Channel
66import Prelude hiding (mapM, sequence, concat, concatMap)
68-- $setup
69-- >>> :m
70-- >>> import qualified Streamly.Data.Stream as Stream
71-- >>> import qualified Streamly.Internal.Data.Stream.Async as Async
73-------------------------------------------------------------------------------
74-- Useful operations
75-------------------------------------------------------------------------------
77-- | Like 'eval' but can specify a config modifier to change the concurrent
78-- channel parameters.
79{-# INLINE evalWith #-}
80evalWith :: MonadAsync m => (Config -> Config) -> Stream m a -> Stream m a
81evalWith cfg stream =
82 -- Stream.fromStreamD $ evalWithD cfg $ Stream.toStreamD stream
83 Stream.fromStreamK $ evalWithK cfg $ Stream.toStreamK stream
85-- | Evaluate a stream asynchronously using a channel and serve the consumer
86-- from the evaluation buffer.
87--
88-- >>> eval = Async.evalWith id
89--
90{-# INLINE eval #-}
91eval :: MonadAsync m => Stream m a -> Stream m a
92eval = evalWith id
94-------------------------------------------------------------------------------
95-- Concurrent combinators for "Stream" type
96-------------------------------------------------------------------------------
98-- | Like 'append' but with a Config modifier.
99{-# INLINE appendWith #-}
100appendWith :: MonadAsync m =>
101 (Config -> Config) -> Stream m a -> Stream m a -> Stream m a
102appendWith modifier stream1 stream2 =
103 Stream.fromStreamK
104 $ appendWithK
105 modifier (Stream.toStreamK stream1) (Stream.toStreamK stream2)
107-- | Binary operation to evaluate two streams concurrently prioritizing the
108-- left stream.
109--
110-- If you want to combine more than two streams you almost always want the
111-- 'concat' operation instead. The performance of this operation degrades
112-- rapidly when more streams are combined as each operation adds one more
113-- concurrent channel. On the other hand, 'concat' uses a single channel for
114-- all streams. However, with this operation you can precisely control the
115-- scheduling by creating arbitrary shape expression trees.
116--
117-- >>> append = Async.appendWith id
118--
119{-# INLINE append #-}
120append :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
121append = appendWith id
123-- concatMapWith modifier f stream = concatWith modifier $ fmap f stream
125-- | Like 'concatMap' but we can also specify the concurrent channel's
126-- configuration parameters using Config modifiers.
127--
128{-# INLINE concatMapWith #-}
129concatMapWith :: MonadAsync m =>
130 (Config -> Config) -> (a -> Stream m b) -> Stream m a -> Stream m b
131concatMapWith modifier f stream =
132 Stream.fromStreamK
133 $ concatMapWithK
134 modifier (Stream.toStreamK . f) (Stream.toStreamK stream)
136-- | Map each element of the input to a stream and then concurrently evaluate
137-- and concatenate the resulting streams.
138--
139-- >>> concatMap = Async.concatMapWith id
140--
141{-# INLINE concatMap #-}
142concatMap :: MonadAsync m => (a -> Stream m b) -> Stream m a -> Stream m b
143concatMap = concatMapWith id
145-- | Like 'concat' but we can also specify the concurrent channel's
146-- configuration parameters using Config modifiers.
147--
148-- >>> concatWith modifier = Async.concatMapWith modifier id
149--
150{-# INLINE concatWith #-}
151concatWith :: MonadAsync m =>
152 (Config -> Config) -> Stream m (Stream m a) -> Stream m a
153concatWith modifier = concatMapWith modifier id
155-- | Evaluate the streams in the input stream concurrently and combine them.
156--
157-- >>> concat = Async.concatWith id
158--
159{-# INLINE concat #-}
160concat :: MonadAsync m => Stream m (Stream m a) -> Stream m a
161concat = concatWith id
163-- | Like 'concat' but works on a list of streams.
164--
165-- >>> concatList = Async.concat . Stream.fromList
166--
167{-# INLINE concatList #-}
168concatList :: MonadAsync m => [Stream m a] -> Stream m a
169concatList = concat . Stream.fromList
171{-# INLINE apply #-}
172{-# SPECIALIZE apply :: Stream IO (a -> b) -> Stream IO a -> Stream IO b #-}
173apply :: MonadAsync m => Stream m (a -> b) -> Stream m a -> Stream m b
174apply stream1 stream2 = concatMap (\g -> concatMap (pure . g) stream2) stream1
176-- |
177-- >>> mapMWith modifier f = Async.concatMapWith modifier (Stream.fromEffect . f)
178--
179{-# INLINE mapMWith #-}
180mapMWith :: MonadAsync m =>
181 (Config -> Config) -> (a -> m b) -> Stream m a -> Stream m b
182mapMWith modifier f = concatMapWith modifier (Stream.fromEffect . f)
184-- |
185-- >>> mapM = Async.mapMWith id
186{-# INLINE mapM #-}
187mapM :: MonadAsync m => (a -> m b) -> Stream m a -> Stream m b
188mapM = mapMWith id
190-- |
191-- >>> sequenceWith modifier = Async.mapMWith modifier id
192--
193{-# INLINE sequenceWith #-}
194sequenceWith :: MonadAsync m =>
195 (Config -> Config) -> Stream m (m a) -> Stream m a
196sequenceWith modifier = mapMWith modifier id
198-- |
199-- >>> sequence = Async.sequenceWith id
200--
201{-# INLINE sequence #-}
202sequence :: MonadAsync m =>
203 Stream m (m a) -> Stream m a
204sequence = sequenceWith id