@@ -0,0 +1,249 @@
1+{-# LANGUAGE UndecidableInstances #-}
4+-- Module : Streamly.Internal.Data.Stream.WSerial
5+-- Copyright : (c) 2017 Composewell Technologies
8+-- Maintainer : streamly@composewell.com
9+-- Stability : experimental
12+-- To run examples in this module:
14+-- >>> import qualified Streamly.Prelude as Stream
16+module Streamly.Internal.Data.Stream.WSerial
18+ -- * Serial interleaving stream
29+import Control.Applicative (liftA2)
30+import Control.DeepSeq (NFData(..))
31+import Control.DeepSeq (NFData1(..))
32+import Control.Monad.Base (MonadBase(..), liftBaseDefault)
33+import Control.Monad.Catch (MonadThrow, throwM)
34+import Control.Monad.IO.Class (MonadIO(..))
35+import Control.Monad.Reader.Class (MonadReader(..))
36+import Control.Monad.State.Class (MonadState(..))
37+import Control.Monad.Trans.Class (MonadTrans(lift))
38+import Data.Foldable (Foldable(foldl'), fold)
39+import Data.Functor.Identity (Identity(..), runIdentity)
40+import Data.Maybe (fromMaybe)
41+import Data.Semigroup (Endo(..))
42+#if __GLASGOW_HASKELL__ < 808
43+import Data.Semigroup (Semigroup(..))
45+import GHC.Exts (IsList(..), IsString(..), oneShot)
47+ ( Lexeme(Ident), lexP, parens, prec, readPrec, readListPrec
48+ , readListPrecDefault)
49+import Streamly.Internal.BaseCompat ((#.))
50+import Streamly.Internal.Data.Maybe.Strict (Maybe'(..), toMaybe)
51+import Streamly.Internal.Data.Stream.StreamK.Type
52+ (Stream, mkStream, foldStream)
54+import qualified Streamly.Internal.Data.Stream.Common as P
55+import qualified Streamly.Internal.Data.Stream.StreamD.Generate as D
56+import qualified Streamly.Internal.Data.Stream.StreamD.Type as D
57+import qualified Streamly.Internal.Data.Stream.StreamK.Type as K
59+import Prelude hiding (map, mapM, repeat, filter)
61+#include "Instances.hs"
65+-- >>> import qualified Streamly.Prelude as Stream
67+------------------------------------------------------------------------------
69+------------------------------------------------------------------------------
71+-- | For 'WSerialT' streams:
74+-- (<>) = 'Streamly.Prelude.wSerial' -- 'Semigroup'
75+-- (>>=) = flip . 'Streamly.Prelude.concatMapWith' 'Streamly.Prelude.wSerial' -- 'Monad'
78+-- Note that '<>' is associative only if we disregard the ordering of elements
79+-- in the resulting stream.
81+-- A single 'Monad' bind behaves like a @for@ loop:
84+-- Stream.toList $ Stream.fromWSerial $ do
85+-- x <- Stream.fromList [1,2] -- foreach x in stream
90+-- Nested monad binds behave like interleaved nested @for@ loops:
93+-- Stream.toList $ Stream.fromWSerial $ do
94+-- x <- Stream.fromList [1,2] -- foreach x in stream
95+-- y <- Stream.fromList [3,4] -- foreach y in stream
98+-- [(1,3),(2,3),(1,4),(2,4)]
100+-- It is a result of interleaving all the nested iterations corresponding to
101+-- element @1@ in the first stream with all the nested iterations of element
104+-- >>> import Streamly.Prelude (wSerial)
105+-- >>> Stream.toList $ Stream.fromList [(1,3),(1,4)] `Stream.wSerial` Stream.fromList [(2,3),(2,4)]
106+-- [(1,3),(2,3),(1,4),(2,4)]
108+-- The @W@ in the name stands for @wide@ or breadth wise scheduling in
109+-- contrast to the depth wise scheduling behavior of 'SerialT'.
111+-- /Since: 0.2.0 ("Streamly")/
114+newtype WSerialT m a = WSerialT {getWSerialT :: Stream m a}
115+ deriving (MonadTrans)
117+-- | An interleaving serial IO stream of elements of type @a@. See 'WSerialT'
118+-- documentation for more details.
120+-- /Since: 0.2.0 ("Streamly")/
123+type WSerial = WSerialT IO
125+{-# INLINE consMWSerial #-}
126+{-# SPECIALIZE consMWSerial :: IO a -> WSerialT IO a -> WSerialT IO a #-}
127+consMWSerial :: Monad m => m a -> WSerialT m a -> WSerialT m a
128+consMWSerial m (WSerialT ms) = WSerialT $ K.consM m ms
130+------------------------------------------------------------------------------
132+------------------------------------------------------------------------------
136+-- Additionally we can have m elements yield from the first stream and n
137+-- elements yielding from the second stream. We can also have time slicing
138+-- variants of positional interleaving, e.g. run first stream for m seconds and
139+-- run the second stream for n seconds.
141+-- Similar combinators can be implemented using WAhead style.
143+{-# INLINE wSerialK #-}
144+wSerialK :: Stream m a -> Stream m a -> Stream m a
145+wSerialK m1 m2 = mkStream $ \st yld sng stp -> do
146+ let stop = foldStream st yld sng stp m2
148+ yieldk a r = yld a (wSerialK m2 r)
149+ foldStream st yieldk single stop m1
151+-- | Interleaves two streams, yielding one element from each stream
152+-- alternately. When one stream stops the rest of the other stream is used in
153+-- the output stream.
157+-- Note that evaluation of @a \`wSerial` b \`wSerial` c@ does not interleave
158+-- @a@, @b@ and @c@ with equal priority. This expression is equivalent to @a
159+-- \`wSerial` (b \`wSerial` c)@, therefore, it fairly interleaves @a@ with the
160+-- result of @b \`wSerial` c@. For example, @Stream.fromList [1,2] \`wSerial`
161+-- Stream.fromList [3,4] \`wSerial` Stream.fromList [5,6]@ would result in
162+-- [1,3,2,5,4,6]. In other words, the leftmost stream gets the same scheduling
163+-- priority as the rest of the streams taken together. The same is true for
164+-- each subexpression on the right.
166+{-# INLINE wSerial #-}
167+wSerial :: WSerialT m a -> WSerialT m a -> WSerialT m a
168+wSerial (WSerialT m1) (WSerialT m2) = WSerialT $ wSerialK m1 m2
170+-- | Like `wSerial` but stops interleaving as soon as the first stream stops.
173+{-# INLINE wSerialFstK #-}
174+wSerialFstK :: Stream m a -> Stream m a -> Stream m a
175+wSerialFstK m1 m2 = mkStream $ \st yld sng stp -> do
176+ let yieldFirst a r = yld a (yieldSecond r m2)
177+ in foldStream st yieldFirst sng stp m1
181+ yieldSecond s1 s2 = mkStream $ \st yld sng stp -> do
182+ let stop = foldStream st yld sng stp s1
184+ yieldk a r = yld a (wSerialK s1 r)
185+ in foldStream st yieldk single stop s2
187+{-# INLINE wSerialFst #-}
188+wSerialFst :: WSerialT m a -> WSerialT m a -> WSerialT m a
189+wSerialFst (WSerialT m1) (WSerialT m2) = WSerialT $ wSerialFstK m1 m2
191+-- | Like `wSerial` but stops interleaving as soon as any of the two streams
195+{-# INLINE wSerialMinK #-}
196+wSerialMinK :: Stream m a -> Stream m a -> Stream m a
197+wSerialMinK m1 m2 = mkStream $ \st yld _ stp -> do
199+ -- "single a" is defined as "yld a (wSerialMin m2 K.nil)" instead of
200+ -- "sng a" to keep the behaviour consistent with the yield continuation.
201+ single a = yld a (wSerialMinK m2 K.nil)
202+ yieldk a r = yld a (wSerialMinK m2 r)
203+ foldStream st yieldk single stop m1
205+{-# INLINE wSerialMin #-}
206+wSerialMin :: WSerialT m a -> WSerialT m a -> WSerialT m a
207+wSerialMin (WSerialT m1) (WSerialT m2) = WSerialT $ wSerialMinK m1 m2
209+instance Semigroup (WSerialT m a) where
212+------------------------------------------------------------------------------
214+------------------------------------------------------------------------------
216+instance Monoid (WSerialT m a) where
217+ mempty = WSerialT K.nil
220+{-# INLINE apWSerial #-}
221+apWSerial :: Monad m => WSerialT m (a -> b) -> WSerialT m a -> WSerialT m b
222+apWSerial (WSerialT m1) (WSerialT m2) =
223+ let f x1 = K.concatMapWith wSerialK (pure . x1) m2
224+ in WSerialT $ K.concatMapWith wSerialK f m1
226+instance Monad m => Applicative (WSerialT m) where
228+ pure = WSerialT . K.fromPure
229+ {-# INLINE (<*>) #-}
232+------------------------------------------------------------------------------
234+------------------------------------------------------------------------------
236+instance Monad m => Monad (WSerialT m) where
238+ {-# INLINE (>>=) #-}
239+ (>>=) (WSerialT m) f = WSerialT $ K.bindWith wSerialK m (getWSerialT . f)
241+------------------------------------------------------------------------------
243+------------------------------------------------------------------------------
245+MONAD_COMMON_INSTANCES(WSerialT,)
246+LIST_INSTANCES(WSerialT)
247+NFDATA1_INSTANCE(WSerialT)
248+FOLDABLE_INSTANCE(WSerialT)
249+TRAVERSABLE_INSTANCE(WSerialT)