git.iko.soy/streamly
on 2ca80e7 help sign in
streamly/src/Streamly/Internal/Data/Stream/Async/Channel.hs
247 lines 8.7 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 , Rate(..)
19 , rate
20 , avgRate
21 , minRate
22 , maxRate
23 , constRate
25 -- * Operations
26 , fromChannel
27 , toChannel
28 , withChannelK
29 , concatMapWithChanK
31 -- XXX Move to the higher level Async module. Low level module would have
32 -- only explicit channel operations.
34 -- * Stream operations
35 , evalWithK
36 , appendWithK
37 , interleaveWithK
38 , concatMapWithK -- appendMap
39 , concatMapInterleaveWithK -- interleaveMap
40 )
41where
43import Control.Monad.IO.Class (MonadIO(liftIO))
44import Streamly.Internal.Control.Concurrent (MonadAsync, askRunInIO)
45import Streamly.Internal.Data.Stream.Async.Channel.Operations
46 (fromChannel, toChannel, toChannelK)
47import Streamly.Internal.Data.SVar.Type (adaptState)
49import qualified Streamly.Internal.Data.Stream as Stream
50import qualified Streamly.Internal.Data.Stream.Async.Channel.Append as Append
51import qualified Streamly.Internal.Data.Stream.Async.Channel.Interleave
52 as Interleave
53import qualified Streamly.Internal.Data.Stream.StreamK.Type as K
55import Streamly.Internal.Data.Stream.Async.Channel.Type
57-- | Allocate a channel and evaluate the stream using the channel and the
58-- supplied evaluator function. The evaluator is run in a worker thread.
59{-# INLINE withChannelK #-}
60withChannelK :: MonadAsync m =>
61 (Channel m b -> K.Stream m a -> K.Stream m b)
62 -> (Config -> Config)
63 -> K.Stream m a
64 -> K.Stream m b
65withChannelK evaluator modifier input = K.concatEffect action
67 where
69 action = do
70 chan <- Append.newChannel modifier
71 toChannelK chan (evaluator chan input)
72 return $ Stream.toStreamK $ fromChannel chan
74{-# INLINE withInterleaveChannelK #-}
75withInterleaveChannelK :: MonadAsync m =>
76 (Channel m b -> K.Stream m a -> K.Stream m b)
77 -> (Config -> Config)
78 -> K.Stream m a
79 -> K.Stream m b
80withInterleaveChannelK evaluator modifier input = K.concatEffect action
82 where
84 action = do
85 chan <- Interleave.newChannel modifier
86 toChannelK chan (evaluator chan input)
87 return $ Stream.toStreamK $ fromChannel chan
89-------------------------------------------------------------------------------
90-- Evaluating a stream
91-------------------------------------------------------------------------------
93-- XXX This should go in the parallel module, this is not specific to async.
95-- | Evaluate the stream asynchronously in a worker thread separate from the
96-- consumer thread.
97--
98-- >> evalWithK = withChannelK (const id)
99--
100{-# INLINABLE evalWithK #-}
101evalWithK :: MonadAsync m => (Config -> Config) -> K.Stream m a -> K.Stream m a
102evalWithK = withChannelK (const id)
104{-
105{-# INLINE_NORMAL evalWithD #-}
106evalWithD :: MonadAsync m => (Config -> Config) -> D.Stream m a -> D.Stream m a
107evalWithD modifier m = D.Stream step Nothing
108 where
110 step _ Nothing = do
111 chan <- newChannel modifier
112 sendFirstWorker chan (D.toStreamK m)
113 -- XXX should use an unfold to make this efficient
114 return $ D.Skip $ Just $ fromChannelD chan
116 step gst (Just (D.UnStream step1 st)) = do
117 r <- step1 gst st
118 return $ case r of
119 D.Yield a s -> D.Yield a (Just $ D.Stream step1 s)
120 D.Skip s -> D.Skip (Just $ D.Stream step1 s)
121 D.Stop -> D.Stop
122-}
124-------------------------------------------------------------------------------
125-- appending two streams
126-------------------------------------------------------------------------------
128-- | Create a new channel and add both the streams to it for async evaluation.
129-- The output stream is the result of the evaluation.
130{-# INLINE appendWithK #-}
131appendWithK :: MonadAsync m =>
132 (Config -> Config) -> K.Stream m a -> K.Stream m a -> K.Stream m a
133appendWithK modifier stream1 stream2 = K.concatEffect action
135 where
137 action = do
138 chan <- Append.newChannel modifier
139 toChannelK chan stream2
140 toChannelK chan stream1
141 return $ Stream.toStreamK $ fromChannel chan
143-- | Create a new channel and add both the streams to it for async evaluation.
144-- The output stream is the result of the evaluation.
145{-# INLINE interleaveWithK #-}
146interleaveWithK :: MonadAsync m =>
147 (Config -> Config) -> K.Stream m a -> K.Stream m a -> K.Stream m a
148interleaveWithK modifier stream1 stream2 = K.concatEffect action
150 where
152 action = do
153 chan <- Interleave.newChannel modifier
154 toChannelK chan stream2
155 toChannelK chan stream1
156 return $ Stream.toStreamK $ fromChannel chan
158-- | Evaluate the first stream in the current thread and add the second stream
159-- to the supplied channel. This is to be used by a worker thread.
160--
161-- This can be used with concatMapWith:
162--
163-- @
164-- concatMap = K.concatMapWith (_appendWithChanK chan) f stream
165-- @
166--
167{-# INLINE _appendWithChanK #-}
168_appendWithChanK :: MonadAsync m =>
169 Channel m a -> K.Stream m a -> K.Stream m a -> K.Stream m a
170_appendWithChanK chan stream1 stream2 =
171 K.before (toChannelK chan stream2) stream1
173-------------------------------------------------------------------------------
174-- concat
175-------------------------------------------------------------------------------
177-- | @concatMapDivK useTail useHead stream@, divides the stream in head and
178-- tail, maps a stream generator on the head and maps an action on the tail of
179-- a stream. Returns the stream generated by the head.
180{-# INLINE concatMapDivK #-}
181concatMapDivK :: Monad m =>
182 (K.Stream m a -> m ())
183 -> (a -> K.Stream m b)
184 -> K.Stream m a
185 -> K.Stream m b
186concatMapDivK useTail useHead stream =
187 K.mkStream $ \st yld sng stp -> do
188 let foldShared = K.foldStreamShared st yld sng stp
189 single a = foldShared $ useHead a
190 yieldk a r = useTail r >> single a
191 in K.foldStreamShared (adaptState st) yieldk single stp stream
193-- | A runner function takes a queuing function @q@ and a stream, it splits the
194-- input stream, queuing the tail and using the head to generate a stream.
195-- 'mkEnqueue' takes a runner function and generates the queuing function @q@.
196-- Note that @q@ and the runner are mutually recursive, mkEnqueue ties the knot
197-- between the two.
198{-# INLINE mkEnqueue #-}
199mkEnqueue :: MonadAsync m =>
200 Channel m b
201 -> ((K.Stream m a -> m ()) -> K.Stream m a -> K.Stream m b)
202 -> m (K.Stream m a -> m ())
203mkEnqueue chan runner = do
204 runInIO <- askRunInIO
205 return
206 $ let q stream = liftIO $ enqueue chan (runInIO, runner q stream) in q
208-- XXX Can be renamed to concatMapWithK if we move concatMapWithK to higher
209-- level module. We can keep only Channel based ops in this module.
211-- | Takes the head element of the input stream and queues the tail of the
212-- stream to the channel, then maps the supplied function on the head and
213-- evaluates the resulting stream.
214--
215-- This function is designed to be used by worker threads on a channel to
216-- concurrently map and evaluate a stream.
217{-# INLINE concatMapWithChanK #-}
218concatMapWithChanK :: MonadAsync m =>
219 Channel m b -> (a -> K.Stream m b) -> K.Stream m a -> K.Stream m b
220concatMapWithChanK chan f stream =
221 let run q = concatMapDivK q f
222 in K.concatMapEffect (`run` stream) (mkEnqueue chan run)
223 -- K.concatMapWith (_appendWithChanK chan) f stream
225-- XXX Add a deep evaluation variant that evaluates individual elements in the
226-- generated streams in parallel.
228-- | Allocate a channel and use it to concurrently evaluate the streams
229-- generated by the mapped function.
230--
231-- >> concatMapWithK modifier f = withChannelK (`concatMapWithChanK` f) modifier
232--
233{-# INLINE concatMapWithK #-}
234concatMapWithK :: MonadAsync m =>
235 (Config -> Config) -> (a -> K.Stream m b) -> K.Stream m a -> K.Stream m b
236concatMapWithK modifier f = withChannelK (`concatMapWithChanK` f) modifier
238-- | Allocate a channel and use it to concurrently evaluate the streams
239-- generated by the mapped function.
240--
241-- >> concatMapInterleaveWithK modifier f = withInterleaveChannelK (`concatMapWithChanK` f) modifier
242--
243{-# INLINE concatMapInterleaveWithK #-}
244concatMapInterleaveWithK :: MonadAsync m =>
245 (Config -> Config) -> (a -> K.Stream m b) -> K.Stream m a -> K.Stream m b
246concatMapInterleaveWithK modifier f =
247 withInterleaveChannelK (`concatMapWithChanK` f) modifier