git.iko.soy/streamly
on master help sign in
split Core.hs into CPS stream and concurrent

Modularize it so that we can have a CPS based stream and a direct stream backend.

commit9ad5b200f10de435739c02d4d5eebe4dfeb5ae67 authorHarendra Kumar <harendra.kumar@gmail.com> dateWed 20 Jun 2018, 23:06 (8 years ago) parent42e5c6b
5 files changed+331−292
compare with master
unifiedsplit
src/Streamly/Core.hs → src/Streamly/CPS/Concurrent.hs+7−154 src/Streamly/CPS/Stream.hs+183 src/Streamly/Prelude.hs+69−69 src/Streamly/Streams.hs+70−68 streamly.cabal+2−1
src/Streamly/Core.hs → src/Streamly/CPS/Concurrent.hsrenamed+7−154
@@ -10,7 +10,7 @@
1010{-# LANGUAGE UndecidableInstances #-} -- XXX
1111
1212-- |
13−-- Module : Streamly.Core
13+-- Module : Streamly.CPS.Concurrent
1414-- Copyright : (c) 2017 Harendra Kumar
1515--
1616-- License : BSD3
@@ -19,28 +19,15 @@
1919-- Portability : GHC
2020--
2121--
22−module Streamly.Core
22+module Streamly.CPS.Concurrent
2323 (
24− -- * Streams
25− Stream (..)
26−
27− -- * Construction (pure)
28− , nil
29− , cons
30− , singleton
31− , once
32− , repeat
33−
3424 -- * Construction (monadic)
35− , consM
36− , consMAhead
25+ consMAhead
3726 , consMAsync
3827 , consMWAsync
3928 , consMParallel
4029
4130 -- * Semigroup Style Composition
42− , serial
43− , wSerial
4431 , ahead
4532 , async
4633 , wAsync
@@ -51,7 +38,6 @@ module Streamly.Core
5138 , runWith
5239
5340 -- * zip
54− , zipWith
5541 , zipAsyncWith
5642
5743 -- * Concurrent Stream Vars (SVars)
@@ -61,10 +47,10 @@ module Streamly.Core
6147 )
6248where
6349
50+--
6451import Control.Concurrent.MVar (newEmptyMVar)
6552import Control.Monad.Catch (throwM)
6653import Control.Monad.IO.Class (MonadIO(..))
67−import Control.Monad.Trans.Class (MonadTrans(lift))
6854import Data.Atomics (atomicModifyIORefCAS_)
6955import Data.Concurrent.Queue.MichaelScott
7056 (LinkedQueue, newQ, nullQ)
@@ -72,14 +58,15 @@ import Data.Functor (void)
7258import Data.Heap (Heap, Entry(..))
7359import Data.IORef (IORef, newIORef, readIORef)
7460import Data.Maybe (fromJust)
75−import Data.Semigroup (Semigroup(..))
7661import GHC.IO (IO(..))
7762import Prelude hiding (repeat, zipWith)
7863
79−import Streamly.SVar
8064import qualified Data.Set as S
8165import qualified Data.Heap as H
8266
67+import Streamly.SVar
68+import Streamly.CPS.Stream
69+
8370-- MVar diagnostics has some overhead - around 5% on asyncly null benchmark, we
8471-- can keep it on in production to debug problems quickly if and when they
8572-- happen, but it may result in unexpected output when threads are left hanging
@@ -90,97 +77,6 @@ import Control.Monad (when)
9077import Data.IORef (writeIORef)
9178#endif
9279
93−
94−------------------------------------------------------------------------------
95−-- The stream type
96−------------------------------------------------------------------------------
97−
98−-- | The type 'Stream m a' represents a monadic stream of values of type 'a'
99−-- constructed using actions in monad 'm'. It uses stop, singleton and yield
100−-- continuations equivalent to the following direct style type:
101−--
102−-- data Stream m a = Stop | Singleton a | Yield a (Stream m a)
103−--
104−-- To facilitate parallel composition we maintain a local state in an SVar that
105−-- is shared across and is used for synchronization of the streams being
106−-- composed.
107−--
108−-- The singleton case can be expressed in terms of stop and yield but we have
109−-- it as a separate case to optimize composition operations for streams with
110−-- single element. We build singleton streams in the implementation of 'pure'
111−-- for Applicative and Monad, and in 'lift' for MonadTrans.
112−--
113−newtype Stream m a =
114− Stream {
115− runStream :: forall r.
116− Maybe (SVar Stream m a) -- local state
117− -> m r -- stop
118− -> (a -> m r) -- singleton
119− -> (a -> Stream m a -> m r) -- yield
120− -> m r
121− }
122−
123−nil :: Stream m a
124−nil = Stream $ \_ stp _ _ -> stp
125−
126−-- | faster than consM because there is no bind.
127−cons :: a -> Stream m a -> Stream m a
128−cons a r = Stream $ \_ _ _ yld -> yld a r
129−
130−-- | Same as @once . return@ but may be faster because there is no bind
131−singleton :: a -> Stream m a
132−singleton a = Stream $ \_ _ single _ -> single a
133−
134−{-# INLINE once #-}
135−once :: Monad m => m a -> Stream m a
136−once m = Stream $ \_ _ single _ -> m >>= single
137−
138−{-# INLINE consM #-}
139−consM :: Monad m => m a -> Stream m a -> Stream m a
140−consM m r = Stream $ \_ _ _ yld -> m >>= \a -> yld a r
141−
142−repeat :: a -> Stream m a
143−repeat a = let x = cons a x in x
144−
145−------------------------------------------------------------------------------
146−-- Semigroup
147−------------------------------------------------------------------------------
148−
149−-- | Concatenates two streams sequentially i.e. the first stream is
150−-- exhausted completely before yielding any element from the second stream.
151−{-# INLINE serial #-}
152−serial :: Stream m a -> Stream m a -> Stream m a
153−serial m1 m2 = go m1
154− where
155− go (Stream m) = Stream $ \_ stp sng yld ->
156− let stop = (runStream m2) Nothing stp sng yld
157− single a = yld a m2
158− yield a r = yld a (go r)
159− in m Nothing stop single yield
160−
161−instance Semigroup (Stream m a) where
162− (<>) = serial
163−
164−------------------------------------------------------------------------------
165−-- Monoid
166−------------------------------------------------------------------------------
167−
168−instance Monoid (Stream m a) where
169− mempty = nil
170− mappend = (<>)
171−
172−------------------------------------------------------------------------------
173−-- Interleave
174−------------------------------------------------------------------------------
175−
176−{-# INLINE wSerial #-}
177−wSerial :: Stream m a -> Stream m a -> Stream m a
178−wSerial m1 m2 = Stream $ \_ stp sng yld -> do
179− let stop = (runStream m2) Nothing stp sng yld
180− single a = yld a m2
181− yield a r = yld a (wSerial m2 r)
182− (runStream m1) Nothing stop single yield
183−
18480-------------------------------------------------------------------------------
18581-- Async
18682-------------------------------------------------------------------------------
@@ -599,25 +495,6 @@ parallel = joinStreamVarPar ParallelVar
599495consMParallel :: MonadAsync m => m a -> Stream m a -> Stream m a
600496consMParallel m r = once m `parallel` r
601497
602−-------------------------------------------------------------------------------
603−-- Functor instace is the same for all types
604−-------------------------------------------------------------------------------
605−
606−instance Monad m => Functor (Stream m) where
607− fmap f m = Stream $ \_ stp sng yld ->
608− let single = sng . f
609− yield a r = yld (f a) (fmap f r)
610− in (runStream m) Nothing stp single yield
611−
612−------------------------------------------------------------------------------
613−-- Alternative & MonadPlus
614−------------------------------------------------------------------------------
615−
616−_alt :: Stream m a -> Stream m a -> Stream m a
617−_alt m1 m2 = Stream $ \_ stp sng yld ->
618− let stop = runStream m2 Nothing stp sng yld
619− in runStream m1 Nothing stop sng yld
620−
621498------------------------------------------------------------------------------
622499-- Stream to stream function application
623500------------------------------------------------------------------------------
@@ -645,23 +522,6 @@ runWith f m = do
645522 pushWorkerPar sv (runOne sv m)
646523 f $ fromStreamVar sv
647524
648−------------------------------------------------------------------------------
649−-- Zipping
650−------------------------------------------------------------------------------
651−
652−{-# INLINE zipWith #-}
653−zipWith :: (a -> b -> c) -> Stream m a -> Stream m b -> Stream m c
654−zipWith f m1 m2 = go m1 m2
655− where
656− go mx my = Stream $ \_ stp sng yld -> do
657− let merge a ra =
658− let single2 b = sng (f a b)
659− yield2 b rb = yld (f a b) (go ra rb)
660− in (runStream my) Nothing stp single2 yield2
661− let single1 a = merge a nil
662− yield1 a ra = merge a ra
663− (runStream mx) Nothing stp single1 yield1
664−
665525{-# INLINE zipAsyncWith #-}
666526zipAsyncWith :: MonadAsync m
667527 => (a -> b -> c) -> Stream m a -> Stream m b -> Stream m c
@@ -676,13 +536,6 @@ zipAsyncWith f m1 m2 = Stream $ \_ stp sng yld -> do
676536 mkAsync m = newAsyncVar m
677537 >>= return . fromStreamVar
678538
679−-------------------------------------------------------------------------------
680−-- Transformers
681−-------------------------------------------------------------------------------
682−
683−instance MonadTrans Stream where
684− lift = once
685−
686539-------------------------------------------------------------------------------
687540-- SVar creation
688541-- This code belongs in SVar.hs but is kept here for perf reasons
src/Streamly/CPS/Stream.hsnew file+183
@@ -0,0 +1,183 @@
1+{-# LANGUAGE CPP #-}
2+{-# LANGUAGE ConstraintKinds #-}
3+{-# LANGUAGE FlexibleContexts #-}
4+{-# LANGUAGE FlexibleInstances #-}
5+{-# LANGUAGE LambdaCase #-}
6+{-# LANGUAGE MagicHash #-}
7+{-# LANGUAGE MultiParamTypeClasses #-}
8+{-# LANGUAGE RankNTypes #-}
9+{-# LANGUAGE UnboxedTuples #-}
10+{-# LANGUAGE UndecidableInstances #-} -- XXX
11+
12+-- |
13+-- Module : Streamly.CPS.Stream
14+-- Copyright : (c) 2017 Harendra Kumar
15+--
16+-- License : BSD3
17+-- Maintainer : harendra.kumar@gmail.com
18+-- Stability : experimental
19+-- Portability : GHC
20+--
21+--
22+module Streamly.CPS.Stream
23+ (
24+ -- * Streams
25+ Stream (..)
26+
27+ -- * Construction (pure)
28+ , nil
29+ , cons
30+ , singleton
31+ , once
32+ , repeat
33+
34+ -- * Construction (monadic)
35+ , consM
36+
37+ -- * Semigroup Style Composition
38+ , serial
39+ , wSerial
40+
41+ -- * zip
42+ , zipWith
43+ )
44+where
45+
46+import Control.Monad.Trans.Class (MonadTrans(lift))
47+import Data.Semigroup (Semigroup(..))
48+import Prelude hiding (repeat, zipWith)
49+
50+import Streamly.SVar
51+
52+------------------------------------------------------------------------------
53+-- The stream type
54+------------------------------------------------------------------------------
55+
56+-- | The type 'Stream m a' represents a monadic stream of values of type 'a'
57+-- constructed using actions in monad 'm'. It uses stop, singleton and yield
58+-- continuations equivalent to the following direct style type:
59+--
60+-- data Stream m a = Stop | Singleton a | Yield a (Stream m a)
61+--
62+-- To facilitate parallel composition we maintain a local state in an SVar that
63+-- is shared across and is used for synchronization of the streams being
64+-- composed.
65+--
66+-- The singleton case can be expressed in terms of stop and yield but we have
67+-- it as a separate case to optimize composition operations for streams with
68+-- single element. We build singleton streams in the implementation of 'pure'
69+-- for Applicative and Monad, and in 'lift' for MonadTrans.
70+--
71+newtype Stream m a =
72+ Stream {
73+ runStream :: forall r.
74+ Maybe (SVar Stream m a) -- local state
75+ -> m r -- stop
76+ -> (a -> m r) -- singleton
77+ -> (a -> Stream m a -> m r) -- yield
78+ -> m r
79+ }
80+
81+nil :: Stream m a
82+nil = Stream $ \_ stp _ _ -> stp
83+
84+-- | faster than consM because there is no bind.
85+cons :: a -> Stream m a -> Stream m a
86+cons a r = Stream $ \_ _ _ yld -> yld a r
87+
88+-- | Same as @once . return@ but may be faster because there is no bind
89+singleton :: a -> Stream m a
90+singleton a = Stream $ \_ _ single _ -> single a
91+
92+{-# INLINE once #-}
93+once :: Monad m => m a -> Stream m a
94+once m = Stream $ \_ _ single _ -> m >>= single
95+
96+{-# INLINE consM #-}
97+consM :: Monad m => m a -> Stream m a -> Stream m a
98+consM m r = Stream $ \_ _ _ yld -> m >>= \a -> yld a r
99+
100+repeat :: a -> Stream m a
101+repeat a = let x = cons a x in x
102+
103+------------------------------------------------------------------------------
104+-- Semigroup
105+------------------------------------------------------------------------------
106+
107+-- | Concatenates two streams sequentially i.e. the first stream is
108+-- exhausted completely before yielding any element from the second stream.
109+{-# INLINE serial #-}
110+serial :: Stream m a -> Stream m a -> Stream m a
111+serial m1 m2 = go m1
112+ where
113+ go (Stream m) = Stream $ \_ stp sng yld ->
114+ let stop = (runStream m2) Nothing stp sng yld
115+ single a = yld a m2
116+ yield a r = yld a (go r)
117+ in m Nothing stop single yield
118+
119+instance Semigroup (Stream m a) where
120+ (<>) = serial
121+
122+------------------------------------------------------------------------------
123+-- Monoid
124+------------------------------------------------------------------------------
125+
126+instance Monoid (Stream m a) where
127+ mempty = nil
128+ mappend = (<>)
129+
130+------------------------------------------------------------------------------
131+-- Interleave
132+------------------------------------------------------------------------------
133+
134+{-# INLINE wSerial #-}
135+wSerial :: Stream m a -> Stream m a -> Stream m a
136+wSerial m1 m2 = Stream $ \_ stp sng yld -> do
137+ let stop = (runStream m2) Nothing stp sng yld
138+ single a = yld a m2
139+ yield a r = yld a (wSerial m2 r)
140+ (runStream m1) Nothing stop single yield
141+
142+-------------------------------------------------------------------------------
143+-- Functor
144+-------------------------------------------------------------------------------
145+
146+instance Monad m => Functor (Stream m) where
147+ fmap f m = Stream $ \_ stp sng yld ->
148+ let single = sng . f
149+ yield a r = yld (f a) (fmap f r)
150+ in (runStream m) Nothing stp single yield
151+
152+------------------------------------------------------------------------------
153+-- Alternative & MonadPlus
154+------------------------------------------------------------------------------
155+
156+_alt :: Stream m a -> Stream m a -> Stream m a
157+_alt m1 m2 = Stream $ \_ stp sng yld ->
158+ let stop = runStream m2 Nothing stp sng yld
159+ in runStream m1 Nothing stop sng yld
160+
161+------------------------------------------------------------------------------
162+-- Zipping
163+------------------------------------------------------------------------------
164+
165+{-# INLINE zipWith #-}
166+zipWith :: (a -> b -> c) -> Stream m a -> Stream m b -> Stream m c
167+zipWith f m1 m2 = go m1 m2
168+ where
169+ go mx my = Stream $ \_ stp sng yld -> do
170+ let merge a ra =
171+ let single2 b = sng (f a b)
172+ yield2 b rb = yld (f a b) (go ra rb)
173+ in (runStream my) Nothing stp single2 yield2
174+ let single1 a = merge a nil
175+ yield1 a ra = merge a ra
176+ (runStream mx) Nothing stp single1 yield1
177+
178+-------------------------------------------------------------------------------
179+-- Transformers
180+-------------------------------------------------------------------------------
181+
182+instance MonadTrans Stream where
183+ lift = once
src/Streamly/Prelude.hs+69−69
@@ -142,8 +142,8 @@ import Prelude hiding (filter, drop, dropWhile, take,
142142import qualified Prelude
143143import qualified System.IO as IO
144144
145−import qualified Streamly.Core as S
146−import Streamly.Core (Stream(Stream))
145+import qualified Streamly.CPS.Stream as C
146+import qualified Streamly.CPS.Concurrent as CC
147147import Streamly.Streams
148148
149149------------------------------------------------------------------------------
@@ -171,7 +171,7 @@ import Streamly.Streams
171171unfoldr :: IsStream t => (b -> Maybe (a, b)) -> b -> t m a
172172unfoldr step = fromStream . go
173173 where
174− go s = Stream $ \_ stp _ yld ->
174+ go s = C.Stream $ \_ stp _ yld ->
175175 case step s of
176176 Nothing -> stp
177177 Just (a, b) -> yld a (go b)
@@ -211,12 +211,12 @@ unfoldr step = fromStream . go
211211unfoldrM :: (IsStream t, MonadAsync m) => (b -> m (Maybe (a, b))) -> b -> t m a
212212unfoldrM step = go
213213 where
214− go s = fromStream $ Stream $ \svr stp sng yld -> do
214+ go s = fromStream $ C.Stream $ \svr stp sng yld -> do
215215 mayb <- step s
216216 case mayb of
217217 Nothing -> stp
218218 Just (a, b) ->
219− S.runStream (toStream (return a |: go b)) svr stp sng yld
219+ C.runStream (toStream (return a |: go b)) svr stp sng yld
220220
221221-- | Construct a stream from a 'Foldable' containing pure values.
222222--
@@ -258,7 +258,7 @@ each = fromFoldable
258258--
259259-- @since 0.2.0
260260once :: (IsStream t, Monad m) => m a -> t m a
261−once = fromStream . S.once
261+once = fromStream . C.once
262262
263263-- | Generate a stream by performing a monadic action @n@ times.
264264--
@@ -296,7 +296,7 @@ repeatM = go
296296iterate :: IsStream t => (a -> a) -> a -> t m a
297297iterate step = fromStream . go
298298 where
299− go s = S.cons s (go (step s))
299+ go s = C.cons s (go (step s))
300300
301301-- | Iterate a monadic function from a seed value, streaming the results
302302-- forever.
@@ -320,9 +320,9 @@ iterate step = fromStream . go
320320iterateM :: (IsStream t, MonadAsync m) => (a -> m a) -> a -> t m a
321321iterateM step = go
322322 where
323− go s = fromStream $ Stream $ \svr stp sng yld -> do
323+ go s = fromStream $ C.Stream $ \svr stp sng yld -> do
324324 next <- step s
325− S.runStream (toStream (return s |: go next)) svr stp sng yld
325+ C.runStream (toStream (return s |: go next)) svr stp sng yld
326326
327327-- | Read lines from an IO Handle into a stream of Strings.
328328--
@@ -330,7 +330,7 @@ iterateM step = go
330330fromHandle :: (IsStream t, MonadIO m) => IO.Handle -> t m String
331331fromHandle h = fromStream go
332332 where
333− go = Stream $ \_ stp _ yld -> do
333+ go = C.Stream $ \_ stp _ yld -> do
334334 eof <- liftIO $ IO.hIsEOF h
335335 if eof
336336 then stp
@@ -357,7 +357,7 @@ foldr step acc m = go (toStream m)
357357 let stop = return acc
358358 single a = return (step a acc)
359359 yield a r = go r >>= \b -> return (step a b)
360− in (S.runStream m1) Nothing stop single yield
360+ in (C.runStream m1) Nothing stop single yield
361361
362362-- | Lazy right fold with a monadic step function. For example, to fold a
363363-- stream into a list:
@@ -376,7 +376,7 @@ foldrM step acc m = go (toStream m)
376376 let stop = return acc
377377 single a = step a acc
378378 yield a r = go r >>= step a
379− in (S.runStream m1) Nothing stop single yield
379+ in (C.runStream m1) Nothing stop single yield
380380
381381-- | Strict left scan with an extraction function. Like 'scanl'', but applies a
382382-- user supplied extraction function (the third argument) at each step. This is
@@ -389,12 +389,12 @@ scanx :: IsStream t => (x -> a -> x) -> x -> (x -> b) -> t m a -> t m b
389389scanx step begin done m =
390390 cons (done begin) $ fromStream $ go (toStream m) begin
391391 where
392− go m1 !acc = Stream $ \_ stp sng yld ->
392+ go m1 !acc = C.Stream $ \_ stp sng yld ->
393393 let single a = sng (done $ step acc a)
394394 yield a r =
395395 let s = step acc a
396396 in yld (done s) (go r s)
397− in S.runStream m1 Nothing stp single yield
397+ in C.runStream m1 Nothing stp single yield
398398
399399-- |
400400-- @since 0.1.1
@@ -425,19 +425,19 @@ foldx step begin done m = get $ go (toStream m) begin
425425 {-# NOINLINE get #-}
426426 get m1 =
427427 let single = return . done
428− in (S.runStream m1) Nothing undefined single undefined
428+ in (C.runStream m1) Nothing undefined single undefined
429429
430430 -- Note, this can be implemented by making a recursive call to "go",
431431 -- however that is more expensive because of unnecessary recursion
432432 -- that cannot be tail call optimized. Unfolding recursion explicitly via
433433 -- continuations is much more efficient.
434− go m1 !acc = Stream $ \_ _ sng yld ->
434+ go m1 !acc = C.Stream $ \_ _ sng yld ->
435435 let stop = sng acc
436436 single a = sng $ step acc a
437437 yield a r =
438438 let stream = go r (step acc a)
439− in (S.runStream stream) Nothing undefined sng yld
440− in (S.runStream m1) Nothing stop single yield
439+ in (C.runStream stream) Nothing undefined sng yld
440+ in (C.runStream m1) Nothing stop single yield
441441
442442-- |
443443-- @since 0.1.0
@@ -463,7 +463,7 @@ foldxM step begin done m = go begin (toStream m)
463463 let stop = acc >>= done
464464 single a = acc >>= \b -> step b a >>= done
465465 yield a r = acc >>= \b -> go (step b a) r
466− in (S.runStream m1) Nothing stop single yield
466+ in (C.runStream m1) Nothing stop single yield
467467
468468-- |
469469-- @since 0.1.0
@@ -487,7 +487,7 @@ uncons m =
487487 let stop = return Nothing
488488 single a = return (Just (a, nil))
489489 yield a r = return (Just (a, fromStream r))
490− in (S.runStream (toStream m)) Nothing stop single yield
490+ in (C.runStream (toStream m)) Nothing stop single yield
491491
492492-- | Write a stream of Strings to an IO Handle.
493493--
@@ -499,7 +499,7 @@ toHandle h m = go (toStream m)
499499 let stop = return ()
500500 single a = liftIO (IO.hPutStrLn h a)
501501 yield a r = liftIO (IO.hPutStrLn h a) >> go r
502− in (S.runStream m1) Nothing stop single yield
502+ in (C.runStream m1) Nothing stop single yield
503503
504504------------------------------------------------------------------------------
505505-- Special folds
@@ -519,9 +519,9 @@ toList = foldrM (\a xs -> return (a : xs)) []
519519take :: IsStream t => Int -> t m a -> t m a
520520take n m = fromStream $ go n (toStream m)
521521 where
522− go n1 m1 = Stream $ \_ stp sng yld ->
522+ go n1 m1 = C.Stream $ \_ stp sng yld ->
523523 let yield a r = yld a (go (n1 - 1) r)
524− in if n1 <= 0 then stp else (S.runStream m1) Nothing stp sng yield
524+ in if n1 <= 0 then stp else (C.runStream m1) Nothing stp sng yield
525525
526526-- | Include only those elements that pass a predicate.
527527--
@@ -530,12 +530,12 @@ take n m = fromStream $ go n (toStream m)
530530filter :: IsStream t => (a -> Bool) -> t m a -> t m a
531531filter p m = fromStream $ go (toStream m)
532532 where
533− go m1 = Stream $ \_ stp sng yld ->
533+ go m1 = C.Stream $ \_ stp sng yld ->
534534 let single a | p a = sng a
535535 | otherwise = stp
536536 yield a r | p a = yld a (go r)
537− | otherwise = (S.runStream r) Nothing stp single yield
538− in (S.runStream m1) Nothing stp single yield
537+ | otherwise = (C.runStream r) Nothing stp single yield
538+ in (C.runStream m1) Nothing stp single yield
539539
540540-- | End the stream as soon as the predicate fails on an element.
541541--
@@ -544,12 +544,12 @@ filter p m = fromStream $ go (toStream m)
544544takeWhile :: IsStream t => (a -> Bool) -> t m a -> t m a
545545takeWhile p m = fromStream $ go (toStream m)
546546 where
547− go m1 = Stream $ \_ stp sng yld ->
547+ go m1 = C.Stream $ \_ stp sng yld ->
548548 let single a | p a = sng a
549549 | otherwise = stp
550550 yield a r | p a = yld a (go r)
551551 | otherwise = stp
552− in (S.runStream m1) Nothing stp single yield
552+ in (C.runStream m1) Nothing stp single yield
553553
554554-- | Discard first 'n' elements from the stream and take the rest.
555555--
@@ -557,13 +557,13 @@ takeWhile p m = fromStream $ go (toStream m)
557557drop :: IsStream t => Int -> t m a -> t m a
558558drop n m = fromStream $ go n (toStream m)
559559 where
560− go n1 m1 = Stream $ \_ stp sng yld ->
560+ go n1 m1 = C.Stream $ \_ stp sng yld ->
561561 let single _ = stp
562− yield _ r = (S.runStream $ go (n1 - 1) r) Nothing stp sng yld
562+ yield _ r = (C.runStream $ go (n1 - 1) r) Nothing stp sng yld
563563 -- Somehow "<=" check performs better than a ">"
564564 in if n1 <= 0
565− then (S.runStream m1) Nothing stp sng yld
566− else (S.runStream m1) Nothing stp single yield
565+ then (C.runStream m1) Nothing stp sng yld
566+ else (C.runStream m1) Nothing stp single yield
567567
568568-- | Drop elements in the stream as long as the predicate succeeds and then
569569-- take the rest of the stream.
@@ -573,12 +573,12 @@ drop n m = fromStream $ go n (toStream m)
573573dropWhile :: IsStream t => (a -> Bool) -> t m a -> t m a
574574dropWhile p m = fromStream $ go (toStream m)
575575 where
576− go m1 = Stream $ \_ stp sng yld ->
576+ go m1 = C.Stream $ \_ stp sng yld ->
577577 let single a | p a = stp
578578 | otherwise = sng a
579− yield a r | p a = (S.runStream r) Nothing stp single yield
579+ yield a r | p a = (C.runStream r) Nothing stp single yield
580580 | otherwise = yld a r
581− in (S.runStream m1) Nothing stp single yield
581+ in (C.runStream m1) Nothing stp single yield
582582
583583-- | Determine whether all elements of a stream satisfy a predicate.
584584--
@@ -591,7 +591,7 @@ all p m = go (toStream m)
591591 | otherwise = return False
592592 yield a r | p a = go r
593593 | otherwise = return False
594− in (S.runStream m1) Nothing (return True) single yield
594+ in (C.runStream m1) Nothing (return True) single yield
595595
596596-- | Determine whether any of the elements of a stream satisfy a predicate.
597597--
@@ -604,7 +604,7 @@ any p m = go (toStream m)
604604 | otherwise = return False
605605 yield a r | p a = return True
606606 | otherwise = go r
607− in (S.runStream m1) Nothing (return False) single yield
607+ in (C.runStream m1) Nothing (return False) single yield
608608
609609-- | Determine the sum of all elements of a stream of numbers
610610--
@@ -626,7 +626,7 @@ head m =
626626 let stop = return Nothing
627627 single a = return (Just a)
628628 yield a _ = return (Just a)
629− in (S.runStream (toStream m)) Nothing stop single yield
629+ in (C.runStream (toStream m)) Nothing stop single yield
630630
631631-- | Extract all but the first element of the stream, if any.
632632--
@@ -636,7 +636,7 @@ tail m =
636636 let stop = return Nothing
637637 single _ = return $ Just nil
638638 yield _ r = return $ Just $ fromStream r
639− in (S.runStream (toStream m)) Nothing stop single yield
639+ in (C.runStream (toStream m)) Nothing stop single yield
640640
641641-- | Extract the last element of the stream, if any.
642642--
@@ -653,7 +653,7 @@ null m =
653653 let stop = return True
654654 single _ = return False
655655 yield _ _ = return False
656− in (S.runStream (toStream m)) Nothing stop single yield
656+ in (C.runStream (toStream m)) Nothing stop single yield
657657
658658-- | Determine whether an element is present in the stream.
659659--
@@ -665,7 +665,7 @@ elem e m = go (toStream m)
665665 let stop = return False
666666 single a = return (a == e)
667667 yield a r = if a == e then return True else go r
668− in (S.runStream m1) Nothing stop single yield
668+ in (C.runStream m1) Nothing stop single yield
669669
670670-- | Determine whether an element is not present in the stream.
671671--
@@ -677,7 +677,7 @@ notElem e m = go (toStream m)
677677 let stop = return True
678678 single a = return (a /= e)
679679 yield a r = if a == e then return False else go r
680− in (S.runStream m1) Nothing stop single yield
680+ in (C.runStream m1) Nothing stop single yield
681681
682682-- | Determine the length of the stream.
683683--
@@ -690,14 +690,14 @@ length = foldl (\n _ -> n + 1) 0 id
690690--
691691-- @since 0.1.1
692692reverse :: (IsStream t) => t m a -> t m a
693−reverse m = fromStream $ go S.nil (toStream m)
693+reverse m = fromStream $ go C.nil (toStream m)
694694 where
695− go rev rest = Stream $ \_ stp sng yld ->
696− let run x = S.runStream x Nothing stp sng yld
695+ go rev rest = C.Stream $ \_ stp sng yld ->
696+ let run x = C.runStream x Nothing stp sng yld
697697 stop = run rev
698− single a = run $ a `S.cons` rev
699− yield a r = run $ go (a `S.cons` rev) r
700− in S.runStream rest Nothing stop single yield
698+ single a = run $ a `C.cons` rev
699+ yield a r = run $ go (a `C.cons` rev) r
700+ in C.runStream rest Nothing stop single yield
701701
702702-- XXX replace the recursive "go" with continuation
703703-- | Determine the minimum element in a stream.
@@ -710,7 +710,7 @@ minimum m = go Nothing (toStream m)
710710 let stop = return res
711711 single a = return $ min_ a res
712712 yield a r = go (min_ a res) r
713− in (S.runStream m1) Nothing stop single yield
713+ in (C.runStream m1) Nothing stop single yield
714714
715715 min_ a res = case res of
716716 Nothing -> Just a
@@ -727,7 +727,7 @@ maximum m = go Nothing (toStream m)
727727 let stop = return res
728728 single a = return $ max_ a res
729729 yield a r = go (max_ a res) r
730− in (S.runStream m1) Nothing stop single yield
730+ in (C.runStream m1) Nothing stop single yield
731731
732732 max_ a res = case res of
733733 Nothing -> Just a
@@ -755,10 +755,10 @@ maximum m = go Nothing (toStream m)
755755mapM :: (IsStream t, MonadAsync m) => (a -> m b) -> t m a -> t m b
756756mapM f m = go (toStream m)
757757 where
758− go m1 = fromStream $ Stream $ \svr stp sng yld ->
758+ go m1 = fromStream $ C.Stream $ \svr stp sng yld ->
759759 let single a = f a >>= sng
760− yield a r = S.runStream (toStream (f a |: (go r))) svr stp sng yld
761− in (S.runStream m1) Nothing stp single yield
760+ yield a r = C.runStream (toStream (f a |: (go r))) svr stp sng yld
761+ in (C.runStream m1) Nothing stp single yield
762762
763763-- | Map a 'Maybe' returning function to a stream, filter out the 'Nothing'
764764-- elements, and return a stream of values extracted from 'Just'.
@@ -768,14 +768,14 @@ mapM f m = go (toStream m)
768768mapMaybe :: (IsStream t) => (a -> Maybe b) -> t m a -> t m b
769769mapMaybe f m = go (toStream m)
770770 where
771− go m1 = fromStream $ Stream $ \_ stp sng yld ->
771+ go m1 = fromStream $ C.Stream $ \_ stp sng yld ->
772772 let single a = case f a of
773773 Just b -> sng b
774774 Nothing -> stp
775775 yield a r = case f a of
776776 Just b -> yld b (toStream $ go r)
777− Nothing -> (S.runStream r) Nothing stp single yield
778− in (S.runStream m1) Nothing stp single yield
777+ Nothing -> (C.runStream r) Nothing stp single yield
778+ in (C.runStream m1) Nothing stp single yield
779779
780780-- | Like 'mapMaybe' but maps a monadic function.
781781--
@@ -799,7 +799,7 @@ mapM_ f m = go (toStream m)
799799 let stop = return ()
800800 single a = void (f a)
801801 yield a r = f a >> go r
802− in (S.runStream m1) Nothing stop single yield
802+ in (C.runStream m1) Nothing stop single yield
803803
804804-- | Reduce a stream of monadic actions to a stream of the output of those
805805-- actions.
@@ -818,10 +818,10 @@ mapM_ f m = go (toStream m)
818818sequence :: (IsStream t, MonadAsync m) => t m (m a) -> t m a
819819sequence m = go (toStream m)
820820 where
821− go m1 = fromStream $ Stream $ \svr stp sng yld ->
821+ go m1 = fromStream $ C.Stream $ \svr stp sng yld ->
822822 let single ma = ma >>= sng
823− yield ma r = S.runStream (toStream $ ma |: go r) svr stp sng yld
824− in (S.runStream m1) Nothing stp single yield
823+ yield ma r = C.runStream (toStream $ ma |: go r) svr stp sng yld
824+ in (C.runStream m1) Nothing stp single yield
825825
826826------------------------------------------------------------------------------
827827-- Serially Zipping Streams
@@ -832,7 +832,7 @@ sequence m = go (toStream m)
832832-- @since 0.1.0
833833{-# INLINABLE zipWith #-}
834834zipWith :: IsStream t => (a -> b -> c) -> t m a -> t m b -> t m c
835−zipWith f m1 m2 = fromStream $ S.zipWith f (toStream m1) (toStream m2)
835+zipWith f m1 m2 = fromStream $ C.zipWith f (toStream m1) (toStream m2)
836836
837837-- | Zip two streams serially using a monadic zipping function.
838838--
@@ -840,15 +840,15 @@ zipWith f m1 m2 = fromStream $ S.zipWith f (toStream m1) (toStream m2)
840840zipWithM :: IsStream t => (a -> b -> t m c) -> t m a -> t m b -> t m c
841841zipWithM f m1 m2 = fromStream $ go (toStream m1) (toStream m2)
842842 where
843− go mx my = Stream $ \_ stp sng yld -> do
843+ go mx my = C.Stream $ \_ stp sng yld -> do
844844 let merge a ra =
845− let run x = S.runStream x Nothing stp sng yld
845+ let run x = C.runStream x Nothing stp sng yld
846846 single2 b = run $ toStream (f a b)
847847 yield2 b rb = run $ toStream (f a b) <> go ra rb
848− in (S.runStream my) Nothing stp single2 yield2
849− let single1 a = merge a S.nil
848+ in (C.runStream my) Nothing stp single2 yield2
849+ let single1 a = merge a C.nil
850850 yield1 a ra = merge a ra
851− (S.runStream mx) Nothing stp single1 yield1
851+ (C.runStream mx) Nothing stp single1 yield1
852852
853853------------------------------------------------------------------------------
854854-- Parallely Zipping Streams
@@ -861,7 +861,7 @@ zipWithM f m1 m2 = fromStream $ go (toStream m1) (toStream m2)
861861zipAsyncWith :: (IsStream t, MonadAsync m)
862862 => (a -> b -> c) -> t m a -> t m b -> t m c
863863zipAsyncWith f m1 m2 =
864− fromStream $ S.zipAsyncWith f (toStream m1) (toStream m2)
864+ fromStream $ CC.zipAsyncWith f (toStream m1) (toStream m2)
865865
866866-- | Zip two streams asyncly (i.e. both the elements being zipped are generated
867867-- concurrently) using a monadic zipping function.
@@ -869,7 +869,7 @@ zipAsyncWith f m1 m2 =
869869-- @since 0.1.0
870870zipAsyncWithM :: (IsStream t, MonadAsync m)
871871 => (a -> b -> t m c) -> t m a -> t m b -> t m c
872−zipAsyncWithM f m1 m2 = fromStream $ Stream $ \_ stp sng yld -> do
872+zipAsyncWithM f m1 m2 = fromStream $ C.Stream $ \_ stp sng yld -> do
873873 ma <- mkAsync m1
874874 mb <- mkAsync m2
875− (S.runStream (toStream (zipWithM f ma mb))) Nothing stp sng yld
875+ (C.runStream (toStream (zipWithM f ma mb))) Nothing stp sng yld
src/Streamly/Streams.hs+70−68
@@ -118,7 +118,9 @@ import Control.Monad.State.Class (MonadState(..))
118118import Control.Monad.Trans.Class (MonadTrans (lift))
119119import Data.Semigroup (Semigroup(..))
120120import Streamly.SVar (MonadAsync, SVar, toStreamVar)
121−import qualified Streamly.Core as S
121+
122+import qualified Streamly.CPS.Stream as C
123+import qualified Streamly.CPS.Concurrent as CC
122124
123125------------------------------------------------------------------------------
124126-- Types that can behave as a Stream
@@ -132,8 +134,8 @@ infixr 5 |:
132134--
133135-- @since 0.2.0
134136class IsStream t where
135− toStream :: t m a -> S.Stream m a
136− fromStream :: S.Stream m a -> t m a
137+ toStream :: t m a -> C.Stream m a
138+ fromStream :: C.Stream m a -> t m a
137139 -- | Constructs a stream by adding a monadic action at the head of an
138140 -- existing stream. For example:
139141 --
@@ -190,7 +192,7 @@ type Streaming = IsStream
190192--
191193-- @since 0.1.0
192194nil :: IsStream t => t m a
193−nil = fromStream S.nil
195+nil = fromStream C.nil
194196
195197-- | Constructs a stream by adding a monadic action at the head of an existing
196198-- stream. For example:
@@ -204,7 +206,7 @@ nil = fromStream S.nil
204206--
205207-- @since 0.2.0
206208consMSerial :: (IsStream t, Monad m) => m a -> t m a -> t m a
207−consMSerial m r = fromStream $ S.consM m (toStream r)
209+consMSerial m r = fromStream $ C.consM m (toStream r)
208210
209211infixr 5 `cons`
210212
@@ -218,7 +220,7 @@ infixr 5 `cons`
218220--
219221-- @since 0.1.0
220222cons :: IsStream t => a -> t m a -> t m a
221−cons a r = fromStream $ S.cons a (toStream r)
223+cons a r = fromStream $ C.cons a (toStream r)
222224
223225infixr 5 .:
224226
@@ -239,23 +241,23 @@ infixr 5 .:
239241-- remaining stream if any otherwise 'Nothing'. The third parameter is to
240242-- represent an "empty" stream.
241243streamBuild :: IsStream t
242− => (forall r. Maybe (SVar S.Stream m a)
244+ => (forall r. Maybe (SVar C.Stream m a)
243245 -> (a -> t m a -> m r)
244246 -> (a -> m r)
245247 -> m r
246248 -> m r)
247249 -> t m a
248−streamBuild k = fromStream $ S.Stream $ \svr stp sng yld ->
250+streamBuild k = fromStream $ C.Stream $ \svr stp sng yld ->
249251 let yield a r = yld a (toStream r)
250252 in k svr yield sng stp
251253
252254-- | Build a singleton stream from a callback function.
253255fromCallback :: IsStream t => (forall r. (a -> m r) -> m r) -> t m a
254−fromCallback k = fromStream $ S.Stream $ \_ _ sng _ -> k sng
256+fromCallback k = fromStream $ C.Stream $ \_ _ sng _ -> k sng
255257
256258-- | Read an SVar to get a stream.
257−fromSVar :: (MonadAsync m, IsStream t) => SVar S.Stream m a -> t m a
258−fromSVar sv = fromStream $ S.fromStreamVar sv
259+fromSVar :: (MonadAsync m, IsStream t) => SVar C.Stream m a -> t m a
260+fromSVar sv = fromStream $ CC.fromStreamVar sv
259261
260262------------------------------------------------------------------------------
261263-- Destroying a stream
@@ -266,7 +268,7 @@ fromSVar sv = fromStream $ S.fromStreamVar sv
266268-- argument is for consuming an "empty" stream that yields nothing.
267269streamFold
268270 :: IsStream t
269− => Maybe (SVar S.Stream m a)
271+ => Maybe (SVar C.Stream m a)
270272 -> (a -> t m a -> m r)
271273 -> (a -> m r)
272274 -> m r
@@ -274,7 +276,7 @@ streamFold
274276 -> m r
275277streamFold svr step single blank m =
276278 let yield a x = step a (fromStream x)
277− in (S.runStream (toStream m)) svr blank single yield
279+ in (C.runStream (toStream m)) svr blank single yield
278280
279281-- | Run a streaming composition, discard the results. By default it interprets
280282-- the stream as 'SerialT', to run other types of streams use the type adapting
@@ -288,7 +290,7 @@ runStream m = go (toStream m)
288290 let stop = return ()
289291 single _ = return ()
290292 yield _ r = go r
291− in (S.runStream m1) Nothing stop single yield
293+ in (C.runStream m1) Nothing stop single yield
292294
293295-- | Same as 'runStream'
294296--
@@ -299,7 +301,7 @@ runStreaming = runStream . adapt
299301
300302-- | Write a stream to an 'SVar' in a non-blocking manner. The stream can then
301303-- be read back from the SVar using 'fromSVar'.
302−toSVar :: (IsStream t, MonadAsync m) => SVar S.Stream m a -> t m a -> m ()
304+toSVar :: (IsStream t, MonadAsync m) => SVar C.Stream m a -> t m a -> m ()
303305toSVar sv m = toStreamVar sv (toStream m)
304306
305307------------------------------------------------------------------------------
@@ -316,17 +318,17 @@ toSVar sv m = toStreamVar sv (toStream m)
316318-- @since 0.2.0
317319mkAsync :: (IsStream t, MonadAsync m) => t m a -> m (t m a)
318320mkAsync m = do
319− s <- S.mkParallel (toStream m)
321+ s <- CC.mkParallel (toStream m)
320322 return $ fromStream s
321323
322324{-# INLINE applyWith #-}
323325applyWith :: (IsStream t, MonadAsync m) => (t m a -> t m b) -> t m a -> t m b
324326applyWith f x = fromStream $
325− S.applyWith (toStream . f . fromStream) (toStream x)
327+ CC.applyWith (toStream . f . fromStream) (toStream x)
326328
327329{-# INLINE runWith #-}
328330runWith :: (IsStream t, MonadAsync m) => (t m a -> m b) -> t m a -> m b
329−runWith f x = S.runWith (f . fromStream) (toStream x)
331+runWith f x = CC.runWith (f . fromStream) (toStream x)
330332
331333infixr 0 |$
332334infixr 0 |$.
@@ -410,21 +412,21 @@ x |&. f = f |$. x
410412-- XXX use template haskell instead and include Monoid and IsStream instances
411413-- as well.
412414
413−withLocal :: MonadReader r m => (r -> r) -> S.Stream m a -> S.Stream m a
415+withLocal :: MonadReader r m => (r -> r) -> C.Stream m a -> C.Stream m a
414416withLocal f m =
415− S.Stream $ \_ stp sng yld ->
417+ C.Stream $ \_ stp sng yld ->
416418 let single = local f . sng
417419 yield a r = local f $ yld a (withLocal f r)
418− in (S.runStream m) Nothing (local f stp) single yield
420+ in (C.runStream m) Nothing (local f stp) single yield
419421
420422{-
421423-- XXX handle and test cross thread state transfer
422424withCatchError
423425 :: MonadError e m
424− => S.Stream m a -> (e -> S.Stream m a) -> S.Stream m a
426+ => C.Stream m a -> (e -> C.Stream m a) -> C.Stream m a
425427withCatchError m h =
426− S.Stream $ \_ stp sng yld ->
427− let run x = S.runStream x Nothing stp sng yield
428+ C.Stream $ \_ stp sng yld ->
429+ let run x = C.runStream x Nothing stp sng yield
428430 handle r = r `catchError` \e -> run $ h e
429431 yield a r = yld a (withCatchError r h)
430432 in handle $ run m
@@ -434,7 +436,7 @@ withCatchError m h =
434436
435437#define MONAD_APPLICATIVE_INSTANCE(STREAM,CONSTRAINT) \
436438instance (Monad m CONSTRAINT) => Applicative (STREAM m) where { \
437− pure = STREAM . S.singleton; \
439+ pure = STREAM . C.singleton; \
438440 (<*>) = ap }
439441
440442#define MONAD_COMMON_INSTANCES(STREAM,CONSTRAINT) \
@@ -523,7 +525,7 @@ instance (MonadState s m CONSTRAINT) => MonadState s (STREAM m) where { \
523525-- time.
524526--
525527-- @since 0.2.0
526−newtype SerialT m a = SerialT {getSerialT :: S.Stream m a}
528+newtype SerialT m a = SerialT {getSerialT :: C.Stream m a}
527529 deriving (Semigroup, Monoid, Functor, MonadTrans)
528530
529531-- |
@@ -556,7 +558,7 @@ instance IsStream SerialT where
556558-- @since 0.2.0
557559{-# INLINE serial #-}
558560serial :: IsStream t => t m a -> t m a -> t m a
559−serial m1 m2 = fromStream $ S.serial (toStream m1) (toStream m2)
561+serial m1 m2 = fromStream $ C.serial (toStream m1) (toStream m2)
560562
561563------------------------------------------------------------------------------
562564-- Monad
@@ -564,8 +566,8 @@ serial m1 m2 = fromStream $ S.serial (toStream m1) (toStream m2)
564566
565567instance Monad m => Monad (SerialT m) where
566568 return = pure
567− (SerialT (S.Stream m)) >>= f = SerialT $ S.Stream $ \_ stp sng yld ->
568− let run x = (S.runStream x) Nothing stp sng yld
569+ (SerialT (C.Stream m)) >>= f = SerialT $ C.Stream $ \_ stp sng yld ->
570+ let run x = (C.runStream x) Nothing stp sng yld
569571 single a = run $ toStream (f a)
570572 yield a r = run $ toStream $ f a <> (fromStream r >>= f)
571573 in m Nothing stp single yield
@@ -618,7 +620,7 @@ MONAD_COMMON_INSTANCES(SerialT,)
618620-- stream.
619621--
620622-- @since 0.2.0
621−newtype WSerialT m a = WSerialT {getWSerialT :: S.Stream m a}
623+newtype WSerialT m a = WSerialT {getWSerialT :: C.Stream m a}
622624 deriving (Functor, MonadTrans)
623625
624626-- |
@@ -650,7 +652,7 @@ instance IsStream WSerialT where
650652-- @since 0.2.0
651653{-# INLINE wSerial #-}
652654wSerial :: IsStream t => t m a -> t m a -> t m a
653−wSerial m1 m2 = fromStream $ S.wSerial (toStream m1) (toStream m2)
655+wSerial m1 m2 = fromStream $ C.wSerial (toStream m1) (toStream m2)
654656
655657instance Semigroup (WSerialT m a) where
656658 (<>) = wSerial
@@ -679,8 +681,8 @@ instance Monoid (WSerialT m a) where
679681
680682instance Monad m => Monad (WSerialT m) where
681683 return = pure
682− (WSerialT (S.Stream m)) >>= f = WSerialT $ S.Stream $ \_ stp sng yld ->
683− let run x = (S.runStream x) Nothing stp sng yld
684+ (WSerialT (C.Stream m)) >>= f = WSerialT $ C.Stream $ \_ stp sng yld ->
685+ let run x = (C.runStream x) Nothing stp sng yld
684686 single a = run $ toStream (f a)
685687 yield a r = run $ toStream $ f a <> (fromStream r >>= f)
686688 in m Nothing stp single yield
@@ -740,7 +742,7 @@ MONAD_COMMON_INSTANCES(WSerialT,)
740742-- streams at a time.
741743--
742744-- @since 0.3.0
743−newtype AheadT m a = AheadT {getAheadT :: S.Stream m a}
745+newtype AheadT m a = AheadT {getAheadT :: C.Stream m a}
744746 deriving (Functor, MonadTrans)
745747
746748instance IsStream AheadT where
@@ -749,7 +751,7 @@ instance IsStream AheadT where
749751
750752 {-# INLINE consM #-}
751753 {-# SPECIALIZE consM :: IO a -> AheadT IO a -> AheadT IO a #-}
752− consM m r = fromStream $ S.consMAhead m (toStream r)
754+ consM m r = fromStream $ CC.consMAhead m (toStream r)
753755
754756 {-# INLINE (|:) #-}
755757 {-# SPECIALIZE (|:) :: IO a -> AheadT IO a -> AheadT IO a #-}
@@ -765,7 +767,7 @@ instance IsStream AheadT where
765767-- @since 0.3.0
766768{-# INLINE ahead #-}
767769ahead :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
768−ahead m1 m2 = fromStream $ S.ahead (toStream m1) (toStream m2)
770+ahead m1 m2 = fromStream $ CC.ahead (toStream m1) (toStream m2)
769771
770772instance MonadAsync m => Semigroup (AheadT m a) where
771773 (<>) = ahead
@@ -785,16 +787,16 @@ instance MonadAsync m => Monoid (AheadT m a) where
785787{-# INLINE aheadbind #-}
786788aheadbind
787789 :: MonadAsync m
788− => S.Stream m a
789− -> (a -> S.Stream m b)
790− -> S.Stream m b
790+ => C.Stream m a
791+ -> (a -> C.Stream m b)
792+ -> C.Stream m b
791793aheadbind m f = go m
792794 where
793− go (S.Stream g) =
794− S.Stream $ \ctx stp sng yld ->
795− let run x = (S.runStream x) ctx stp sng yld
795+ go (C.Stream g) =
796+ C.Stream $ \ctx stp sng yld ->
797+ let run x = (C.runStream x) ctx stp sng yld
796798 single a = run $ f a
797− yield a r = run $ f a `S.ahead` go r
799+ yield a r = run $ f a `CC.ahead` go r
798800 in g Nothing stp single yield
799801
800802instance MonadAsync m => Monad (AheadT m) where
@@ -861,7 +863,7 @@ MONAD_COMMON_INSTANCES(AheadT, MONADPARALLEL)
861863-- streams at a time.
862864--
863865-- @since 0.1.0
864−newtype AsyncT m a = AsyncT {getAsyncT :: S.Stream m a}
866+newtype AsyncT m a = AsyncT {getAsyncT :: C.Stream m a}
865867 deriving (Functor, MonadTrans)
866868
867869instance IsStream AsyncT where
@@ -870,7 +872,7 @@ instance IsStream AsyncT where
870872
871873 {-# INLINE consM #-}
872874 {-# SPECIALIZE consM :: IO a -> AsyncT IO a -> AsyncT IO a #-}
873− consM m r = fromStream $ S.consMAsync m (toStream r)
875+ consM m r = fromStream $ CC.consMAsync m (toStream r)
874876
875877 {-# INLINE (|:) #-}
876878 {-# SPECIALIZE (|:) :: IO a -> AsyncT IO a -> AsyncT IO a #-}
@@ -887,7 +889,7 @@ instance IsStream AsyncT where
887889-- @since 0.2.0
888890{-# INLINE async #-}
889891async :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
890−async m1 m2 = fromStream $ S.async (toStream m1) (toStream m2)
892+async m1 m2 = fromStream $ CC.async (toStream m1) (toStream m2)
891893
892894-- | Same as 'async'.
893895--
@@ -914,22 +916,22 @@ instance MonadAsync m => Monoid (AsyncT m a) where
914916
915917{-# INLINE parbind #-}
916918parbind
917− :: (forall c. S.Stream m c -> S.Stream m c -> S.Stream m c)
918− -> S.Stream m a
919− -> (a -> S.Stream m b)
920− -> S.Stream m b
919+ :: (forall c. C.Stream m c -> C.Stream m c -> C.Stream m c)
920+ -> C.Stream m a
921+ -> (a -> C.Stream m b)
922+ -> C.Stream m b
921923parbind par m f = go m
922924 where
923− go (S.Stream g) =
924− S.Stream $ \ctx stp sng yld ->
925− let run x = (S.runStream x) ctx stp sng yld
925+ go (C.Stream g) =
926+ C.Stream $ \ctx stp sng yld ->
927+ let run x = (C.runStream x) ctx stp sng yld
926928 single a = run $ f a
927929 yield a r = run $ f a `par` go r
928930 in g Nothing stp single yield
929931
930932instance MonadAsync m => Monad (AsyncT m) where
931933 return = pure
932− (AsyncT m) >>= f = AsyncT $ parbind S.async m (getAsyncT . f)
934+ (AsyncT m) >>= f = AsyncT $ parbind CC.async m (getAsyncT . f)
933935
934936------------------------------------------------------------------------------
935937-- Other instances
@@ -989,7 +991,7 @@ MONAD_COMMON_INSTANCES(AsyncT, MONADPARALLEL)
989991-- stream.
990992--
991993-- @since 0.2.0
992−newtype WAsyncT m a = WAsyncT {getWAsyncT :: S.Stream m a}
994+newtype WAsyncT m a = WAsyncT {getWAsyncT :: C.Stream m a}
993995 deriving (Functor, MonadTrans)
994996
995997instance IsStream WAsyncT where
@@ -998,7 +1000,7 @@ instance IsStream WAsyncT where
9981000
9991001 {-# INLINE consM #-}
10001002 {-# SPECIALIZE consM :: IO a -> WAsyncT IO a -> WAsyncT IO a #-}
1001− consM m r = fromStream $ S.consMWAsync m (toStream r)
1003+ consM m r = fromStream $ CC.consMWAsync m (toStream r)
10021004
10031005 {-# INLINE (|:) #-}
10041006 {-# SPECIALIZE (|:) :: IO a -> WAsyncT IO a -> WAsyncT IO a #-}
@@ -1014,7 +1016,7 @@ instance IsStream WAsyncT where
10141016-- @since 0.2.0
10151017{-# INLINE wAsync #-}
10161018wAsync :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
1017−wAsync m1 m2 = fromStream $ S.wAsync (toStream m1) (toStream m2)
1019+wAsync m1 m2 = fromStream $ CC.wAsync (toStream m1) (toStream m2)
10181020
10191021instance MonadAsync m => Semigroup (WAsyncT m a) where
10201022 (<>) = wAsync
@@ -1034,7 +1036,7 @@ instance MonadAsync m => Monoid (WAsyncT m a) where
10341036instance MonadAsync m => Monad (WAsyncT m) where
10351037 return = pure
10361038 (WAsyncT m) >>= f =
1037− WAsyncT $ parbind S.wAsync m (getWAsyncT . f)
1039+ WAsyncT $ parbind CC.wAsync m (getWAsyncT . f)
10381040
10391041------------------------------------------------------------------------------
10401042-- Other instances
@@ -1132,7 +1134,7 @@ MONAD_COMMON_INSTANCES(WAsyncT, MONADPARALLEL)
11321134-- streams as it needs to retain state for each unfinished stream.
11331135--
11341136-- @since 0.1.0
1135−newtype ParallelT m a = ParallelT {getParallelT :: S.Stream m a}
1137+newtype ParallelT m a = ParallelT {getParallelT :: C.Stream m a}
11361138 deriving (Functor, MonadTrans)
11371139
11381140instance IsStream ParallelT where
@@ -1141,7 +1143,7 @@ instance IsStream ParallelT where
11411143
11421144 {-# INLINE consM #-}
11431145 {-# SPECIALIZE consM :: IO a -> ParallelT IO a -> ParallelT IO a #-}
1144− consM m r = fromStream $ S.consMParallel m (toStream r)
1146+ consM m r = fromStream $ CC.consMParallel m (toStream r)
11451147
11461148 {-# INLINE (|:) #-}
11471149 {-# SPECIALIZE (|:) :: IO a -> ParallelT IO a -> ParallelT IO a #-}
@@ -1157,7 +1159,7 @@ instance IsStream ParallelT where
11571159-- @since 0.2.0
11581160{-# INLINE parallel #-}
11591161parallel :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
1160−parallel m1 m2 = fromStream $ S.parallel (toStream m1) (toStream m2)
1162+parallel m1 m2 = fromStream $ CC.parallel (toStream m1) (toStream m2)
11611163
11621164instance MonadAsync m => Semigroup (ParallelT m a) where
11631165 (<>) = parallel
@@ -1176,7 +1178,7 @@ instance MonadAsync m => Monoid (ParallelT m a) where
11761178
11771179instance MonadAsync m => Monad (ParallelT m) where
11781180 return = pure
1179− (ParallelT m) >>= f = ParallelT $ parbind S.parallel m (getParallelT . f)
1181+ (ParallelT m) >>= f = ParallelT $ parbind CC.parallel m (getParallelT . f)
11801182
11811183------------------------------------------------------------------------------
11821184-- Other instances
@@ -1207,7 +1209,7 @@ MONAD_COMMON_INSTANCES(ParallelT, MONADPARALLEL)
12071209-- 'SerialT'.
12081210--
12091211-- @since 0.2.0
1210−newtype ZipSerialM m a = ZipSerialM {getZipSerialM :: S.Stream m a}
1212+newtype ZipSerialM m a = ZipSerialM {getZipSerialM :: C.Stream m a}
12111213 deriving (Functor, Semigroup, Monoid)
12121214
12131215-- |
@@ -1230,8 +1232,8 @@ instance IsStream ZipSerialM where
12301232 (|:) = consMSerial
12311233
12321234instance Monad m => Applicative (ZipSerialM m) where
1233− pure = ZipSerialM . S.repeat
1234− m1 <*> m2 = fromStream $ S.zipWith id (toStream m1) (toStream m2)
1235+ pure = ZipSerialM . C.repeat
1236+ m1 <*> m2 = fromStream $ C.zipWith id (toStream m1) (toStream m2)
12351237
12361238------------------------------------------------------------------------------
12371239-- Parallely Zipping Streams
@@ -1254,7 +1256,7 @@ instance Monad m => Applicative (ZipSerialM m) where
12541256-- 'SerialT'.
12551257--
12561258-- @since 0.2.0
1257−newtype ZipAsyncM m a = ZipAsyncM {getZipAsyncM :: S.Stream m a}
1259+newtype ZipAsyncM m a = ZipAsyncM {getZipAsyncM :: C.Stream m a}
12581260 deriving (Functor, Semigroup, Monoid)
12591261
12601262instance IsStream ZipAsyncM where
@@ -1272,8 +1274,8 @@ instance IsStream ZipAsyncM where
12721274 (|:) = consMSerial
12731275
12741276instance MonadAsync m => Applicative (ZipAsyncM m) where
1275− pure = ZipAsyncM . S.repeat
1276− m1 <*> m2 = fromStream $ S.zipAsyncWith id (toStream m1) (toStream m2)
1277+ pure = ZipAsyncM . C.repeat
1278+ m1 <*> m2 = fromStream $ CC.zipAsyncWith id (toStream m1) (toStream m2)
12771279
12781280-------------------------------------------------------------------------------
12791281-- Type adapting combinators
streamly.cabal+2−1
@@ -107,7 +107,8 @@ flag examples-sdl
107107library
108108 hs-source-dirs: src
109109 other-modules: Streamly.SVar
110− , Streamly.Core
110+ , Streamly.CPS.Stream
111+ , Streamly.CPS.Concurrent
111112 , Streamly.Streams
112113
113114 exposed-modules: Streamly.Prelude