git.iko.soy/streamly
on 0945bc3 help sign in
streamly/src/Streamly/Internal/Data/Stream/Async/Channel.hs
244 lines 8.6 kB history raw
1-- |
2-- Module : Streamly.Internal.Data.Stream.Async.Channel
3-- Copyright : (c) 2017 Composewell Technologies
4-- License : BSD-3-Clause
5-- Maintainer : streamly@composewell.com
6-- Stability : experimental
7-- Portability : GHC
9module Streamly.Internal.Data.Stream.Async.Channel
10 (
11 -- * Channel
12 Channel
14 -- * Configuration
15 , Config
16 , maxThreads
17 , maxBuffer
18 , eagerEval
19 , Rate(..)
20 , rate
21 , avgRate
22 , minRate
23 , maxRate
24 , constRate
25 , inspectMode
27 -- * Operations
28 , fromChannel
29 , toChannel
30 , withChannelK
31 , concatMapWithChanK
33 -- XXX Move to the higher level Async module. Low level module would have
34 -- only explicit channel operations.
36 -- * Stream operations
37 , evalWithK
38 , appendWithK
39 , interleaveWithK
40 , concatMapWithK -- appendMap
41 , concatMapInterleaveWithK -- interleaveMap
42 )
43where
45import Control.Monad.IO.Class (MonadIO(liftIO))
46import Streamly.Internal.Control.Concurrent (MonadAsync, askRunInIO)
47import Streamly.Internal.Data.Stream.Async.Channel.Operations
48 (fromChannel, fromChannelK, toChannel, toChannelK)
50import qualified Streamly.Internal.Data.Stream as Stream
51import qualified Streamly.Internal.Data.Stream.Async.Channel.Append as Append
52import qualified Streamly.Internal.Data.Stream.Async.Channel.Interleave
53 as Interleave
54import qualified Streamly.Internal.Data.Stream.StreamK.Type as K
56import Streamly.Internal.Data.Stream.Async.Channel.Type
57import Streamly.Internal.Data.Stream.Channel.Types
59-- | Allocate a channel and evaluate the stream using the channel and the
60-- supplied evaluator function. The evaluator is run in a worker thread.
61{-# INLINE withChannelK #-}
62withChannelK :: MonadAsync m =>
63 (Channel m b -> K.Stream m a -> K.Stream m b)
64 -> (Config -> Config)
65 -> K.Stream m a
66 -> K.Stream m b
67withChannelK evaluator modifier input = K.concatEffect action
69 where
71 action = do
72 chan <- Append.newChannel modifier
73 toChannelK chan (evaluator chan input)
74 return $ fromChannelK chan
76-- XXX Dedup with withChannelK
78{-# INLINE withInterleaveChannelK #-}
79withInterleaveChannelK :: MonadAsync m =>
80 (Channel m b -> K.Stream m a -> K.Stream m b)
81 -> (Config -> Config)
82 -> K.Stream m a
83 -> K.Stream m b
84withInterleaveChannelK evaluator modifier input = K.concatEffect action
86 where
88 action = do
89 chan <- Interleave.newChannel modifier
90 toChannelK chan (evaluator chan input)
91 return $ Stream.toStreamK $ fromChannel chan
93-------------------------------------------------------------------------------
94-- Evaluating a stream
95-------------------------------------------------------------------------------
97-- XXX This should go in the parallel module, this is not specific to async.
99-- | Evaluate the stream asynchronously in a worker thread separate from the
100-- consumer thread.
101--
102-- >> evalWithK = withChannelK (const id)
103--
104{-# INLINABLE evalWithK #-}
105evalWithK :: MonadAsync m => (Config -> Config) -> K.Stream m a -> K.Stream m a
106evalWithK = withChannelK (const id)
108{-
109{-# INLINE_NORMAL evalWithD #-}
110evalWithD :: MonadAsync m => (Config -> Config) -> D.Stream m a -> D.Stream m a
111evalWithD modifier m = D.Stream step Nothing
112 where
114 step _ Nothing = do
115 chan <- newChannel modifier
116 sendFirstWorker chan (D.toStreamK m)
117 -- XXX should use an unfold to make this efficient
118 return $ D.Skip $ Just $ fromChannelD chan
120 step gst (Just (D.UnStream step1 st)) = do
121 r <- step1 gst st
122 return $ case r of
123 D.Yield a s -> D.Yield a (Just $ D.Stream step1 s)
124 D.Skip s -> D.Skip (Just $ D.Stream step1 s)
125 D.Stop -> D.Stop
126-}
128-------------------------------------------------------------------------------
129-- appending two streams
130-------------------------------------------------------------------------------
132-- | Create a new channel and add both the streams to it for async evaluation.
133-- The output stream is the result of the evaluation.
134{-# INLINE appendWithK #-}
135appendWithK :: MonadAsync m =>
136 (Config -> Config) -> K.Stream m a -> K.Stream m a -> K.Stream m a
137appendWithK modifier stream1 stream2 = K.concatEffect action
139 where
141 action = do
142 chan <- Append.newChannel modifier
143 toChannelK chan stream2
144 toChannelK chan stream1
145 return $ Stream.toStreamK $ fromChannel chan
147-- | Create a new channel and add both the streams to it for async evaluation.
148-- The output stream is the result of the evaluation.
149{-# INLINE interleaveWithK #-}
150interleaveWithK :: MonadAsync m =>
151 (Config -> Config) -> K.Stream m a -> K.Stream m a -> K.Stream m a
152interleaveWithK modifier stream1 stream2 = K.concatEffect action
154 where
156 action = do
157 chan <- Interleave.newChannel modifier
158 toChannelK chan stream2
159 toChannelK chan stream1
160 return $ Stream.toStreamK $ fromChannel chan
162-- | Evaluate the first stream in the current thread and add the second stream
163-- to the supplied channel. This is to be used by a worker thread.
164--
165-- This can be used with concatMapWith:
166--
167-- @
168-- concatMap = K.concatMapWith (_appendWithChanK chan) f stream
169-- @
170--
171{-# INLINE _appendWithChanK #-}
172_appendWithChanK :: MonadAsync m =>
173 Channel m a -> K.Stream m a -> K.Stream m a -> K.Stream m a
174_appendWithChanK chan stream1 stream2 =
175 K.before (toChannelK chan stream2) stream1
177-------------------------------------------------------------------------------
178-- concat
179-------------------------------------------------------------------------------
181-- | A runner function takes a queuing function @q@ and a stream, it splits the
182-- input stream, queuing the tail and using the head to generate a stream.
183-- 'mkEnqueue' takes a runner function and generates the queuing function @q@.
184-- Note that @q@ and the runner are mutually recursive, mkEnqueue ties the knot
185-- between the two.
186{-# INLINE mkEnqueue #-}
187mkEnqueue :: MonadAsync m =>
188 Channel m b
189 -> ((K.Stream m a -> m ()) -> K.Stream m a -> K.Stream m b)
190 -> m (K.Stream m a -> m ())
191mkEnqueue chan runner = do
192 runInIO <- askRunInIO
193 return
194 $ let q stream = do
195 -- Enqueue the outer loop
196 liftIO $ enqueue chan False (runInIO, runner q stream)
197 -- XXX In case of eager dispatch we can just directly dispatch
198 -- a worker with the tail stream here rather than first queuing
199 -- and then dispatching a worker which dequeues the work. The
200 -- older implementation did a direct dispatch here and its perf
201 -- characterstics looked much better.
202 eagerDispatch chan
203 in q
205-- XXX Can be renamed to concatMapWithK if we move concatMapWithK to higher
206-- level module. We can keep only Channel based ops in this module.
208-- | Takes the head element of the input stream and queues the tail of the
209-- stream to the channel, then maps the supplied function on the head and
210-- evaluates the resulting stream.
211--
212-- This function is designed to be used by worker threads on a channel to
213-- concurrently map and evaluate a stream.
214{-# INLINE concatMapWithChanK #-}
215concatMapWithChanK :: MonadAsync m =>
216 Channel m b -> (a -> K.Stream m b) -> K.Stream m a -> K.Stream m b
217concatMapWithChanK chan f stream =
218 let run q = concatMapDivK q f
219 in K.concatMapEffect (`run` stream) (mkEnqueue chan run)
220 -- K.concatMapWith (_appendWithChanK chan) f stream
222-- XXX Add a deep evaluation variant that evaluates individual elements in the
223-- generated streams in parallel.
225-- | Allocate a channel and use it to concurrently evaluate the streams
226-- generated by the mapped function.
227--
228-- >> concatMapWithK modifier f = withChannelK (`concatMapWithChanK` f) modifier
229--
230{-# INLINE concatMapWithK #-}
231concatMapWithK :: MonadAsync m =>
232 (Config -> Config) -> (a -> K.Stream m b) -> K.Stream m a -> K.Stream m b
233concatMapWithK modifier f = withChannelK (`concatMapWithChanK` f) modifier
235-- | Allocate a channel and use it to concurrently evaluate the streams
236-- generated by the mapped function.
237--
238-- >> concatMapInterleaveWithK modifier f = withInterleaveChannelK (`concatMapWithChanK` f) modifier
239--
240{-# INLINE concatMapInterleaveWithK #-}
241concatMapInterleaveWithK :: MonadAsync m =>
242 (Config -> Config) -> (a -> K.Stream m b) -> K.Stream m a -> K.Stream m b
243concatMapInterleaveWithK modifier f =
244 withInterleaveChannelK (`concatMapWithChanK` f) modifier