@@ -18,28 +18,13 @@
19module Asyncly.Prelude
19module Asyncly.Prelude
21− -- * Product Style Composition
30− -- * Stream Type Adapters
39 -- * Construction
21 -- * Construction
42 -- * Elimination
25 -- * Elimination
45 -- * Special folds
30 -- * Special folds
@@ -54,462 +39,62 @@ module Asyncly.Prelude
57−import Control.Applicative (Alternative (..), liftA2)
58−import Control.Monad (MonadPlus(..), ap, liftM)
59−import Control.Monad.Base (MonadBase (..))
60−import Control.Monad.Catch (MonadThrow)
61−import Control.Monad.Error.Class (MonadError(..))
62−import Control.Monad.IO.Class (MonadIO(..))
63−import Control.Monad.Reader.Class (MonadReader(..))
64−import Control.Monad.State.Class (MonadState(..))
65−import Control.Monad.Trans.Class (MonadTrans)
66−import Data.Semigroup (Semigroup(..), cycle1)
42+import Data.Semigroup (Semigroup(..))
67import Prelude hiding (drop, take, zipWith)
43import Prelude hiding (drop, take, zipWith)
68import Asyncly.Core
44import Asyncly.Core
72−-- Different styles of streams represented by different types dictate the
73−-- manner in which a stream is processed in monadic or applicative composition.
74−-- Each type differs in the processing order of the elements or processing
75−-- parallelism. The types have the same underlying representation and
76−-- therefore can be converted back and forth easily to switch the composition
77−-- style at any point.
79−-- 'StreamT' and 'InterleavedT' process the streams serially but differ in the
80−-- order in which the elements of the streams are processed. 'StreamT' goes
81−-- depth first that is it combine one element of the first stream to every
82−-- element of the second stream before it comes back to process the second
83−-- element of the first stream. 'InterleavedT' takes turns when processing
84−-- elements from each stream and therefore all iterations finish at the same
85−-- time rather than one iteration at a time as in the case of 'StreamT'.
86−-- Depending on the cache behavior or IO behavior 'InterleavedT' might be
87−-- slightly more efficient than 'StreamT' or vice-versa.
89−-- 'AsyncT' and 'ParallelT' can process the individual elements of the streams
90−-- in parallel. 'AsyncT' is the parallel version of 'StreamT', it follows the
91−-- order of 'StreamT', processing multiple elements of a stream concurrently.
92−-- 'ParallelT' follows the order of 'InterleavedT', processing and interleaving
93−-- both streams concurrently. Also, note that 'fmap' on these streams would
94−-- also work in a concurrent fashion i.e. multiple elements in a stream or
95−-- multiple streams can be mapped concurrently.
97−-- Zipping types have a different style of composition altogether, they zip the
98−-- streams rather than multiplying them. These types provide only Applicative
99−-- composition and no Monad instances. 'ZipStream' zips serially i.e. it
100−-- produces one element from each stream in a serial manner and then zips the
101−-- two items. 'ZipAsync' on the other hand zips in parallel, it produces one
102−-- element from each stream concurrently and then zips the two.
104−-- It may help develop a better understanding if we think in terms of stream
105−-- generation (production) side parallelism versus stream processing
106−-- (consumption) side parallelism. The sum style operators (e.g. '<|' or '<|>')
107−-- generate streams in parallel and then combine the results into a unified
108−-- stream. On the other hand the product style composition (e.g. 'AsyncT' or
109−-- 'ParallelT' or 'ZipAsync') dictate how the elements of a stream are
110−-- processed when it is consumed.
112−------------------------------------------------------------------------------
113−-- StreamT iterates serially in the Monad and Applicative compositions
114−------------------------------------------------------------------------------
116−-- | Note that 'StreamT' style composition works well operationally for
117−-- infinite streams. However if the second stream is infinite we will never
118−-- come back to the second iteration of the first stream.
120−newtype StreamT m a = StreamT {getStreamT :: Stream m a}
121− deriving (Semigroup, Monoid, MonadTrans, MonadIO, MonadThrow)
123−deriving instance MonadAsync m => Alternative (StreamT m)
124−deriving instance MonadAsync m => MonadPlus (StreamT m)
125−deriving instance (MonadBase b m, Monad m) => MonadBase b (StreamT m)
126−deriving instance MonadError e m => MonadError e (StreamT m)
127−deriving instance MonadReader r m => MonadReader r (StreamT m)
128−deriving instance MonadState s m => MonadState s (StreamT m)
130−instance Streaming StreamT where
131− toStream = getStreamT
132− fromStream = StreamT
134−-- XXX The Functor/Applicative/Num instances for all the types are exactly the
135−-- same, how can we reduce this boilerplate (use TH)? We cannot derive them
136−-- from a single base type because they depend on the Monad instance which is
137−-- different for each type.
139−------------------------------------------------------------------------------
141−------------------------------------------------------------------------------
143−-- | Execute a monadic action sequentially for each element in the 'AsyncT'
144−-- stream, i.e. an iteration finishes completely before the next one starts.
145−instance Monad m => Monad (StreamT m) where
147− (StreamT (Stream m)) >>= f = StreamT $ Stream $ \_ stp yld ->
148− let run x = (runStream x) Nothing stp yld
149− yield a Nothing = run $ getStreamT (f a)
150− yield a (Just r) = run $ getStreamT (f a)
151− <> getStreamT (StreamT r >>= f)
152− in m Nothing stp yield
154−------------------------------------------------------------------------------
156−------------------------------------------------------------------------------
158−instance Monad m => Applicative (StreamT m) where
159− pure = StreamT . yielding
162−------------------------------------------------------------------------------
164−------------------------------------------------------------------------------
166−instance Monad m => Functor (StreamT m) where
169−------------------------------------------------------------------------------
171−------------------------------------------------------------------------------
173−instance (Monad m, Num a) => Num (StreamT m a) where
174− fromInteger n = pure (fromInteger n)
176− negate = fmap negate
178− signum = fmap signum
184−instance (Monad m, Fractional a) => Fractional (StreamT m a) where
185− fromRational n = pure (fromRational n)
191−instance (Monad m, Floating a) => Floating (StreamT m a) where
211− logBase = liftA2 logBase
213−------------------------------------------------------------------------------
214−-- InterleavedT interleaves the iterations in the Monad and Applicative
216------------------------------------------------------------------------------
46------------------------------------------------------------------------------
217−
47+-- Construct and deconstruct
218−-- | Note that 'InterleavedT' style composition cannot work well operationally
219−-- on infinite streams. It will keep accumulating the elements of the stream
220−-- resulting in continuous increase in space consumption.
222−newtype InterleavedT m a = InterleavedT {getInterleavedT :: Stream m a}
223− deriving (Semigroup, Monoid, MonadTrans, MonadIO, MonadThrow)
225−deriving instance MonadAsync m => Alternative (InterleavedT m)
226−deriving instance MonadAsync m => MonadPlus (InterleavedT m)
227−deriving instance (MonadBase b m, Monad m) => MonadBase b (InterleavedT m)
228−deriving instance MonadError e m => MonadError e (InterleavedT m)
229−deriving instance MonadReader r m => MonadReader r (InterleavedT m)
230−deriving instance MonadState s m => MonadState s (InterleavedT m)
232−instance Streaming InterleavedT where
233− toStream = getInterleavedT
234− fromStream = InterleavedT
236−-- | Execute a monadic action for each element in the stream, in a fairly
237−-- interleaved manner i.e. iterations yield alternately.
238−instance Monad m => Monad (InterleavedT m) where
240− (InterleavedT (Stream m)) >>= f = InterleavedT $ Stream $ \_ stp yld ->
241− let run x = (runStream x) Nothing stp yld
242− yield a Nothing = run $ getInterleavedT (f a)
243− yield a (Just r) = run $ getInterleavedT (f a)
245− getInterleavedT (InterleavedT r >>= f)
246− in m Nothing stp yield
248−------------------------------------------------------------------------------
250−------------------------------------------------------------------------------
252−instance Monad m => Applicative (InterleavedT m) where
253− pure = InterleavedT . yielding
256−------------------------------------------------------------------------------
258−------------------------------------------------------------------------------
260−instance Monad m => Functor (InterleavedT m) where
263−------------------------------------------------------------------------------
265−------------------------------------------------------------------------------
267−instance (Monad m, Num a) => Num (InterleavedT m a) where
268− fromInteger n = pure (fromInteger n)
270− negate = fmap negate
272− signum = fmap signum
278−instance (Monad m, Fractional a) => Fractional (InterleavedT m a) where
279− fromRational n = pure (fromRational n)
285−instance (Monad m, Floating a) => Floating (InterleavedT m a) where
305− logBase = liftA2 logBase
307−------------------------------------------------------------------------------
308−-- AsyncT runs the iterations asynchronously in the Monad and Applicative
309−-- compositions. More iterations are started in parallel on demand. It can work
310−-- with infinite streams.
311------------------------------------------------------------------------------
48------------------------------------------------------------------------------
313−-- | Just like 'StreamT', 'AsyncT' style composition works well operationally
50+-- | Decompose a stream into its head and tail. If the stream is empty, returns
314−-- for infinite streams. However, if the second stream is infinite we will
51+-- 'Nothing'. If the stream is non-empty, returns 'Just (a, ma)', where 'a' is
315−-- never be able to finish the first stream.
52+-- the head of the stream and 'ma' its tail.
316−--
53+uncons :: (Streaming t, Monad m, Monoid (t m a))
317−-- Note that 'fmap' on 'AsyncT' would map on multiple elements concurrently.
54+ => t m a -> m (Maybe (a, t m a))
318−
55+uncons m = (runStream (toStream m)) Nothing stop yield
319−newtype AsyncT m a = AsyncT {getAsyncT :: Stream m a}
320− deriving (Semigroup, Monoid, MonadTrans)
322−deriving instance MonadAsync m => Alternative (AsyncT m)
323−deriving instance MonadAsync m => MonadPlus (AsyncT m)
324−deriving instance MonadAsync m => MonadIO (AsyncT m)
325−deriving instance MonadAsync m => MonadThrow (AsyncT m)
326−deriving instance (MonadBase b m, MonadAsync m) => MonadBase b (AsyncT m)
327−deriving instance (MonadError e m, MonadAsync m) => MonadError e (AsyncT m)
328−deriving instance (MonadReader r m, MonadAsync m) => MonadReader r (AsyncT m)
329−deriving instance (MonadState s m, MonadAsync m) => MonadState s (AsyncT m)
331−instance Streaming AsyncT where
332− toStream = getAsyncT
335−{-# INLINE parbind #-}
337− :: (forall c. Stream m c -> Stream m c -> Stream m c)
339− -> (a -> Stream m b)
344− Stream $ \ctx stp yld ->
345− let run x = (runStream x) ctx stp yld
346− yield a Nothing = run $ f a
347− yield a (Just r) = run $ f a `k` (go r)
348− in g Nothing stp yield
350−-- | Execute a monadic action for each element in the stream, running
351−-- iterations in parallel, but giving preference to iterations started
353−instance MonadAsync m => Monad (AsyncT m) where
355− (AsyncT m) >>= f = AsyncT $ parbind par m g
356− where g x = getAsyncT (f x)
357− par = parallel (CtxType Conjunction LIFO)
359−------------------------------------------------------------------------------
361−------------------------------------------------------------------------------
363−instance MonadAsync m => Applicative (AsyncT m) where
364− pure = AsyncT . yielding
367−------------------------------------------------------------------------------
369−------------------------------------------------------------------------------
371−instance MonadAsync m => Functor (AsyncT m) where
374−------------------------------------------------------------------------------
376−------------------------------------------------------------------------------
378−instance (MonadAsync m, Num a) => Num (AsyncT m a) where
379− fromInteger n = pure (fromInteger n)
381− negate = fmap negate
383− signum = fmap signum
389−instance (MonadAsync m, Fractional a) => Fractional (AsyncT m a) where
390− fromRational n = pure (fromRational n)
396−instance (MonadAsync m, Floating a) => Floating (AsyncT m a) where
416− logBase = liftA2 logBase
418−------------------------------------------------------------------------------
419−-- ParallelT runs all iterations in parallel in the Monad and Applicative
420−-- compositions. Iterations are scheduled in a round-robin fashion.
421−------------------------------------------------------------------------------
423−-- | Note that, just like 'InterleavedT', 'ParallelT' style composition cannot
424−-- work well operationally on infinite streams. It will keep accumulating the
425−-- elements of the stream resulting in continuous increase in space
428−-- Note that 'fmap' on 'ParallelT' would map on both streams concurrently.
430−newtype ParallelT m a = ParallelT {getParallelT :: Stream m a}
431− deriving (Semigroup, Monoid, MonadTrans)
433−deriving instance MonadAsync m => Alternative (ParallelT m)
434−deriving instance MonadAsync m => MonadPlus (ParallelT m)
435−deriving instance MonadAsync m => MonadIO (ParallelT m)
436−deriving instance MonadAsync m => MonadThrow (ParallelT m)
437−deriving instance (MonadBase b m, MonadAsync m) => MonadBase b (ParallelT m)
438−deriving instance (MonadError e m, MonadAsync m) => MonadError e (ParallelT m)
439−deriving instance (MonadReader r m, MonadAsync m) => MonadReader r (ParallelT m)
440−deriving instance (MonadState s m, MonadAsync m) => MonadState s (ParallelT m)
442−instance Streaming ParallelT where
443− toStream = getParallelT
444− fromStream = ParallelT
446−-- | Execute a monadic action for each element in the stream, running
447−-- iterations in a fairly parallel manner, i.e. all iterations are equally
449−instance MonadAsync m => Monad (ParallelT m) where
450− return = pure
57+ stop = return Nothing
451− (ParallelT m) >>= f = ParallelT $ parbind par m g
452− where g x = getParallelT (f x)
453− par = parallel (CtxType Conjunction FIFO)
455−------------------------------------------------------------------------------
457−------------------------------------------------------------------------------
459−instance MonadAsync m => Applicative (ParallelT m) where
460− pure = ParallelT . yielding
463−------------------------------------------------------------------------------
59+ {-# INLINE yield #-}
464−-- Functor
60+ yield a Nothing = return (Just (a, mempty))
465−------------------------------------------------------------------------------
61+ yield a (Just x) = return (Just (a, (fromStream x)))
467−instance MonadAsync m => Functor (ParallelT m) where
63+-- | Build a Stream by unfolding steps starting from a seed.
468− fmap = liftM
64+unfoldr :: (Streaming t, Monad m) => (b -> m (Maybe (a, b))) -> b -> t m a
65+unfoldr step = fromStream . go
67+ go s = Stream $ \_ stp yld -> do
71+ Just (a, b) -> yld a (Just (go b))
470------------------------------------------------------------------------------
73------------------------------------------------------------------------------
471−-- Num
74+-- Special folds
472------------------------------------------------------------------------------
75------------------------------------------------------------------------------
474−instance (MonadAsync m, Num a) => Num (ParallelT m a) where
77+-- | Take first 'n' elements from the stream and discard the rest.
475− fromInteger n = pure (fromInteger n)
78+take :: Streaming t => Int -> t m a -> t m a
476−
79+take n m = fromStream $ go n (toStream m)
477− negate = fmap negate
80+ where
478− abs = fmap abs
81+ go n1 m1 = Stream $ \ctx stp yld -> do
479− signum = fmap signum
82+ let yield a Nothing = yld a Nothing
480−
83+ yield a (Just x) = yld a (Just (go (n1 - 1) x))
481− (+) = liftA2 (+)
84+ if (n1 <= 0)
482− (*) = liftA2 (*)
85+ then stp
483− (-) = liftA2 (-)
86+ else (runStream m1) ctx stp yield
485−instance (MonadAsync m, Fractional a) => Fractional (ParallelT m a) where
486− fromRational n = pure (fromRational n)
492−instance (MonadAsync m, Floating a) => Floating (ParallelT m a) where
511− (**) = liftA2 (**)
88+-- | Discard first 'n' elements from the stream and take the rest.
512− logBase = liftA2 logBase
89+drop :: Streaming t => Int -> t m a -> t m a
90+drop n m = fromStream $ go n (toStream m)
92+ go n1 m1 = Stream $ \ctx stp yld -> do
93+ let yield _ Nothing = stp
94+ yield _ (Just x) = (runStream $ go (n1 - 1) x) ctx stp yld
96+ then (runStream m1) ctx stp yld
97+ else (runStream m1) ctx stp yield
514------------------------------------------------------------------------------
99------------------------------------------------------------------------------
515-- Serially Zipping Streams
100-- Serially Zipping Streams
@@ -543,68 +128,6 @@ zipWith f m1 m2 = fromStream $ go (toStream m1) (toStream m2)
543 yield1 a (Just ra) = merge a ra
128 yield1 a (Just ra) = merge a ra
544 (runStream mx) Nothing stp yield1
129 (runStream mx) Nothing stp yield1
546−-- | 'ZipStream' zips serially i.e. it produces one element from each stream in
547−-- a serial manner and then zips the two elements.
548−newtype ZipStream m a = ZipStream {getZipStream :: Stream m a}
549− deriving (Semigroup, Monoid)
551−deriving instance MonadAsync m => Alternative (ZipStream m)
553−instance Monad m => Functor (ZipStream m) where
554− fmap f (ZipStream (Stream m)) = ZipStream $ Stream $ \_ stp yld ->
555− let yield a Nothing = yld (f a) Nothing
556− yield a (Just r) = yld (f a)
557− (Just (getZipStream (fmap f (ZipStream r))))
558− in m Nothing stp yield
560−instance Monad m => Applicative (ZipStream m) where
561− pure a = ZipStream $ cycle1 (pure a)
564−instance Streaming ZipStream where
565− toStream = getZipStream
566− fromStream = ZipStream
568−instance (Monad m, Num a) => Num (ZipStream m a) where
569− fromInteger n = pure (fromInteger n)
571− negate = fmap negate
573− signum = fmap signum
579−instance (Monad m, Fractional a) => Fractional (ZipStream m a) where
580− fromRational n = pure (fromRational n)
586−instance (Monad m, Floating a) => Floating (ZipStream m a) where
606− logBase = liftA2 logBase
608------------------------------------------------------------------------------
131------------------------------------------------------------------------------
609-- Parallely Zipping Streams
132-- Parallely Zipping Streams
610------------------------------------------------------------------------------
133------------------------------------------------------------------------------
@@ -626,150 +149,3 @@ zipAsyncWith f m1 m2 = fromStream $ Stream $ \_ stp yld -> do
626 ma <- async m1
149 ma <- async m1
627 mb <- async m2
150 mb <- async m2
628 (runStream (toStream (zipWith f ma mb))) Nothing stp yld
151 (runStream (toStream (zipWith f ma mb))) Nothing stp yld
630−-- | 'ZipAsync' zips in parallel, it produces one element from each stream
631−-- concurrently and then zips the two.
632−newtype ZipAsync m a = ZipAsync {getZipAsync :: Stream m a}
633− deriving (Semigroup, Monoid)
635−deriving instance MonadAsync m => Alternative (ZipAsync m)
637−instance Monad m => Functor (ZipAsync m) where
638− fmap f (ZipAsync (Stream m)) = ZipAsync $ Stream $ \_ stp yld ->
639− let yield a Nothing = yld (f a) Nothing
640− yield a (Just r) = yld (f a)
641− (Just (getZipAsync (fmap f (ZipAsync r))))
642− in m Nothing stp yield
644−instance MonadAsync m => Applicative (ZipAsync m) where
645− pure a = ZipAsync $ cycle1 (pure a)
646− (<*>) = zipAsyncWith id
648−instance Streaming ZipAsync where
649− toStream = getZipAsync
650− fromStream = ZipAsync
652−instance (MonadAsync m, Num a) => Num (ZipAsync m a) where
653− fromInteger n = pure (fromInteger n)
655− negate = fmap negate
657− signum = fmap signum
663−instance (MonadAsync m, Fractional a) => Fractional (ZipAsync m a) where
664− fromRational n = pure (fromRational n)
670−instance (MonadAsync m, Floating a) => Floating (ZipAsync m a) where
690− logBase = liftA2 logBase
692−-------------------------------------------------------------------------------
693−-- Type adapting combinators
694−-------------------------------------------------------------------------------
696−-- | Adapt one streaming type to another.
697−adapt :: (Streaming t1, Streaming t2) => t1 m a -> t2 m a
698−adapt = fromStream . toStream
700−-- | Interpret an ambiguously typed stream as 'StreamT'.
701−serially :: StreamT m a -> StreamT m a
704−-- | Interpret an ambiguously typed stream as 'InterleavedT'.
705−interleaving :: InterleavedT m a -> InterleavedT m a
708−-- | Interpret an ambiguously typed stream as 'AsyncT'.
709−asyncly :: AsyncT m a -> AsyncT m a
712−-- | Interpret an ambiguously typed stream as 'ParallelT'.
713−parallely :: ParallelT m a -> ParallelT m a
716−-- | Interpret an ambiguously typed stream as 'ZipStream'.
717−zipping :: ZipStream m a -> ZipStream m a
720−-- | Interpret an ambiguously typed stream as 'ZipAsync'.
721−zippingAsync :: ZipAsync m a -> ZipAsync m a
724−------------------------------------------------------------------------------
725−-- Construct and deconstruct
726−------------------------------------------------------------------------------
728−-- | Decompose a stream into its head and tail. If the stream is empty, returns
729−-- 'Nothing'. If the stream is non-empty, returns 'Just (a, ma)', where 'a' is
730−-- the head of the stream and 'ma' its tail.
731−uncons :: (Streaming t, Monad m, Monoid (t m a))
732− => t m a -> m (Maybe (a, t m a))
733−uncons m = (runStream (toStream m)) Nothing stop yield
735− stop = return Nothing
737− {-# INLINE yield #-}
738− yield a Nothing = return (Just (a, mempty))
739− yield a (Just x) = return (Just (a, (fromStream x)))
741−-- | Build a Stream by unfolding steps starting from a seed.
742−unfoldr :: (Streaming t, Monad m) => (b -> m (Maybe (a, b))) -> b -> t m a
743−unfoldr step = fromStream . go
745− go s = Stream $ \_ stp yld -> do
749− Just (a, b) -> yld a (Just (go b))
751−------------------------------------------------------------------------------
753−------------------------------------------------------------------------------
755−-- | Take first 'n' elements from the stream and discard the rest.
756−take :: Streaming t => Int -> t m a -> t m a
757−take n m = fromStream $ go n (toStream m)
759− go n1 m1 = Stream $ \ctx stp yld -> do
760− let yield a Nothing = yld a Nothing
761− yield a (Just x) = yld a (Just (go (n1 - 1) x))
764− else (runStream m1) ctx stp yield
766−-- | Discard first 'n' elements from the stream and take the rest.
767−drop :: Streaming t => Int -> t m a -> t m a
768−drop n m = fromStream $ go n (toStream m)
770− go n1 m1 = Stream $ \ctx stp yld -> do
771− let yield _ Nothing = stp
772− yield _ (Just x) = (runStream $ go (n1 - 1) x) ctx stp yld
774− then (runStream m1) ctx stp yld
775− else (runStream m1) ctx stp yield