git.iko.soy/streamly
on cee4d0f help sign in
streamly/src/Streamly/Internal/Data/Stream/Async/Channel.hs
248 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, fromChannelK, 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
56import Streamly.Internal.Data.Stream.Channel.Types
58-- | Allocate a channel and evaluate the stream using the channel and the
59-- supplied evaluator function. The evaluator is run in a worker thread.
60{-# INLINE withChannelK #-}
61withChannelK :: MonadAsync m =>
62 (Channel m b -> K.Stream m a -> K.Stream m b)
63 -> (Config -> Config)
64 -> K.Stream m a
65 -> K.Stream m b
66withChannelK evaluator modifier input = K.concatEffect action
68 where
70 action = do
71 chan <- Append.newChannel modifier
72 toChannelK chan (evaluator chan input)
73 return $ fromChannelK chan
75{-# INLINE withInterleaveChannelK #-}
76withInterleaveChannelK :: MonadAsync m =>
77 (Channel m b -> K.Stream m a -> K.Stream m b)
78 -> (Config -> Config)
79 -> K.Stream m a
80 -> K.Stream m b
81withInterleaveChannelK evaluator modifier input = K.concatEffect action
83 where
85 action = do
86 chan <- Interleave.newChannel modifier
87 toChannelK chan (evaluator chan input)
88 return $ Stream.toStreamK $ fromChannel chan
90-------------------------------------------------------------------------------
91-- Evaluating a stream
92-------------------------------------------------------------------------------
94-- XXX This should go in the parallel module, this is not specific to async.
96-- | Evaluate the stream asynchronously in a worker thread separate from the
97-- consumer thread.
98--
99-- >> evalWithK = withChannelK (const id)
100--
101{-# INLINABLE evalWithK #-}
102evalWithK :: MonadAsync m => (Config -> Config) -> K.Stream m a -> K.Stream m a
103evalWithK = withChannelK (const id)
105{-
106{-# INLINE_NORMAL evalWithD #-}
107evalWithD :: MonadAsync m => (Config -> Config) -> D.Stream m a -> D.Stream m a
108evalWithD modifier m = D.Stream step Nothing
109 where
111 step _ Nothing = do
112 chan <- newChannel modifier
113 sendFirstWorker chan (D.toStreamK m)
114 -- XXX should use an unfold to make this efficient
115 return $ D.Skip $ Just $ fromChannelD chan
117 step gst (Just (D.UnStream step1 st)) = do
118 r <- step1 gst st
119 return $ case r of
120 D.Yield a s -> D.Yield a (Just $ D.Stream step1 s)
121 D.Skip s -> D.Skip (Just $ D.Stream step1 s)
122 D.Stop -> D.Stop
123-}
125-------------------------------------------------------------------------------
126-- appending two streams
127-------------------------------------------------------------------------------
129-- | Create a new channel and add both the streams to it for async evaluation.
130-- The output stream is the result of the evaluation.
131{-# INLINE appendWithK #-}
132appendWithK :: MonadAsync m =>
133 (Config -> Config) -> K.Stream m a -> K.Stream m a -> K.Stream m a
134appendWithK modifier stream1 stream2 = K.concatEffect action
136 where
138 action = do
139 chan <- Append.newChannel modifier
140 toChannelK chan stream2
141 toChannelK chan stream1
142 return $ Stream.toStreamK $ fromChannel chan
144-- | Create a new channel and add both the streams to it for async evaluation.
145-- The output stream is the result of the evaluation.
146{-# INLINE interleaveWithK #-}
147interleaveWithK :: MonadAsync m =>
148 (Config -> Config) -> K.Stream m a -> K.Stream m a -> K.Stream m a
149interleaveWithK modifier stream1 stream2 = K.concatEffect action
151 where
153 action = do
154 chan <- Interleave.newChannel modifier
155 toChannelK chan stream2
156 toChannelK chan stream1
157 return $ Stream.toStreamK $ fromChannel chan
159-- | Evaluate the first stream in the current thread and add the second stream
160-- to the supplied channel. This is to be used by a worker thread.
161--
162-- This can be used with concatMapWith:
163--
164-- @
165-- concatMap = K.concatMapWith (_appendWithChanK chan) f stream
166-- @
167--
168{-# INLINE _appendWithChanK #-}
169_appendWithChanK :: MonadAsync m =>
170 Channel m a -> K.Stream m a -> K.Stream m a -> K.Stream m a
171_appendWithChanK chan stream1 stream2 =
172 K.before (toChannelK chan stream2) stream1
174-------------------------------------------------------------------------------
175-- concat
176-------------------------------------------------------------------------------
178-- | @concatMapDivK useTail useHead stream@, divides the stream in head and
179-- tail, maps a stream generator on the head and maps an action on the tail of
180-- a stream. Returns the stream generated by the head.
181{-# INLINE concatMapDivK #-}
182concatMapDivK :: Monad m =>
183 (K.Stream m a -> m ())
184 -> (a -> K.Stream m b)
185 -> K.Stream m a
186 -> K.Stream m b
187concatMapDivK useTail useHead stream =
188 K.mkStream $ \st yld sng stp -> do
189 let foldShared = K.foldStreamShared st yld sng stp
190 single a = foldShared $ useHead a
191 yieldk a r = useTail r >> single a
192 in K.foldStreamShared (adaptState st) yieldk single stp stream
194-- | A runner function takes a queuing function @q@ and a stream, it splits the
195-- input stream, queuing the tail and using the head to generate a stream.
196-- 'mkEnqueue' takes a runner function and generates the queuing function @q@.
197-- Note that @q@ and the runner are mutually recursive, mkEnqueue ties the knot
198-- between the two.
199{-# INLINE mkEnqueue #-}
200mkEnqueue :: MonadAsync m =>
201 Channel m b
202 -> ((K.Stream m a -> m ()) -> K.Stream m a -> K.Stream m b)
203 -> m (K.Stream m a -> m ())
204mkEnqueue chan runner = do
205 runInIO <- askRunInIO
206 return
207 $ let q stream = liftIO $ enqueue chan (runInIO, runner q stream) in q
209-- XXX Can be renamed to concatMapWithK if we move concatMapWithK to higher
210-- level module. We can keep only Channel based ops in this module.
212-- | Takes the head element of the input stream and queues the tail of the
213-- stream to the channel, then maps the supplied function on the head and
214-- evaluates the resulting stream.
215--
216-- This function is designed to be used by worker threads on a channel to
217-- concurrently map and evaluate a stream.
218{-# INLINE concatMapWithChanK #-}
219concatMapWithChanK :: MonadAsync m =>
220 Channel m b -> (a -> K.Stream m b) -> K.Stream m a -> K.Stream m b
221concatMapWithChanK chan f stream =
222 let run q = concatMapDivK q f
223 in K.concatMapEffect (`run` stream) (mkEnqueue chan run)
224 -- K.concatMapWith (_appendWithChanK chan) f stream
226-- XXX Add a deep evaluation variant that evaluates individual elements in the
227-- generated streams in parallel.
229-- | Allocate a channel and use it to concurrently evaluate the streams
230-- generated by the mapped function.
231--
232-- >> concatMapWithK modifier f = withChannelK (`concatMapWithChanK` f) modifier
233--
234{-# INLINE concatMapWithK #-}
235concatMapWithK :: MonadAsync m =>
236 (Config -> Config) -> (a -> K.Stream m b) -> K.Stream m a -> K.Stream m b
237concatMapWithK modifier f = withChannelK (`concatMapWithChanK` f) modifier
239-- | Allocate a channel and use it to concurrently evaluate the streams
240-- generated by the mapped function.
241--
242-- >> concatMapInterleaveWithK modifier f = withInterleaveChannelK (`concatMapWithChanK` f) modifier
243--
244{-# INLINE concatMapInterleaveWithK #-}
245concatMapInterleaveWithK :: MonadAsync m =>
246 (Config -> Config) -> (a -> K.Stream m b) -> K.Stream m a -> K.Stream m b
247concatMapInterleaveWithK modifier f =
248 withInterleaveChannelK (`concatMapWithChanK` f) modifier