@@ -1,3 +1,5 @@
1+{-# LANGUAGE UndecidableInstances #-}
2+
13-- |
24-- Module : Streamly.Internal.Data.Stream.Serial
35-- Copyright : (c) 2017 Composewell Technologies
@@ -10,6 +12,7 @@
1012-- To run examples in this module:
1113--
1214-- >>> import qualified Streamly.Data.Stream as Stream
15+-- >>> import qualified Streamly.Prelude as IsStream
1316--
1417module Streamly.Internal.Data.Stream.Serial
1518 (
@@ -18,6 +21,14 @@ module Streamly.Internal.Data.Stream.Serial
1821 , Serial
1922 , serial
2023
24+ -- * Serial interleaving stream
25+ , WSerialT(..)
26+ , WSerial
27+ , wSerial
28+ , wSerialFst
29+ , wSerialMin
30+ , consMWSerial
31+
2132 -- * Construction
2233 , Stream.cons
2334 , Stream.consM
@@ -34,15 +45,42 @@ module Streamly.Internal.Data.Stream.Serial
3445 )
3546where
3647
37−import GHC.Exts (IsList(..))
48+import Control.Applicative (liftA2)
49+import Control.DeepSeq (NFData(..), NFData1(..))
50+import Control.Monad.Base (MonadBase(..), liftBaseDefault)
51+import Control.Monad.Catch (MonadThrow, throwM)
52+import Control.Monad.IO.Class (MonadIO(..))
53+import Control.Monad.Reader.Class (MonadReader(..))
54+import Control.Monad.State.Class (MonadState(..))
55+import Control.Monad.Trans.Class (MonadTrans(lift))
56+import Data.Foldable (Foldable(foldl'), fold)
57+import Data.Functor.Identity (Identity(..), runIdentity)
58+import Data.Maybe (fromMaybe)
59+import Data.Semigroup (Endo(..))
60+import GHC.Exts (IsList(..), IsString(..), oneShot)
61+import Text.Read
62+ ( Lexeme(Ident), lexP, parens, prec, readPrec, readListPrec
63+ , readListPrecDefault)
64+import Streamly.Internal.BaseCompat ((#.))
65+import Streamly.Internal.Data.Maybe.Strict (Maybe'(..), toMaybe)
3866import Streamly.Data.Stream (Stream)
3967
4068import qualified Streamly.Data.Stream as Stream
69+import qualified Streamly.Internal.Data.Stream.Common as P
70+import qualified Streamly.Internal.Data.Stream.StreamD as D
71+ (fromStreamK, toStreamK, mapM)
72+import qualified Streamly.Internal.Data.Stream.StreamK.Type as K
73+ (Stream, cons, consM, nil, concatMapWith, fromPure, bindWith
74+ , withLocal, interleave, interleaveFst, interleaveMin)
4175
42−import Prelude hiding (map)
76+import Prelude hiding (map, mapM, repeat, filter)
77+
78+#include "Instances.hs"
79+#include "inline.hs"
4380
4481-- $setup
4582-- >>> import qualified Streamly.Data.Stream as Stream
83+-- >>> import qualified Streamly.Prelude as IsStream
4684
4785------------------------------------------------------------------------------
4886-- SerialT
@@ -82,3 +120,146 @@ serial = (<>)
82120{-# INLINE map #-}
83121map :: Monad m => (a -> b) -> SerialT m a -> SerialT m b
84122map f = Stream.mapM (return . f)
123+
124+------------------------------------------------------------------------------
125+-- WSerialT
126+------------------------------------------------------------------------------
127+
128+-- | For 'WSerialT' streams:
129+--
130+-- @
131+-- (<>) = 'Streamly.Prelude.wSerial' -- 'Semigroup'
132+-- (>>=) = flip . 'Streamly.Prelude.concatMapWith' 'Streamly.Prelude.wSerial' -- 'Monad'
133+-- @
134+--
135+-- Note that '<>' is associative only if we disregard the ordering of elements
136+-- in the resulting stream.
137+--
138+-- A single 'Monad' bind behaves like a @for@ loop:
139+--
140+-- >>> :{
141+-- IsStream.toList $ IsStream.fromWSerial $ do
142+-- x <- IsStream.fromList [1,2] -- foreach x in stream
143+-- return x
144+-- :}
145+-- [1,2]
146+--
147+-- Nested monad binds behave like interleaved nested @for@ loops:
148+--
149+-- >>> :{
150+-- IsStream.toList $ IsStream.fromWSerial $ do
151+-- x <- IsStream.fromList [1,2] -- foreach x in stream
152+-- y <- IsStream.fromList [3,4] -- foreach y in stream
153+-- return (x, y)
154+-- :}
155+-- [(1,3),(2,3),(1,4),(2,4)]
156+--
157+-- It is a result of interleaving all the nested iterations corresponding to
158+-- element @1@ in the first stream with all the nested iterations of element
159+-- @2@:
160+--
161+-- >>> import Streamly.Prelude (wSerial)
162+-- >>> IsStream.toList $ IsStream.fromList [(1,3),(1,4)] `IsStream.wSerial` IsStream.fromList [(2,3),(2,4)]
163+-- [(1,3),(2,3),(1,4),(2,4)]
164+--
165+-- The @W@ in the name stands for @wide@ or breadth wise scheduling in
166+-- contrast to the depth wise scheduling behavior of 'Stream'.
167+--
168+-- /Since: 0.2.0 ("Streamly")/
169+--
170+-- @since 0.8.0
171+newtype WSerialT m a = WSerialT {getWSerialT :: K.Stream m a}
172+ deriving (MonadTrans)
173+
174+-- | An interleaving serial IO stream of elements of type @a@. See 'WSerialT'
175+-- documentation for more details.
176+--
177+-- /Since: 0.2.0 ("Streamly")/
178+--
179+-- @since 0.8.0
180+type WSerial = WSerialT IO
181+
182+{-# INLINE consMWSerial #-}
183+{-# SPECIALIZE consMWSerial :: IO a -> WSerialT IO a -> WSerialT IO a #-}
184+consMWSerial :: Monad m => m a -> WSerialT m a -> WSerialT m a
185+consMWSerial m (WSerialT ms) = WSerialT $ K.consM m ms
186+
187+------------------------------------------------------------------------------
188+-- Semigroup
189+------------------------------------------------------------------------------
190+
191+infixr 6 `wSerial`
192+
193+-- | Interleaves two streams, yielding one element from each stream
194+-- alternately. When one stream stops the rest of the other stream is used in
195+-- the output stream.
196+--
197+-- This gives exponential priority to earlier streams than the ones joining
198+-- later. Because of exponential weighting it can be used with 'concatMapWith'.
199+--
200+-- /Not fused/
201+
202+-- NOTE:
203+--
204+-- Note that evaluation of @a \`wSerial` b \`wSerial` c@ does not interleave
205+-- @a@, @b@ and @c@ with equal priority. This expression is equivalent to @a
206+-- \`wSerial` (b \`wSerial` c)@, therefore, it fairly interleaves @a@ with the
207+-- result of @b \`wSerial` c@. For example, @Stream.fromList [1,2] \`wSerial`
208+-- Stream.fromList [3,4] \`wSerial` Stream.fromList [5,6]@ would result in
209+-- [1,3,2,5,4,6]. In other words, the leftmost stream gets the same scheduling
210+-- priority as the rest of the streams taken together. The same is true for
211+-- each subexpression on the right.
212+--
213+{-# INLINE wSerial #-}
214+wSerial :: WSerialT m a -> WSerialT m a -> WSerialT m a
215+wSerial (WSerialT m1) (WSerialT m2) = WSerialT $ K.interleave m1 m2
216+
217+{-# INLINE wSerialFst #-}
218+wSerialFst :: WSerialT m a -> WSerialT m a -> WSerialT m a
219+wSerialFst (WSerialT m1) (WSerialT m2) = WSerialT $ K.interleaveFst m1 m2
220+
221+{-# INLINE wSerialMin #-}
222+wSerialMin :: WSerialT m a -> WSerialT m a -> WSerialT m a
223+wSerialMin (WSerialT m1) (WSerialT m2) = WSerialT $ K.interleaveMin m1 m2
224+
225+instance Semigroup (WSerialT m a) where
226+ (<>) = wSerial
227+
228+------------------------------------------------------------------------------
229+-- Monoid
230+------------------------------------------------------------------------------
231+
232+instance Monoid (WSerialT m a) where
233+ mempty = WSerialT K.nil
234+ mappend = (<>)
235+
236+{-# INLINE apWSerial #-}
237+apWSerial :: Monad m => WSerialT m (a -> b) -> WSerialT m a -> WSerialT m b
238+apWSerial (WSerialT m1) (WSerialT m2) =
239+ let f x1 = K.concatMapWith K.interleave (pure . x1) m2
240+ in WSerialT $ K.concatMapWith K.interleave f m1
241+
242+instance Monad m => Applicative (WSerialT m) where
243+ {-# INLINE pure #-}
244+ pure = WSerialT . K.fromPure
245+ {-# INLINE (<*>) #-}
246+ (<*>) = apWSerial
247+
248+------------------------------------------------------------------------------
249+-- Monad
250+------------------------------------------------------------------------------
251+
252+instance Monad m => Monad (WSerialT m) where
253+ return = pure
254+ {-# INLINE (>>=) #-}
255+ (>>=) (WSerialT m) f = WSerialT $ K.bindWith K.interleave m (getWSerialT . f)
256+
257+------------------------------------------------------------------------------
258+-- Other instances
259+------------------------------------------------------------------------------
260+
261+MONAD_COMMON_INSTANCES(WSerialT,)
262+LIST_INSTANCES(WSerialT)
263+NFDATA1_INSTANCE(WSerialT)
264+FOLDABLE_INSTANCE(WSerialT)
265+TRAVERSABLE_INSTANCE(WSerialT)