git.iko.soy/streamly
on 0945bc3 help sign in
streamly/src/Streamly/Internal/Data/Stream/WSerial.hs
205 lines 7.0 kB history raw
1{-# LANGUAGE UndecidableInstances #-}
3-- |
4-- Module : Streamly.Internal.Data.Stream.WSerial
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.WSerial
17 (
18 -- * Serial interleaving stream
19 WSerialT(..)
20 , WSerial
21 , wSerial
22 , wSerialFst
23 , wSerialMin
24 , consMWSerial
25 )
26where
28import Control.Applicative (liftA2)
29import Control.DeepSeq (NFData(..))
30import Control.DeepSeq (NFData1(..))
31import Control.Monad.Base (MonadBase(..), liftBaseDefault)
32import Control.Monad.Catch (MonadThrow, throwM)
33import Control.Monad.IO.Class (MonadIO(..))
34import Control.Monad.Reader.Class (MonadReader(..))
35import Control.Monad.State.Class (MonadState(..))
36import Control.Monad.Trans.Class (MonadTrans(lift))
37import Data.Foldable (Foldable(foldl'), fold)
38import Data.Functor.Identity (Identity(..), runIdentity)
39import Data.Maybe (fromMaybe)
40import Data.Semigroup (Endo(..))
41import GHC.Exts (IsList(..), IsString(..), oneShot)
42import Text.Read
43 ( Lexeme(Ident), lexP, parens, prec, readPrec, readListPrec
44 , readListPrecDefault)
45import Streamly.Internal.BaseCompat ((#.))
46import Streamly.Internal.Data.Maybe.Strict (Maybe'(..), toMaybe)
47import Streamly.Internal.Data.Stream.StreamK.Type (Stream)
49import qualified Streamly.Internal.Data.Stream.Common as P
50import qualified Streamly.Internal.Data.Stream.StreamD as D
51 (fromStreamK, toStreamK, mapM)
52import qualified Streamly.Internal.Data.Stream.StreamK.Type as K
53 (cons, consM, nil, concatMapWith, fromPure, bindWith
54 , withLocal, interleave, interleaveFst, interleaveMin)
56import Prelude hiding (map, mapM, repeat, filter)
58#include "Instances.hs"
59#include "inline.hs"
61-- $setup
62-- >>> import qualified Streamly.Prelude as Stream
64------------------------------------------------------------------------------
65-- WSerialT
66------------------------------------------------------------------------------
68-- | For 'WSerialT' streams:
69--
70-- @
71-- (<>) = 'Streamly.Prelude.wSerial' -- 'Semigroup'
72-- (>>=) = flip . 'Streamly.Prelude.concatMapWith' 'Streamly.Prelude.wSerial' -- 'Monad'
73-- @
74--
75-- Note that '<>' is associative only if we disregard the ordering of elements
76-- in the resulting stream.
77--
78-- A single 'Monad' bind behaves like a @for@ loop:
79--
80-- >>> :{
81-- Stream.toList $ Stream.fromWSerial $ do
82-- x <- Stream.fromList [1,2] -- foreach x in stream
83-- return x
84-- :}
85-- [1,2]
86--
87-- Nested monad binds behave like interleaved nested @for@ loops:
88--
89-- >>> :{
90-- Stream.toList $ Stream.fromWSerial $ do
91-- x <- Stream.fromList [1,2] -- foreach x in stream
92-- y <- Stream.fromList [3,4] -- foreach y in stream
93-- return (x, y)
94-- :}
95-- [(1,3),(2,3),(1,4),(2,4)]
96--
97-- It is a result of interleaving all the nested iterations corresponding to
98-- element @1@ in the first stream with all the nested iterations of element
99-- @2@:
100--
101-- >>> import Streamly.Prelude (wSerial)
102-- >>> Stream.toList $ Stream.fromList [(1,3),(1,4)] `Stream.wSerial` Stream.fromList [(2,3),(2,4)]
103-- [(1,3),(2,3),(1,4),(2,4)]
104--
105-- The @W@ in the name stands for @wide@ or breadth wise scheduling in
106-- contrast to the depth wise scheduling behavior of 'Stream'.
107--
108-- /Since: 0.2.0 ("Streamly")/
109--
110-- @since 0.8.0
111newtype WSerialT m a = WSerialT {getWSerialT :: Stream m a}
112 deriving (MonadTrans)
114-- | An interleaving serial IO stream of elements of type @a@. See 'WSerialT'
115-- documentation for more details.
116--
117-- /Since: 0.2.0 ("Streamly")/
118--
119-- @since 0.8.0
120type WSerial = WSerialT IO
122{-# INLINE consMWSerial #-}
123{-# SPECIALIZE consMWSerial :: IO a -> WSerialT IO a -> WSerialT IO a #-}
124consMWSerial :: Monad m => m a -> WSerialT m a -> WSerialT m a
125consMWSerial m (WSerialT ms) = WSerialT $ K.consM m ms
127------------------------------------------------------------------------------
128-- Semigroup
129------------------------------------------------------------------------------
131infixr 6 `wSerial`
133-- | Interleaves two streams, yielding one element from each stream
134-- alternately. When one stream stops the rest of the other stream is used in
135-- the output stream.
136--
137-- This gives exponential priority to earlier streams than the ones joining
138-- later. Because of exponential weighting it can be used with 'concatMapWith'.
139--
140-- /Not fused/
142-- NOTE:
143--
144-- Note that evaluation of @a \`wSerial` b \`wSerial` c@ does not interleave
145-- @a@, @b@ and @c@ with equal priority. This expression is equivalent to @a
146-- \`wSerial` (b \`wSerial` c)@, therefore, it fairly interleaves @a@ with the
147-- result of @b \`wSerial` c@. For example, @Stream.fromList [1,2] \`wSerial`
148-- Stream.fromList [3,4] \`wSerial` Stream.fromList [5,6]@ would result in
149-- [1,3,2,5,4,6]. In other words, the leftmost stream gets the same scheduling
150-- priority as the rest of the streams taken together. The same is true for
151-- each subexpression on the right.
152--
153{-# INLINE wSerial #-}
154wSerial :: WSerialT m a -> WSerialT m a -> WSerialT m a
155wSerial (WSerialT m1) (WSerialT m2) = WSerialT $ K.interleave m1 m2
157{-# INLINE wSerialFst #-}
158wSerialFst :: WSerialT m a -> WSerialT m a -> WSerialT m a
159wSerialFst (WSerialT m1) (WSerialT m2) = WSerialT $ K.interleaveFst m1 m2
161{-# INLINE wSerialMin #-}
162wSerialMin :: WSerialT m a -> WSerialT m a -> WSerialT m a
163wSerialMin (WSerialT m1) (WSerialT m2) = WSerialT $ K.interleaveMin m1 m2
165instance Semigroup (WSerialT m a) where
166 (<>) = wSerial
168------------------------------------------------------------------------------
169-- Monoid
170------------------------------------------------------------------------------
172instance Monoid (WSerialT m a) where
173 mempty = WSerialT K.nil
174 mappend = (<>)
176{-# INLINE apWSerial #-}
177apWSerial :: Monad m => WSerialT m (a -> b) -> WSerialT m a -> WSerialT m b
178apWSerial (WSerialT m1) (WSerialT m2) =
179 let f x1 = K.concatMapWith K.interleave (pure . x1) m2
180 in WSerialT $ K.concatMapWith K.interleave f m1
182instance Monad m => Applicative (WSerialT m) where
183 {-# INLINE pure #-}
184 pure = WSerialT . K.fromPure
185 {-# INLINE (<*>) #-}
186 (<*>) = apWSerial
188------------------------------------------------------------------------------
189-- Monad
190------------------------------------------------------------------------------
192instance Monad m => Monad (WSerialT m) where
193 return = pure
194 {-# INLINE (>>=) #-}
195 (>>=) (WSerialT m) f = WSerialT $ K.bindWith K.interleave m (getWSerialT . f)
197------------------------------------------------------------------------------
198-- Other instances
199------------------------------------------------------------------------------
201MONAD_COMMON_INSTANCES(WSerialT,)
202LIST_INSTANCES(WSerialT)
203NFDATA1_INSTANCE(WSerialT)
204FOLDABLE_INSTANCE(WSerialT)
205TRAVERSABLE_INSTANCE(WSerialT)