git.iko.soy/streamly
on 6bde698 help sign in
streamly/src/Streamly/Pipe/Types.hs
447 lines 18 kB history raw
1{-# LANGUAGE CPP #-}
2{-# LANGUAGE ExistentialQuantification #-}
4#include "inline.hs"
6-- |
7-- Module : Streamly.Pipe.Types
8-- Copyright : (c) 2019 Composewell Technologies
9-- License : BSD3
10-- Maintainer : harendra.kumar@gmail.com
11-- Stability : experimental
12-- Portability : GHC
14module Streamly.Pipe.Types
15 ( Step (..)
16 , Pipe (..)
17 , PipeState (..)
18 , zipWith
19 , tee
20 , map
21 , compose
22 )
23where
25import Control.Arrow (Arrow(..))
26import Control.Category (Category(..))
27import Data.Maybe (isJust)
28import Data.Semigroup (Semigroup(..))
29import Prelude hiding (zipWith, map, id, unzip, null)
30import Streamly.Strict (Tuple'(..), Tuple3'(..))
32import qualified Prelude
34------------------------------------------------------------------------------
35-- Pipes
36------------------------------------------------------------------------------
38-- A scan is a much simpler version of pipes. A scan always produces an output
39-- on an input whereas a pipe does not necessarily produce an output on an
40-- input, it might consume multiple inputs before producing an output. That way
41-- it can implement filtering. Similarly, it can produce more than one output
42-- on an single input.
43--
44-- Therefore when two pipes are composed in parallel formation, one may run
45-- slower or faster than the other. If all of them are being fed from the same
46-- source, we may have to buffer the input to match the speeds. In case of
47-- scans we do not have that problem.
48--
49-- We may also need a "Stop" constructor to indicate that we are not generating
50-- any more values and we can have a "Done" constructor to indicate that we are
51-- not consuming any more values. Similarly we can have a stop with error or
52-- exception and a done with error or leftover values.
53--
54-- In generator mode, Continue means no output/continue. In fold mode Continue means
55-- need more input to produce result. we can perhaps call it Continue instead.
56--
57data Step s a =
58 Yield a s
59 | Continue s
61-- | Represents a stateful transformation over an input stream of values of
62-- type @a@ to outputs of type @b@ in 'Monad' @m@.
64-- A pipe uses a consume function and a produce function. It can switch from
65-- consume/fold mode to a produce/source mode. The first step function is a
66-- fold function while the seocnd one is a stream generator function.
67--
68-- We can upgrade a stream or a fold into a pipe. However, streams are more
69-- efficient in generation and folds are more efficient in consumption.
70--
71-- For pure transformation we can have a 'Scan' type. A Scan would be more
72-- efficient in zipping whereas pipes are useful for merging and zipping where
73-- we know buffering can occur. A Scan type can be upgraded to a pipe.
74--
75-- XXX In general the starting state could either be for generation or for
76-- consumption. Currently we are only starting with a consumption state.
77--
78-- An explicit either type for better readability of the code
79data PipeState s1 s2 = Consume s1 | Produce s2
81isProduce :: PipeState s1 s2 -> Bool
82isProduce s =
83 case s of
84 Produce _ -> True
85 Consume _ -> False
87data Pipe m a b =
88 forall s1 s2. Pipe (s1 -> a -> m (Step (PipeState s1 s2) b))
89 (s2 -> m (Step (PipeState s1 s2) b)) s1
91instance Monad m => Functor (Pipe m a) where
92 {-# INLINE_NORMAL fmap #-}
93 fmap f (Pipe consume produce initial) = Pipe consume' produce' initial
94 where
95 {-# INLINE_LATE consume' #-}
96 consume' st a = do
97 r <- consume st a
98 return $ case r of
99 Yield x s -> Yield (f x) s
100 Continue s -> Continue s
102 {-# INLINE_LATE produce' #-}
103 produce' st = do
104 r <- produce st
105 return $ case r of
106 Yield x s -> Yield (f x) s
107 Continue s -> Continue s
109-- XXX move this to a separate module
110data Deque a = Deque [a] [a]
112{-# INLINE null #-}
113null :: Deque a -> Bool
114null (Deque [] []) = True
115null _ = False
117{-# INLINE snoc #-}
118snoc :: a -> Deque a -> Deque a
119snoc a (Deque snocList consList) = Deque (a : snocList) consList
121{-# INLINE uncons #-}
122uncons :: Deque a -> Maybe (a, Deque a)
123uncons (Deque snocList consList) =
124 case consList of
125 h : t -> Just (h, Deque snocList t)
126 _ ->
127 case Prelude.reverse snocList of
128 h : t -> Just (h, Deque [] t)
129 _ -> Nothing
131-- | The composed pipe distributes the input to both the constituent pipes and
132-- zips the output of the two using a supplied zipping function.
133--
134-- @since 0.7.0
135{-# INLINE_NORMAL zipWith #-}
136zipWith :: Monad m => (a -> b -> c) -> Pipe m i a -> Pipe m i b -> Pipe m i c
137zipWith f (Pipe consumeL produceL stateL) (Pipe consumeR produceR stateR) =
138 Pipe consume produce state
139 where
141 -- Left state means we need to consume input from the source. A Right
142 -- state means we either have buffered input or we are in generation
143 -- mode so we do not need input from source in either case.
144 --
145 state = Tuple' (Consume stateL, Nothing, Nothing)
146 (Consume stateR, Nothing, Nothing)
148 -- XXX for heavy buffering we need to have the (ring) buffer in pinned
149 -- memory using the Storable instance.
150 {-# INLINE_LATE consume #-}
151 consume (Tuple' (sL, resL, lq) (sR, resR, rq)) a = do
152 s1 <- drive sL resL lq consumeL produceL a
153 s2 <- drive sR resR rq consumeR produceR a
154 yieldOutput s1 s2
156 where
158 {-# INLINE drive #-}
159 drive st res queue fConsume fProduce val = do
160 case res of
161 Nothing -> goConsume st queue val fConsume fProduce
162 Just x -> return $
163 case queue of
164 Nothing -> (st, Just x, Just $ (Deque [val] []))
165 Just q -> (st, Just x, Just $ snoc val q)
167 {-# INLINE goConsume #-}
168 goConsume stt queue val fConsume stp2 = do
169 case stt of
170 Consume st -> do
171 case queue of
172 Nothing -> do
173 r <- fConsume st val
174 return $ case r of
175 Yield x s -> (s, Just x, Nothing)
176 Continue s -> (s, Nothing, Nothing)
177 Just queue' ->
178 case uncons queue' of
179 Just (v, q) -> do
180 r <- fConsume st v
181 let q' = snoc val q
182 return $ case r of
183 Yield x s -> (s, Just x, Just q')
184 Continue s -> (s, Nothing, Just q')
185 Nothing -> undefined -- never occurs
186 Produce st -> do
187 r <- stp2 st
188 return $ case r of
189 Yield x s -> (s, Just x, queue)
190 Continue s -> (s, Nothing, queue)
192 {-# INLINE_LATE produce #-}
193 produce (Tuple' (sL, resL, lq) (sR, resR, rq)) = do
194 s1 <- drive sL resL lq consumeL produceL
195 s2 <- drive sR resR rq consumeR produceR
196 yieldOutput s1 s2
198 where
200 {-# INLINE drive #-}
201 drive stt res q fConsume fProduce = do
202 case res of
203 Nothing -> goProduce stt q fConsume fProduce
204 Just x -> return (stt, Just x, q)
206 {-# INLINE goProduce #-}
207 goProduce stt queue fConsume fProduce = do
208 case stt of
209 Consume st -> do
210 case queue of
211 -- See yieldOutput. We enter produce mode only when
212 -- each pipe is either in Produce state or the
213 -- queue is non-empty. So this case cannot occur.
214 Nothing -> undefined
215 Just queue' ->
216 case uncons queue' of
217 Just (v, q) -> do
218 r <- fConsume st v
219 -- We provide a guarantee that if the
220 -- queue is "Just" it is always
221 -- non-empty. yieldOutput and goConsume
222 -- depend on it.
223 let q' = if null q
224 then Nothing
225 else Just q
226 return $ case r of
227 Yield x s -> (s, Just x, q')
228 Continue s -> (s, Nothing, q')
229 Nothing -> return (stt, Nothing, Nothing)
230 Produce st -> do
231 r <- fProduce st
232 return $ case r of
233 Yield x s -> (s, Just x, queue)
234 Continue s -> (s, Nothing, queue)
236 {-# INLINE yieldOutput #-}
237 yieldOutput s1@(sL', resL', lq') s2@(sR', resR', rq') = return $
238 -- switch to produce mode if we do not need input
239 if (isProduce sL' || isJust lq') && (isProduce sR' || isJust rq')
240 then
241 case (resL', resR') of
242 (Just xL, Just xR) ->
243 Yield (f xL xR) (Produce (Tuple' (clear s1) (clear s2)))
244 _ -> Continue (Produce (Tuple' s1 s2))
245 else
246 case (resL', resR') of
247 (Just xL, Just xR) ->
248 Yield (f xL xR) (Consume (Tuple' (clear s1) (clear s2)))
249 _ -> Continue (Consume (Tuple' s1 s2))
250 where clear (s, _, q) = (s, Nothing, q)
252instance Monad m => Applicative (Pipe m a) where
253 {-# INLINE pure #-}
254 pure b = Pipe (\_ _ -> pure $ Yield b (Consume ())) undefined ()
256 (<*>) = zipWith id
258-- | The composed pipe distributes the input to both the constituent pipes and
259-- merges the outputs of the two.
260--
261-- @since 0.7.0
262{-# INLINE_NORMAL tee #-}
263tee :: Monad m => Pipe m a b -> Pipe m a b -> Pipe m a b
264tee (Pipe consumeL produceL stateL) (Pipe consumeR produceR stateR) =
265 Pipe consume produce state
266 where
268 state = Tuple' (Consume stateL) (Consume stateR)
270 consume (Tuple' sL sR) a = do
271 case sL of
272 Consume st -> do
273 r <- consumeL st a
274 return $ case r of
275 Yield x s -> Yield x (Produce (Tuple3' (Just a) s sR))
276 Continue s -> Continue (Produce (Tuple3' (Just a) s sR))
277 -- XXX we should never come here unless the initial state of the
278 -- first pipe is set to "Right".
279 Produce _st -> undefined -- do
280 {-
281 r <- produceL st
282 return $ case r of
283 Yield x s -> Yield x (Right (Tuple3' (Just a) s sR))
284 Continue s -> Continue (Right (Tuple3' (Just a) s sR))
285 -}
287 produce (Tuple3' (Just a) sL sR) = do
288 case sL of
289 Consume _ -> do
290 case sR of
291 Consume st -> do
292 r <- consumeR st a
293 let nextL s = Consume (Tuple' sL s)
294 let nextR s = Produce (Tuple3' Nothing sL s)
295 return $ case r of
296 Yield x s@(Consume _) -> Yield x (nextL s)
297 Yield x s@(Produce _) -> Yield x (nextR s)
298 Continue s@(Consume _) -> Continue (nextL s)
299 Continue s@(Produce _) -> Continue (nextR s)
300 -- We will never come here unless the initial state of
301 -- second pipe is set to "Right".
302 Produce _ -> undefined
303 Produce st -> do
304 r <- produceL st
305 let next s = Produce (Tuple3' (Just a) s sR)
306 return $ case r of
307 Yield x s -> Yield x (next s)
308 Continue s -> Continue (next s)
310 produce (Tuple3' Nothing sL sR) = do
311 case sR of
312 Consume _ -> undefined -- should never occur
313 Produce st -> do
314 r <- produceR st
315 return $ case r of
316 Yield x s@(Consume _) ->
317 Yield x (Consume (Tuple' sL s))
318 Yield x s@(Produce _) ->
319 Yield x (Produce (Tuple3' Nothing sL s))
320 Continue s@(Consume _) ->
321 Continue (Consume (Tuple' sL s))
322 Continue s@(Produce _) ->
323 Continue (Produce (Tuple3' Nothing sL s))
325instance Monad m => Semigroup (Pipe m a b) where
326 {-# INLINE (<>) #-}
327 (<>) = tee
329-- | Lift a pure function to a 'Pipe'.
330--
331-- @since 0.7.0
332{-# INLINE map #-}
333map :: Monad m => (a -> b) -> Pipe m a b
334map f = Pipe consume undefined ()
335 where
336 consume _ a = return $ Yield (f a) (Consume ())
338{-
339-- | A hollow or identity 'Pipe' passes through everything that comes in.
340--
341-- @since 0.7.0
342{-# INLINE id #-}
343id :: Monad m => Pipe m a a
344id = map Prelude.id
345-}
347-- | Compose two pipes such that the output of the second pipe is attached to
348-- the input of the first pipe.
349--
350-- @since 0.7.0
351{-# INLINE_NORMAL compose #-}
352compose :: Monad m => Pipe m b c -> Pipe m a b -> Pipe m a c
353compose (Pipe consumeL produceL stateL) (Pipe consumeR produceR stateR) =
354 Pipe consume produce state
356 where
358 state = Tuple' (Consume stateL) (Consume stateR)
360 consume (Tuple' sL sR) a = do
361 case sL of
362 Consume stt ->
363 case sR of
364 Consume st -> do
365 rres <- consumeR st a
366 case rres of
367 Yield x sR' -> do
368 let next s =
369 if isProduce sR'
370 then Produce s
371 else Consume s
372 lres <- consumeL stt x
373 return $ case lres of
374 Yield y s1@(Consume _) ->
375 Yield y (next $ Tuple' s1 sR')
376 Yield y s1@(Produce _) ->
377 Yield y (Produce $ Tuple' s1 sR')
378 Continue s1@(Consume _) ->
379 Continue (next $ Tuple' s1 sR')
380 Continue s1@(Produce _) ->
381 Continue (Produce $ Tuple' s1 sR')
382 Continue s1@(Consume _) ->
383 return $ Continue (Consume $ Tuple' sL s1)
384 Continue s1@(Produce _) ->
385 return $ Continue (Produce $ Tuple' sL s1)
386 Produce _ -> undefined
387 -- XXX we should never come here unless the initial state of the
388 -- first pipe is set to "Right".
389 Produce _ -> undefined
391 -- XXX we need to write the code in mor optimized fashion. Use Continue
392 -- more and less yield points.
393 produce (Tuple' sL sR) = do
394 case sL of
395 Produce st -> do
396 r <- produceL st
397 let next s = if isProduce sR then Produce s else Consume s
398 return $ case r of
399 Yield x s@(Consume _) -> Yield x (next $ Tuple' s sR)
400 Yield x s@(Produce _) -> Yield x (Produce $ Tuple' s sR)
401 Continue s@(Consume _) -> Continue (next $ Tuple' s sR)
402 Continue s@(Produce _) -> Continue (Produce $ Tuple' s sR)
403 Consume stt ->
404 case sR of
405 Produce st -> do
406 rR <- produceR st
407 case rR of
408 Yield x sR' -> do
409 let next s =
410 if isProduce sR'
411 then Produce s
412 else Consume s
413 rL <- consumeL stt x
414 return $ case rL of
415 Yield y s1@(Consume _) ->
416 Yield y (next $ Tuple' s1 sR')
417 Yield y s1@(Produce _) ->
418 Yield y (Produce $ Tuple' s1 sR')
419 Continue s1@(Consume _) ->
420 Continue (next $ Tuple' s1 sR')
421 Continue s1@(Produce _) ->
422 Continue (Produce $ Tuple' s1 sR')
423 Continue s1@(Consume _) ->
424 return $ Continue (Consume $ Tuple' sL s1)
425 Continue s1@(Produce _) ->
426 return $ Continue (Produce $ Tuple' sL s1)
427 Consume _ -> return $ Continue (Consume $ Tuple' sL sR)
429instance Monad m => Category (Pipe m) where
430 {-# INLINE id #-}
431 id = map Prelude.id
433 {-# INLINE (.) #-}
434 (.) = compose
436unzip :: Pipe m a x -> Pipe m b y -> Pipe m (a, b) (x, y)
437unzip = undefined
439instance Monad m => Arrow (Pipe m) where
440 {-# INLINE arr #-}
441 arr = map
443 {-# INLINE (***) #-}
444 (***) = unzip
446 {-# INLINE (&&&) #-}
447 (&&&) = zipWith (,)