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#if __GLASGOW_HASKELL__ >= 806
12{-# LANGUAGE QuantifiedConstraints #-}
14{-# LANGUAGE RankNTypes #-}
15{-# LANGUAGE UndecidableInstances #-} -- XXX
17#include "../inline.hs"
20-- Module : Streamly.Streams.StreamK.Type
21-- Copyright : (c) 2017 Harendra Kumar
24-- Maintainer : streamly@composewell.com
25-- Stability : experimental
29-- Continuation passing style (CPS) stream implementation. The symbol 'K' below
30-- denotes a function as well as a Kontinuation.
32module Streamly.Streams.StreamK.Type
34 -- * A class for streams
84 , Streaming -- deprecated
88import Control.Monad (void, ap, (>=>))
89import Control.Monad.IO.Class (MonadIO(liftIO))
90import Control.Monad.Trans.Class (MonadTrans(lift))
91#if __GLASGOW_HASKELL__ >= 800
92import Data.Kind (Type)
94#if __GLASGOW_HASKELL__ < 808
95import Data.Semigroup (Semigroup(..))
97import Prelude hiding (map, mapM, concatMap, foldr)
99import Streamly.Internal.Data.SVar
101------------------------------------------------------------------------------
103------------------------------------------------------------------------------
105-- | The type @Stream m a@ represents a monadic stream of values of type 'a'
106-- constructed using actions in monad 'm'. It uses stop, singleton and yield
107-- continuations equivalent to the following direct style type:
110-- data Stream m a = Stop | Singleton a | Yield a (Stream m a)
113-- To facilitate parallel composition we maintain a local state in an 'SVar'
114-- that is shared across and is used for synchronization of the streams being
117-- The singleton case can be expressed in terms of stop and yield but we have
118-- it as a separate case to optimize composition operations for streams with
119-- single element. We build singleton streams in the implementation of 'pure'
120-- for Applicative and Monad, and in 'lift' for MonadTrans.
122-- XXX remove the Stream type parameter from State as it is always constant.
123-- We can remove it from SVar as well
127 State Stream m a -- state
128 -> (a -> Stream m a -> m r) -- yield
129 -> (a -> m r) -- singleton
134------------------------------------------------------------------------------
135-- Types that can behave as a Stream
136------------------------------------------------------------------------------
141-- XXX Use a different SVar based on the stream type. But we need to make sure
142-- that we do not lose performance due to polymorphism.
144-- | Class of types that can represent a stream of elements of some type 'a' in
149#if __GLASGOW_HASKELL__ >= 806
150 ( forall m a. MonadAsync m => Semigroup (t m a)
151 , forall m a. MonadAsync m => Monoid (t m a)
152 , forall m. Monad m => Functor (t m)
153 , forall m. MonadAsync m => Applicative (t m)
157 toStream :: t m a -> Stream m a
158 fromStream :: Stream m a -> t m a
159 -- | Constructs a stream by adding a monadic action at the head of an
160 -- existing stream. For example:
163 -- > toList $ getLine \`consM` getLine \`consM` nil
169 -- /Concurrent (do not use 'parallely' to construct infinite streams)/
172 consM :: MonadAsync m => m a -> t m a -> t m a
173 -- | Operator equivalent of 'consM'. We can read it as "@parallel colon@"
174 -- to remember that @|@ comes before ':'.
177 -- > toList $ getLine |: getLine |: nil
184 -- let delay = threadDelay 1000000 >> print 1
185 -- drain $ serially $ delay |: delay |: delay |: nil
186 -- drain $ parallely $ delay |: delay |: delay |: nil
189 -- /Concurrent (do not use 'parallely' to construct infinite streams)/
192 (|:) :: MonadAsync m => m a -> t m a -> t m a
193 -- We can define (|:) just as 'consM' but it is defined explicitly for each
194 -- type because we want to use SPECIALIZE pragma on the definition.
196-- | Same as 'IsStream'.
199{-# DEPRECATED Streaming "Please use IsStream instead." #-}
200type Streaming = IsStream
202-------------------------------------------------------------------------------
203-- Type adapting combinators
204-------------------------------------------------------------------------------
206-- XXX Move/reset the State here by reconstructing the stream with cleared
207-- state. Can we make sure we do not do that when t1 = t2? If we do this then
208-- we do not need to do that explicitly using svarStyle. It would act as
209-- unShare when the stream type is the same.
211-- | Adapt any specific stream type to any other specific stream type.
214adapt :: (IsStream t1, IsStream t2) => t1 m a -> t2 m a
215adapt = fromStream . toStream
217------------------------------------------------------------------------------
219------------------------------------------------------------------------------
221-- XXX The State is always parameterized by "Stream" which means State is not
222-- different for different stream types. So we have to manually make sure that
223-- when converting from one stream to another we migrate the state correctly.
224-- This can be fixed if we use a different SVar type for different streams.
225-- Currently we always use "SVar Stream" and therefore a different State type
226-- parameterized by that stream.
228-- XXX Since t is coercible we should be able to coerce k
229-- mkStream k = fromStream $ MkStream $ coerce k
231-- | Build a stream from an 'SVar', a stop continuation, a singleton stream
232-- continuation and a yield continuation.
233{-# INLINE_EARLY mkStream #-}
234mkStream :: IsStream t
235 => (forall r. State Stream m a
236 -> (a -> t m a -> m r)
241mkStream k = fromStream $ MkStream $ \st yld sng stp ->
242 let yieldk a r = yld a (toStream r)
243 in k st yieldk sng stp
245{-# RULES "mkStream from stream" mkStream = mkStreamFromStream #-}
246mkStreamFromStream :: IsStream t
247 => (forall r. State Stream m a
248 -> (a -> Stream m a -> m r)
253mkStreamFromStream k = fromStream $ MkStream k
255{-# RULES "mkStream stream" mkStream = mkStreamStream #-}
257 :: (forall r. State Stream m a
258 -> (a -> Stream m a -> m r)
263mkStreamStream = MkStream
265-- | A terminal function that has no continuation to follow.
266type StopK m = forall r. m r -> m r
268-- | A monadic continuation, it is a function that yields a value of type "a"
269-- and calls the argument (a -> m r) as a continuation with that value. We can
270-- also think of it as a callback with a handler (a -> m r). Category
271-- theorists call it a codensity type, a special type of right kan extension.
272type YieldK m a = forall r. (a -> m r) -> m r
274_wrapM :: Monad m => m a -> YieldK m a
275_wrapM m = \k -> m >>= k
277-- | Make an empty stream from a stop function.
278fromStopK :: IsStream t => StopK m -> t m a
279fromStopK k = mkStream $ \_ _ _ stp -> k stp
281-- | Make a singleton stream from a yield function.
282fromYieldK :: IsStream t => YieldK m a -> t m a
283fromYieldK k = mkStream $ \_ _ sng _ -> k sng
285-- | Add a yield function at the head of the stream.
286consK :: IsStream t => YieldK m a -> t m a -> t m a
287consK k r = mkStream $ \_ yld _ _ -> k (\x -> yld x r)
289-- XXX Build a stream from a repeating callback function.
291------------------------------------------------------------------------------
293------------------------------------------------------------------------------
297-- faster than consM because there is no bind.
298-- | Construct a stream by adding a pure value at the head of an existing
299-- stream. For serial streams this is the same as @(return a) \`consM` r@ but
300-- more efficient. For concurrent streams this is not concurrent whereas
301-- 'consM' is concurrent. For example:
304-- > toList $ 1 \`cons` 2 \`cons` 3 \`cons` nil
309{-# INLINE_NORMAL cons #-}
310cons :: IsStream t => a -> t m a -> t m a
311cons a r = mkStream $ \_ yld _ _ -> yld a r
315-- | Operator equivalent of 'cons'.
318-- > toList $ 1 .: 2 .: 3 .: nil
324(.:) :: IsStream t => a -> t m a -> t m a
335{-# INLINE_NORMAL nil #-}
336nil :: IsStream t => t m a
337nil = mkStream $ \_ _ _ stp -> stp
339-- | An empty stream producing a side effect.
342-- > toList (nilM (print "nil"))
348{-# INLINE_NORMAL nilM #-}
349nilM :: (IsStream t, Monad m) => m b -> t m a
350nilM m = mkStream $ \_ _ _ stp -> m >> stp
352{-# INLINE_NORMAL yield #-}
353yield :: IsStream t => a -> t m a
354yield a = mkStream $ \_ _ single _ -> single a
356{-# INLINE_NORMAL yieldM #-}
357yieldM :: (Monad m, IsStream t) => m a -> t m a
358yieldM m = fromStream $ mkStream $ \_ _ single _ -> m >>= single
360-- XXX specialize to IO?
361{-# INLINE consMBy #-}
362consMBy :: (IsStream t, MonadAsync m) => (t m a -> t m a -> t m a)
363 -> m a -> t m a -> t m a
364consMBy f m r = (fromStream $ yieldM m) `f` r
366------------------------------------------------------------------------------
368------------------------------------------------------------------------------
370-- | Fold a stream by providing an SVar, a stop continuation, a singleton
371-- continuation and a yield continuation. The stream would share the current
372-- SVar passed via the State.
373{-# INLINE_EARLY foldStreamShared #-}
377 -> (a -> t m a -> m r)
382foldStreamShared st yld sng stp m =
383 let yieldk a x = yld a (fromStream x)
384 MkStream k = toStream m
385 in k st yieldk sng stp
387-- XXX write a similar rule for foldStream as well?
388{-# RULES "foldStreamShared from stream"
389 foldStreamShared = foldStreamSharedStream #-}
390foldStreamSharedStream
392 -> (a -> Stream m a -> m r)
397foldStreamSharedStream st yld sng stp m =
398 let MkStream k = toStream m
401-- | Fold a stream by providing a State, stop continuation, a singleton
402-- continuation and a yield continuation. The stream will not use the SVar
404{-# INLINE foldStream #-}
408 -> (a -> t m a -> m r)
413foldStream st yld sng stp m =
414 let yieldk a x = yld a (fromStream x)
415 MkStream k = toStream m
416 in k (adaptState st) yieldk sng stp
418-- Run the stream using a run function associated with the SVar that runs the
419-- streams with a captured snapshot of the monadic state.
420{-# INLINE foldStreamSVar #-}
422 :: (IsStream t, MonadIO m)
424 -> State Stream m a -- state
425 -> (a -> t m a -> m r) -- yield
426 -> (a -> m r) -- singleton
430foldStreamSVar sv st yld sng stp m =
431 let mrun = runInIO $ svarMrun sv
432 in void $ liftIO $ mrun $ foldStreamShared st yld sng stp m
434-------------------------------------------------------------------------------
436-------------------------------------------------------------------------------
438-- NOTE: specializing the function outside the instance definition seems to
439-- improve performance quite a bit at times, even if we have the same
440-- SPECIALIZE in the instance definition.
441{-# INLINE consMStream #-}
442{-# SPECIALIZE consMStream :: IO a -> Stream IO a -> Stream IO a #-}
443consMStream :: (Monad m) => m a -> Stream m a -> Stream m a
444consMStream m r = MkStream $ \_ yld _ _ -> m >>= \a -> yld a r
446-------------------------------------------------------------------------------
448-------------------------------------------------------------------------------
450instance IsStream Stream where
455 {-# SPECIALIZE consM :: IO a -> Stream IO a -> Stream IO a #-}
456 consM :: Monad m => m a -> Stream m a -> Stream m a
460 {-# SPECIALIZE (|:) :: IO a -> Stream IO a -> Stream IO a #-}
461 (|:) :: Monad m => m a -> Stream m a -> Stream m a
464-------------------------------------------------------------------------------
466-------------------------------------------------------------------------------
468-- XXX perhaps we can just use foldrSM/buildM everywhere as they are more
469-- general and cover foldrS/buildS as well.
471-- | The function 'f' decides how to reconstruct the stream. We could
472-- reconstruct using a shared state (SVar) or without sharing the state.
474{-# INLINE foldrSWith #-}
475foldrSWith :: IsStream t
476 => (forall r. State Stream m b
477 -> (b -> t m b -> m r)
482 -> (a -> t m b -> t m b) -> t m b -> t m a -> t m b
483foldrSWith f step final m = go m
485 go m1 = mkStream $ \st yld sng stp ->
486 let run x = f st yld sng stp x
488 single a = run $ step a final
489 yieldk a r = run $ step a (go r)
490 -- XXX if type a and b are the same we do not need adaptState, can we
491 -- save some perf with that?
492 -- XXX since we are using adaptState anyway here we can use
493 -- foldStreamShared instead, will that save some perf?
494 in foldStream (adaptState st) yieldk single stop m1
496-- XXX we can use rewrite rules just for foldrSWith, if the function f is the
497-- same we can rewrite it.
499-- | Fold sharing the SVar state within the reconstructed stream
500{-# INLINE_NORMAL foldrSShared #-}
501foldrSShared :: IsStream t => (a -> t m b -> t m b) -> t m b -> t m a -> t m b
502foldrSShared = foldrSWith foldStreamShared
504-- XXX consM is a typeclass method, therefore rewritten already. Instead maybe
505-- we can make consM polymorphic using rewrite rules.
506-- {-# RULES "foldrSShared/id" foldrSShared consM nil = \x -> x #-}
507{-# RULES "foldrSShared/nil"
508 forall k z. foldrSShared k z nil = z #-}
509{-# RULES "foldrSShared/single"
510 forall k z x. foldrSShared k z (yield x) = k x z #-}
511-- {-# RULES "foldrSShared/app" [1]
512-- forall ys. foldrSShared consM ys = \xs -> xs `conjoin` ys #-}
514-- | Lazy right associative fold to a stream.
515{-# INLINE_NORMAL foldrS #-}
516foldrS :: IsStream t => (a -> t m b -> t m b) -> t m b -> t m a -> t m b
517foldrS = foldrSWith foldStream
519{-# RULES "foldrS/id" foldrS cons nil = \x -> x #-}
520{-# RULES "foldrS/nil" forall k z. foldrS k z nil = z #-}
521-- See notes in GHC.Base about this rule
522-- {-# RULES "foldr/cons"
523-- forall k z x xs. foldrS k z (x `cons` xs) = k x (foldrS k z xs) #-}
524{-# RULES "foldrS/single" forall k z x. foldrS k z (yield x) = k x z #-}
525-- {-# RULES "foldrS/app" [1]
526-- forall ys. foldrS cons ys = \xs -> xs `conjoin` ys #-}
528-------------------------------------------------------------------------------
529-- foldrS with monadic cons i.e. consM
530-------------------------------------------------------------------------------
532{-# INLINE foldrSMWith #-}
533foldrSMWith :: (IsStream t, Monad m)
534 => (forall r. State Stream m b
535 -> (b -> t m b -> m r)
540 -> (m a -> t m b -> t m b) -> t m b -> t m a -> t m b
541foldrSMWith f step final m = go m
543 go m1 = mkStream $ \st yld sng stp ->
544 let run x = f st yld sng stp x
546 single a = run $ step (return a) final
547 yieldk a r = run $ step (return a) (go r)
548 in foldStream (adaptState st) yieldk single stop m1
550{-# INLINE_NORMAL foldrSM #-}
551foldrSM :: (IsStream t, Monad m)
552 => (m a -> t m b -> t m b) -> t m b -> t m a -> t m b
553foldrSM = foldrSMWith foldStream
555-- {-# RULES "foldrSM/id" foldrSM consM nil = \x -> x #-}
556{-# RULES "foldrSM/nil" forall k z. foldrSM k z nil = z #-}
557{-# RULES "foldrSM/single" forall k z x. foldrSM k z (yieldM x) = k x z #-}
558-- {-# RULES "foldrSM/app" [1]
559-- forall ys. foldrSM consM ys = \xs -> xs `conjoin` ys #-}
561-- Like foldrSM but sharing the SVar state within the recostructed stream.
562{-# INLINE_NORMAL foldrSMShared #-}
563foldrSMShared :: (IsStream t, Monad m)
564 => (m a -> t m b -> t m b) -> t m b -> t m a -> t m b
565foldrSMShared = foldrSMWith foldStreamShared
567-- {-# RULES "foldrSM/id" foldrSM consM nil = \x -> x #-}
568{-# RULES "foldrSMShared/nil"
569 forall k z. foldrSMShared k z nil = z #-}
570{-# RULES "foldrSMShared/single"
571 forall k z x. foldrSMShared k z (yieldM x) = k x z #-}
572-- {-# RULES "foldrSM/app" [1]
573-- forall ys. foldrSM consM ys = \xs -> xs `conjoin` ys #-}
575-------------------------------------------------------------------------------
577-------------------------------------------------------------------------------
579{-# INLINE_NORMAL build #-}
580build :: IsStream t => forall a. (forall b. (a -> b -> b) -> b -> b) -> t m a
583{-# RULES "foldrM/build"
584 forall k z (g :: forall b. (a -> b -> b) -> b -> b).
585 foldrM k z (build g) = g k z #-}
587{-# RULES "foldrS/build"
588 forall k z (g :: forall b. (a -> b -> b) -> b -> b).
589 foldrS k z (build g) = g k z #-}
591{-# RULES "foldrS/cons/build"
592 forall k z x (g :: forall b. (a -> b -> b) -> b -> b).
593 foldrS k z (x `cons` build g) = k x (g k z) #-}
595{-# RULES "foldrSShared/build"
596 forall k z (g :: forall b. (a -> b -> b) -> b -> b).
597 foldrSShared k z (build g) = g k z #-}
599{-# RULES "foldrSShared/cons/build"
600 forall k z x (g :: forall b. (a -> b -> b) -> b -> b).
601 foldrSShared k z (x `cons` build g) = k x (g k z) #-}
603-- build a stream by applying cons and nil to a build function
604{-# INLINE_NORMAL buildS #-}
605buildS :: IsStream t => ((a -> t m a -> t m a) -> t m a -> t m a) -> t m a
608{-# RULES "foldrS/buildS"
609 forall k z (g :: (a -> t m a -> t m a) -> t m a -> t m a).
610 foldrS k z (buildS g) = g k z #-}
612{-# RULES "foldrS/cons/buildS"
613 forall k z x (g :: (a -> t m a -> t m a) -> t m a -> t m a).
614 foldrS k z (x `cons` buildS g) = k x (g k z) #-}
616{-# RULES "foldrSShared/buildS"
617 forall k z (g :: (a -> t m a -> t m a) -> t m a -> t m a).
618 foldrSShared k z (buildS g) = g k z #-}
620{-# RULES "foldrSShared/cons/buildS"
621 forall k z x (g :: (a -> t m a -> t m a) -> t m a -> t m a).
622 foldrSShared k z (x `cons` buildS g) = k x (g k z) #-}
624-- build a stream by applying consM and nil to a build function
625{-# INLINE_NORMAL buildSM #-}
626buildSM :: (IsStream t, MonadAsync m)
627 => ((m a -> t m a -> t m a) -> t m a -> t m a) -> t m a
628buildSM g = g consM nil
630{-# RULES "foldrSM/buildSM"
631 forall k z (g :: (m a -> t m a -> t m a) -> t m a -> t m a).
632 foldrSM k z (buildSM g) = g k z #-}
634{-# RULES "foldrSMShared/buildSM"
635 forall k z (g :: (m a -> t m a -> t m a) -> t m a -> t m a).
636 foldrSMShared k z (buildSM g) = g k z #-}
638-- Disabled because this may not fire as consM is a class Op
640{-# RULES "foldrS/consM/buildSM"
641 forall k z x (g :: (m a -> t m a -> t m a) -> t m a -> t m a)
642 . foldrSM k z (x `consM` buildSM g)
647-- Build using monadic build functions (continuations) instead of
648-- reconstructing a stream.
649{-# INLINE_NORMAL buildM #-}
650buildM :: (IsStream t, MonadAsync m)
651 => (forall r. (a -> t m a -> m r)
657buildM g = mkStream $ \st yld sng stp ->
658 g (\a r -> foldStream st yld sng stp (return a `consM` r)) sng stp
660-- | Like 'buildM' but shares the SVar state across computations.
661{-# INLINE_NORMAL sharedM #-}
662sharedM :: (IsStream t, MonadAsync m)
663 => (forall r. (a -> t m a -> m r)
669sharedM g = mkStream $ \st yld sng stp ->
670 g (\a r -> foldStreamShared st yld sng stp (return a `consM` r)) sng stp
672-------------------------------------------------------------------------------
674-------------------------------------------------------------------------------
676{-# INLINE_NORMAL augmentS #-}
677augmentS :: IsStream t
678 => ((a -> t m a -> t m a) -> t m a -> t m a) -> t m a -> t m a
679augmentS g xs = g cons xs
681{-# RULES "augmentS/nil"
682 forall (g :: (a -> t m a -> t m a) -> t m a -> t m a).
683 augmentS g nil = buildS g
686{-# RULES "foldrS/augmentS"
687 forall k z xs (g :: (a -> t m a -> t m a) -> t m a -> t m a).
688 foldrS k z (augmentS g xs) = g k (foldrS k z xs)
691{-# RULES "augmentS/buildS"
692 forall (g :: (a -> t m a -> t m a) -> t m a -> t m a)
693 (h :: (a -> t m a -> t m a) -> t m a -> t m a).
694 augmentS g (buildS h) = buildS (\c n -> g c (h c n))
697{-# INLINE_NORMAL augmentSM #-}
698augmentSM :: (IsStream t, MonadAsync m)
699 => ((m a -> t m a -> t m a) -> t m a -> t m a) -> t m a -> t m a
700augmentSM g xs = g consM xs
702{-# RULES "augmentSM/nil"
703 forall (g :: (m a -> t m a -> t m a) -> t m a -> t m a).
704 augmentSM g nil = buildSM g
707{-# RULES "foldrSM/augmentSM"
708 forall k z xs (g :: (m a -> t m a -> t m a) -> t m a -> t m a).
709 foldrSM k z (augmentSM g xs) = g k (foldrSM k z xs)
712{-# RULES "augmentSM/buildSM"
713 forall (g :: (m a -> t m a -> t m a) -> t m a -> t m a)
714 (h :: (m a -> t m a -> t m a) -> t m a -> t m a).
715 augmentSM g (buildSM h) = buildSM (\c n -> g c (h c n))
718-------------------------------------------------------------------------------
719-- Experimental foldrM/buildM
720-------------------------------------------------------------------------------
722-- | Lazy right fold with a monadic step function.
723{-# INLINE_NORMAL foldrM #-}
724foldrM :: IsStream t => (a -> m b -> m b) -> m b -> t m a -> m b
725foldrM step acc m = go m
729 single a = step a acc
730 yieldk a r = step a (go r)
731 in foldStream defState yieldk single stop m1
733{-# INLINE_NORMAL foldrMKWith #-}
736 -> (a -> t m a -> m b)
743 -> ((a -> t m a -> m b) -> (a -> m b) -> m b -> m b)
745foldrMKWith f step acc g = go g
749 single a = step a acc
750 yieldk a r = step a (go (\yld sng stp -> f defState yld sng stp r))
751 in k yieldk single stop
754{-# RULES "foldrM/buildS"
755 forall k z (g :: (a -> t m a -> t m a) -> t m a -> t m a)
756 . foldrM k z (buildS g)
760-- XXX in which case will foldrM/buildM fusion be useful?
761{-# RULES "foldrM/buildM"
762 forall step acc (g :: (forall r.
768 foldrM step acc (buildM g) = foldrMKWith foldStream step acc g
771{-# RULES "foldrM/sharedM"
772 forall step acc (g :: (forall r.
778 foldrM step acc (sharedM g) = foldrMKWith foldStreamShared step acc g
781------------------------------------------------------------------------------
783------------------------------------------------------------------------------
785-- | Polymorphic version of the 'Semigroup' operation '<>' of 'SerialT'.
786-- Appends two streams sequentially, yielding all elements from the first
787-- stream, and then all elements from the second stream.
791serial :: IsStream t => t m a -> t m a -> t m a
792serial xs ys = augmentS (\c n -> foldrS c n xs) ys
796 go m = mkStream $ \st yld sng stp ->
797 let stop = foldStream st yld sng stp m2
799 yieldk a r = yld a (go r)
800 in foldStream st yieldk single stop m
803-- join/merge/append streams depending on consM
804{-# INLINE conjoin #-}
805conjoin :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
806conjoin xs ys = augmentSM (\c n -> foldrSM c n xs) ys
808instance Semigroup (Stream m a) where
811------------------------------------------------------------------------------
813------------------------------------------------------------------------------
815instance Monoid (Stream m a) where
819-------------------------------------------------------------------------------
821-------------------------------------------------------------------------------
823#if __GLASGOW_HASKELL__ < 800
827{-# INLINE_LATE mapFB #-}
828mapFB :: forall (t :: (Type -> Type) -> Type -> Type) b m a.
829 (b -> t m b -> t m b) -> (a -> b) -> a -> t m b -> t m b
830mapFB c f = \x ys -> c (f x) ys
834"mapFB/mapFB" forall c f g. mapFB (mapFB c f) g = mapFB c (f . g)
835"mapFB/id" forall c. mapFB c (\x -> x) = c
839map :: IsStream t => (a -> b) -> t m a -> t m b
840map f xs = buildS (\c n -> foldrS (mapFB c f) n xs)
842-- XXX This definition might potentially be more efficient, but the cost in the
843-- benchmark is dominated by unfoldrM cost so we cannot correctly determine
844-- differences in the mapping cost. We should perhaps deduct the cost of
845-- unfoldrM from the benchmarks and then compare.
850 mkStream $ \st yld sng stp ->
852 yieldk a r = yld (f a) (go r)
853 in foldStream (adaptState st) yieldk single stp m1
856{-# INLINE_LATE mapMFB #-}
857mapMFB :: Monad m => (m b -> t m b -> t m b) -> (a -> m b) -> m a -> t m b -> t m b
858mapMFB c f = \x ys -> c (x >>= f) ys
861 "mapMFB/mapMFB" forall c f g. mapMFB (mapMFB c f) g = mapMFB c (f >=> g)
863-- XXX These rules may never fire because pure/return type class rules will
866"mapMFB/pure" forall c. mapMFB c (\x -> pure x) = c
867"mapMFB/return" forall c. mapMFB c (\x -> return x) = c
870-- Be careful when modifying this, this uses a consM (|:) deliberately to allow
871-- other stream types to overload it.
873mapM :: (IsStream t, MonadAsync m) => (a -> m b) -> t m a -> t m b
874mapM f = foldrSShared (\x xs -> f x `consM` xs) nil
875-- See note under map definition above.
879 go m1 = mkStream $ \st yld sng stp ->
880 let single a = f a >>= sng
881 yieldk a r = foldStreamShared st yld sng stp $ f a |: go r
882 in foldStream (adaptState st) yieldk single stp m1
885-- This is experimental serial version supporting fusion.
887-- XXX what if we do not want to fuse two concurrent mapMs?
888-- XXX we can combine two concurrent mapM only if the SVar is of the same type
889-- So for now we use it only for serial streams.
890-- XXX fusion would be easier for monomoprhic stream types.
891-- {-# RULES "mapM serial" mapM = mapMSerial #-}
892{-# INLINE mapMSerial #-}
893mapMSerial :: MonadAsync m => (a -> m b) -> Stream m a -> Stream m b
894mapMSerial f xs = buildSM (\c n -> foldrSMShared (mapMFB c f) n xs)
896-- XXX in fact use the Stream type everywhere and only use polymorphism in the
897-- high level modules/prelude.
898instance Monad m => Functor (Stream m) where
901-------------------------------------------------------------------------------
903-------------------------------------------------------------------------------
905instance MonadTrans Stream where
908-------------------------------------------------------------------------------
910-------------------------------------------------------------------------------
912-- | Detach a stream from an SVar
913{-# INLINE unShare #-}
914unShare :: IsStream t => t m a -> t m a
915unShare x = mkStream $ \st yld sng stp ->
916 foldStream st yld sng stp x
918-- XXX This is just concatMapBy with arguments flipped. We need to keep this
919-- instead of using a concatMap style definition because the bind
920-- implementation in Async and WAsync streams show significant perf degradation
921-- if the argument order is changed.
922{-# INLINE bindWith #-}
925 => (forall c. t m c -> t m c -> t m c)
929bindWith par m1 f = go m1
932 mkStream $ \st yld sng stp ->
933 let foldShared = foldStreamShared st yld sng stp
934 single a = foldShared $ unShare (f a)
935 yieldk a r = foldShared $ unShare (f a) `par` go r
936 in foldStream (adaptState st) yieldk single stp m
938-- XXX express in terms of foldrS?
939-- XXX can we use a different stream type for the generated stream being
940-- falttened so that we can combine them differently and keep the resulting
942-- XXX do we need specialize to IO?
943-- XXX can we optimize when c and a are same, by removing the forall using
944-- rewrite rules with type applications?
946-- | Perform a 'concatMap' using a specified concat strategy. The first
947-- argument specifies a merge or concat function that is used to merge the
948-- streams generated by the map function. For example, the concat function
949-- could be 'serial', 'parallel', 'async', 'ahead' or any other zip or merge
953{-# INLINE concatMapBy #-}
956 => (forall c. t m c -> t m c -> t m c)
960concatMapBy par f xs = bindWith par xs f
962{-# INLINE concatMap #-}
963concatMap :: IsStream t => (a -> t m b) -> t m a -> t m b
964concatMap f m = fromStream $
966 (\a -> adapt $ toStream $ f a)
971-- XXX This fuses but when the stream is nil this performs poorly.
972-- The filterAllOut benchmark degrades. Need to investigate and fix that.
973{-# INLINE concatMap #-}
974concatMap :: IsStream t => (a -> t m b) -> t m a -> t m b
975concatMap f xs = buildS
976 (\c n -> foldrS (\x b -> foldrS c b (f x)) n xs)
978-- Stream polymorphic concatMap implementation
979-- XXX need to use buildSM/foldrSMShared for parallel behavior
980-- XXX unShare seems to degrade the fused performance
981{-# INLINE_EARLY concatMap_ #-}
982concatMap_ :: IsStream t => (a -> t m b) -> t m a -> t m b
983concatMap_ f xs = buildS
984 (\c n -> foldrSShared (\x b -> foldrSShared c b (unShare $ f x)) n xs)
987instance Monad m => Applicative (Stream m) where
993-- NOTE: even though concatMap for StreamD is 3x faster compared to StreamK,
994-- the monad instance of StreamD is slower than StreamK after foldr/build
996instance Monad m => Monad (Stream m) where
997 {-# INLINE return #-}
1000 (>>=) = flip concatMap
1003-- ConcatMap recursively on itself using a merge strategy.
1004concatLoopBy :: IsStream t
1005 => (forall c. t m c -> t m c -> t m c)
1006 -> (a -> t m a) -> t m a
1007concatLoopBy = undefined
1009-- This is mfix. Put another way, concatMap recursively on the output of a
1010-- stream. Compare with iterate.
1011concatLoop :: IsStream t => (a -> t m a) -> t m a
1012concatLoop = concatLoopBy serial
1014instance MonadFix (Stream m) where
1017-- The SVar implementation is something similar to concatFeedBack, so maybe
1018-- there is an opportunity to share the implementation here. In an SVar we run
1019-- a part of an action (a stream), it yields an output and the rest of the
1020-- stream, we yield the output and queue back the rest of the stream for
1021-- further evaluation. Also see unfoldrA.
1023-- There could be multiple variants of this combinator, for example use a
1024-- specific way of concating i.e. concatLoopBy. Which also includes combinators
1025-- with different stop behaviors. For example if the Left values are errors we
1026-- can stop the whole composition on errors or on specific errors.
1028-- We can also flip the serial/ahead append behavior e.g. instead of processing
1029-- the Left output after the original stream we can reverse the order.
1031-- | Concat map on the 'Left' output of a stream and merge it back into the
1032-- stream. The right output is yielded in the output stream.
1033concatFeedBack :: IsStream t
1034 => (b -> t m (Either a b)) -> t m (Either a b) -> t m a
1035concatFeedBack = undefined
1037-- Compare this with unfoldr. Start with a seed stream and generate a stream
1038-- with a value and new seeds. The new seeds are fed back to generate a seed
1039-- stream and so on. This is a stream level unfoldr.
1040concatUnfoldr :: IsStream t
1041 => (b -> t m (Maybe (a, b))) -> t m (Maybe (a, b)) -> t m a
1042concatUnfoldr = 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