1{-# LANGUAGE BangPatterns #-}
3{-# LANGUAGE ConstraintKinds #-}
4{-# LANGUAGE FlexibleContexts #-}
5{-# LANGUAGE FlexibleInstances #-}
6{-# LANGUAGE InstanceSigs #-}
7{-# LANGUAGE MultiParamTypeClasses #-}
8{-# LANGUAGE PatternSynonyms #-}
9{-# LANGUAGE KindSignatures #-}
10{-# LANGUAGE ViewPatterns #-}
11{-# LANGUAGE RankNTypes #-}
12{-# LANGUAGE UndecidableInstances #-} -- XXX
14#include "../inline.hs"
17-- Module : Streamly.Streams.StreamK.Type
18-- Copyright : (c) 2017 Harendra Kumar
21-- Maintainer : streamly@composewell.com
22-- Stability : experimental
26-- Continuation passing style (CPS) stream implementation. The symbol 'K' below
27-- denotes a function as well as a Kontinuation.
29module Streamly.Streams.StreamK.Type
31 -- * A class for streams
81 , Streaming -- deprecated
85import Control.Monad (void, ap, (>=>))
86import Control.Monad.IO.Class (MonadIO(liftIO))
87import Control.Monad.Trans.Class (MonadTrans(lift))
88#if __GLASGOW_HASKELL__ >= 800
89import Data.Kind (Type)
91#if __GLASGOW_HASKELL__ < 808
92import Data.Semigroup (Semigroup(..))
94import Prelude hiding (map, mapM, concatMap, foldr)
96import Streamly.Internal.Data.SVar
98------------------------------------------------------------------------------
100------------------------------------------------------------------------------
102-- | The type @Stream m a@ represents a monadic stream of values of type 'a'
103-- constructed using actions in monad 'm'. It uses stop, singleton and yield
104-- continuations equivalent to the following direct style type:
107-- data Stream m a = Stop | Singleton a | Yield a (Stream m a)
110-- To facilitate parallel composition we maintain a local state in an 'SVar'
111-- that is shared across and is used for synchronization of the streams being
114-- The singleton case can be expressed in terms of stop and yield but we have
115-- it as a separate case to optimize composition operations for streams with
116-- single element. We build singleton streams in the implementation of 'pure'
117-- for Applicative and Monad, and in 'lift' for MonadTrans.
119-- XXX remove the Stream type parameter from State as it is always constant.
120-- We can remove it from SVar as well
124 State Stream m a -- state
125 -> (a -> Stream m a -> m r) -- yield
126 -> (a -> m r) -- singleton
131------------------------------------------------------------------------------
132-- Types that can behave as a Stream
133------------------------------------------------------------------------------
138-- XXX Use a different SVar based on the stream type. But we need to make sure
139-- that we do not lose performance due to polymorphism.
141-- | Class of types that can represent a stream of elements of some type 'a' in
145class IsStream t where
146 toStream :: t m a -> Stream m a
147 fromStream :: Stream m a -> t m a
148 -- | Constructs a stream by adding a monadic action at the head of an
149 -- existing stream. For example:
152 -- > toList $ getLine \`consM` getLine \`consM` nil
158 -- /Concurrent (do not use 'parallely' to construct infinite streams)/
161 consM :: MonadAsync m => m a -> t m a -> t m a
162 -- | Operator equivalent of 'consM'. We can read it as "@parallel colon@"
163 -- to remember that @|@ comes before ':'.
166 -- > toList $ getLine |: getLine |: nil
173 -- let delay = threadDelay 1000000 >> print 1
174 -- drain $ serially $ delay |: delay |: delay |: nil
175 -- drain $ parallely $ delay |: delay |: delay |: nil
178 -- /Concurrent (do not use 'parallely' to construct infinite streams)/
181 (|:) :: MonadAsync m => m a -> t m a -> t m a
182 -- We can define (|:) just as 'consM' but it is defined explicitly for each
183 -- type because we want to use SPECIALIZE pragma on the definition.
185-- | Same as 'IsStream'.
188{-# DEPRECATED Streaming "Please use IsStream instead." #-}
189type Streaming = IsStream
191-------------------------------------------------------------------------------
192-- Type adapting combinators
193-------------------------------------------------------------------------------
195-- XXX Move/reset the State here by reconstructing the stream with cleared
196-- state. Can we make sure we do not do that when t1 = t2? If we do this then
197-- we do not need to do that explicitly using svarStyle. It would act as
198-- unShare when the stream type is the same.
200-- | Adapt any specific stream type to any other specific stream type.
203adapt :: (IsStream t1, IsStream t2) => t1 m a -> t2 m a
204adapt = fromStream . toStream
206------------------------------------------------------------------------------
208------------------------------------------------------------------------------
210-- XXX The State is always parameterized by "Stream" which means State is not
211-- different for different stream types. So we have to manually make sure that
212-- when converting from one stream to another we migrate the state correctly.
213-- This can be fixed if we use a different SVar type for different streams.
214-- Currently we always use "SVar Stream" and therefore a different State type
215-- parameterized by that stream.
217-- XXX Since t is coercible we should be able to coerce k
218-- mkStream k = fromStream $ MkStream $ coerce k
220-- | Build a stream from an 'SVar', a stop continuation, a singleton stream
221-- continuation and a yield continuation.
222{-# INLINE_EARLY mkStream #-}
223mkStream :: IsStream t
224 => (forall r. State Stream m a
225 -> (a -> t m a -> m r)
230mkStream k = fromStream $ MkStream $ \st yld sng stp ->
231 let yieldk a r = yld a (toStream r)
232 in k st yieldk sng stp
234{-# RULES "mkStream from stream" mkStream = mkStreamFromStream #-}
235mkStreamFromStream :: IsStream t
236 => (forall r. State Stream m a
237 -> (a -> Stream m a -> m r)
242mkStreamFromStream k = fromStream $ MkStream k
244{-# RULES "mkStream stream" mkStream = mkStreamStream #-}
246 :: (forall r. State Stream m a
247 -> (a -> Stream m a -> m r)
252mkStreamStream = MkStream
254-- | A terminal function that has no continuation to follow.
255type StopK m = forall r. m r -> m r
257-- | A monadic continuation, it is a function that yields a value of type "a"
258-- and calls the argument (a -> m r) as a continuation with that value. We can
259-- also think of it as a callback with a handler (a -> m r). Category
260-- theorists call it a codensity type, a special type of right kan extension.
261type YieldK m a = forall r. (a -> m r) -> m r
263_wrapM :: Monad m => m a -> YieldK m a
264_wrapM m = \k -> m >>= k
266-- | Make an empty stream from a stop function.
267fromStopK :: IsStream t => StopK m -> t m a
268fromStopK k = mkStream $ \_ _ _ stp -> k stp
270-- | Make a singleton stream from a yield function.
271fromYieldK :: IsStream t => YieldK m a -> t m a
272fromYieldK k = mkStream $ \_ _ sng _ -> k sng
274-- | Add a yield function at the head of the stream.
275consK :: IsStream t => YieldK m a -> t m a -> t m a
276consK k r = mkStream $ \_ yld _ _ -> k (\x -> yld x r)
278-- XXX Build a stream from a repeating callback function.
280------------------------------------------------------------------------------
282------------------------------------------------------------------------------
286-- faster than consM because there is no bind.
287-- | Construct a stream by adding a pure value at the head of an existing
288-- stream. For serial streams this is the same as @(return a) \`consM` r@ but
289-- more efficient. For concurrent streams this is not concurrent whereas
290-- 'consM' is concurrent. For example:
293-- > toList $ 1 \`cons` 2 \`cons` 3 \`cons` nil
298{-# INLINE_NORMAL cons #-}
299cons :: IsStream t => a -> t m a -> t m a
300cons a r = mkStream $ \_ yld _ _ -> yld a r
304-- | Operator equivalent of 'cons'.
307-- > toList $ 1 .: 2 .: 3 .: nil
313(.:) :: IsStream t => a -> t m a -> t m a
324{-# INLINE_NORMAL nil #-}
325nil :: IsStream t => t m a
326nil = mkStream $ \_ _ _ stp -> stp
328-- | An empty stream producing a side effect.
331-- > toList (nilM (print "nil"))
337{-# INLINE_NORMAL nilM #-}
338nilM :: (IsStream t, Monad m) => m b -> t m a
339nilM m = mkStream $ \_ _ _ stp -> m >> stp
341{-# INLINE_NORMAL yield #-}
342yield :: IsStream t => a -> t m a
343yield a = mkStream $ \_ _ single _ -> single a
345{-# INLINE_NORMAL yieldM #-}
346yieldM :: (Monad m, IsStream t) => m a -> t m a
347yieldM m = fromStream $ mkStream $ \_ _ single _ -> m >>= single
349-- XXX specialize to IO?
350{-# INLINE consMBy #-}
351consMBy :: (IsStream t, MonadAsync m) => (t m a -> t m a -> t m a)
352 -> m a -> t m a -> t m a
353consMBy f m r = (fromStream $ yieldM m) `f` r
355------------------------------------------------------------------------------
357------------------------------------------------------------------------------
359-- | Fold a stream by providing an SVar, a stop continuation, a singleton
360-- continuation and a yield continuation. The stream would share the current
361-- SVar passed via the State.
362{-# INLINE_EARLY foldStreamShared #-}
366 -> (a -> t m a -> m r)
371foldStreamShared st yld sng stp m =
372 let yieldk a x = yld a (fromStream x)
373 MkStream k = toStream m
374 in k st yieldk sng stp
376-- XXX write a similar rule for foldStream as well?
377{-# RULES "foldStreamShared from stream"
378 foldStreamShared = foldStreamSharedStream #-}
379foldStreamSharedStream
381 -> (a -> Stream m a -> m r)
386foldStreamSharedStream st yld sng stp m =
387 let MkStream k = toStream m
390-- | Fold a stream by providing a State, stop continuation, a singleton
391-- continuation and a yield continuation. The stream will not use the SVar
393{-# INLINE foldStream #-}
397 -> (a -> t m a -> m r)
402foldStream st yld sng stp m =
403 let yieldk a x = yld a (fromStream x)
404 MkStream k = toStream m
405 in k (adaptState st) yieldk sng stp
407-- Run the stream using a run function associated with the SVar that runs the
408-- streams with a captured snapshot of the monadic state.
409{-# INLINE foldStreamSVar #-}
411 :: (IsStream t, MonadIO m)
413 -> State Stream m a -- state
414 -> (a -> t m a -> m r) -- yield
415 -> (a -> m r) -- singleton
419foldStreamSVar sv st yld sng stp m =
420 let mrun = runInIO $ svarMrun sv
421 in void $ liftIO $ mrun $ foldStreamShared st yld sng stp m
423-------------------------------------------------------------------------------
425-------------------------------------------------------------------------------
427-- NOTE: specializing the function outside the instance definition seems to
428-- improve performance quite a bit at times, even if we have the same
429-- SPECIALIZE in the instance definition.
430{-# INLINE consMStream #-}
431{-# SPECIALIZE consMStream :: IO a -> Stream IO a -> Stream IO a #-}
432consMStream :: (Monad m) => m a -> Stream m a -> Stream m a
433consMStream m r = MkStream $ \_ yld _ _ -> m >>= \a -> yld a r
435-------------------------------------------------------------------------------
437-------------------------------------------------------------------------------
439instance IsStream Stream where
444 {-# SPECIALIZE consM :: IO a -> Stream IO a -> Stream IO a #-}
445 consM :: Monad m => m a -> Stream m a -> Stream m a
449 {-# SPECIALIZE (|:) :: IO a -> Stream IO a -> Stream IO a #-}
450 (|:) :: Monad m => m a -> Stream m a -> Stream m a
453-------------------------------------------------------------------------------
455-------------------------------------------------------------------------------
457-- XXX perhaps we can just use foldrSM/buildM everywhere as they are more
458-- general and cover foldrS/buildS as well.
460-- | The function 'f' decides how to reconstruct the stream. We could
461-- reconstruct using a shared state (SVar) or without sharing the state.
463{-# INLINE foldrSWith #-}
464foldrSWith :: IsStream t
465 => (forall r. State Stream m b
466 -> (b -> t m b -> m r)
471 -> (a -> t m b -> t m b) -> t m b -> t m a -> t m b
472foldrSWith f step final m = go m
474 go m1 = mkStream $ \st yld sng stp ->
475 let run x = f st yld sng stp x
477 single a = run $ step a final
478 yieldk a r = run $ step a (go r)
479 -- XXX if type a and b are the same we do not need adaptState, can we
480 -- save some perf with that?
481 -- XXX since we are using adaptState anyway here we can use
482 -- foldStreamShared instead, will that save some perf?
483 in foldStream (adaptState st) yieldk single stop m1
485-- XXX we can use rewrite rules just for foldrSWith, if the function f is the
486-- same we can rewrite it.
488-- | Fold sharing the SVar state within the reconstructed stream
489{-# INLINE_NORMAL foldrSShared #-}
490foldrSShared :: IsStream t => (a -> t m b -> t m b) -> t m b -> t m a -> t m b
491foldrSShared = foldrSWith foldStreamShared
493-- XXX consM is a typeclass method, therefore rewritten already. Instead maybe
494-- we can make consM polymorphic using rewrite rules.
495-- {-# RULES "foldrSShared/id" foldrSShared consM nil = \x -> x #-}
496{-# RULES "foldrSShared/nil"
497 forall k z. foldrSShared k z nil = z #-}
498{-# RULES "foldrSShared/single"
499 forall k z x. foldrSShared k z (yield x) = k x z #-}
500-- {-# RULES "foldrSShared/app" [1]
501-- forall ys. foldrSShared consM ys = \xs -> xs `conjoin` ys #-}
503-- | Lazy right associative fold to a stream.
504{-# INLINE_NORMAL foldrS #-}
505foldrS :: IsStream t => (a -> t m b -> t m b) -> t m b -> t m a -> t m b
506foldrS = foldrSWith foldStream
508{-# RULES "foldrS/id" foldrS cons nil = \x -> x #-}
509{-# RULES "foldrS/nil" forall k z. foldrS k z nil = z #-}
510-- See notes in GHC.Base about this rule
511-- {-# RULES "foldr/cons"
512-- forall k z x xs. foldrS k z (x `cons` xs) = k x (foldrS k z xs) #-}
513{-# RULES "foldrS/single" forall k z x. foldrS k z (yield x) = k x z #-}
514-- {-# RULES "foldrS/app" [1]
515-- forall ys. foldrS cons ys = \xs -> xs `conjoin` ys #-}
517-------------------------------------------------------------------------------
518-- foldrS with monadic cons i.e. consM
519-------------------------------------------------------------------------------
521{-# INLINE foldrSMWith #-}
522foldrSMWith :: (IsStream t, Monad m)
523 => (forall r. State Stream m b
524 -> (b -> t m b -> m r)
529 -> (m a -> t m b -> t m b) -> t m b -> t m a -> t m b
530foldrSMWith f step final m = go m
532 go m1 = mkStream $ \st yld sng stp ->
533 let run x = f st yld sng stp x
535 single a = run $ step (return a) final
536 yieldk a r = run $ step (return a) (go r)
537 in foldStream (adaptState st) yieldk single stop m1
539{-# INLINE_NORMAL foldrSM #-}
540foldrSM :: (IsStream t, Monad m)
541 => (m a -> t m b -> t m b) -> t m b -> t m a -> t m b
542foldrSM = foldrSMWith foldStream
544-- {-# RULES "foldrSM/id" foldrSM consM nil = \x -> x #-}
545{-# RULES "foldrSM/nil" forall k z. foldrSM k z nil = z #-}
546{-# RULES "foldrSM/single" forall k z x. foldrSM k z (yieldM x) = k x z #-}
547-- {-# RULES "foldrSM/app" [1]
548-- forall ys. foldrSM consM ys = \xs -> xs `conjoin` ys #-}
550-- Like foldrSM but sharing the SVar state within the recostructed stream.
551{-# INLINE_NORMAL foldrSMShared #-}
552foldrSMShared :: (IsStream t, Monad m)
553 => (m a -> t m b -> t m b) -> t m b -> t m a -> t m b
554foldrSMShared = foldrSMWith foldStreamShared
556-- {-# RULES "foldrSM/id" foldrSM consM nil = \x -> x #-}
557{-# RULES "foldrSMShared/nil"
558 forall k z. foldrSMShared k z nil = z #-}
559{-# RULES "foldrSMShared/single"
560 forall k z x. foldrSMShared k z (yieldM x) = k x z #-}
561-- {-# RULES "foldrSM/app" [1]
562-- forall ys. foldrSM consM ys = \xs -> xs `conjoin` ys #-}
564-------------------------------------------------------------------------------
566-------------------------------------------------------------------------------
568{-# INLINE_NORMAL build #-}
569build :: IsStream t => forall a. (forall b. (a -> b -> b) -> b -> b) -> t m a
572{-# RULES "foldrM/build"
573 forall k z (g :: forall b. (a -> b -> b) -> b -> b).
574 foldrM k z (build g) = g k z #-}
576{-# RULES "foldrS/build"
577 forall k z (g :: forall b. (a -> b -> b) -> b -> b).
578 foldrS k z (build g) = g k z #-}
580{-# RULES "foldrS/cons/build"
581 forall k z x (g :: forall b. (a -> b -> b) -> b -> b).
582 foldrS k z (x `cons` build g) = k x (g k z) #-}
584{-# RULES "foldrSShared/build"
585 forall k z (g :: forall b. (a -> b -> b) -> b -> b).
586 foldrSShared k z (build g) = g k z #-}
588{-# RULES "foldrSShared/cons/build"
589 forall k z x (g :: forall b. (a -> b -> b) -> b -> b).
590 foldrSShared k z (x `cons` build g) = k x (g k z) #-}
592-- build a stream by applying cons and nil to a build function
593{-# INLINE_NORMAL buildS #-}
594buildS :: IsStream t => ((a -> t m a -> t m a) -> t m a -> t m a) -> t m a
597{-# RULES "foldrS/buildS"
598 forall k z (g :: (a -> t m a -> t m a) -> t m a -> t m a).
599 foldrS k z (buildS g) = g k z #-}
601{-# RULES "foldrS/cons/buildS"
602 forall k z x (g :: (a -> t m a -> t m a) -> t m a -> t m a).
603 foldrS k z (x `cons` buildS g) = k x (g k z) #-}
605{-# RULES "foldrSShared/buildS"
606 forall k z (g :: (a -> t m a -> t m a) -> t m a -> t m a).
607 foldrSShared k z (buildS g) = g k z #-}
609{-# RULES "foldrSShared/cons/buildS"
610 forall k z x (g :: (a -> t m a -> t m a) -> t m a -> t m a).
611 foldrSShared k z (x `cons` buildS g) = k x (g k z) #-}
613-- build a stream by applying consM and nil to a build function
614{-# INLINE_NORMAL buildSM #-}
615buildSM :: (IsStream t, MonadAsync m)
616 => ((m a -> t m a -> t m a) -> t m a -> t m a) -> t m a
617buildSM g = g consM nil
619{-# RULES "foldrSM/buildSM"
620 forall k z (g :: (m a -> t m a -> t m a) -> t m a -> t m a).
621 foldrSM k z (buildSM g) = g k z #-}
623{-# RULES "foldrSMShared/buildSM"
624 forall k z (g :: (m a -> t m a -> t m a) -> t m a -> t m a).
625 foldrSMShared k z (buildSM g) = g k z #-}
627-- Disabled because this may not fire as consM is a class Op
629{-# RULES "foldrS/consM/buildSM"
630 forall k z x (g :: (m a -> t m a -> t m a) -> t m a -> t m a)
631 . foldrSM k z (x `consM` buildSM g)
636-- Build using monadic build functions (continuations) instead of
637-- reconstructing a stream.
638{-# INLINE_NORMAL buildM #-}
639buildM :: (IsStream t, MonadAsync m)
640 => (forall r. (a -> t m a -> m r)
646buildM g = mkStream $ \st yld sng stp ->
647 g (\a r -> foldStream st yld sng stp (return a `consM` r)) sng stp
649-- | Like 'buildM' but shares the SVar state across computations.
650{-# INLINE_NORMAL sharedM #-}
651sharedM :: (IsStream t, MonadAsync m)
652 => (forall r. (a -> t m a -> m r)
658sharedM g = mkStream $ \st yld sng stp ->
659 g (\a r -> foldStreamShared st yld sng stp (return a `consM` r)) sng stp
661-------------------------------------------------------------------------------
663-------------------------------------------------------------------------------
665{-# INLINE_NORMAL augmentS #-}
666augmentS :: IsStream t
667 => ((a -> t m a -> t m a) -> t m a -> t m a) -> t m a -> t m a
668augmentS g xs = g cons xs
670{-# RULES "augmentS/nil"
671 forall (g :: (a -> t m a -> t m a) -> t m a -> t m a).
672 augmentS g nil = buildS g
675{-# RULES "foldrS/augmentS"
676 forall k z xs (g :: (a -> t m a -> t m a) -> t m a -> t m a).
677 foldrS k z (augmentS g xs) = g k (foldrS k z xs)
680{-# RULES "augmentS/buildS"
681 forall (g :: (a -> t m a -> t m a) -> t m a -> t m a)
682 (h :: (a -> t m a -> t m a) -> t m a -> t m a).
683 augmentS g (buildS h) = buildS (\c n -> g c (h c n))
686{-# INLINE_NORMAL augmentSM #-}
687augmentSM :: (IsStream t, MonadAsync m)
688 => ((m a -> t m a -> t m a) -> t m a -> t m a) -> t m a -> t m a
689augmentSM g xs = g consM xs
691{-# RULES "augmentSM/nil"
692 forall (g :: (m a -> t m a -> t m a) -> t m a -> t m a).
693 augmentSM g nil = buildSM g
696{-# RULES "foldrSM/augmentSM"
697 forall k z xs (g :: (m a -> t m a -> t m a) -> t m a -> t m a).
698 foldrSM k z (augmentSM g xs) = g k (foldrSM k z xs)
701{-# RULES "augmentSM/buildSM"
702 forall (g :: (m a -> t m a -> t m a) -> t m a -> t m a)
703 (h :: (m a -> t m a -> t m a) -> t m a -> t m a).
704 augmentSM g (buildSM h) = buildSM (\c n -> g c (h c n))
707-------------------------------------------------------------------------------
708-- Experimental foldrM/buildM
709-------------------------------------------------------------------------------
711-- | Lazy right fold with a monadic step function.
712{-# INLINE_NORMAL foldrM #-}
713foldrM :: IsStream t => (a -> m b -> m b) -> m b -> t m a -> m b
714foldrM step acc m = go m
718 single a = step a acc
719 yieldk a r = step a (go r)
720 in foldStream defState yieldk single stop m1
722{-# INLINE_NORMAL foldrMKWith #-}
725 -> (a -> t m a -> m b)
732 -> ((a -> t m a -> m b) -> (a -> m b) -> m b -> m b)
734foldrMKWith f step acc g = go g
738 single a = step a acc
739 yieldk a r = step a (go (\yld sng stp -> f defState yld sng stp r))
740 in k yieldk single stop
743{-# RULES "foldrM/buildS"
744 forall k z (g :: (a -> t m a -> t m a) -> t m a -> t m a)
745 . foldrM k z (buildS g)
749-- XXX in which case will foldrM/buildM fusion be useful?
750{-# RULES "foldrM/buildM"
751 forall step acc (g :: (forall r.
757 foldrM step acc (buildM g) = foldrMKWith foldStream step acc g
760{-# RULES "foldrM/sharedM"
761 forall step acc (g :: (forall r.
767 foldrM step acc (sharedM g) = foldrMKWith foldStreamShared step acc g
770------------------------------------------------------------------------------
772------------------------------------------------------------------------------
774-- | Polymorphic version of the 'Semigroup' operation '<>' of 'SerialT'.
775-- Appends two streams sequentially, yielding all elements from the first
776-- stream, and then all elements from the second stream.
780serial :: IsStream t => t m a -> t m a -> t m a
781serial xs ys = augmentS (\c n -> foldrS c n xs) ys
785 go m = mkStream $ \st yld sng stp ->
786 let stop = foldStream st yld sng stp m2
788 yieldk a r = yld a (go r)
789 in foldStream st yieldk single stop m
792-- join/merge/append streams depending on consM
793{-# INLINE conjoin #-}
794conjoin :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
795conjoin xs ys = augmentSM (\c n -> foldrSM c n xs) ys
797instance Semigroup (Stream m a) where
800------------------------------------------------------------------------------
802------------------------------------------------------------------------------
804instance Monoid (Stream m a) where
808-------------------------------------------------------------------------------
810-------------------------------------------------------------------------------
812#if __GLASGOW_HASKELL__ < 800
816{-# INLINE_LATE mapFB #-}
817mapFB :: forall (t :: (Type -> Type) -> Type -> Type) b m a.
818 (b -> t m b -> t m b) -> (a -> b) -> a -> t m b -> t m b
819mapFB c f = \x ys -> c (f x) ys
823"mapFB/mapFB" forall c f g. mapFB (mapFB c f) g = mapFB c (f . g)
824"mapFB/id" forall c. mapFB c (\x -> x) = c
828map :: IsStream t => (a -> b) -> t m a -> t m b
829map f xs = buildS (\c n -> foldrS (mapFB c f) n xs)
831-- XXX This definition might potentially be more efficient, but the cost in the
832-- benchmark is dominated by unfoldrM cost so we cannot correctly determine
833-- differences in the mapping cost. We should perhaps deduct the cost of
834-- unfoldrM from the benchmarks and then compare.
839 mkStream $ \st yld sng stp ->
841 yieldk a r = yld (f a) (go r)
842 in foldStream (adaptState st) yieldk single stp m1
845{-# INLINE_LATE mapMFB #-}
846mapMFB :: Monad m => (m b -> t m b -> t m b) -> (a -> m b) -> m a -> t m b -> t m b
847mapMFB c f = \x ys -> c (x >>= f) ys
850 "mapMFB/mapMFB" forall c f g. mapMFB (mapMFB c f) g = mapMFB c (f >=> g)
852-- XXX These rules may never fire because pure/return type class rules will
855"mapMFB/pure" forall c. mapMFB c (\x -> pure x) = c
856"mapMFB/return" forall c. mapMFB c (\x -> return x) = c
859-- Be careful when modifying this, this uses a consM (|:) deliberately to allow
860-- other stream types to overload it.
862mapM :: (IsStream t, MonadAsync m) => (a -> m b) -> t m a -> t m b
863mapM f = foldrSShared (\x xs -> f x `consM` xs) nil
864-- See note under map definition above.
868 go m1 = mkStream $ \st yld sng stp ->
869 let single a = f a >>= sng
870 yieldk a r = foldStreamShared st yld sng stp $ f a |: go r
871 in foldStream (adaptState st) yieldk single stp m1
874-- This is experimental serial version supporting fusion.
876-- XXX what if we do not want to fuse two concurrent mapMs?
877-- XXX we can combine two concurrent mapM only if the SVar is of the same type
878-- So for now we use it only for serial streams.
879-- XXX fusion would be easier for monomoprhic stream types.
880-- {-# RULES "mapM serial" mapM = mapMSerial #-}
881{-# INLINE mapMSerial #-}
882mapMSerial :: MonadAsync m => (a -> m b) -> Stream m a -> Stream m b
883mapMSerial f xs = buildSM (\c n -> foldrSMShared (mapMFB c f) n xs)
885-- XXX in fact use the Stream type everywhere and only use polymorphism in the
886-- high level modules/prelude.
887instance Monad m => Functor (Stream m) where
890-------------------------------------------------------------------------------
892-------------------------------------------------------------------------------
894instance MonadTrans Stream where
897-------------------------------------------------------------------------------
899-------------------------------------------------------------------------------
901-- | Detach a stream from an SVar
902{-# INLINE unShare #-}
903unShare :: IsStream t => t m a -> t m a
904unShare x = mkStream $ \st yld sng stp ->
905 foldStream st yld sng stp x
907-- XXX This is just concatMapBy with arguments flipped. We need to keep this
908-- instead of using a concatMap style definition because the bind
909-- implementation in Async and WAsync streams show significant perf degradation
910-- if the argument order is changed.
911{-# INLINE bindWith #-}
914 => (forall c. t m c -> t m c -> t m c)
918bindWith par m1 f = go m1
921 mkStream $ \st yld sng stp ->
922 let foldShared = foldStreamShared st yld sng stp
923 single a = foldShared $ unShare (f a)
924 yieldk a r = foldShared $ unShare (f a) `par` go r
925 in foldStream (adaptState st) yieldk single stp m
927-- XXX express in terms of foldrS?
928-- XXX can we use a different stream type for the generated stream being
929-- falttened so that we can combine them differently and keep the resulting
931-- XXX do we need specialize to IO?
932-- XXX can we optimize when c and a are same, by removing the forall using
933-- rewrite rules with type applications?
935-- | Perform a 'concatMap' using a specified concat strategy. The first
936-- argument specifies a merge or concat function that is used to merge the
937-- streams generated by the map function. For example, the concat function
938-- could be 'serial', 'parallel', 'async', 'ahead' or any other zip or merge
942{-# INLINE concatMapBy #-}
945 => (forall c. t m c -> t m c -> t m c)
949concatMapBy par f xs = bindWith par xs f
951{-# INLINE concatMap #-}
952concatMap :: IsStream t => (a -> t m b) -> t m a -> t m b
953concatMap f m = fromStream $
955 (\a -> adapt $ toStream $ f a)
960-- XXX This fuses but when the stream is nil this performs poorly.
961-- The filterAllOut benchmark degrades. Need to investigate and fix that.
962{-# INLINE concatMap #-}
963concatMap :: IsStream t => (a -> t m b) -> t m a -> t m b
964concatMap f xs = buildS
965 (\c n -> foldrS (\x b -> foldrS c b (f x)) n xs)
967-- Stream polymorphic concatMap implementation
968-- XXX need to use buildSM/foldrSMShared for parallel behavior
969-- XXX unShare seems to degrade the fused performance
970{-# INLINE_EARLY concatMap_ #-}
971concatMap_ :: IsStream t => (a -> t m b) -> t m a -> t m b
972concatMap_ f xs = buildS
973 (\c n -> foldrSShared (\x b -> foldrSShared c b (unShare $ f x)) n xs)
976instance Monad m => Applicative (Stream m) where
982-- NOTE: even though concatMap for StreamD is 3x faster compared to StreamK,
983-- the monad instance of StreamD is slower than StreamK after foldr/build
985instance Monad m => Monad (Stream m) where
986 {-# INLINE return #-}
989 (>>=) = flip concatMap
992-- ConcatMap recursively on itself using a merge strategy.
993concatLoopBy :: IsStream t
994 => (forall c. t m c -> t m c -> t m c)
995 -> (a -> t m a) -> t m a
996concatLoopBy = undefined
998-- This is mfix. Put another way, concatMap recursively on the output of a
999-- stream. Compare with iterate.
1000concatLoop :: IsStream t => (a -> t m a) -> t m a
1001concatLoop = concatLoopBy serial
1003instance MonadFix (Stream m) where
1006-- The SVar implementation is something similar to concatFeedBack, so maybe
1007-- there is an opportunity to share the implementation here. In an SVar we run
1008-- a part of an action (a stream), it yields an output and the rest of the
1009-- stream, we yield the output and queue back the rest of the stream for
1010-- further evaluation. Also see unfoldrA.
1012-- There could be multiple variants of this combinator, for example use a
1013-- specific way of concating i.e. concatLoopBy. Which also includes combinators
1014-- with different stop behaviors. For example if the Left values are errors we
1015-- can stop the whole composition on errors or on specific errors.
1017-- We can also flip the serial/ahead append behavior e.g. instead of processing
1018-- the Left output after the original stream we can reverse the order.
1020-- | Concat map on the 'Left' output of a stream and merge it back into the
1021-- stream. The right output is yielded in the output stream.
1022concatFeedBack :: IsStream t
1023 => (b -> t m (Either a b)) -> t m (Either a b) -> t m a
1024concatFeedBack = undefined
1026-- Compare this with unfoldr. Start with a seed stream and generate a stream
1027-- with a value and new seeds. The new seeds are fed back to generate a seed
1028-- stream and so on. This is a stream level unfoldr.
1029concatUnfoldr :: IsStream t
1030 => (b -> t m (Maybe (a, b))) -> t m (Maybe (a, b)) -> t m a
1031concatUnfoldr = undefined
keysesc to close
- j k
- move down, up
- enter
- open
- esc
- back, or up a level
- 1 2 3 4
- readme, files, commits, refs
- /
- filter
- y
- copy the clone command
- c
- compare the chosen commit with another (or shift-click both)
- ?
- show or hide this list