git.iko.soy/streamly
on d4e6ca6 help sign in
streamly/src/Streamly/Internal/Data/Stream/Serial.hs
478 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 , serial
23 -- * Serial interleaving stream
24 , WSerialT(..)
25 , WSerial
26 , wSerialK
27 , wSerial
28 , wSerialFst
29 , wSerialMin
30 , consMWSerial
32 -- * Construction
33 , cons
34 , consM
35 , repeat
36 , unfoldrM
37 , fromList
39 -- * Elimination
40 , toList
42 -- * Transformation
43 , map
44 , mapM
46 -- * Deprecated
47 , StreamT
48 , InterleavedT
49 )
50where
52import Control.Applicative (liftA2)
53import Control.DeepSeq (NFData(..))
54#if MIN_VERSION_deepseq(1,4,3)
55import Control.DeepSeq (NFData1(..))
56#endif
57import Control.Monad.Base (MonadBase(..), liftBaseDefault)
58import Control.Monad.Catch (MonadThrow, throwM)
59import Control.Monad.IO.Class (MonadIO(..))
60import Control.Monad.Reader.Class (MonadReader(..))
61import Control.Monad.State.Class (MonadState(..))
62import Control.Monad.Trans.Class (MonadTrans(lift))
63import Data.Foldable (Foldable(foldl'), fold)
64import Data.Functor.Identity (Identity(..), runIdentity)
65import Data.Maybe (fromMaybe)
66import Data.Semigroup (Endo(..))
67#if __GLASGOW_HASKELL__ < 808
68import Data.Semigroup (Semigroup(..))
69#endif
70import GHC.Exts (IsList(..), IsString(..))
71import Text.Read
72 ( Lexeme(Ident), lexP, parens, prec, readPrec, readListPrec
73 , readListPrecDefault)
74import Streamly.Internal.BaseCompat ((#.), errorWithoutStackTrace, oneShot)
75import Streamly.Internal.Data.Maybe.Strict (Maybe'(..), toMaybe)
76import Streamly.Internal.Data.Stream.StreamK.Type
77 (Stream, mkStream, foldStream)
79import qualified Streamly.Internal.Data.Stream.Prelude as P
80import qualified Streamly.Internal.Data.Stream.StreamD.Generate as D
81import qualified Streamly.Internal.Data.Stream.StreamD.Type as D
82import qualified Streamly.Internal.Data.Stream.StreamK.Type as K
84import Prelude hiding (map, mapM, errorWithoutStackTrace, repeat)
86#include "Instances.hs"
87#include "inline.hs"
89-- $setup
90-- >>> import qualified Streamly.Prelude as Stream
92------------------------------------------------------------------------------
93-- SerialT
94------------------------------------------------------------------------------
96-- | For 'SerialT' streams:
97--
98-- @
99-- (<>) = 'Streamly.Prelude.serial' -- 'Semigroup'
100-- (>>=) = flip . 'Streamly.Prelude.concatMapWith' 'Streamly.Prelude.serial' -- 'Monad'
101-- @
102--
103-- A single 'Monad' bind behaves like a @for@ loop:
104--
105-- >>> :{
106-- Stream.toList $ do
107-- x <- Stream.fromList [1,2] -- foreach x in stream
108-- return x
109-- :}
110-- [1,2]
111--
112-- Nested monad binds behave like nested @for@ loops:
113--
114-- >>> :{
115-- Stream.toList $ do
116-- x <- Stream.fromList [1,2] -- foreach x in stream
117-- y <- Stream.fromList [3,4] -- foreach y in stream
118-- return (x, y)
119-- :}
120-- [(1,3),(1,4),(2,3),(2,4)]
121--
122-- /Since: 0.2.0 ("Streamly")/
123--
124-- @since 0.8.0
125newtype SerialT m a = SerialT {getSerialT :: Stream m a}
126 -- XXX when deriving do we inherit an INLINE?
127 deriving (Semigroup, Monoid, MonadTrans)
129-- | A serial IO stream of elements of type @a@. See 'SerialT' documentation
130-- for more details.
131--
132-- /Since: 0.2.0 ("Streamly")/
133--
134-- @since 0.8.0
135type Serial = SerialT IO
137-- |
138-- @since 0.1.0
139{-# DEPRECATED StreamT "Please use 'SerialT' instead." #-}
140type StreamT = SerialT
142------------------------------------------------------------------------------
143-- Generation
144------------------------------------------------------------------------------
146{-# INLINE cons #-}
147cons :: a -> SerialT m a -> SerialT m a
148cons x (SerialT ms) = SerialT $ K.cons x ms
150{-# INLINE consM #-}
151{-# SPECIALIZE consM :: IO a -> SerialT IO a -> SerialT IO a #-}
152consM :: Monad m => m a -> SerialT m a -> SerialT m a
153consM m (SerialT ms) = SerialT $ K.consM m ms
155-- |
156-- Generate an infinite stream by repeating a pure value.
157--
158{-# INLINE_NORMAL repeat #-}
159repeat :: Monad m => a -> SerialT m a
160repeat = SerialT . D.toStreamK . D.repeat
162------------------------------------------------------------------------------
163-- Combining
164------------------------------------------------------------------------------
166{-# INLINE serial #-}
167serial :: SerialT m a -> SerialT m a -> SerialT m a
168serial = (<>)
170------------------------------------------------------------------------------
171-- Monad
172------------------------------------------------------------------------------
174instance Monad m => Monad (SerialT m) where
175 return = pure
177 -- Benchmarks better with StreamD bind and pure:
178 -- toList, filterAllout, *>, *<, >> (~2x)
179 --
180 -- pure = SerialT . D.fromStreamD . D.fromPure
181 -- m >>= f = D.fromStreamD $ D.concatMap (D.toStreamD . f) (D.toStreamD m)
183 -- Benchmarks better with CPS bind and pure:
184 -- Prime sieve (25x)
185 -- n binds, breakAfterSome, filterAllIn, state transformer (~2x)
186 --
187 {-# INLINE (>>=) #-}
188 (>>=) (SerialT m) f = SerialT $ K.bindWith K.serial m (getSerialT . f)
190 {-# INLINE (>>) #-}
191 (>>) = (*>)
193------------------------------------------------------------------------------
194-- Other instances
195------------------------------------------------------------------------------
197{-# INLINE mapM #-}
198mapM :: Monad m => (a -> m b) -> SerialT m a -> SerialT m b
199mapM f (SerialT m) = SerialT $ D.toStreamK $ D.mapM f $ D.fromStreamK m
201-- |
202-- @
203-- map = fmap
204-- @
205--
206-- Same as 'fmap'.
207--
208-- @
209-- > S.toList $ S.map (+1) $ S.fromList [1,2,3]
210-- [2,3,4]
211-- @
212--
213-- @since 0.4.0
214{-# INLINE map #-}
215map :: Monad m => (a -> b) -> SerialT m a -> SerialT m b
216map f = mapM (return . f)
218{-# INLINE apSerial #-}
219apSerial :: Monad m => SerialT m (a -> b) -> SerialT m a -> SerialT m b
220apSerial (SerialT m1) (SerialT m2) =
221 SerialT $ D.toStreamK $ D.fromStreamK m1 <*> D.fromStreamK m2
223{-# INLINE apSequence #-}
224apSequence :: Monad m => SerialT m a -> SerialT m b -> SerialT m b
225apSequence (SerialT m1) (SerialT m2) =
226 SerialT $ D.toStreamK $ D.fromStreamK m1 *> D.fromStreamK m2
228{-# INLINE apDiscardSnd #-}
229apDiscardSnd :: Monad m => SerialT m a -> SerialT m b -> SerialT m a
230apDiscardSnd (SerialT m1) (SerialT m2) =
231 SerialT $ D.toStreamK $ D.fromStreamK m1 <* D.fromStreamK m2
233-- Note: we need to define all the typeclass operations because we want to
234-- INLINE them.
235instance Monad m => Applicative (SerialT m) where
236 {-# INLINE pure #-}
237 pure = SerialT . K.fromPure
239 {-# INLINE (<*>) #-}
240 (<*>) = apSerial
241 -- (<*>) = K.apSerial
243#if MIN_VERSION_base(4,10,0)
244 {-# INLINE liftA2 #-}
245 liftA2 f x = (<*>) (fmap f x)
246#endif
248 {-# INLINE (*>) #-}
249 (*>) = apSequence
250 -- (*>) = K.apSerialDiscardFst
252 {-# INLINE (<*) #-}
253 (<*) = apDiscardSnd
254 -- (<*) = K.apSerialDiscardSnd
256MONAD_COMMON_INSTANCES(SerialT,)
257LIST_INSTANCES(SerialT)
258NFDATA1_INSTANCE(SerialT)
259FOLDABLE_INSTANCE(SerialT)
260TRAVERSABLE_INSTANCE(SerialT)
262------------------------------------------------------------------------------
263-- WSerialT
264------------------------------------------------------------------------------
266-- | For 'WSerialT' streams:
267--
268-- @
269-- (<>) = 'Streamly.Prelude.wSerial' -- 'Semigroup'
270-- (>>=) = flip . 'Streamly.Prelude.concatMapWith' 'Streamly.Prelude.wSerial' -- 'Monad'
271-- @
272--
273-- Note that '<>' is associative only if we disregard the ordering of elements
274-- in the resulting stream.
275--
276-- A single 'Monad' bind behaves like a @for@ loop:
277--
278-- >>> :{
279-- Stream.toList $ Stream.fromWSerial $ do
280-- x <- Stream.fromList [1,2] -- foreach x in stream
281-- return x
282-- :}
283-- [1,2]
284--
285-- Nested monad binds behave like interleaved nested @for@ loops:
286--
287-- >>> :{
288-- Stream.toList $ Stream.fromWSerial $ do
289-- x <- Stream.fromList [1,2] -- foreach x in stream
290-- y <- Stream.fromList [3,4] -- foreach y in stream
291-- return (x, y)
292-- :}
293-- [(1,3),(2,3),(1,4),(2,4)]
294--
295-- It is a result of interleaving all the nested iterations corresponding to
296-- element @1@ in the first stream with all the nested iterations of element
297-- @2@:
298--
299-- >>> import Streamly.Prelude (wSerial)
300-- >>> Stream.toList $ Stream.fromList [(1,3),(1,4)] `Stream.wSerial` Stream.fromList [(2,3),(2,4)]
301-- [(1,3),(2,3),(1,4),(2,4)]
302--
303-- The @W@ in the name stands for @wide@ or breadth wise scheduling in
304-- contrast to the depth wise scheduling behavior of 'SerialT'.
305--
306-- /Since: 0.2.0 ("Streamly")/
307--
308-- @since 0.8.0
309newtype WSerialT m a = WSerialT {getWSerialT :: Stream m a}
310 deriving (MonadTrans)
312-- | An interleaving serial IO stream of elements of type @a@. See 'WSerialT'
313-- documentation for more details.
314--
315-- /Since: 0.2.0 ("Streamly")/
316--
317-- @since 0.8.0
318type WSerial = WSerialT IO
320-- |
321-- @since 0.1.0
322{-# DEPRECATED InterleavedT "Please use 'WSerialT' instead." #-}
323type InterleavedT = WSerialT
325{-# INLINE consMWSerial #-}
326{-# SPECIALIZE consMWSerial :: IO a -> WSerialT IO a -> WSerialT IO a #-}
327consMWSerial :: Monad m => m a -> WSerialT m a -> WSerialT m a
328consMWSerial m (WSerialT ms) = WSerialT $ K.consM m ms
330------------------------------------------------------------------------------
331-- Semigroup
332------------------------------------------------------------------------------
334infixr 6 `wSerial`
336-- Additionally we can have m elements yield from the first stream and n
337-- elements yielding from the second stream. We can also have time slicing
338-- variants of positional interleaving, e.g. run first stream for m seconds and
339-- run the second stream for n seconds.
340--
341-- Similar combinators can be implemented using WAhead style.
343{-# INLINE wSerialK #-}
344wSerialK :: Stream m a -> Stream m a -> Stream m a
345wSerialK m1 m2 = mkStream $ \st yld sng stp -> do
346 let stop = foldStream st yld sng stp m2
347 single a = yld a m2
348 yieldk a r = yld a (wSerialK m2 r)
349 foldStream st yieldk single stop m1
351-- | Interleaves two streams, yielding one element from each stream
352-- alternately. When one stream stops the rest of the other stream is used in
353-- the output stream.
355-- Scheduling Notes:
356--
357-- Note that evaluation of @a \`wSerial` b \`wSerial` c@ does not interleave
358-- @a@, @b@ and @c@ with equal priority. This expression is equivalent to @a
359-- \`wSerial` (b \`wSerial` c)@, therefore, it fairly interleaves @a@ with the
360-- result of @b \`wSerial` c@. For example, @Stream.fromList [1,2] \`wSerial`
361-- Stream.fromList [3,4] \`wSerial` Stream.fromList [5,6]@ would result in
362-- [1,3,2,5,4,6]. In other words, the leftmost stream gets the same scheduling
363-- priority as the rest of the streams taken together. The same is true for
364-- each subexpression on the right.
365--
366{-# INLINE wSerial #-}
367wSerial :: WSerialT m a -> WSerialT m a -> WSerialT m a
368wSerial (WSerialT m1) (WSerialT m2) = WSerialT $ wSerialK m1 m2
370-- | Like `wSerial` but stops interleaving as soon as the first stream stops.
371--
372-- @since 0.7.0
373{-# INLINE wSerialFstK #-}
374wSerialFstK :: Stream m a -> Stream m a -> Stream m a
375wSerialFstK m1 m2 = mkStream $ \st yld sng stp -> do
376 let yieldFirst a r = yld a (yieldSecond r m2)
377 in foldStream st yieldFirst sng stp m1
379 where
381 yieldSecond s1 s2 = mkStream $ \st yld sng stp -> do
382 let stop = foldStream st yld sng stp s1
383 single a = yld a s1
384 yieldk a r = yld a (wSerialK s1 r)
385 in foldStream st yieldk single stop s2
387{-# INLINE wSerialFst #-}
388wSerialFst :: WSerialT m a -> WSerialT m a -> WSerialT m a
389wSerialFst (WSerialT m1) (WSerialT m2) = WSerialT $ wSerialFstK m1 m2
391-- | Like `wSerial` but stops interleaving as soon as any of the two streams
392-- stops.
393--
394-- @since 0.7.0
395{-# INLINE wSerialMinK #-}
396wSerialMinK :: Stream m a -> Stream m a -> Stream m a
397wSerialMinK m1 m2 = mkStream $ \st yld _ stp -> do
398 let stop = stp
399 -- "single a" is defined as "yld a (wSerialMin m2 K.nil)" instead of
400 -- "sng a" to keep the behaviour consistent with the yield continuation.
401 single a = yld a (wSerialMinK m2 K.nil)
402 yieldk a r = yld a (wSerialMinK m2 r)
403 foldStream st yieldk single stop m1
405{-# INLINE wSerialMin #-}
406wSerialMin :: WSerialT m a -> WSerialT m a -> WSerialT m a
407wSerialMin (WSerialT m1) (WSerialT m2) = WSerialT $ wSerialMinK m1 m2
409instance Semigroup (WSerialT m a) where
410 (<>) = wSerial
412------------------------------------------------------------------------------
413-- Monoid
414------------------------------------------------------------------------------
416instance Monoid (WSerialT m a) where
417 mempty = WSerialT K.nil
418 mappend = (<>)
420{-# INLINE apWSerial #-}
421apWSerial :: Monad m => WSerialT m (a -> b) -> WSerialT m a -> WSerialT m b
422apWSerial (WSerialT m1) (WSerialT m2) =
423 let f x1 = K.concatMapWith wSerialK (pure . x1) m2
424 in WSerialT $ K.concatMapWith wSerialK f m1
426instance Monad m => Applicative (WSerialT m) where
427 {-# INLINE pure #-}
428 pure = WSerialT . K.fromPure
429 {-# INLINE (<*>) #-}
430 (<*>) = apWSerial
432------------------------------------------------------------------------------
433-- Monad
434------------------------------------------------------------------------------
436instance Monad m => Monad (WSerialT m) where
437 return = pure
438 {-# INLINE (>>=) #-}
439 (>>=) (WSerialT m) f = WSerialT $ K.bindWith wSerialK m (getWSerialT . f)
441------------------------------------------------------------------------------
442-- Other instances
443------------------------------------------------------------------------------
445MONAD_COMMON_INSTANCES(WSerialT,)
446LIST_INSTANCES(WSerialT)
447NFDATA1_INSTANCE(WSerialT)
448FOLDABLE_INSTANCE(WSerialT)
449TRAVERSABLE_INSTANCE(WSerialT)
451------------------------------------------------------------------------------
452-- Construction
453------------------------------------------------------------------------------
455-- | Build a stream by unfolding a /monadic/ step function starting from a
456-- seed. The step function returns the next element in the stream and the next
457-- seed value. When it is done it returns 'Nothing' and the stream ends. For
458-- example,
459--
460-- @
461-- let f b =
462-- if b > 3
463-- then return Nothing
464-- else print b >> return (Just (b, b + 1))
465-- in drain $ unfoldrM f 0
466-- @
467-- @
468-- 0
469-- 1
470-- 2
471-- 3
472-- @
473--
474-- /Pre-release/
475--
476{-# INLINE unfoldrM #-}
477unfoldrM :: Monad m => (b -> m (Maybe (a, b))) -> b -> SerialT m a
478unfoldrM step seed = SerialT $ D.toStreamK (D.unfoldrM step seed)