git.iko.soy/streamly
on aaa775b help sign in
streamly/core/src/Streamly/Internal/Data/Stream/WSerial.hs
249 lines 8.6 kB history raw
1{-# LANGUAGE UndecidableInstances #-}
3-- |
4-- Module : Streamly.Internal.Data.Stream.WSerial
5-- Copyright : (c) 2017 Composewell Technologies
6--
7-- License : BSD3
8-- Maintainer : streamly@composewell.com
9-- Stability : experimental
10-- Portability : GHC
11--
12-- To run examples in this module:
13--
14-- >>> import qualified Streamly.Prelude as Stream
15--
16module Streamly.Internal.Data.Stream.WSerial
17 (
18 -- * Serial interleaving stream
19 WSerialT(..)
20 , WSerial
21 , wSerialK
22 , wSerial
23 , wSerialFst
24 , wSerialMin
25 , consMWSerial
26 )
27where
29import Control.Applicative (liftA2)
30import Control.DeepSeq (NFData(..))
31import Control.DeepSeq (NFData1(..))
32import Control.Monad.Base (MonadBase(..), liftBaseDefault)
33import Control.Monad.Catch (MonadThrow, throwM)
34import Control.Monad.IO.Class (MonadIO(..))
35import Control.Monad.Reader.Class (MonadReader(..))
36import Control.Monad.State.Class (MonadState(..))
37import Control.Monad.Trans.Class (MonadTrans(lift))
38import Data.Foldable (Foldable(foldl'), fold)
39import Data.Functor.Identity (Identity(..), runIdentity)
40import Data.Maybe (fromMaybe)
41import Data.Semigroup (Endo(..))
42#if __GLASGOW_HASKELL__ < 808
43import Data.Semigroup (Semigroup(..))
44#endif
45import GHC.Exts (IsList(..), IsString(..), oneShot)
46import Text.Read
47 ( Lexeme(Ident), lexP, parens, prec, readPrec, readListPrec
48 , readListPrecDefault)
49import Streamly.Internal.BaseCompat ((#.))
50import Streamly.Internal.Data.Maybe.Strict (Maybe'(..), toMaybe)
51import Streamly.Internal.Data.Stream.StreamK.Type
52 (Stream, mkStream, foldStream)
54import qualified Streamly.Internal.Data.Stream.Common as P
55import qualified Streamly.Internal.Data.Stream.StreamD.Generate as D
56import qualified Streamly.Internal.Data.Stream.StreamD.Type as D
57import qualified Streamly.Internal.Data.Stream.StreamK.Type as K
59import Prelude hiding (map, mapM, repeat, filter)
61#include "Instances.hs"
62#include "inline.hs"
64-- $setup
65-- >>> import qualified Streamly.Prelude as Stream
67------------------------------------------------------------------------------
68-- WSerialT
69------------------------------------------------------------------------------
71-- | For 'WSerialT' streams:
72--
73-- @
74-- (<>) = 'Streamly.Prelude.wSerial' -- 'Semigroup'
75-- (>>=) = flip . 'Streamly.Prelude.concatMapWith' 'Streamly.Prelude.wSerial' -- 'Monad'
76-- @
77--
78-- Note that '<>' is associative only if we disregard the ordering of elements
79-- in the resulting stream.
80--
81-- A single 'Monad' bind behaves like a @for@ loop:
82--
83-- >>> :{
84-- Stream.toList $ Stream.fromWSerial $ do
85-- x <- Stream.fromList [1,2] -- foreach x in stream
86-- return x
87-- :}
88-- [1,2]
89--
90-- Nested monad binds behave like interleaved nested @for@ loops:
91--
92-- >>> :{
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
96-- return (x, y)
97-- :}
98-- [(1,3),(2,3),(1,4),(2,4)]
99--
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
102-- @2@:
103--
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)]
107--
108-- The @W@ in the name stands for @wide@ or breadth wise scheduling in
109-- contrast to the depth wise scheduling behavior of 'SerialT'.
110--
111-- /Since: 0.2.0 ("Streamly")/
112--
113-- @since 0.8.0
114newtype 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.
119--
120-- /Since: 0.2.0 ("Streamly")/
121--
122-- @since 0.8.0
123type WSerial = WSerialT IO
125{-# INLINE consMWSerial #-}
126{-# SPECIALIZE consMWSerial :: IO a -> WSerialT IO a -> WSerialT IO a #-}
127consMWSerial :: Monad m => m a -> WSerialT m a -> WSerialT m a
128consMWSerial m (WSerialT ms) = WSerialT $ K.consM m ms
130------------------------------------------------------------------------------
131-- Semigroup
132------------------------------------------------------------------------------
134infixr 6 `wSerial`
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.
140--
141-- Similar combinators can be implemented using WAhead style.
143{-# INLINE wSerialK #-}
144wSerialK :: Stream m a -> Stream m a -> Stream m a
145wSerialK m1 m2 = mkStream $ \st yld sng stp -> do
146 let stop = foldStream st yld sng stp m2
147 single a = yld a 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.
155-- Scheduling Notes:
156--
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.
165--
166{-# INLINE wSerial #-}
167wSerial :: WSerialT m a -> WSerialT m a -> WSerialT m a
168wSerial (WSerialT m1) (WSerialT m2) = WSerialT $ wSerialK m1 m2
170-- | Like `wSerial` but stops interleaving as soon as the first stream stops.
171--
172-- @since 0.7.0
173{-# INLINE wSerialFstK #-}
174wSerialFstK :: Stream m a -> Stream m a -> Stream m a
175wSerialFstK 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
179 where
181 yieldSecond s1 s2 = mkStream $ \st yld sng stp -> do
182 let stop = foldStream st yld sng stp s1
183 single a = yld a s1
184 yieldk a r = yld a (wSerialK s1 r)
185 in foldStream st yieldk single stop s2
187{-# INLINE wSerialFst #-}
188wSerialFst :: WSerialT m a -> WSerialT m a -> WSerialT m a
189wSerialFst (WSerialT m1) (WSerialT m2) = WSerialT $ wSerialFstK m1 m2
191-- | Like `wSerial` but stops interleaving as soon as any of the two streams
192-- stops.
193--
194-- @since 0.7.0
195{-# INLINE wSerialMinK #-}
196wSerialMinK :: Stream m a -> Stream m a -> Stream m a
197wSerialMinK m1 m2 = mkStream $ \st yld _ stp -> do
198 let stop = stp
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 #-}
206wSerialMin :: WSerialT m a -> WSerialT m a -> WSerialT m a
207wSerialMin (WSerialT m1) (WSerialT m2) = WSerialT $ wSerialMinK m1 m2
209instance Semigroup (WSerialT m a) where
210 (<>) = wSerial
212------------------------------------------------------------------------------
213-- Monoid
214------------------------------------------------------------------------------
216instance Monoid (WSerialT m a) where
217 mempty = WSerialT K.nil
218 mappend = (<>)
220{-# INLINE apWSerial #-}
221apWSerial :: Monad m => WSerialT m (a -> b) -> WSerialT m a -> WSerialT m b
222apWSerial (WSerialT m1) (WSerialT m2) =
223 let f x1 = K.concatMapWith wSerialK (pure . x1) m2
224 in WSerialT $ K.concatMapWith wSerialK f m1
226instance Monad m => Applicative (WSerialT m) where
227 {-# INLINE pure #-}
228 pure = WSerialT . K.fromPure
229 {-# INLINE (<*>) #-}
230 (<*>) = apWSerial
232------------------------------------------------------------------------------
233-- Monad
234------------------------------------------------------------------------------
236instance Monad m => Monad (WSerialT m) where
237 return = pure
238 {-# INLINE (>>=) #-}
239 (>>=) (WSerialT m) f = WSerialT $ K.bindWith wSerialK m (getWSerialT . f)
241------------------------------------------------------------------------------
242-- Other instances
243------------------------------------------------------------------------------
245MONAD_COMMON_INSTANCES(WSerialT,)
246LIST_INSTANCES(WSerialT)
247NFDATA1_INSTANCE(WSerialT)
248FOLDABLE_INSTANCE(WSerialT)
249TRAVERSABLE_INSTANCE(WSerialT)