git.iko.soy/streamly
on c57d6db help sign in
streamly/src/Streamly/Internal/Data/Stream/Serial.hs
504 lines 15 kB history raw
1{-# LANGUAGE UndecidableInstances #-}
3-- |
4-- Module : Streamly.Internal.Data.Stream.Serial
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.Serial
17 (
18 -- * Serial appending stream
19 SerialT
20 , Serial
21 , K.serial
22 , fromSerial
24 -- * Serial interleaving stream
25 , WSerialT
26 , WSerial
27 , wSerial
28 , wSerialFst
29 , wSerialMin
30 , fromWSerial
32 -- * Construction
33 , unfoldrM
35 -- * Transformation
36 , map
37 , mapM
39 -- * Deprecated
40 , StreamT
41 , InterleavedT
42 , (<=>)
43 , interleaving
44 )
45where
47import Control.Applicative (liftA2)
48import Control.DeepSeq (NFData(..))
49#if MIN_VERSION_deepseq(1,4,3)
50import Control.DeepSeq (NFData1(..))
51#endif
52import Control.Monad.Base (MonadBase(..), liftBaseDefault)
53import Control.Monad.Catch (MonadThrow, throwM)
54import Control.Monad.IO.Class (MonadIO(..))
55import Control.Monad.Reader.Class (MonadReader(..))
56import Control.Monad.State.Class (MonadState(..))
57import Control.Monad.Trans.Class (MonadTrans(lift))
58import Data.Foldable (Foldable(foldl'), fold)
59import Data.Functor.Identity (Identity(..), runIdentity)
60import Data.Maybe (fromMaybe)
61import Data.Semigroup (Endo(..))
62#if __GLASGOW_HASKELL__ < 808
63import Data.Semigroup (Semigroup(..))
64#endif
65import GHC.Exts (IsList(..), IsString(..))
66import Text.Read
67 ( Lexeme(Ident), lexP, parens, prec, readPrec, readListPrec
68 , readListPrecDefault)
69import Streamly.Internal.BaseCompat ((#.), errorWithoutStackTrace)
70import Streamly.Internal.Data.Stream.StreamK.Type
71 (IsStream(..), adapt, Stream, mkStream, foldStream)
72import Streamly.Internal.Data.Maybe.Strict (Maybe'(..), toMaybe)
74import qualified Streamly.Internal.Data.Stream.Prelude as P
75 (cmpBy, foldl', foldr, eqBy, fromList, toList)
76import qualified Streamly.Internal.Data.Stream.StreamK as K (withLocal)
77import qualified Streamly.Internal.Data.Stream.StreamK.Type as K
78import qualified Streamly.Internal.Data.Stream.StreamD.Generate as D (unfoldrM)
79import qualified Streamly.Internal.Data.Stream.StreamD.Type as D
81import Prelude hiding (map, mapM, errorWithoutStackTrace)
83#include "Instances.hs"
84#include "inline.hs"
86------------------------------------------------------------------------------
87-- SerialT
88------------------------------------------------------------------------------
90-- | For 'SerialT' streams:
91--
92-- @
93-- (<>) = 'Streamly.Prelude.serial' -- 'Semigroup'
94-- (>>=) = flip . 'Streamly.Prelude.concatMapWith' 'Streamly.Prelude.serial' -- 'Monad'
95-- @
96--
97-- A single 'Monad' bind behaves like a @for@ loop:
98--
99-- >>> :{
100-- Stream.toList $ do
101-- x <- Stream.fromList [1,2] -- foreach x in stream
102-- return x
103-- :}
104-- [1,2]
105--
106-- Nested monad binds behave like nested @for@ loops:
107--
108-- >>> :{
109-- Stream.toList $ do
110-- x <- Stream.fromList [1,2] -- foreach x in stream
111-- y <- Stream.fromList [3,4] -- foreach y in stream
112-- return (x, y)
113-- :}
114-- [(1,3),(1,4),(2,3),(2,4)]
115--
116-- /Since: 0.2.0 ("Streamly")/
117--
118-- @since 0.8.0
119newtype SerialT m a = SerialT {getSerialT :: Stream m a}
120 -- XXX when deriving do we inherit an INLINE?
121 deriving (Semigroup, Monoid, MonadTrans)
123-- | A serial IO stream of elements of type @a@. See 'SerialT' documentation
124-- for more details.
125--
126-- /Since: 0.2.0 ("Streamly")/
127--
128-- @since 0.8.0
129type Serial = SerialT IO
131-- |
132-- @since 0.1.0
133{-# DEPRECATED StreamT "Please use 'SerialT' instead." #-}
134type StreamT = SerialT
136-- | Fix the type of a polymorphic stream as 'SerialT'.
137--
138-- /Since: 0.1.0 ("Streamly")/
139--
140-- @since 0.8.0
141fromSerial :: IsStream t => SerialT m a -> t m a
142fromSerial = adapt
144{-# INLINE consMSerial #-}
145{-# SPECIALIZE consMSerial :: IO a -> SerialT IO a -> SerialT IO a #-}
146consMSerial :: Monad m => m a -> SerialT m a -> SerialT m a
147consMSerial m ms = fromStream $ K.consMStream m (toStream ms)
149instance IsStream SerialT where
150 toStream = getSerialT
151 fromStream = SerialT
152 consM = consMSerial
153 (|:) = consMSerial
155------------------------------------------------------------------------------
156-- Monad
157------------------------------------------------------------------------------
159instance Monad m => Monad (SerialT m) where
160 return = pure
162 -- Benchmarks better with StreamD bind and pure:
163 -- toList, filterAllout, *>, *<, >> (~2x)
164 --
165 -- pure = SerialT . D.fromStreamD . D.fromPure
166 -- m >>= f = D.fromStreamD $ D.concatMap (D.toStreamD . f) (D.toStreamD m)
168 -- Benchmarks better with CPS bind and pure:
169 -- Prime sieve (25x)
170 -- n binds, breakAfterSome, filterAllIn, state transformer (~2x)
171 --
172 {-# INLINE (>>=) #-}
173 (>>=) = K.bindWith K.serial
175 {-# INLINE (>>) #-}
176 (>>) = (*>)
178------------------------------------------------------------------------------
179-- Other instances
180------------------------------------------------------------------------------
182{-# INLINE mapM #-}
183mapM :: (IsStream t, Monad m) => (a -> m b) -> t m a -> t m b
184mapM f m = D.fromStreamD $ D.mapM f $ D.toStreamD m
186-- |
187-- @
188-- map = fmap
189-- @
190--
191-- Same as 'fmap'.
192--
193-- @
194-- > S.toList $ S.map (+1) $ S.fromList [1,2,3]
195-- [2,3,4]
196-- @
197--
198-- @since 0.4.0
199{-# INLINE map #-}
200map :: (IsStream t, Monad m) => (a -> b) -> t m a -> t m b
201map f = mapM (return . f)
203{-# INLINE apSerial #-}
204apSerial :: Monad m => SerialT m (a -> b) -> SerialT m a -> SerialT m b
205apSerial (SerialT m1) (SerialT m2) =
206 D.fromStreamD $ D.toStreamD m1 <*> D.toStreamD m2
208{-# INLINE apSequence #-}
209apSequence :: Monad m => SerialT m a -> SerialT m b -> SerialT m b
210apSequence (SerialT m1) (SerialT m2) =
211 D.fromStreamD $ D.toStreamD m1 *> D.toStreamD m2
213{-# INLINE apDiscardSnd #-}
214apDiscardSnd :: Monad m => SerialT m a -> SerialT m b -> SerialT m a
215apDiscardSnd (SerialT m1) (SerialT m2) =
216 D.fromStreamD $ D.toStreamD m1 <* D.toStreamD m2
218-- Note: we need to define all the typeclass operations because we want to
219-- INLINE them.
220instance Monad m => Applicative (SerialT m) where
221 {-# INLINE pure #-}
222 pure = SerialT . K.fromPure
224 {-# INLINE (<*>) #-}
225 (<*>) = apSerial
226 -- (<*>) = K.apSerial
228#if MIN_VERSION_base(4,10,0)
229 {-# INLINE liftA2 #-}
230 liftA2 f x = (<*>) (fmap f x)
231#endif
233 {-# INLINE (*>) #-}
234 (*>) = apSequence
235 -- (*>) = K.apSerialDiscardFst
237 {-# INLINE (<*) #-}
238 (<*) = apDiscardSnd
239 -- (<*) = K.apSerialDiscardSnd
241MONAD_COMMON_INSTANCES(SerialT,)
242LIST_INSTANCES(SerialT)
243NFDATA1_INSTANCE(SerialT)
244FOLDABLE_INSTANCE(SerialT)
245TRAVERSABLE_INSTANCE(SerialT)
247------------------------------------------------------------------------------
248-- WSerialT
249------------------------------------------------------------------------------
251-- | For 'WSerialT' streams:
252--
253-- @
254-- (<>) = 'Streamly.Prelude.wSerial' -- 'Semigroup'
255-- (>>=) = flip . 'Streamly.Prelude.concatMapWith' 'Streamly.Prelude.wSerial' -- 'Monad'
256-- @
257--
258-- Note that '<>' is associative only if we disregard the ordering of elements
259-- in the resulting stream.
260--
261-- A single 'Monad' bind behaves like a @for@ loop:
262--
263-- >>> :{
264-- Stream.toList $ Stream.fromWSerial $ do
265-- x <- Stream.fromList [1,2] -- foreach x in stream
266-- return x
267-- :}
268-- [1,2]
269--
270-- Nested monad binds behave like interleaved nested @for@ loops:
271--
272-- >>> :{
273-- Stream.toList $ Stream.fromWSerial $ do
274-- x <- Stream.fromList [1,2] -- foreach x in stream
275-- y <- Stream.fromList [3,4] -- foreach y in stream
276-- return (x, y)
277-- :}
278-- [(1,3),(2,3),(1,4),(2,4)]
279--
280-- It is a result of interleaving all the nested iterations corresponding to
281-- element @1@ in the first stream with all the nested iterations of element
282-- @2@:
283--
284-- >>> import Streamly.Prelude (wSerial)
285-- >>> Stream.toList $ Stream.fromList [(1,3),(1,4)] `wSerial` Stream.fromList [(2,3),(2,4)]
286-- [(1,3),(2,3),(1,4),(2,4)]
287--
288-- The @W@ in the name stands for @wide@ or breadth wise scheduling in
289-- contrast to the depth wise scheduling behavior of 'SerialT'.
290--
291-- /Since: 0.2.0 ("Streamly")/
292--
293-- @since 0.8.0
294newtype WSerialT m a = WSerialT {getWSerialT :: Stream m a}
295 deriving (MonadTrans)
297-- | An interleaving serial IO stream of elements of type @a@. See 'WSerialT'
298-- documentation for more details.
299--
300-- /Since: 0.2.0 ("Streamly")/
301--
302-- @since 0.8.0
303type WSerial = WSerialT IO
305-- |
306-- @since 0.1.0
307{-# DEPRECATED InterleavedT "Please use 'WSerialT' instead." #-}
308type InterleavedT = WSerialT
310-- | Fix the type of a polymorphic stream as 'WSerialT'.
311--
312-- /Since: 0.2.0 ("Streamly")/
313--
314-- @since 0.8.0
315fromWSerial :: IsStream t => WSerialT m a -> t m a
316fromWSerial = adapt
318-- | Same as 'fromWSerial'.
319--
320-- @since 0.1.0
321{-# DEPRECATED interleaving "Please use fromWSerial instead." #-}
322interleaving :: IsStream t => WSerialT m a -> t m a
323interleaving = fromWSerial
325consMWSerial :: Monad m => m a -> WSerialT m a -> WSerialT m a
326consMWSerial m ms = fromStream $ K.consMStream m (toStream ms)
328instance IsStream WSerialT where
329 toStream = getWSerialT
330 fromStream = WSerialT
332 {-# INLINE consM #-}
333 {-# SPECIALIZE consM :: IO a -> WSerialT IO a -> WSerialT IO a #-}
334 consM :: Monad m => m a -> WSerialT m a -> WSerialT m a
335 consM = consMWSerial
337 {-# INLINE (|:) #-}
338 {-# SPECIALIZE (|:) :: IO a -> WSerialT IO a -> WSerialT IO a #-}
339 (|:) :: Monad m => m a -> WSerialT m a -> WSerialT m a
340 (|:) = consMWSerial
342------------------------------------------------------------------------------
343-- Semigroup
344------------------------------------------------------------------------------
346infixr 6 `wSerial`
348-- Additionally we can have m elements yield from the first stream and n
349-- elements yielding from the second stream. We can also have time slicing
350-- variants of positional interleaving, e.g. run first stream for m seconds and
351-- run the second stream for n seconds.
352--
353-- Similar combinators can be implemented using WAhead style.
355-- | Interleaves two streams, yielding one element from each stream
356-- alternately. When one stream stops the rest of the other stream is used in
357-- the output stream.
358--
359-- >>> import Streamly.Prelude (wSerial)
360-- >>> stream1 = Stream.fromList [1,2]
361-- >>> stream2 = Stream.fromList [3,4]
362-- >>> Stream.toList $ Stream.fromWSerial $ stream1 `wSerial` stream2
363-- [1,3,2,4]
364--
365-- Note, for singleton streams 'wSerial' and 'serial' are identical.
366--
367-- Note that this operation cannot be used to fold a container of infinite
368-- streams but it can be used for very large streams as the state that it needs
369-- to maintain is proportional to the logarithm of the number of streams.
370--
371-- @since 0.8.0
372--
373-- /Since: 0.2.0 ("Streamly")/
375-- Scheduling Notes:
376--
377-- Note that evaluation of @a \`wSerial` b \`wSerial` c@ does not interleave
378-- @a@, @b@ and @c@ with equal priority. This expression is equivalent to @a
379-- \`wSerial` (b \`wSerial` c)@, therefore, it fairly interleaves @a@ with the
380-- result of @b \`wSerial` c@. For example, @Stream.fromList [1,2] \`wSerial`
381-- Stream.fromList [3,4] \`wSerial` Stream.fromList [5,6]@ would result in
382-- [1,3,2,5,4,6]. In other words, the leftmost stream gets the same scheduling
383-- priority as the rest of the streams taken together. The same is true for
384-- each subexpression on the right.
385--
386{-# INLINE wSerial #-}
387wSerial :: IsStream t => t m a -> t m a -> t m a
388wSerial m1 m2 = mkStream $ \st yld sng stp -> do
389 let stop = foldStream st yld sng stp m2
390 single a = yld a m2
391 yieldk a r = yld a (wSerial m2 r)
392 foldStream st yieldk single stop m1
394-- | Like `wSerial` but stops interleaving as soon as the first stream stops.
395--
396-- @since 0.7.0
397{-# INLINE wSerialFst #-}
398wSerialFst :: IsStream t => t m a -> t m a -> t m a
399wSerialFst m1 m2 = mkStream $ \st yld sng stp -> do
400 let yieldFirst a r = yld a (yieldSecond r m2)
401 in foldStream st yieldFirst sng stp m1
403 where
405 yieldSecond s1 s2 = mkStream $ \st yld sng stp -> do
406 let stop = foldStream st yld sng stp s1
407 single a = yld a s1
408 yieldk a r = yld a (wSerial s1 r)
409 in foldStream st yieldk single stop s2
411-- | Like `wSerial` but stops interleaving as soon as any of the two streams
412-- stops.
413--
414-- @since 0.7.0
415{-# INLINE wSerialMin #-}
416wSerialMin :: IsStream t => t m a -> t m a -> t m a
417wSerialMin m1 m2 = mkStream $ \st yld _ stp -> do
418 let stop = stp
419 -- "single a" is defined as "yld a (wSerialMin m2 K.nil)" instead of
420 -- "sng a" to keep the behaviour consistent with the yield continuation.
421 single a = yld a (wSerialMin m2 K.nil)
422 yieldk a r = yld a (wSerialMin m2 r)
423 foldStream st yieldk single stop m1
425instance Semigroup (WSerialT m a) where
426 (<>) = wSerial
428infixr 5 <=>
430-- | Same as 'wSerial'.
431--
432-- @since 0.1.0
433{-# DEPRECATED (<=>) "Please use 'wSerial' instead." #-}
434{-# INLINE (<=>) #-}
435(<=>) :: IsStream t => t m a -> t m a -> t m a
436(<=>) = wSerial
438------------------------------------------------------------------------------
439-- Monoid
440------------------------------------------------------------------------------
442instance Monoid (WSerialT m a) where
443 mempty = K.nil
444 mappend = (<>)
446{-# INLINE apWSerial #-}
447apWSerial :: Monad m => WSerialT m (a -> b) -> WSerialT m a -> WSerialT m b
448apWSerial (WSerialT m1) (WSerialT m2) =
449 let f x1 = K.concatMapBy wSerial (pure . x1) m2
450 in WSerialT $ K.concatMapBy wSerial f m1
452instance Monad m => Applicative (WSerialT m) where
453 {-# INLINE pure #-}
454 pure = WSerialT . K.fromPure
455 {-# INLINE (<*>) #-}
456 (<*>) = apWSerial
458------------------------------------------------------------------------------
459-- Monad
460------------------------------------------------------------------------------
462instance Monad m => Monad (WSerialT m) where
463 return = pure
464 {-# INLINE (>>=) #-}
465 (>>=) = K.bindWith wSerial
467------------------------------------------------------------------------------
468-- Other instances
469------------------------------------------------------------------------------
471MONAD_COMMON_INSTANCES(WSerialT,)
472LIST_INSTANCES(WSerialT)
473NFDATA1_INSTANCE(WSerialT)
474FOLDABLE_INSTANCE(WSerialT)
475TRAVERSABLE_INSTANCE(WSerialT)
477------------------------------------------------------------------------------
478-- Construction
479------------------------------------------------------------------------------
481-- | Build a stream by unfolding a /monadic/ step function starting from a
482-- seed. The step function returns the next element in the stream and the next
483-- seed value. When it is done it returns 'Nothing' and the stream ends. For
484-- example,
485--
486-- @
487-- let f b =
488-- if b > 3
489-- then return Nothing
490-- else print b >> return (Just (b, b + 1))
491-- in drain $ unfoldrM f 0
492-- @
493-- @
494-- 0
495-- 1
496-- 2
497-- 3
498-- @
499--
500-- /Pre-release/
501--
502{-# INLINE unfoldrM #-}
503unfoldrM :: (IsStream t, Monad m) => (b -> m (Maybe (a, b))) -> b -> t m a
504unfoldrM step seed = D.fromStreamD (D.unfoldrM step seed)