@@ -9,49 +9,53 @@
99module Streamly.Internal.Data.Stream.Concurrent.Channel
1010 (
1111 -- * Channel
12− Channel
12+ Channel (..)
1313 , newChannel
14+ , withChannel
15+ , withChannelK
16+ , fromChannel
17+ , toChannel
18+ , toChannelK
19+ , stopChannel
1420
1521 -- * Configuration
1622 , Config
23+ , defaultConfig
24+
25+ -- ** Limits
1726 , maxThreads
1827 , maxBuffer
19− , eagerEval
28+
29+ -- ** Rate Control
2030 , Rate(..)
2131 , rate
2232 , avgRate
2333 , minRate
2434 , maxRate
2535 , constRate
26− , inspectMode
36+
37+ -- ** Stop behavior
2738 , StopWhen (..)
2839 , stopWhen
40+ , getStopWhen
41+
42+ -- ** Scheduling behavior
43+ , eagerEval
2944 , ordered
3045 , interleaved
3146
32− -- * Operations
33− , fromChannel
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
47+ -- ** Diagnostics
48+ , inspectMode
4549 )
4650where
4751
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)
5054import Streamly.Internal.Data.Stream.Concurrent.Channel.Operations
5155 (fromChannel, fromChannelK, toChannel, toChannelK)
5256
53−import qualified Streamly.Internal.Data.Stream as Stream
54−import qualified Streamly.Internal.Data.Stream.Concurrent.Channel.Append as Append
57+import qualified Streamly.Internal.Data.Stream.Concurrent.Channel.Append
58+ as Append
5559import qualified Streamly.Internal.Data.Stream.Concurrent.Channel.Interleave
5660 as Interleave
5761import qualified Streamly.Internal.Data.Stream.StreamK.Type as K
@@ -60,7 +64,7 @@ import Streamly.Internal.Data.Stream.Concurrent.Channel.Type
6064import Streamly.Internal.Data.Stream.Channel.Types
6165
6266-- | 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
6468-- newChannel.
6569{-# INLINE newChannel #-}
6670newChannel :: MonadAsync m =>
@@ -88,191 +92,14 @@ withChannelK evaluator modifier input = K.concatEffect action
8892 toChannelK chan (evaluator chan input)
8993 return $ fromChannelK chan
9094
91−-------------------------------------------------------------------------------
92−-- Evaluating a stream
93−-------------------------------------------------------------------------------
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 =>
132− ((Config -> Config) -> m (Channel m a))
95+-- | Allocate a channel and evaluate the stream using the channel and the
96+-- supplied evaluator function. The evaluator is run in a worker thread.
97+{-# INLINE withChannel #-}
98+withChannel :: MonadAsync m =>
99+ (Channel m b -> Stream m a -> Stream m b)
133100 -> (Config -> Config)
134− -> K.Stream m a
135− -> K.Stream m a
136− -> K.Stream m a
137−_appendGeneric newChan modifier stream1 stream2 = K.concatEffect action
138−
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
101+ -> Stream m a
102+ -> Stream m b
103+withChannel evaluator modifier input =
104+ let f chan stream = toStreamK $ evaluator chan (fromStreamK stream)
105+ in fromStreamK $ withChannelK f modifier (toStreamK input)