1{-# LANGUAGE FlexibleContexts #-}
2{-# LANGUAGE FlexibleInstances #-}
3{-# LANGUAGE GeneralizedNewtypeDeriving#-}
4{-# LANGUAGE MultiParamTypeClasses #-}
5{-# LANGUAGE RankNTypes #-}
6{-# LANGUAGE StandaloneDeriving #-}
7{-# LANGUAGE UndecidableInstances #-} -- XXX
10-- Module : Asyncly.Prelude
11-- Copyright : (c) 2017 Harendra Kumar
14-- Maintainer : harendra.kumar@gmail.com
15-- Stability : experimental
21 -- * Product Style Composition
30 -- * Stream Type Adapters
57import Control.Applicative (Alternative (..), liftA2)
58import Control.Monad (MonadPlus(..), ap, liftM)
59import Control.Monad.Base (MonadBase (..))
60import Control.Monad.Catch (MonadThrow)
61import Control.Monad.Error.Class (MonadError(..))
62import Control.Monad.IO.Class (MonadIO(..))
63import Control.Monad.Reader.Class (MonadReader(..))
64import Control.Monad.State.Class (MonadState(..))
65import Control.Monad.Trans.Class (MonadTrans)
66import Data.Semigroup (Semigroup(..), cycle1)
67import Prelude hiding (drop, take, zipWith)
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
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.
120newtype StreamT m a = StreamT {getStreamT :: Stream m a}
121 deriving (Semigroup, Monoid, MonadTrans, MonadIO, MonadThrow)
123deriving instance MonadAsync m => Alternative (StreamT m)
124deriving instance MonadAsync m => MonadPlus (StreamT m)
125deriving instance (MonadBase b m, Monad m) => MonadBase b (StreamT m)
126deriving instance MonadError e m => MonadError e (StreamT m)
127deriving instance MonadReader r m => MonadReader r (StreamT m)
128deriving instance MonadState s m => MonadState s (StreamT m)
130instance Streaming StreamT where
131 toStream = getStreamT
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.
145instance 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------------------------------------------------------------------------------
158instance Monad m => Applicative (StreamT m) where
159 pure = StreamT . yielding
162------------------------------------------------------------------------------
164------------------------------------------------------------------------------
166instance Monad m => Functor (StreamT m) where
169------------------------------------------------------------------------------
171------------------------------------------------------------------------------
173instance (Monad m, Num a) => Num (StreamT m a) where
174 fromInteger n = pure (fromInteger n)
184instance (Monad m, Fractional a) => Fractional (StreamT m a) where
185 fromRational n = pure (fromRational n)
191instance (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------------------------------------------------------------------------------
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.
222newtype InterleavedT m a = InterleavedT {getInterleavedT :: Stream m a}
223 deriving (Semigroup, Monoid, MonadTrans, MonadIO, MonadThrow)
225deriving instance MonadAsync m => Alternative (InterleavedT m)
226deriving instance MonadAsync m => MonadPlus (InterleavedT m)
227deriving instance (MonadBase b m, Monad m) => MonadBase b (InterleavedT m)
228deriving instance MonadError e m => MonadError e (InterleavedT m)
229deriving instance MonadReader r m => MonadReader r (InterleavedT m)
230deriving instance MonadState s m => MonadState s (InterleavedT m)
232instance 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.
238instance 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------------------------------------------------------------------------------
252instance Monad m => Applicative (InterleavedT m) where
253 pure = InterleavedT . yielding
256------------------------------------------------------------------------------
258------------------------------------------------------------------------------
260instance Monad m => Functor (InterleavedT m) where
263------------------------------------------------------------------------------
265------------------------------------------------------------------------------
267instance (Monad m, Num a) => Num (InterleavedT m a) where
268 fromInteger n = pure (fromInteger n)
278instance (Monad m, Fractional a) => Fractional (InterleavedT m a) where
279 fromRational n = pure (fromRational n)
285instance (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------------------------------------------------------------------------------
313-- | Just like 'StreamT', 'AsyncT' style composition works well operationally
314-- for infinite streams. However, if the second stream is infinite we will
315-- never be able to finish the first stream.
317-- Note that 'fmap' on 'AsyncT' would map on multiple elements concurrently.
319newtype AsyncT m a = AsyncT {getAsyncT :: Stream m a}
320 deriving (Semigroup, Monoid, MonadTrans)
322deriving instance MonadAsync m => Alternative (AsyncT m)
323deriving instance MonadAsync m => MonadPlus (AsyncT m)
324deriving instance MonadAsync m => MonadIO (AsyncT m)
325deriving instance MonadAsync m => MonadThrow (AsyncT m)
326deriving instance (MonadBase b m, MonadAsync m) => MonadBase b (AsyncT m)
327deriving instance (MonadError e m, MonadAsync m) => MonadError e (AsyncT m)
328deriving instance (MonadReader r m, MonadAsync m) => MonadReader r (AsyncT m)
329deriving instance (MonadState s m, MonadAsync m) => MonadState s (AsyncT m)
331instance Streaming AsyncT where
335{-# INLINE parbind #-}
337 :: (forall c. Stream m c -> Stream m c -> Stream m c)
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
353instance 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------------------------------------------------------------------------------
363instance MonadAsync m => Applicative (AsyncT m) where
364 pure = AsyncT . yielding
367------------------------------------------------------------------------------
369------------------------------------------------------------------------------
371instance MonadAsync m => Functor (AsyncT m) where
374------------------------------------------------------------------------------
376------------------------------------------------------------------------------
378instance (MonadAsync m, Num a) => Num (AsyncT m a) where
379 fromInteger n = pure (fromInteger n)
389instance (MonadAsync m, Fractional a) => Fractional (AsyncT m a) where
390 fromRational n = pure (fromRational n)
396instance (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.
430newtype ParallelT m a = ParallelT {getParallelT :: Stream m a}
431 deriving (Semigroup, Monoid, MonadTrans)
433deriving instance MonadAsync m => Alternative (ParallelT m)
434deriving instance MonadAsync m => MonadPlus (ParallelT m)
435deriving instance MonadAsync m => MonadIO (ParallelT m)
436deriving instance MonadAsync m => MonadThrow (ParallelT m)
437deriving instance (MonadBase b m, MonadAsync m) => MonadBase b (ParallelT m)
438deriving instance (MonadError e m, MonadAsync m) => MonadError e (ParallelT m)
439deriving instance (MonadReader r m, MonadAsync m) => MonadReader r (ParallelT m)
440deriving instance (MonadState s m, MonadAsync m) => MonadState s (ParallelT m)
442instance 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
449instance MonadAsync m => Monad (ParallelT m) where
451 (ParallelT m) >>= f = ParallelT $ parbind par m g
452 where g x = getParallelT (f x)
453 par = parallel (CtxType Conjunction FIFO)
455------------------------------------------------------------------------------
457------------------------------------------------------------------------------
459instance MonadAsync m => Applicative (ParallelT m) where
460 pure = ParallelT . yielding
463------------------------------------------------------------------------------
465------------------------------------------------------------------------------
467instance MonadAsync m => Functor (ParallelT m) where
470------------------------------------------------------------------------------
472------------------------------------------------------------------------------
474instance (MonadAsync m, Num a) => Num (ParallelT m a) where
475 fromInteger n = pure (fromInteger n)
485instance (MonadAsync m, Fractional a) => Fractional (ParallelT m a) where
486 fromRational n = pure (fromRational n)
492instance (MonadAsync m, Floating a) => Floating (ParallelT m a) where
512 logBase = liftA2 logBase
514------------------------------------------------------------------------------
515-- Serially Zipping Streams
516------------------------------------------------------------------------------
518-- | Zip two AsyncT streams serially using a monadic zipping function.
519zipWithM :: Streaming t => (a -> b -> t m c) -> t m a -> t m b -> t m c
520zipWithM f m1 m2 = fromStream $ go (toStream m1) (toStream m2)
522 go mx my = Stream $ \_ stp yld -> do
524 let yield2 b Nothing = (runStream (g a b)) Nothing stp yld
526 (runStream ((g a b) <> (go ra rb))) Nothing stp yld
527 in (runStream my) Nothing stp yield2
528 let yield1 a Nothing = merge a mempty
529 yield1 a (Just ra) = merge a ra
530 (runStream mx) Nothing stp yield1
531 g a b = toStream $ f a b
533-- | Zip two AsyncT streams serially using a pure zipping function.
534zipWith :: Streaming t => (a -> b -> c) -> t m a -> t m b -> t m c
535zipWith f m1 m2 = fromStream $ go (toStream m1) (toStream m2)
537 go mx my = Stream $ \_ stp yld -> do
539 let yield2 b Nothing = yld (f a b) Nothing
540 yield2 b (Just rb) = yld (f a b) (Just (go ra rb))
541 in (runStream my) Nothing stp yield2
542 let yield1 a Nothing = merge a mempty
543 yield1 a (Just ra) = merge a ra
544 (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.
548newtype ZipStream m a = ZipStream {getZipStream :: Stream m a}
549 deriving (Semigroup, Monoid)
551deriving instance MonadAsync m => Alternative (ZipStream m)
553instance 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
560instance Monad m => Applicative (ZipStream m) where
561 pure a = ZipStream $ cycle1 (pure a)
564instance Streaming ZipStream where
565 toStream = getZipStream
566 fromStream = ZipStream
568instance (Monad m, Num a) => Num (ZipStream m a) where
569 fromInteger n = pure (fromInteger n)
579instance (Monad m, Fractional a) => Fractional (ZipStream m a) where
580 fromRational n = pure (fromRational n)
586instance (Monad m, Floating a) => Floating (ZipStream m a) where
606 logBase = liftA2 logBase
608------------------------------------------------------------------------------
609-- Parallely Zipping Streams
610------------------------------------------------------------------------------
612-- | Zip two AsyncT streams asyncly (i.e. both the streams are generated
613-- concurrently) using a monadic zipping function.
614zipAsyncWithM :: (Streaming t, MonadAsync m)
615 => (a -> b -> t m c) -> t m a -> t m b -> t m c
616zipAsyncWithM f m1 m2 = fromStream $ Stream $ \_ stp yld -> do
619 (runStream (toStream (zipWithM f ma mb))) Nothing stp yld
621-- | Zip two AsyncT streams asyncly (i.e. both the streams are generated
622-- concurrently) using a pure zipping function.
623zipAsyncWith :: (Streaming t, MonadAsync m)
624 => (a -> b -> c) -> t m a -> t m b -> t m c
625zipAsyncWith f m1 m2 = fromStream $ Stream $ \_ stp yld -> do
628 (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.
632newtype ZipAsync m a = ZipAsync {getZipAsync :: Stream m a}
633 deriving (Semigroup, Monoid)
635deriving instance MonadAsync m => Alternative (ZipAsync m)
637instance 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
644instance MonadAsync m => Applicative (ZipAsync m) where
645 pure a = ZipAsync $ cycle1 (pure a)
646 (<*>) = zipAsyncWith id
648instance Streaming ZipAsync where
649 toStream = getZipAsync
650 fromStream = ZipAsync
652instance (MonadAsync m, Num a) => Num (ZipAsync m a) where
653 fromInteger n = pure (fromInteger n)
663instance (MonadAsync m, Fractional a) => Fractional (ZipAsync m a) where
664 fromRational n = pure (fromRational n)
670instance (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.
697adapt :: (Streaming t1, Streaming t2) => t1 m a -> t2 m a
698adapt = fromStream . toStream
700-- | Interpret an ambiguously typed stream as 'StreamT'.
701serially :: StreamT m a -> StreamT m a
704-- | Interpret an ambiguously typed stream as 'InterleavedT'.
705interleaving :: InterleavedT m a -> InterleavedT m a
708-- | Interpret an ambiguously typed stream as 'AsyncT'.
709asyncly :: AsyncT m a -> AsyncT m a
712-- | Interpret an ambiguously typed stream as 'ParallelT'.
713parallely :: ParallelT m a -> ParallelT m a
716-- | Interpret an ambiguously typed stream as 'ZipStream'.
717zipping :: ZipStream m a -> ZipStream m a
720-- | Interpret an ambiguously typed stream as 'ZipAsync'.
721zippingAsync :: 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.
731uncons :: (Streaming t, Monad m, Monoid (t m a))
732 => t m a -> m (Maybe (a, t m a))
733uncons m = (runStream (toStream m)) Nothing stop yield
735 stop = return Nothing
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.
742unfoldr :: (Streaming t, Monad m) => (b -> m (Maybe (a, b))) -> b -> t m a
743unfoldr 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.
756take :: Streaming t => Int -> t m a -> t m a
757take 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.
767drop :: Streaming t => Int -> t m a -> t m a
768drop 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
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