git.iko.soy/streamly
on b0fdda4 help sign in
streamly/src/Streamly/Internal/Data/Stream/Serial.hs
503 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-- >>> Stream.toList $ Stream.fromList [(1,3),(1,4)] `wSerial` Stream.fromList [(2,3),(2,4)]
285-- [(1,3),(2,3),(1,4),(2,4)]
286--
287-- The @W@ in the name stands for @wide@ or breadth wise scheduling in
288-- contrast to the depth wise scheduling behavior of 'SerialT'.
289--
290-- /Since: 0.2.0 ("Streamly")/
291--
292-- @since 0.8.0
293newtype WSerialT m a = WSerialT {getWSerialT :: Stream m a}
294 deriving (MonadTrans)
296-- | An interleaving serial IO stream of elements of type @a@. See 'WSerialT'
297-- documentation for more details.
298--
299-- /Since: 0.2.0 ("Streamly")/
300--
301-- @since 0.8.0
302type WSerial = WSerialT IO
304-- |
305-- @since 0.1.0
306{-# DEPRECATED InterleavedT "Please use 'WSerialT' instead." #-}
307type InterleavedT = WSerialT
309-- | Fix the type of a polymorphic stream as 'WSerialT'.
310--
311-- /Since: 0.2.0 ("Streamly")/
312--
313-- @since 0.8.0
314fromWSerial :: IsStream t => WSerialT m a -> t m a
315fromWSerial = adapt
317-- | Same as 'fromWSerial'.
318--
319-- @since 0.1.0
320{-# DEPRECATED interleaving "Please use fromWSerial instead." #-}
321interleaving :: IsStream t => WSerialT m a -> t m a
322interleaving = fromWSerial
324consMWSerial :: Monad m => m a -> WSerialT m a -> WSerialT m a
325consMWSerial m ms = fromStream $ K.consMStream m (toStream ms)
327instance IsStream WSerialT where
328 toStream = getWSerialT
329 fromStream = WSerialT
331 {-# INLINE consM #-}
332 {-# SPECIALIZE consM :: IO a -> WSerialT IO a -> WSerialT IO a #-}
333 consM :: Monad m => m a -> WSerialT m a -> WSerialT m a
334 consM = consMWSerial
336 {-# INLINE (|:) #-}
337 {-# SPECIALIZE (|:) :: IO a -> WSerialT IO a -> WSerialT IO a #-}
338 (|:) :: Monad m => m a -> WSerialT m a -> WSerialT m a
339 (|:) = consMWSerial
341------------------------------------------------------------------------------
342-- Semigroup
343------------------------------------------------------------------------------
345infixr 6 `wSerial`
347-- Additionally we can have m elements yield from the first stream and n
348-- elements yielding from the second stream. We can also have time slicing
349-- variants of positional interleaving, e.g. run first stream for m seconds and
350-- run the second stream for n seconds.
351--
352-- Similar combinators can be implemented using WAhead style.
354-- | Interleaves two streams, yielding one element from each stream
355-- alternately. When one stream stops the rest of the other stream is used in
356-- the output stream.
357--
358-- >>> import Streamly.Prelude (wSerial)
359-- >>> stream1 = Stream.fromList [1,2]
360-- >>> stream2 = Stream.fromList [3,4]
361-- >>> Stream.toList $ Stream.fromWSerial $ stream1 `wSerial` stream2
362-- [1,3,2,4]
363--
364-- Note, for singleton streams 'wSerial' and 'serial' are identical.
365--
366-- Note that this operation cannot be used to fold a container of infinite
367-- streams but it can be used for very large streams as the state that it needs
368-- to maintain is proportional to the logarithm of the number of streams.
369--
370-- @since 0.8.0
371--
372-- /Since: 0.2.0 ("Streamly")/
374-- Scheduling Notes:
375--
376-- Note that evaluation of @a \`wSerial` b \`wSerial` c@ does not interleave
377-- @a@, @b@ and @c@ with equal priority. This expression is equivalent to @a
378-- \`wSerial` (b \`wSerial` c)@, therefore, it fairly interleaves @a@ with the
379-- result of @b \`wSerial` c@. For example, @Stream.fromList [1,2] \`wSerial`
380-- Stream.fromList [3,4] \`wSerial` Stream.fromList [5,6]@ would result in
381-- [1,3,2,5,4,6]. In other words, the leftmost stream gets the same scheduling
382-- priority as the rest of the streams taken together. The same is true for
383-- each subexpression on the right.
384--
385{-# INLINE wSerial #-}
386wSerial :: IsStream t => t m a -> t m a -> t m a
387wSerial m1 m2 = mkStream $ \st yld sng stp -> do
388 let stop = foldStream st yld sng stp m2
389 single a = yld a m2
390 yieldk a r = yld a (wSerial m2 r)
391 foldStream st yieldk single stop m1
393-- | Like `wSerial` but stops interleaving as soon as the first stream stops.
394--
395-- @since 0.7.0
396{-# INLINE wSerialFst #-}
397wSerialFst :: IsStream t => t m a -> t m a -> t m a
398wSerialFst m1 m2 = mkStream $ \st yld sng stp -> do
399 let yieldFirst a r = yld a (yieldSecond r m2)
400 in foldStream st yieldFirst sng stp m1
402 where
404 yieldSecond s1 s2 = mkStream $ \st yld sng stp -> do
405 let stop = foldStream st yld sng stp s1
406 single a = yld a s1
407 yieldk a r = yld a (wSerial s1 r)
408 in foldStream st yieldk single stop s2
410-- | Like `wSerial` but stops interleaving as soon as any of the two streams
411-- stops.
412--
413-- @since 0.7.0
414{-# INLINE wSerialMin #-}
415wSerialMin :: IsStream t => t m a -> t m a -> t m a
416wSerialMin m1 m2 = mkStream $ \st yld _ stp -> do
417 let stop = stp
418 -- "single a" is defined as "yld a (wSerialMin m2 K.nil)" instead of
419 -- "sng a" to keep the behaviour consistent with the yield continuation.
420 single a = yld a (wSerialMin m2 K.nil)
421 yieldk a r = yld a (wSerialMin m2 r)
422 foldStream st yieldk single stop m1
424instance Semigroup (WSerialT m a) where
425 (<>) = wSerial
427infixr 5 <=>
429-- | Same as 'wSerial'.
430--
431-- @since 0.1.0
432{-# DEPRECATED (<=>) "Please use 'wSerial' instead." #-}
433{-# INLINE (<=>) #-}
434(<=>) :: IsStream t => t m a -> t m a -> t m a
435(<=>) = wSerial
437------------------------------------------------------------------------------
438-- Monoid
439------------------------------------------------------------------------------
441instance Monoid (WSerialT m a) where
442 mempty = K.nil
443 mappend = (<>)
445{-# INLINE apWSerial #-}
446apWSerial :: Monad m => WSerialT m (a -> b) -> WSerialT m a -> WSerialT m b
447apWSerial (WSerialT m1) (WSerialT m2) =
448 let f x1 = K.concatMapBy wSerial (pure . x1) m2
449 in WSerialT $ K.concatMapBy wSerial f m1
451instance Monad m => Applicative (WSerialT m) where
452 {-# INLINE pure #-}
453 pure = WSerialT . K.fromPure
454 {-# INLINE (<*>) #-}
455 (<*>) = apWSerial
457------------------------------------------------------------------------------
458-- Monad
459------------------------------------------------------------------------------
461instance Monad m => Monad (WSerialT m) where
462 return = pure
463 {-# INLINE (>>=) #-}
464 (>>=) = K.bindWith wSerial
466------------------------------------------------------------------------------
467-- Other instances
468------------------------------------------------------------------------------
470MONAD_COMMON_INSTANCES(WSerialT,)
471LIST_INSTANCES(WSerialT)
472NFDATA1_INSTANCE(WSerialT)
473FOLDABLE_INSTANCE(WSerialT)
474TRAVERSABLE_INSTANCE(WSerialT)
476------------------------------------------------------------------------------
477-- Construction
478------------------------------------------------------------------------------
480-- | Build a stream by unfolding a /monadic/ step function starting from a
481-- seed. The step function returns the next element in the stream and the next
482-- seed value. When it is done it returns 'Nothing' and the stream ends. For
483-- example,
484--
485-- @
486-- let f b =
487-- if b > 3
488-- then return Nothing
489-- else print b >> return (Just (b, b + 1))
490-- in drain $ unfoldrM f 0
491-- @
492-- @
493-- 0
494-- 1
495-- 2
496-- 3
497-- @
498--
499-- /Pre-release/
500--
501{-# INLINE unfoldrM #-}
502unfoldrM :: (IsStream t, Monad m) => (b -> m (Maybe (a, b))) -> b -> t m a
503unfoldrM step seed = D.fromStreamD (D.unfoldrM step seed)