git.iko.soy/streamly
on master help sign in
Cleanly separate the Channel and stream modules
ee2fd69Harendra Kumar
src/Streamly/Internal/Data/Stream/Concurrent.hs+182−5
compare with master
unifiedsplit
@@ -86,12 +86,16 @@ module Streamly.Internal.Data.Stream.Concurrent
86 )
86 )
87where
87where
88
88
89+import Control.Monad.IO.Class (MonadIO(liftIO))
89−import Streamly.Internal.Control.Concurrent (MonadAsync)
90+import Streamly.Internal.Control.Concurrent (MonadAsync, askRunInIO)
91+import Streamly.Internal.Data.Stream.Channel.Types (concatMapDivK)
90import Streamly.Internal.Data.Stream.Type (Stream)
92import Streamly.Internal.Data.Stream.Type (Stream)
91
93
92import qualified Streamly.Internal.Data.Stream as Stream
94import qualified Streamly.Internal.Data.Stream as Stream
95+import qualified Streamly.Internal.Data.Stream.StreamK.Type as K
93
96
94import Streamly.Internal.Data.Stream.Concurrent.Channel
97import Streamly.Internal.Data.Stream.Concurrent.Channel
98+
95import Prelude hiding (mapM, sequence, concat, concatMap)
99import Prelude hiding (mapM, sequence, concat, concatMap)
96
100
97-- $setup
101-- $setup
@@ -114,16 +118,38 @@ import Prelude hiding (mapM, sequence, concat, concatMap)
114-- :}
118-- :}
115
119
116-------------------------------------------------------------------------------
120-------------------------------------------------------------------------------
117−-- Useful operations
121+-- Evaluating a stream
118-------------------------------------------------------------------------------
122-------------------------------------------------------------------------------
119
123
124+{-
125+{-# INLINE_NORMAL evalWithD #-}
126+evalWithD :: MonadAsync m => (Config -> Config) -> D.Stream m a -> D.Stream m a
127+evalWithD modifier m = D.Stream step Nothing
128+ where
129+
130+ step _ Nothing = do
131+ chan <- newChannel modifier
132+ sendFirstWorker chan (D.toStreamK m)
133+ -- XXX should use an unfold to make this efficient
134+ return $ D.Skip $ Just $ fromChannelD chan
135+
136+ step gst (Just (D.UnStream step1 st)) = do
137+ r <- step1 gst st
138+ return $ case r of
139+ D.Yield a s -> D.Yield a (Just $ D.Stream step1 s)
140+ D.Skip s -> D.Skip (Just $ D.Stream step1 s)
141+ D.Stop -> D.Stop
142+-}
143+
120-- | Like 'eval' but can specify a config modifier to change the concurrent
144-- | Like 'eval' but can specify a config modifier to change the concurrent
121-- channel parameters.
145-- channel parameters.
146+--
147+-- > evalWith = withChannel (const id)
148+--
122{-# INLINE evalWith #-}
149{-# INLINE evalWith #-}
123evalWith :: MonadAsync m => (Config -> Config) -> Stream m a -> Stream m a
150evalWith :: MonadAsync m => (Config -> Config) -> Stream m a -> Stream m a
124−evalWith cfg stream =
151+evalWith = withChannel (const id)
125 -- Stream.fromStreamD $ evalWithD cfg $ Stream.toStreamD stream
152 -- Stream.fromStreamD $ evalWithD cfg $ Stream.toStreamD stream
126− Stream.fromStreamK $ evalWithK cfg $ Stream.toStreamK stream
127
153
128-- | Evaluate a stream asynchronously using a channel and serve the consumer
154-- | Evaluate a stream asynchronously using a channel and serve the consumer
129-- from the evaluation buffer.
155-- from the evaluation buffer.
@@ -135,9 +161,65 @@ eval :: MonadAsync m => Stream m a -> Stream m a
135eval = evalWith id
161eval = evalWith id
136
162
137-------------------------------------------------------------------------------
163-------------------------------------------------------------------------------
138−-- Concurrent combinators for "Stream" type
164+-- appending two streams
139-------------------------------------------------------------------------------
165-------------------------------------------------------------------------------
140
166
167+{-# INLINE _appendGeneric #-}
168+_appendGeneric :: MonadAsync m =>
169+ ((Config -> Config) -> m (Channel m a))
170+ -> (Config -> Config)
171+ -> K.Stream m a
172+ -> K.Stream m a
173+ -> K.Stream m a
174+_appendGeneric newChan modifier stream1 stream2 = K.concatEffect action
175+
176+ where
177+
178+ action = do
179+ chan <- newChan modifier
180+ let cfg = modifier defaultConfig
181+ done = K.nilM (stopChannel chan)
182+ case getStopWhen cfg of
183+ AllStop -> do
184+ toChannelK chan stream2
185+ toChannelK chan stream1
186+ FirstStops -> do
187+ toChannelK chan stream2
188+ toChannelK chan (K.serial stream1 done)
189+ AnyStops -> do
190+ toChannelK chan (K.serial stream2 done)
191+ toChannelK chan (K.serial stream1 done)
192+ return $ Stream.toStreamK $ fromChannel chan
193+
194+-- | Create a new channel and add both the streams to it for async evaluation.
195+-- The output stream is the result of the evaluation.
196+{-# INLINE appendWithK #-}
197+appendWithK :: MonadAsync m =>
198+ (Config -> Config) -> K.Stream m a -> K.Stream m a -> K.Stream m a
199+appendWithK modifier stream1 stream2 =
200+{-
201+ if getOrdered (modifier defaultConfig)
202+ then concatMapWithK modifier id (stream1 `K.cons` K.fromPure stream2)
203+ else _appendGeneric Append.newChannel modifier stream1 stream2
204+-}
205+ concatMapWithK modifier id (stream1 `K.cons` K.fromPure stream2)
206+
207+-- | Evaluate the first stream in the current thread and add the second stream
208+-- to the supplied channel. This is to be used by a worker thread.
209+--
210+-- This can be used with concatMapWith:
211+--
212+-- @
213+-- concatMap = K.concatMapWith (_appendWithChanK chan) f stream
214+-- @
215+--
216+{-# INLINE _appendWithChanK #-}
217+_appendWithChanK :: MonadAsync m =>
218+ Channel m a -> K.Stream m a -> K.Stream m a -> K.Stream m a
219+_appendWithChanK chan stream1 stream2 =
220+ K.before (toChannelK chan stream2) stream1
221+
222+
141-- | Like 'append' but with a Config modifier.
223-- | Like 'append' but with a Config modifier.
142{-# INLINE combineWith #-}
224{-# INLINE combineWith #-}
143combineWith :: MonadAsync m =>
225combineWith :: MonadAsync m =>
@@ -217,6 +299,101 @@ parallelFst = combineWith (eagerEval . stopWhen FirstStops)
217parallelMin :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
299parallelMin :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
218parallelMin = combineWith (eagerEval . stopWhen AnyStops)
300parallelMin = combineWith (eagerEval . stopWhen AnyStops)
219
301
302+-------------------------------------------------------------------------------
303+-- concat
304+-------------------------------------------------------------------------------
305+
306+-- | A runner function takes a queuing function @q@ and a stream, it splits the
307+-- input stream, queuing the tail and using the head to generate a stream.
308+-- 'mkEnqueue' takes a runner function and generates the queuing function @q@.
309+-- Note that @q@ and the runner are mutually recursive, mkEnqueue ties the knot
310+-- between the two.
311+{-# INLINE mkEnqueue #-}
312+mkEnqueue :: MonadAsync m =>
313+ Channel m b
314+ -> ((K.Stream m a -> m ()) -> K.Stream m a -> K.Stream m b)
315+ -> m (K.Stream m a -> m ())
316+mkEnqueue chan runner = do
317+ runInIO <- askRunInIO
318+ return
319+ $ let q stream = do
320+ -- Enqueue the outer loop
321+ liftIO $ enqueue chan False (runInIO, runner q stream)
322+ -- XXX In case of eager dispatch we can just directly dispatch
323+ -- a worker with the tail stream here rather than first queuing
324+ -- and then dispatching a worker which dequeues the work. The
325+ -- older implementation did a direct dispatch here and its perf
326+ -- characterstics looked much better.
327+ eagerDispatch chan
328+ in q
329+
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+
333+-- | Takes the head element of the input stream and queues the tail of the
334+-- stream to the channel, then maps the supplied function on the head and
335+-- evaluates the resulting stream.
336+--
337+-- This function is designed to be used by worker threads on a channel to
338+-- concurrently map and evaluate a stream.
339+{-# INLINE concatMapWithChanK #-}
340+concatMapWithChanK :: MonadAsync m =>
341+ Channel m b -> (a -> K.Stream m b) -> K.Stream m a -> K.Stream m b
342+concatMapWithChanK chan f stream =
343+ let run q = concatMapDivK q f
344+ in K.concatMapEffect (`run` stream) (mkEnqueue chan run)
345+ -- K.concatMapWith (_appendWithChanK chan) f stream
346+
347+{-# INLINE concatMapWithChanKAny #-}
348+concatMapWithChanKAny :: MonadAsync m =>
349+ Channel m b -> (a -> K.Stream m b) -> K.Stream m a -> K.Stream m b
350+concatMapWithChanKAny chan f stream =
351+ let done = K.nilM (stopChannel chan)
352+ run q = concatMapDivK q (\x -> K.serial (f x) done)
353+ in K.concatMapEffect (`run` stream) (mkEnqueue chan run)
354+
355+{-# INLINE concatMapWithChanKFirst #-}
356+concatMapWithChanKFirst :: MonadAsync m =>
357+ Channel m b -> (a -> K.Stream m b) -> K.Stream m a -> K.Stream m b
358+concatMapWithChanKFirst chan f stream =
359+ let done = K.nilM (stopChannel chan)
360+ run q = concatMapDivK q f
361+ in K.concatEffect $ do
362+ res <- K.uncons stream
363+ case res of
364+ Nothing -> return K.nil
365+ Just (h, t) -> do
366+ q <- mkEnqueue chan run
367+ q t
368+ return $ K.serial (f h) done
369+
370+{-# INLINE concatMapWithChanKGeneric #-}
371+concatMapWithChanKGeneric :: MonadAsync m =>
372+ (Config -> Config)
373+ -> Channel m b
374+ -> (a -> K.Stream m b)
375+ -> K.Stream m a
376+ -> K.Stream m b
377+concatMapWithChanKGeneric modifier chan f stream = do
378+ let cfg = modifier defaultConfig
379+ case getStopWhen cfg of
380+ AllStop -> concatMapWithChanK chan f stream
381+ FirstStops -> concatMapWithChanKFirst chan f stream
382+ AnyStops -> concatMapWithChanKAny chan f stream
383+
384+-- XXX Add a deep evaluation variant that evaluates individual elements in the
385+-- generated streams in parallel.
386+
387+-- | Allocate a channel and use it to concurrently evaluate the streams
388+-- generated by the mapped function.
389+--
390+{-# INLINE concatMapWithK #-}
391+concatMapWithK :: MonadAsync m =>
392+ (Config -> Config) -> (a -> K.Stream m b) -> K.Stream m a -> K.Stream m b
393+concatMapWithK modifier f =
394+ let g = concatMapWithChanKGeneric modifier
395+ in withChannelK (`g` f) modifier
396+
220-- concatMapWith modifier f stream = concatWith modifier $ fmap f stream
397-- concatMapWith modifier f stream = concatWith modifier $ fmap f stream
221
398
222-- | Like 'concatMap' but we can also specify the concurrent channel's
399-- | Like 'concatMap' but we can also specify the concurrent channel's
src/Streamly/Internal/Data/Stream/Concurrent/Channel.hs+35−208
@@ -9,49 +9,53 @@
9module Streamly.Internal.Data.Stream.Concurrent.Channel
9module Streamly.Internal.Data.Stream.Concurrent.Channel
10 (
10 (
11 -- * Channel
11 -- * Channel
12− Channel
12+ Channel (..)
13 , newChannel
13 , newChannel
14+ , withChannel
15+ , withChannelK
16+ , fromChannel
17+ , toChannel
18+ , toChannelK
19+ , stopChannel
14
20
15 -- * Configuration
21 -- * Configuration
16 , Config
22 , Config
23+ , defaultConfig
24+
25+ -- ** Limits
17 , maxThreads
26 , maxThreads
18 , maxBuffer
27 , maxBuffer
19− , eagerEval
28+
29+ -- ** Rate Control
20 , Rate(..)
30 , Rate(..)
21 , rate
31 , rate
22 , avgRate
32 , avgRate
23 , minRate
33 , minRate
24 , maxRate
34 , maxRate
25 , constRate
35 , constRate
26− , inspectMode
36+
37+ -- ** Stop behavior
27 , StopWhen (..)
38 , StopWhen (..)
28 , stopWhen
39 , stopWhen
40+ , getStopWhen
41+
42+ -- ** Scheduling behavior
43+ , eagerEval
29 , ordered
44 , ordered
30 , interleaved
45 , interleaved
31
46
32− -- * Operations
47+ -- ** Diagnostics
33− , fromChannel
48+ , inspectMode
34− , toChannel
35− , withChannelK
36− , concatMapWithChanK
37−
38− -- XXX Move to the higher level Async module. Low level module would have
39− -- only explicit channel operations.
40−
41− -- * Stream operations
42− , evalWithK
43− , appendWithK
44− , concatMapWithK -- appendMap
45 )
49 )
46where
50where
47
51
48−import Control.Monad.IO.Class (MonadIO(liftIO))
49−import Streamly.Internal.Control.Concurrent (MonadAsync, askRunInIO)
52+import Streamly.Internal.Control.Concurrent (MonadAsync)
53+import Streamly.Internal.Data.Stream (Stream, fromStreamK, toStreamK)
50import Streamly.Internal.Data.Stream.Concurrent.Channel.Operations
54import Streamly.Internal.Data.Stream.Concurrent.Channel.Operations
51 (fromChannel, fromChannelK, toChannel, toChannelK)
55 (fromChannel, fromChannelK, toChannel, toChannelK)
52
56
53−import qualified Streamly.Internal.Data.Stream as Stream
57+import qualified Streamly.Internal.Data.Stream.Concurrent.Channel.Append
54−import qualified Streamly.Internal.Data.Stream.Concurrent.Channel.Append as Append
58+ as Append
55import qualified Streamly.Internal.Data.Stream.Concurrent.Channel.Interleave
59import qualified Streamly.Internal.Data.Stream.Concurrent.Channel.Interleave
56 as Interleave
60 as Interleave
57import qualified Streamly.Internal.Data.Stream.StreamK.Type as K
61import qualified Streamly.Internal.Data.Stream.StreamK.Type as K
@@ -60,7 +64,7 @@ import Streamly.Internal.Data.Stream.Concurrent.Channel.Type
60import Streamly.Internal.Data.Stream.Channel.Types
64import Streamly.Internal.Data.Stream.Channel.Types
61
65
62-- | Create a new concurrent stream evaluation channel. The monad
66-- | Create a new concurrent stream evaluation channel. The monad
63−-- state used to run the stream actions is taken from the call site of
67+-- state used to run the stream actions is captured from the call site of
64-- newChannel.
68-- newChannel.
65{-# INLINE newChannel #-}
69{-# INLINE newChannel #-}
66newChannel :: MonadAsync m =>
70newChannel :: MonadAsync m =>
@@ -88,191 +92,14 @@ withChannelK evaluator modifier input = K.concatEffect action
88 toChannelK chan (evaluator chan input)
92 toChannelK chan (evaluator chan input)
89 return $ fromChannelK chan
93 return $ fromChannelK chan
90
94
91−-------------------------------------------------------------------------------
95+-- | Allocate a channel and evaluate the stream using the channel and the
92−-- Evaluating a stream
96+-- supplied evaluator function. The evaluator is run in a worker thread.
93−-------------------------------------------------------------------------------
97+{-# INLINE withChannel #-}
94−
95−-- XXX This should go in the parallel module, this is not specific to async.
96−
97−-- | Evaluate the stream asynchronously in a worker thread separate from the
98−-- consumer thread.
99−--
100−-- >> evalWithK = withChannelK (const id)
101−--
102−{-# INLINABLE evalWithK #-}
103−evalWithK :: MonadAsync m => (Config -> Config) -> K.Stream m a -> K.Stream m a
104−evalWithK = withChannelK (const id)
105−
106−{-
107−{-# INLINE_NORMAL evalWithD #-}
108−evalWithD :: MonadAsync m => (Config -> Config) -> D.Stream m a -> D.Stream m a
109−evalWithD modifier m = D.Stream step Nothing
110− where
111−
112− step _ Nothing = do
113− chan <- newChannel modifier
114− sendFirstWorker chan (D.toStreamK m)
115− -- XXX should use an unfold to make this efficient
116− return $ D.Skip $ Just $ fromChannelD chan
117−
118− step gst (Just (D.UnStream step1 st)) = do
119− r <- step1 gst st
120− return $ case r of
121− D.Yield a s -> D.Yield a (Just $ D.Stream step1 s)
122− D.Skip s -> D.Skip (Just $ D.Stream step1 s)
123− D.Stop -> D.Stop
124−-}
125−
126−-------------------------------------------------------------------------------
127−-- appending two streams
128−-------------------------------------------------------------------------------
129−
130−{-# INLINE _appendGeneric #-}
131−_appendGeneric :: MonadAsync m =>
98+withChannel :: MonadAsync m =>
132− ((Config -> Config) -> m (Channel m a))
99+ (Channel m b -> Stream m a -> Stream m b)
133 -> (Config -> Config)
100 -> (Config -> Config)
134− -> K.Stream m a
101+ -> Stream m a
135− -> K.Stream m a
102+ -> Stream m b
136− -> K.Stream m a
103+withChannel evaluator modifier input =
137−_appendGeneric newChan modifier stream1 stream2 = K.concatEffect action
104+ let f chan stream = toStreamK $ evaluator chan (fromStreamK stream)
138−
105+ in fromStreamK $ withChannelK f modifier (toStreamK input)
139− where
140−
141− action = do
142− chan <- newChan modifier
143− let cfg = modifier defaultConfig
144− done = K.nilM (stopChannel chan)
145− case getStopWhen cfg of
146− AllStop -> do
147− toChannelK chan stream2
148− toChannelK chan stream1
149− FirstStops -> do
150− toChannelK chan stream2
151− toChannelK chan (K.serial stream1 done)
152− AnyStops -> do
153− toChannelK chan (K.serial stream2 done)
154− toChannelK chan (K.serial stream1 done)
155− return $ Stream.toStreamK $ fromChannel chan
156−
157−-- | Create a new channel and add both the streams to it for async evaluation.
158−-- The output stream is the result of the evaluation.
159−{-# INLINE appendWithK #-}
160−appendWithK :: MonadAsync m =>
161− (Config -> Config) -> K.Stream m a -> K.Stream m a -> K.Stream m a
162−appendWithK modifier stream1 stream2 =
163−{-
164− if getOrdered (modifier defaultConfig)
165− then concatMapWithK modifier id (stream1 `K.cons` K.fromPure stream2)
166− else _appendGeneric Append.newChannel modifier stream1 stream2
167−-}
168− concatMapWithK modifier id (stream1 `K.cons` K.fromPure stream2)
169−
170−-- | Evaluate the first stream in the current thread and add the second stream
171−-- to the supplied channel. This is to be used by a worker thread.
172−--
173−-- This can be used with concatMapWith:
174−--
175−-- @
176−-- concatMap = K.concatMapWith (_appendWithChanK chan) f stream
177−-- @
178−--
179−{-# INLINE _appendWithChanK #-}
180−_appendWithChanK :: MonadAsync m =>
181− Channel m a -> K.Stream m a -> K.Stream m a -> K.Stream m a
182−_appendWithChanK chan stream1 stream2 =
183− K.before (toChannelK chan stream2) stream1
184−
185−-------------------------------------------------------------------------------
186−-- concat
187−-------------------------------------------------------------------------------
188−
189−-- | A runner function takes a queuing function @q@ and a stream, it splits the
190−-- input stream, queuing the tail and using the head to generate a stream.
191−-- 'mkEnqueue' takes a runner function and generates the queuing function @q@.
192−-- Note that @q@ and the runner are mutually recursive, mkEnqueue ties the knot
193−-- between the two.
194−{-# INLINE mkEnqueue #-}
195−mkEnqueue :: MonadAsync m =>
196− Channel m b
197− -> ((K.Stream m a -> m ()) -> K.Stream m a -> K.Stream m b)
198− -> m (K.Stream m a -> m ())
199−mkEnqueue chan runner = do
200− runInIO <- askRunInIO
201− return
202− $ let q stream = do
203− -- Enqueue the outer loop
204− liftIO $ enqueue chan False (runInIO, runner q stream)
205− -- XXX In case of eager dispatch we can just directly dispatch
206− -- a worker with the tail stream here rather than first queuing
207− -- and then dispatching a worker which dequeues the work. The
208− -- older implementation did a direct dispatch here and its perf
209− -- characterstics looked much better.
210− eagerDispatch chan
211− in q
212−
213−-- XXX Can be renamed to concatMapWithK if we move concatMapWithK to higher
214−-- level module. We can keep only Channel based ops in this module.
215−
216−-- | Takes the head element of the input stream and queues the tail of the
217−-- stream to the channel, then maps the supplied function on the head and
218−-- evaluates the resulting stream.
219−--
220−-- This function is designed to be used by worker threads on a channel to
221−-- concurrently map and evaluate a stream.
222−{-# INLINE concatMapWithChanK #-}
223−concatMapWithChanK :: MonadAsync m =>
224− Channel m b -> (a -> K.Stream m b) -> K.Stream m a -> K.Stream m b
225−concatMapWithChanK chan f stream =
226− let run q = concatMapDivK q f
227− in K.concatMapEffect (`run` stream) (mkEnqueue chan run)
228− -- K.concatMapWith (_appendWithChanK chan) f stream
229−
230−{-# INLINE concatMapWithChanKAny #-}
231−concatMapWithChanKAny :: MonadAsync m =>
232− Channel m b -> (a -> K.Stream m b) -> K.Stream m a -> K.Stream m b
233−concatMapWithChanKAny chan f stream =
234− let done = K.nilM (stopChannel chan)
235− run q = concatMapDivK q (\x -> K.serial (f x) done)
236− in K.concatMapEffect (`run` stream) (mkEnqueue chan run)
237−
238−{-# INLINE concatMapWithChanKFirst #-}
239−concatMapWithChanKFirst :: MonadAsync m =>
240− Channel m b -> (a -> K.Stream m b) -> K.Stream m a -> K.Stream m b
241−concatMapWithChanKFirst chan f stream =
242− let done = K.nilM (stopChannel chan)
243− run q = concatMapDivK q f
244− in K.concatEffect $ do
245− res <- K.uncons stream
246− case res of
247− Nothing -> return K.nil
248− Just (h, t) -> do
249− q <- mkEnqueue chan run
250− q t
251− return $ K.serial (f h) done
252−
253−{-# INLINE concatMapWithChanKGeneric #-}
254−concatMapWithChanKGeneric :: MonadAsync m =>
255− (Config -> Config)
256− -> Channel m b
257− -> (a -> K.Stream m b)
258− -> K.Stream m a
259− -> K.Stream m b
260−concatMapWithChanKGeneric modifier chan f stream = do
261− let cfg = modifier defaultConfig
262− case getStopWhen cfg of
263− AllStop -> concatMapWithChanK chan f stream
264− FirstStops -> concatMapWithChanKFirst chan f stream
265− AnyStops -> concatMapWithChanKAny chan f stream
266−
267−-- XXX Add a deep evaluation variant that evaluates individual elements in the
268−-- generated streams in parallel.
269−
270−-- | Allocate a channel and use it to concurrently evaluate the streams
271−-- generated by the mapped function.
272−--
273−{-# INLINE concatMapWithK #-}
274−concatMapWithK :: MonadAsync m =>
275− (Config -> Config) -> (a -> K.Stream m b) -> K.Stream m a -> K.Stream m b
276−concatMapWithK modifier f =
277− let g = concatMapWithChanKGeneric modifier
278− in withChannelK (`g` f) modifier