git.iko.soy/streamly
on 7bf60ff help sign in
streamly/src/Streamly/Internal/Data/Stream/Serial.hs
507 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, oneShot)
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-- $setup
87-- >>> import qualified Streamly.Prelude as Stream
89------------------------------------------------------------------------------
90-- SerialT
91------------------------------------------------------------------------------
93-- | For 'SerialT' streams:
94--
95-- @
96-- (<>) = 'Streamly.Prelude.serial' -- 'Semigroup'
97-- (>>=) = flip . 'Streamly.Prelude.concatMapWith' 'Streamly.Prelude.serial' -- 'Monad'
98-- @
99--
100-- A single 'Monad' bind behaves like a @for@ loop:
101--
102-- >>> :{
103-- Stream.toList $ do
104-- x <- Stream.fromList [1,2] -- foreach x in stream
105-- return x
106-- :}
107-- [1,2]
108--
109-- Nested monad binds behave like nested @for@ loops:
110--
111-- >>> :{
112-- Stream.toList $ do
113-- x <- Stream.fromList [1,2] -- foreach x in stream
114-- y <- Stream.fromList [3,4] -- foreach y in stream
115-- return (x, y)
116-- :}
117-- [(1,3),(1,4),(2,3),(2,4)]
118--
119-- /Since: 0.2.0 ("Streamly")/
120--
121-- @since 0.8.0
122newtype SerialT m a = SerialT {getSerialT :: Stream m a}
123 -- XXX when deriving do we inherit an INLINE?
124 deriving (Semigroup, Monoid, MonadTrans)
126-- | A serial IO stream of elements of type @a@. See 'SerialT' documentation
127-- for more details.
128--
129-- /Since: 0.2.0 ("Streamly")/
130--
131-- @since 0.8.0
132type Serial = SerialT IO
134-- |
135-- @since 0.1.0
136{-# DEPRECATED StreamT "Please use 'SerialT' instead." #-}
137type StreamT = SerialT
139-- | Fix the type of a polymorphic stream as 'SerialT'.
140--
141-- /Since: 0.1.0 ("Streamly")/
142--
143-- @since 0.8.0
144fromSerial :: IsStream t => SerialT m a -> t m a
145fromSerial = adapt
147{-# INLINE consMSerial #-}
148{-# SPECIALIZE consMSerial :: IO a -> SerialT IO a -> SerialT IO a #-}
149consMSerial :: Monad m => m a -> SerialT m a -> SerialT m a
150consMSerial m ms = fromStream $ K.consMStream m (toStream ms)
152instance IsStream SerialT where
153 toStream = getSerialT
154 fromStream = SerialT
155 consM = consMSerial
156 (|:) = consMSerial
158------------------------------------------------------------------------------
159-- Monad
160------------------------------------------------------------------------------
162instance Monad m => Monad (SerialT m) where
163 return = pure
165 -- Benchmarks better with StreamD bind and pure:
166 -- toList, filterAllout, *>, *<, >> (~2x)
167 --
168 -- pure = SerialT . D.fromStreamD . D.fromPure
169 -- m >>= f = D.fromStreamD $ D.concatMap (D.toStreamD . f) (D.toStreamD m)
171 -- Benchmarks better with CPS bind and pure:
172 -- Prime sieve (25x)
173 -- n binds, breakAfterSome, filterAllIn, state transformer (~2x)
174 --
175 {-# INLINE (>>=) #-}
176 (>>=) = K.bindWith K.serial
178 {-# INLINE (>>) #-}
179 (>>) = (*>)
181------------------------------------------------------------------------------
182-- Other instances
183------------------------------------------------------------------------------
185{-# INLINE mapM #-}
186mapM :: (IsStream t, Monad m) => (a -> m b) -> t m a -> t m b
187mapM f m = D.fromStreamD $ D.mapM f $ D.toStreamD m
189-- |
190-- @
191-- map = fmap
192-- @
193--
194-- Same as 'fmap'.
195--
196-- @
197-- > S.toList $ S.map (+1) $ S.fromList [1,2,3]
198-- [2,3,4]
199-- @
200--
201-- @since 0.4.0
202{-# INLINE map #-}
203map :: (IsStream t, Monad m) => (a -> b) -> t m a -> t m b
204map f = mapM (return . f)
206{-# INLINE apSerial #-}
207apSerial :: Monad m => SerialT m (a -> b) -> SerialT m a -> SerialT m b
208apSerial (SerialT m1) (SerialT m2) =
209 D.fromStreamD $ D.toStreamD m1 <*> D.toStreamD m2
211{-# INLINE apSequence #-}
212apSequence :: Monad m => SerialT m a -> SerialT m b -> SerialT m b
213apSequence (SerialT m1) (SerialT m2) =
214 D.fromStreamD $ D.toStreamD m1 *> D.toStreamD m2
216{-# INLINE apDiscardSnd #-}
217apDiscardSnd :: Monad m => SerialT m a -> SerialT m b -> SerialT m a
218apDiscardSnd (SerialT m1) (SerialT m2) =
219 D.fromStreamD $ D.toStreamD m1 <* D.toStreamD m2
221-- Note: we need to define all the typeclass operations because we want to
222-- INLINE them.
223instance Monad m => Applicative (SerialT m) where
224 {-# INLINE pure #-}
225 pure = SerialT . K.fromPure
227 {-# INLINE (<*>) #-}
228 (<*>) = apSerial
229 -- (<*>) = K.apSerial
231#if MIN_VERSION_base(4,10,0)
232 {-# INLINE liftA2 #-}
233 liftA2 f x = (<*>) (fmap f x)
234#endif
236 {-# INLINE (*>) #-}
237 (*>) = apSequence
238 -- (*>) = K.apSerialDiscardFst
240 {-# INLINE (<*) #-}
241 (<*) = apDiscardSnd
242 -- (<*) = K.apSerialDiscardSnd
244MONAD_COMMON_INSTANCES(SerialT,)
245LIST_INSTANCES(SerialT)
246NFDATA1_INSTANCE(SerialT)
247FOLDABLE_INSTANCE(SerialT)
248TRAVERSABLE_INSTANCE(SerialT)
250------------------------------------------------------------------------------
251-- WSerialT
252------------------------------------------------------------------------------
254-- | For 'WSerialT' streams:
255--
256-- @
257-- (<>) = 'Streamly.Prelude.wSerial' -- 'Semigroup'
258-- (>>=) = flip . 'Streamly.Prelude.concatMapWith' 'Streamly.Prelude.wSerial' -- 'Monad'
259-- @
260--
261-- Note that '<>' is associative only if we disregard the ordering of elements
262-- in the resulting stream.
263--
264-- A single 'Monad' bind behaves like a @for@ loop:
265--
266-- >>> :{
267-- Stream.toList $ Stream.fromWSerial $ do
268-- x <- Stream.fromList [1,2] -- foreach x in stream
269-- return x
270-- :}
271-- [1,2]
272--
273-- Nested monad binds behave like interleaved nested @for@ loops:
274--
275-- >>> :{
276-- Stream.toList $ Stream.fromWSerial $ do
277-- x <- Stream.fromList [1,2] -- foreach x in stream
278-- y <- Stream.fromList [3,4] -- foreach y in stream
279-- return (x, y)
280-- :}
281-- [(1,3),(2,3),(1,4),(2,4)]
282--
283-- It is a result of interleaving all the nested iterations corresponding to
284-- element @1@ in the first stream with all the nested iterations of element
285-- @2@:
286--
287-- >>> import Streamly.Prelude (wSerial)
288-- >>> Stream.toList $ Stream.fromList [(1,3),(1,4)] `wSerial` Stream.fromList [(2,3),(2,4)]
289-- [(1,3),(2,3),(1,4),(2,4)]
290--
291-- The @W@ in the name stands for @wide@ or breadth wise scheduling in
292-- contrast to the depth wise scheduling behavior of 'SerialT'.
293--
294-- /Since: 0.2.0 ("Streamly")/
295--
296-- @since 0.8.0
297newtype WSerialT m a = WSerialT {getWSerialT :: Stream m a}
298 deriving (MonadTrans)
300-- | An interleaving serial IO stream of elements of type @a@. See 'WSerialT'
301-- documentation for more details.
302--
303-- /Since: 0.2.0 ("Streamly")/
304--
305-- @since 0.8.0
306type WSerial = WSerialT IO
308-- |
309-- @since 0.1.0
310{-# DEPRECATED InterleavedT "Please use 'WSerialT' instead." #-}
311type InterleavedT = WSerialT
313-- | Fix the type of a polymorphic stream as 'WSerialT'.
314--
315-- /Since: 0.2.0 ("Streamly")/
316--
317-- @since 0.8.0
318fromWSerial :: IsStream t => WSerialT m a -> t m a
319fromWSerial = adapt
321-- | Same as 'fromWSerial'.
322--
323-- @since 0.1.0
324{-# DEPRECATED interleaving "Please use fromWSerial instead." #-}
325interleaving :: IsStream t => WSerialT m a -> t m a
326interleaving = fromWSerial
328consMWSerial :: Monad m => m a -> WSerialT m a -> WSerialT m a
329consMWSerial m ms = fromStream $ K.consMStream m (toStream ms)
331instance IsStream WSerialT where
332 toStream = getWSerialT
333 fromStream = WSerialT
335 {-# INLINE consM #-}
336 {-# SPECIALIZE consM :: IO a -> WSerialT IO a -> WSerialT IO a #-}
337 consM :: Monad m => m a -> WSerialT m a -> WSerialT m a
338 consM = consMWSerial
340 {-# INLINE (|:) #-}
341 {-# SPECIALIZE (|:) :: IO a -> WSerialT IO a -> WSerialT IO a #-}
342 (|:) :: Monad m => m a -> WSerialT m a -> WSerialT m a
343 (|:) = consMWSerial
345------------------------------------------------------------------------------
346-- Semigroup
347------------------------------------------------------------------------------
349infixr 6 `wSerial`
351-- Additionally we can have m elements yield from the first stream and n
352-- elements yielding from the second stream. We can also have time slicing
353-- variants of positional interleaving, e.g. run first stream for m seconds and
354-- run the second stream for n seconds.
355--
356-- Similar combinators can be implemented using WAhead style.
358-- | Interleaves two streams, yielding one element from each stream
359-- alternately. When one stream stops the rest of the other stream is used in
360-- the output stream.
361--
362-- >>> import Streamly.Prelude (wSerial)
363-- >>> stream1 = Stream.fromList [1,2]
364-- >>> stream2 = Stream.fromList [3,4]
365-- >>> Stream.toList $ Stream.fromWSerial $ stream1 `wSerial` stream2
366-- [1,3,2,4]
367--
368-- Note, for singleton streams 'wSerial' and 'serial' are identical.
369--
370-- Note that this operation cannot be used to fold a container of infinite
371-- streams but it can be used for very large streams as the state that it needs
372-- to maintain is proportional to the logarithm of the number of streams.
373--
374-- @since 0.8.0
375--
376-- /Since: 0.2.0 ("Streamly")/
378-- Scheduling Notes:
379--
380-- Note that evaluation of @a \`wSerial` b \`wSerial` c@ does not interleave
381-- @a@, @b@ and @c@ with equal priority. This expression is equivalent to @a
382-- \`wSerial` (b \`wSerial` c)@, therefore, it fairly interleaves @a@ with the
383-- result of @b \`wSerial` c@. For example, @Stream.fromList [1,2] \`wSerial`
384-- Stream.fromList [3,4] \`wSerial` Stream.fromList [5,6]@ would result in
385-- [1,3,2,5,4,6]. In other words, the leftmost stream gets the same scheduling
386-- priority as the rest of the streams taken together. The same is true for
387-- each subexpression on the right.
388--
389{-# INLINE wSerial #-}
390wSerial :: IsStream t => t m a -> t m a -> t m a
391wSerial m1 m2 = mkStream $ \st yld sng stp -> do
392 let stop = foldStream st yld sng stp m2
393 single a = yld a m2
394 yieldk a r = yld a (wSerial m2 r)
395 foldStream st yieldk single stop m1
397-- | Like `wSerial` but stops interleaving as soon as the first stream stops.
398--
399-- @since 0.7.0
400{-# INLINE wSerialFst #-}
401wSerialFst :: IsStream t => t m a -> t m a -> t m a
402wSerialFst m1 m2 = mkStream $ \st yld sng stp -> do
403 let yieldFirst a r = yld a (yieldSecond r m2)
404 in foldStream st yieldFirst sng stp m1
406 where
408 yieldSecond s1 s2 = mkStream $ \st yld sng stp -> do
409 let stop = foldStream st yld sng stp s1
410 single a = yld a s1
411 yieldk a r = yld a (wSerial s1 r)
412 in foldStream st yieldk single stop s2
414-- | Like `wSerial` but stops interleaving as soon as any of the two streams
415-- stops.
416--
417-- @since 0.7.0
418{-# INLINE wSerialMin #-}
419wSerialMin :: IsStream t => t m a -> t m a -> t m a
420wSerialMin m1 m2 = mkStream $ \st yld _ stp -> do
421 let stop = stp
422 -- "single a" is defined as "yld a (wSerialMin m2 K.nil)" instead of
423 -- "sng a" to keep the behaviour consistent with the yield continuation.
424 single a = yld a (wSerialMin m2 K.nil)
425 yieldk a r = yld a (wSerialMin m2 r)
426 foldStream st yieldk single stop m1
428instance Semigroup (WSerialT m a) where
429 (<>) = wSerial
431infixr 5 <=>
433-- | Same as 'wSerial'.
434--
435-- @since 0.1.0
436{-# DEPRECATED (<=>) "Please use 'wSerial' instead." #-}
437{-# INLINE (<=>) #-}
438(<=>) :: IsStream t => t m a -> t m a -> t m a
439(<=>) = wSerial
441------------------------------------------------------------------------------
442-- Monoid
443------------------------------------------------------------------------------
445instance Monoid (WSerialT m a) where
446 mempty = K.nil
447 mappend = (<>)
449{-# INLINE apWSerial #-}
450apWSerial :: Monad m => WSerialT m (a -> b) -> WSerialT m a -> WSerialT m b
451apWSerial (WSerialT m1) (WSerialT m2) =
452 let f x1 = K.concatMapBy wSerial (pure . x1) m2
453 in WSerialT $ K.concatMapBy wSerial f m1
455instance Monad m => Applicative (WSerialT m) where
456 {-# INLINE pure #-}
457 pure = WSerialT . K.fromPure
458 {-# INLINE (<*>) #-}
459 (<*>) = apWSerial
461------------------------------------------------------------------------------
462-- Monad
463------------------------------------------------------------------------------
465instance Monad m => Monad (WSerialT m) where
466 return = pure
467 {-# INLINE (>>=) #-}
468 (>>=) = K.bindWith wSerial
470------------------------------------------------------------------------------
471-- Other instances
472------------------------------------------------------------------------------
474MONAD_COMMON_INSTANCES(WSerialT,)
475LIST_INSTANCES(WSerialT)
476NFDATA1_INSTANCE(WSerialT)
477FOLDABLE_INSTANCE(WSerialT)
478TRAVERSABLE_INSTANCE(WSerialT)
480------------------------------------------------------------------------------
481-- Construction
482------------------------------------------------------------------------------
484-- | Build a stream by unfolding a /monadic/ step function starting from a
485-- seed. The step function returns the next element in the stream and the next
486-- seed value. When it is done it returns 'Nothing' and the stream ends. For
487-- example,
488--
489-- @
490-- let f b =
491-- if b > 3
492-- then return Nothing
493-- else print b >> return (Just (b, b + 1))
494-- in drain $ unfoldrM f 0
495-- @
496-- @
497-- 0
498-- 1
499-- 2
500-- 3
501-- @
502--
503-- /Pre-release/
504--
505{-# INLINE unfoldrM #-}
506unfoldrM :: (IsStream t, Monad m) => (b -> m (Maybe (a, b))) -> b -> t m a
507unfoldrM step seed = D.fromStreamD (D.unfoldrM step seed)