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 : harendra.kumar@gmail.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
80 , Streaming -- deprecated
84import Control.Monad (void, ap, (>=>))
85import Control.Monad.IO.Class (MonadIO(liftIO))
86import Control.Monad.Trans.Class (MonadTrans(lift))
87#if __GLASGOW_HASKELL__ >= 800
88import Data.Kind (Type)
90#if __GLASGOW_HASKELL__ < 808
91import Data.Semigroup (Semigroup(..))
93import Prelude hiding (map, mapM, concatMap, foldr)
97------------------------------------------------------------------------------
99------------------------------------------------------------------------------
101-- | The type @Stream m a@ represents a monadic stream of values of type 'a'
102-- constructed using actions in monad 'm'. It uses stop, singleton and yield
103-- continuations equivalent to the following direct style type:
106-- data Stream m a = Stop | Singleton a | Yield a (Stream m a)
109-- To facilitate parallel composition we maintain a local state in an 'SVar'
110-- that is shared across and is used for synchronization of the streams being
113-- The singleton case can be expressed in terms of stop and yield but we have
114-- it as a separate case to optimize composition operations for streams with
115-- single element. We build singleton streams in the implementation of 'pure'
116-- for Applicative and Monad, and in 'lift' for MonadTrans.
118-- XXX remove the Stream type parameter from State as it is always constant.
119-- We can remove it from SVar as well
123 State Stream m a -- state
124 -> (a -> Stream m a -> m r) -- yield
125 -> (a -> m r) -- singleton
130------------------------------------------------------------------------------
131-- Types that can behave as a Stream
132------------------------------------------------------------------------------
137-- XXX Use a different SVar based on the stream type. But we need to make sure
138-- that we do not lose performance due to polymorphism.
140-- | Class of types that can represent a stream of elements of some type 'a' in
144class IsStream t where
145 toStream :: t m a -> Stream m a
146 fromStream :: Stream m a -> t m a
147 -- | Constructs a stream by adding a monadic action at the head of an
148 -- existing stream. For example:
151 -- > toList $ getLine \`consM` getLine \`consM` nil
157 -- /Concurrent (do not use 'parallely' to construct infinite streams)/
160 consM :: MonadAsync m => m a -> t m a -> t m a
161 -- | Operator equivalent of 'consM'. We can read it as "@parallel colon@"
162 -- to remember that @|@ comes before ':'.
165 -- > toList $ getLine |: getLine |: nil
172 -- let delay = threadDelay 1000000 >> print 1
173 -- drain $ serially $ delay |: delay |: delay |: nil
174 -- drain $ parallely $ delay |: delay |: delay |: nil
177 -- /Concurrent (do not use 'parallely' to construct infinite streams)/
180 (|:) :: MonadAsync m => m a -> t m a -> t m a
181 -- We can define (|:) just as 'consM' but it is defined explicitly for each
182 -- type because we want to use SPECIALIZE pragma on the definition.
184-- | Same as 'IsStream'.
187{-# DEPRECATED Streaming "Please use IsStream instead." #-}
188type Streaming = IsStream
190-------------------------------------------------------------------------------
191-- Type adapting combinators
192-------------------------------------------------------------------------------
194-- XXX Move/reset the State here by reconstructing the stream with cleared
195-- state. Can we make sure we do not do that when t1 = t2? If we do this then
196-- we do not need to do that explicitly using svarStyle. It would act as
197-- unShare when the stream type is the same.
199-- | Adapt any specific stream type to any other specific stream type.
202adapt :: (IsStream t1, IsStream t2) => t1 m a -> t2 m a
203adapt = fromStream . toStream
205------------------------------------------------------------------------------
207------------------------------------------------------------------------------
209-- XXX The State is always parameterized by "Stream" which means State is not
210-- different for different stream types. So we have to manually make sure that
211-- when converting from one stream to another we migrate the state correctly.
212-- This can be fixed if we use a different SVar type for different streams.
213-- Currently we always use "SVar Stream" and therefore a different State type
214-- parameterized by that stream.
216-- XXX Since t is coercible we should be able to coerce k
217-- mkStream k = fromStream $ MkStream $ coerce k
219-- | Build a stream from an 'SVar', a stop continuation, a singleton stream
220-- continuation and a yield continuation.
221{-# INLINE_EARLY mkStream #-}
222mkStream :: IsStream t
223 => (forall r. State Stream m a
224 -> (a -> t m a -> m r)
229mkStream k = fromStream $ MkStream $ \st yld sng stp ->
230 let yieldk a r = yld a (toStream r)
231 in k st yieldk sng stp
233{-# RULES "mkStream from stream" mkStream = mkStreamFromStream #-}
234mkStreamFromStream :: IsStream t
235 => (forall r. State Stream m a
236 -> (a -> Stream m a -> m r)
241mkStreamFromStream k = fromStream $ MkStream k
243{-# RULES "mkStream stream" mkStream = mkStreamStream #-}
245 :: (forall r. State Stream m a
246 -> (a -> Stream m a -> m r)
251mkStreamStream = MkStream
253-- | A terminal function that has no continuation to follow.
254type StopK m = forall r. m r -> m r
256-- | A monadic continuation, it is a function that yields a value of type "a"
257-- and calls the argument (a -> m r) as a continuation with that value. We can
258-- also think of it as a callback with a handler (a -> m r). Category
259-- theorists call it a codensity type, a special type of right kan extension.
260type YieldK m a = forall r. (a -> m r) -> m r
262_wrapM :: Monad m => m a -> YieldK m a
263_wrapM m = \k -> m >>= k
265-- | Make an empty stream from a stop function.
266fromStopK :: IsStream t => StopK m -> t m a
267fromStopK k = mkStream $ \_ _ _ stp -> k stp
269-- | Make a singleton stream from a yield function.
270fromYieldK :: IsStream t => YieldK m a -> t m a
271fromYieldK k = mkStream $ \_ _ sng _ -> k sng
273-- | Add a yield function at the head of the stream.
274consK :: IsStream t => YieldK m a -> t m a -> t m a
275consK k r = mkStream $ \_ yld _ _ -> k (\x -> yld x r)
277-- XXX Build a stream from a repeating callback function.
279------------------------------------------------------------------------------
281------------------------------------------------------------------------------
285-- faster than consM because there is no bind.
286-- | Construct a stream by adding a pure value at the head of an existing
287-- stream. For serial streams this is the same as @(return a) \`consM` r@ but
288-- more efficient. For concurrent streams this is not concurrent whereas
289-- 'consM' is concurrent. For example:
292-- > toList $ 1 \`cons` 2 \`cons` 3 \`cons` nil
297{-# INLINE_NORMAL cons #-}
298cons :: IsStream t => a -> t m a -> t m a
299cons a r = mkStream $ \_ yld _ _ -> yld a r
303-- | Operator equivalent of 'cons'.
306-- > toList $ 1 .: 2 .: 3 .: nil
312(.:) :: IsStream t => a -> t m a -> t m a
323{-# INLINE_NORMAL nil #-}
324nil :: IsStream t => t m a
325nil = mkStream $ \_ _ _ stp -> stp
327{-# INLINE_NORMAL yield #-}
328yield :: IsStream t => a -> t m a
329yield a = mkStream $ \_ _ single _ -> single a
331{-# INLINE_NORMAL yieldM #-}
332yieldM :: (Monad m, IsStream t) => m a -> t m a
333yieldM m = fromStream $ mkStream $ \_ _ single _ -> m >>= single
335-- XXX specialize to IO?
336{-# INLINE consMBy #-}
337consMBy :: (IsStream t, MonadAsync m) => (t m a -> t m a -> t m a)
338 -> m a -> t m a -> t m a
339consMBy f m r = (fromStream $ yieldM m) `f` r
341------------------------------------------------------------------------------
343------------------------------------------------------------------------------
345-- | Fold a stream by providing an SVar, a stop continuation, a singleton
346-- continuation and a yield continuation. The stream would share the current
347-- SVar passed via the State.
348{-# INLINE_EARLY foldStreamShared #-}
352 -> (a -> t m a -> m r)
357foldStreamShared st yld sng stp m =
358 let yieldk a x = yld a (fromStream x)
359 MkStream k = toStream m
360 in k st yieldk sng stp
362-- XXX write a similar rule for foldStream as well?
363{-# RULES "foldStreamShared from stream"
364 foldStreamShared = foldStreamSharedStream #-}
365foldStreamSharedStream
367 -> (a -> Stream m a -> m r)
372foldStreamSharedStream st yld sng stp m =
373 let MkStream k = toStream m
376-- | Fold a stream by providing a State, stop continuation, a singleton
377-- continuation and a yield continuation. The stream will not use the SVar
379{-# INLINE foldStream #-}
383 -> (a -> t m a -> m r)
388foldStream st yld sng stp m =
389 let yieldk a x = yld a (fromStream x)
390 MkStream k = toStream m
391 in k (adaptState st) yieldk sng stp
393-- Run the stream using a run function associated with the SVar that runs the
394-- streams with a captured snapshot of the monadic state.
395{-# INLINE foldStreamSVar #-}
397 :: (IsStream t, MonadIO m)
399 -> State Stream m a -- state
400 -> (a -> t m a -> m r) -- yield
401 -> (a -> m r) -- singleton
405foldStreamSVar sv st yld sng stp m =
406 let mrun = runInIO $ svarMrun sv
407 in void $ liftIO $ mrun $ foldStreamShared st yld sng stp m
409-------------------------------------------------------------------------------
411-------------------------------------------------------------------------------
413-- NOTE: specializing the function outside the instance definition seems to
414-- improve performance quite a bit at times, even if we have the same
415-- SPECIALIZE in the instance definition.
416{-# INLINE consMStream #-}
417{-# SPECIALIZE consMStream :: IO a -> Stream IO a -> Stream IO a #-}
418consMStream :: (Monad m) => m a -> Stream m a -> Stream m a
419consMStream m r = MkStream $ \_ yld _ _ -> m >>= \a -> yld a r
421-------------------------------------------------------------------------------
423-------------------------------------------------------------------------------
425instance IsStream Stream where
430 {-# SPECIALIZE consM :: IO a -> Stream IO a -> Stream IO a #-}
431 consM :: Monad m => m a -> Stream m a -> Stream m a
435 {-# SPECIALIZE (|:) :: IO a -> Stream IO a -> Stream IO a #-}
436 (|:) :: Monad m => m a -> Stream m a -> Stream m a
439-------------------------------------------------------------------------------
441-------------------------------------------------------------------------------
443-- XXX perhaps we can just use foldrSM/buildM everywhere as they are more
444-- general and cover foldrS/buildS as well.
446-- | The function 'f' decides how to reconstruct the stream. We could
447-- reconstruct using a shared state (SVar) or without sharing the state.
449{-# INLINE foldrSWith #-}
450foldrSWith :: IsStream t
451 => (forall r. State Stream m b
452 -> (b -> t m b -> m r)
457 -> (a -> t m b -> t m b) -> t m b -> t m a -> t m b
458foldrSWith f step final m = go m
460 go m1 = mkStream $ \st yld sng stp ->
461 let run x = f st yld sng stp x
463 single a = run $ step a final
464 yieldk a r = run $ step a (go r)
465 -- XXX if type a and b are the same we do not need adaptState, can we
466 -- save some perf with that?
467 -- XXX since we are using adaptState anyway here we can use
468 -- foldStreamShared instead, will that save some perf?
469 in foldStream (adaptState st) yieldk single stop m1
471-- XXX we can use rewrite rules just for foldrSWith, if the function f is the
472-- same we can rewrite it.
474-- | Fold sharing the SVar state within the reconstructed stream
475{-# INLINE_NORMAL foldrSShared #-}
476foldrSShared :: IsStream t => (a -> t m b -> t m b) -> t m b -> t m a -> t m b
477foldrSShared = foldrSWith foldStreamShared
479-- XXX consM is a typeclass method, therefore rewritten already. Instead maybe
480-- we can make consM polymorphic using rewrite rules.
481-- {-# RULES "foldrSShared/id" foldrSShared consM nil = \x -> x #-}
482{-# RULES "foldrSShared/nil"
483 forall k z. foldrSShared k z nil = z #-}
484{-# RULES "foldrSShared/single"
485 forall k z x. foldrSShared k z (yield x) = k x z #-}
486-- {-# RULES "foldrSShared/app" [1]
487-- forall ys. foldrSShared consM ys = \xs -> xs `conjoin` ys #-}
489-- | Lazy right associative fold to a stream.
490{-# INLINE_NORMAL foldrS #-}
491foldrS :: IsStream t => (a -> t m b -> t m b) -> t m b -> t m a -> t m b
492foldrS = foldrSWith foldStream
494{-# RULES "foldrS/id" foldrS cons nil = \x -> x #-}
495{-# RULES "foldrS/nil" forall k z. foldrS k z nil = z #-}
496-- See notes in GHC.Base about this rule
497-- {-# RULES "foldr/cons"
498-- forall k z x xs. foldrS k z (x `cons` xs) = k x (foldrS k z xs) #-}
499{-# RULES "foldrS/single" forall k z x. foldrS k z (yield x) = k x z #-}
500-- {-# RULES "foldrS/app" [1]
501-- forall ys. foldrS cons ys = \xs -> xs `conjoin` ys #-}
503-------------------------------------------------------------------------------
504-- foldrS with monadic cons i.e. consM
505-------------------------------------------------------------------------------
507{-# INLINE foldrSMWith #-}
508foldrSMWith :: (IsStream t, Monad m)
509 => (forall r. State Stream m b
510 -> (b -> t m b -> m r)
515 -> (m a -> t m b -> t m b) -> t m b -> t m a -> t m b
516foldrSMWith f step final m = go m
518 go m1 = mkStream $ \st yld sng stp ->
519 let run x = f st yld sng stp x
521 single a = run $ step (return a) final
522 yieldk a r = run $ step (return a) (go r)
523 in foldStream (adaptState st) yieldk single stop m1
525{-# INLINE_NORMAL foldrSM #-}
526foldrSM :: (IsStream t, Monad m)
527 => (m a -> t m b -> t m b) -> t m b -> t m a -> t m b
528foldrSM = foldrSMWith foldStream
530-- {-# RULES "foldrSM/id" foldrSM consM nil = \x -> x #-}
531{-# RULES "foldrSM/nil" forall k z. foldrSM k z nil = z #-}
532{-# RULES "foldrSM/single" forall k z x. foldrSM k z (yieldM x) = k x z #-}
533-- {-# RULES "foldrSM/app" [1]
534-- forall ys. foldrSM consM ys = \xs -> xs `conjoin` ys #-}
536-- Like foldrSM but sharing the SVar state within the recostructed stream.
537{-# INLINE_NORMAL foldrSMShared #-}
538foldrSMShared :: (IsStream t, Monad m)
539 => (m a -> t m b -> t m b) -> t m b -> t m a -> t m b
540foldrSMShared = foldrSMWith foldStreamShared
542-- {-# RULES "foldrSM/id" foldrSM consM nil = \x -> x #-}
543{-# RULES "foldrSMShared/nil"
544 forall k z. foldrSMShared k z nil = z #-}
545{-# RULES "foldrSMShared/single"
546 forall k z x. foldrSMShared k z (yieldM x) = k x z #-}
547-- {-# RULES "foldrSM/app" [1]
548-- forall ys. foldrSM consM ys = \xs -> xs `conjoin` ys #-}
550-------------------------------------------------------------------------------
552-------------------------------------------------------------------------------
554{-# INLINE_NORMAL build #-}
555build :: IsStream t => forall a. (forall b. (a -> b -> b) -> b -> b) -> t m a
558{-# RULES "foldrM/build"
559 forall k z (g :: forall b. (a -> b -> b) -> b -> b).
560 foldrM k z (build g) = g k z #-}
562{-# RULES "foldrS/build"
563 forall k z (g :: forall b. (a -> b -> b) -> b -> b).
564 foldrS k z (build g) = g k z #-}
566{-# RULES "foldrS/cons/build"
567 forall k z x (g :: forall b. (a -> b -> b) -> b -> b).
568 foldrS k z (x `cons` build g) = k x (g k z) #-}
570{-# RULES "foldrSShared/build"
571 forall k z (g :: forall b. (a -> b -> b) -> b -> b).
572 foldrSShared k z (build g) = g k z #-}
574{-# RULES "foldrSShared/cons/build"
575 forall k z x (g :: forall b. (a -> b -> b) -> b -> b).
576 foldrSShared k z (x `cons` build g) = k x (g k z) #-}
578-- build a stream by applying cons and nil to a build function
579{-# INLINE_NORMAL buildS #-}
580buildS :: IsStream t => ((a -> t m a -> t m a) -> t m a -> t m a) -> t m a
583{-# RULES "foldrS/buildS"
584 forall k z (g :: (a -> t m a -> t m a) -> t m a -> t m a).
585 foldrS k z (buildS g) = g k z #-}
587{-# RULES "foldrS/cons/buildS"
588 forall k z x (g :: (a -> t m a -> t m a) -> t m a -> t m a).
589 foldrS k z (x `cons` buildS g) = k x (g k z) #-}
591{-# RULES "foldrSShared/buildS"
592 forall k z (g :: (a -> t m a -> t m a) -> t m a -> t m a).
593 foldrSShared k z (buildS g) = g k z #-}
595{-# RULES "foldrSShared/cons/buildS"
596 forall k z x (g :: (a -> t m a -> t m a) -> t m a -> t m a).
597 foldrSShared k z (x `cons` buildS g) = k x (g k z) #-}
599-- build a stream by applying consM and nil to a build function
600{-# INLINE_NORMAL buildSM #-}
601buildSM :: (IsStream t, MonadAsync m)
602 => ((m a -> t m a -> t m a) -> t m a -> t m a) -> t m a
603buildSM g = g consM nil
605{-# RULES "foldrSM/buildSM"
606 forall k z (g :: (m a -> t m a -> t m a) -> t m a -> t m a).
607 foldrSM k z (buildSM g) = g k z #-}
609{-# RULES "foldrSMShared/buildSM"
610 forall k z (g :: (m a -> t m a -> t m a) -> t m a -> t m a).
611 foldrSMShared k z (buildSM g) = g k z #-}
613-- Disabled because this may not fire as consM is a class Op
615{-# RULES "foldrS/consM/buildSM"
616 forall k z x (g :: (m a -> t m a -> t m a) -> t m a -> t m a)
617 . foldrSM k z (x `consM` buildSM g)
622-- Build using monadic build functions (continuations) instead of
623-- reconstructing a stream.
624{-# INLINE_NORMAL buildM #-}
625buildM :: (IsStream t, MonadAsync m)
626 => (forall r. (a -> t m a -> m r)
632buildM g = mkStream $ \st yld sng stp ->
633 g (\a r -> foldStream st yld sng stp (return a `consM` r)) sng stp
635-- | Like 'buildM' but shares the SVar state across computations.
636{-# INLINE_NORMAL sharedM #-}
637sharedM :: (IsStream t, MonadAsync m)
638 => (forall r. (a -> t m a -> m r)
644sharedM g = mkStream $ \st yld sng stp ->
645 g (\a r -> foldStreamShared st yld sng stp (return a `consM` r)) sng stp
647-------------------------------------------------------------------------------
649-------------------------------------------------------------------------------
651{-# INLINE_NORMAL augmentS #-}
652augmentS :: IsStream t
653 => ((a -> t m a -> t m a) -> t m a -> t m a) -> t m a -> t m a
654augmentS g xs = g cons xs
656{-# RULES "augmentS/nil"
657 forall (g :: (a -> t m a -> t m a) -> t m a -> t m a).
658 augmentS g nil = buildS g
661{-# RULES "foldrS/augmentS"
662 forall k z xs (g :: (a -> t m a -> t m a) -> t m a -> t m a).
663 foldrS k z (augmentS g xs) = g k (foldrS k z xs)
666{-# RULES "augmentS/buildS"
667 forall (g :: (a -> t m a -> t m a) -> t m a -> t m a)
668 (h :: (a -> t m a -> t m a) -> t m a -> t m a).
669 augmentS g (buildS h) = buildS (\c n -> g c (h c n))
672{-# INLINE_NORMAL augmentSM #-}
673augmentSM :: (IsStream t, MonadAsync m)
674 => ((m a -> t m a -> t m a) -> t m a -> t m a) -> t m a -> t m a
675augmentSM g xs = g consM xs
677{-# RULES "augmentSM/nil"
678 forall (g :: (m a -> t m a -> t m a) -> t m a -> t m a).
679 augmentSM g nil = buildSM g
682{-# RULES "foldrSM/augmentSM"
683 forall k z xs (g :: (m a -> t m a -> t m a) -> t m a -> t m a).
684 foldrSM k z (augmentSM g xs) = g k (foldrSM k z xs)
687{-# RULES "augmentSM/buildSM"
688 forall (g :: (m a -> t m a -> t m a) -> t m a -> t m a)
689 (h :: (m a -> t m a -> t m a) -> t m a -> t m a).
690 augmentSM g (buildSM h) = buildSM (\c n -> g c (h c n))
693-------------------------------------------------------------------------------
694-- Experimental foldrM/buildM
695-------------------------------------------------------------------------------
697-- | Lazy right fold with a monadic step function.
698{-# INLINE_NORMAL foldrM #-}
699foldrM :: IsStream t => (a -> m b -> m b) -> m b -> t m a -> m b
700foldrM step acc m = go m
704 single a = step a acc
705 yieldk a r = step a (go r)
706 in foldStream defState yieldk single stop m1
708{-# INLINE_NORMAL foldrMKWith #-}
711 -> (a -> t m a -> m b)
718 -> ((a -> t m a -> m b) -> (a -> m b) -> m b -> m b)
720foldrMKWith f step acc g = go g
724 single a = step a acc
725 yieldk a r = step a (go (\yld sng stp -> f defState yld sng stp r))
726 in k yieldk single stop
729{-# RULES "foldrM/buildS"
730 forall k z (g :: (a -> t m a -> t m a) -> t m a -> t m a)
731 . foldrM k z (buildS g)
735-- XXX in which case will foldrM/buildM fusion be useful?
736{-# RULES "foldrM/buildM"
737 forall step acc (g :: (forall r.
743 foldrM step acc (buildM g) = foldrMKWith foldStream step acc g
746{-# RULES "foldrM/sharedM"
747 forall step acc (g :: (forall r.
753 foldrM step acc (sharedM g) = foldrMKWith foldStreamShared step acc g
756------------------------------------------------------------------------------
758------------------------------------------------------------------------------
760-- | Polymorphic version of the 'Semigroup' operation '<>' of 'SerialT'.
761-- Appends two streams sequentially, yielding all elements from the first
762-- stream, and then all elements from the second stream.
766serial :: IsStream t => t m a -> t m a -> t m a
767serial xs ys = augmentS (\c n -> foldrS c n xs) ys
771 go m = mkStream $ \st yld sng stp ->
772 let stop = foldStream st yld sng stp m2
774 yieldk a r = yld a (go r)
775 in foldStream st yieldk single stop m
778-- join/merge/append streams depending on consM
779{-# INLINE conjoin #-}
780conjoin :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
781conjoin xs ys = augmentSM (\c n -> foldrSM c n xs) ys
783instance Semigroup (Stream m a) where
786------------------------------------------------------------------------------
788------------------------------------------------------------------------------
790instance Monoid (Stream m a) where
794-------------------------------------------------------------------------------
796-------------------------------------------------------------------------------
798#if __GLASGOW_HASKELL__ < 800
802{-# INLINE_LATE mapFB #-}
803mapFB :: forall (t :: (Type -> Type) -> Type -> Type) b m a.
804 (b -> t m b -> t m b) -> (a -> b) -> a -> t m b -> t m b
805mapFB c f = \x ys -> c (f x) ys
809"mapFB/mapFB" forall c f g. mapFB (mapFB c f) g = mapFB c (f . g)
810"mapFB/id" forall c. mapFB c (\x -> x) = c
814map :: IsStream t => (a -> b) -> t m a -> t m b
815map f xs = buildS (\c n -> foldrS (mapFB c f) n xs)
817-- XXX This definition might potentially be more efficient, but the cost in the
818-- benchmark is dominated by unfoldrM cost so we cannot correctly determine
819-- differences in the mapping cost. We should perhaps deduct the cost of
820-- unfoldrM from the benchmarks and then compare.
825 mkStream $ \st yld sng stp ->
827 yieldk a r = yld (f a) (go r)
828 in foldStream (adaptState st) yieldk single stp m1
831{-# INLINE_LATE mapMFB #-}
832mapMFB :: Monad m => (m b -> t m b -> t m b) -> (a -> m b) -> m a -> t m b -> t m b
833mapMFB c f = \x ys -> c (x >>= f) ys
836 "mapMFB/mapMFB" forall c f g. mapMFB (mapMFB c f) g = mapMFB c (f >=> g)
838-- XXX These rules may never fire because pure/return type class rules will
841"mapMFB/pure" forall c. mapMFB c (\x -> pure x) = c
842"mapMFB/return" forall c. mapMFB c (\x -> return x) = c
845-- Be careful when modifying this, this uses a consM (|:) deliberately to allow
846-- other stream types to overload it.
848mapM :: (IsStream t, MonadAsync m) => (a -> m b) -> t m a -> t m b
849mapM f = foldrSShared (\x xs -> f x `consM` xs) nil
850-- See note under map definition above.
854 go m1 = mkStream $ \st yld sng stp ->
855 let single a = f a >>= sng
856 yieldk a r = foldStreamShared st yld sng stp $ f a |: go r
857 in foldStream (adaptState st) yieldk single stp m1
860-- This is experimental serial version supporting fusion.
862-- XXX what if we do not want to fuse two concurrent mapMs?
863-- XXX we can combine two concurrent mapM only if the SVar is of the same type
864-- So for now we use it only for serial streams.
865-- XXX fusion would be easier for monomoprhic stream types.
866-- {-# RULES "mapM serial" mapM = mapMSerial #-}
867{-# INLINE mapMSerial #-}
868mapMSerial :: MonadAsync m => (a -> m b) -> Stream m a -> Stream m b
869mapMSerial f xs = buildSM (\c n -> foldrSMShared (mapMFB c f) n xs)
871-- XXX in fact use the Stream type everywhere and only use polymorphism in the
872-- high level modules/prelude.
873instance Monad m => Functor (Stream m) where
876-------------------------------------------------------------------------------
878-------------------------------------------------------------------------------
880instance MonadTrans Stream where
883-------------------------------------------------------------------------------
885-------------------------------------------------------------------------------
887-- | Detach a stream from an SVar
888{-# INLINE unShare #-}
889unShare :: IsStream t => t m a -> t m a
890unShare x = mkStream $ \st yld sng stp ->
891 foldStream st yld sng stp x
893-- XXX This is just concatMapBy with arguments flipped. We need to keep this
894-- instead of using a concatMap style definition because the bind
895-- implementation in Async and WAsync streams show significant perf degradation
896-- if the argument order is changed.
897{-# INLINE bindWith #-}
900 => (forall c. t m c -> t m c -> t m c)
904bindWith par m1 f = go m1
907 mkStream $ \st yld sng stp ->
908 let foldShared = foldStreamShared st yld sng stp
909 single a = foldShared $ unShare (f a)
910 yieldk a r = foldShared $ unShare (f a) `par` go r
911 in foldStream (adaptState st) yieldk single stp m
913-- XXX express in terms of foldrS?
914-- XXX can we use a different stream type for the generated stream being
915-- falttened so that we can combine them differently and keep the resulting
917-- XXX do we need specialize to IO?
918-- XXX can we optimize when c and a are same, by removing the forall using
919-- rewrite rules with type applications?
921-- | Perform a 'concatMap' using a specified concat strategy. The first
922-- argument specifies a merge or concat function that is used to merge the
923-- streams generated by the map function. For example, the concat function
924-- could be 'serial', 'parallel', 'async', 'ahead' or any other zip or merge
928{-# INLINE concatMapBy #-}
931 => (forall c. t m c -> t m c -> t m c)
935concatMapBy par f xs = bindWith par xs f
937{-# INLINE concatMap #-}
938concatMap :: IsStream t => (a -> t m b) -> t m a -> t m b
939concatMap f m = fromStream $
941 (\a -> adapt $ toStream $ f a)
946-- XXX This fuses but when the stream is nil this performs poorly.
947-- The filterAllOut benchmark degrades. Need to investigate and fix that.
948{-# INLINE concatMap #-}
949concatMap :: IsStream t => (a -> t m b) -> t m a -> t m b
950concatMap f xs = buildS
951 (\c n -> foldrS (\x b -> foldrS c b (f x)) n xs)
953-- Stream polymorphic concatMap implementation
954-- XXX need to use buildSM/foldrSMShared for parallel behavior
955-- XXX unShare seems to degrade the fused performance
956{-# INLINE_EARLY concatMap_ #-}
957concatMap_ :: IsStream t => (a -> t m b) -> t m a -> t m b
958concatMap_ f xs = buildS
959 (\c n -> foldrSShared (\x b -> foldrSShared c b (unShare $ f x)) n xs)
962instance Monad m => Applicative (Stream m) where
968-- NOTE: even though concatMap for StreamD is 3x faster compared to StreamK,
969-- the monad instance of StreamD is slower than StreamK after foldr/build
971instance Monad m => Monad (Stream m) where
972 {-# INLINE return #-}
975 (>>=) = flip concatMap
978-- ConcatMap recursively on itself using a merge strategy.
979concatLoopBy :: IsStream t
980 => (forall c. t m c -> t m c -> t m c)
981 -> (a -> t m a) -> t m a
982concatLoopBy = undefined
984-- This is mfix. Put another way, concatMap recursively on the output of a
985-- stream. Compare with iterate.
986concatLoop :: IsStream t => (a -> t m a) -> t m a
987concatLoop = concatLoopBy serial
989instance MonadFix (Stream m) where
992-- The SVar implementation is something similar to concatFeedBack, so maybe
993-- there is an opportunity to share the implementation here. In an SVar we run
994-- a part of an action (a stream), it yields an output and the rest of the
995-- stream, we yield the output and queue back the rest of the stream for
996-- further evaluation. Also see unfoldrA.
998-- There could be multiple variants of this combinator, for example use a
999-- specific way of concating i.e. concatLoopBy. Which also includes combinators
1000-- with different stop behaviors. For example if the Left values are errors we
1001-- can stop the whole composition on errors or on specific errors.
1003-- We can also flip the serial/ahead append behavior e.g. instead of processing
1004-- the Left output after the original stream we can reverse the order.
1006-- | Concat map on the 'Left' output of a stream and merge it back into the
1007-- stream. The right output is yielded in the output stream.
1008concatFeedBack :: IsStream t
1009 => (b -> t m (Either a b)) -> t m (Either a b) -> t m a
1010concatFeedBack = undefined
1012-- Compare this with unfoldr. Start with a seed stream and generate a stream
1013-- with a value and new seeds. The new seeds are fed back to generate a seed
1014-- stream and so on. This is a stream level unfoldr.
1015concatUnfoldr :: IsStream t
1016 => (b -> t m (Maybe (a, b))) -> t m (Maybe (a, b)) -> t m a
1017concatUnfoldr = 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