git.iko.soy/streamly
on 2ca80e7 help sign in
streamly/src/Streamly/Internal/Data/Stream/WSerial.hs
208 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(..))
41#if __GLASGOW_HASKELL__ < 808
42import Data.Semigroup (Semigroup(..))
43#endif
44import GHC.Exts (IsList(..), IsString(..), oneShot)
45import Text.Read
46 ( Lexeme(Ident), lexP, parens, prec, readPrec, readListPrec
47 , readListPrecDefault)
48import Streamly.Internal.BaseCompat ((#.))
49import Streamly.Internal.Data.Maybe.Strict (Maybe'(..), toMaybe)
50import Streamly.Internal.Data.Stream.StreamK.Type (Stream)
52import qualified Streamly.Internal.Data.Stream.Common as P
53import qualified Streamly.Internal.Data.Stream.StreamD as D
54 (fromStreamK, toStreamK, mapM)
55import qualified Streamly.Internal.Data.Stream.StreamK.Type as K
56 (cons, consM, nil, concatMapWith, fromPure, bindWith
57 , withLocal, interleave, interleaveFst, interleaveMin)
59import Prelude hiding (map, mapM, repeat, filter)
61#include "Instances.hs"
62#include "inline.hs"
64-- $setup
65-- >>> import qualified Streamly.Prelude as Stream
67------------------------------------------------------------------------------
68-- WSerialT
69------------------------------------------------------------------------------
71-- | For 'WSerialT' streams:
72--
73-- @
74-- (<>) = 'Streamly.Prelude.wSerial' -- 'Semigroup'
75-- (>>=) = flip . 'Streamly.Prelude.concatMapWith' 'Streamly.Prelude.wSerial' -- 'Monad'
76-- @
77--
78-- Note that '<>' is associative only if we disregard the ordering of elements
79-- in the resulting stream.
80--
81-- A single 'Monad' bind behaves like a @for@ loop:
82--
83-- >>> :{
84-- Stream.toList $ Stream.fromWSerial $ do
85-- x <- Stream.fromList [1,2] -- foreach x in stream
86-- return x
87-- :}
88-- [1,2]
89--
90-- Nested monad binds behave like interleaved nested @for@ loops:
91--
92-- >>> :{
93-- Stream.toList $ Stream.fromWSerial $ do
94-- x <- Stream.fromList [1,2] -- foreach x in stream
95-- y <- Stream.fromList [3,4] -- foreach y in stream
96-- return (x, y)
97-- :}
98-- [(1,3),(2,3),(1,4),(2,4)]
99--
100-- It is a result of interleaving all the nested iterations corresponding to
101-- element @1@ in the first stream with all the nested iterations of element
102-- @2@:
103--
104-- >>> import Streamly.Prelude (wSerial)
105-- >>> Stream.toList $ Stream.fromList [(1,3),(1,4)] `Stream.wSerial` Stream.fromList [(2,3),(2,4)]
106-- [(1,3),(2,3),(1,4),(2,4)]
107--
108-- The @W@ in the name stands for @wide@ or breadth wise scheduling in
109-- contrast to the depth wise scheduling behavior of 'Stream'.
110--
111-- /Since: 0.2.0 ("Streamly")/
112--
113-- @since 0.8.0
114newtype WSerialT m a = WSerialT {getWSerialT :: Stream m a}
115 deriving (MonadTrans)
117-- | An interleaving serial IO stream of elements of type @a@. See 'WSerialT'
118-- documentation for more details.
119--
120-- /Since: 0.2.0 ("Streamly")/
121--
122-- @since 0.8.0
123type WSerial = WSerialT IO
125{-# INLINE consMWSerial #-}
126{-# SPECIALIZE consMWSerial :: IO a -> WSerialT IO a -> WSerialT IO a #-}
127consMWSerial :: Monad m => m a -> WSerialT m a -> WSerialT m a
128consMWSerial m (WSerialT ms) = WSerialT $ K.consM m ms
130------------------------------------------------------------------------------
131-- Semigroup
132------------------------------------------------------------------------------
134infixr 6 `wSerial`
136-- | Interleaves two streams, yielding one element from each stream
137-- alternately. When one stream stops the rest of the other stream is used in
138-- the output stream.
139--
140-- This gives exponential priority to earlier streams than the ones joining
141-- later. Because of exponential weighting it can be used with 'concatMapWith'.
142--
143-- /Not fused/
145-- NOTE:
146--
147-- Note that evaluation of @a \`wSerial` b \`wSerial` c@ does not interleave
148-- @a@, @b@ and @c@ with equal priority. This expression is equivalent to @a
149-- \`wSerial` (b \`wSerial` c)@, therefore, it fairly interleaves @a@ with the
150-- result of @b \`wSerial` c@. For example, @Stream.fromList [1,2] \`wSerial`
151-- Stream.fromList [3,4] \`wSerial` Stream.fromList [5,6]@ would result in
152-- [1,3,2,5,4,6]. In other words, the leftmost stream gets the same scheduling
153-- priority as the rest of the streams taken together. The same is true for
154-- each subexpression on the right.
155--
156{-# INLINE wSerial #-}
157wSerial :: WSerialT m a -> WSerialT m a -> WSerialT m a
158wSerial (WSerialT m1) (WSerialT m2) = WSerialT $ K.interleave m1 m2
160{-# INLINE wSerialFst #-}
161wSerialFst :: WSerialT m a -> WSerialT m a -> WSerialT m a
162wSerialFst (WSerialT m1) (WSerialT m2) = WSerialT $ K.interleaveFst m1 m2
164{-# INLINE wSerialMin #-}
165wSerialMin :: WSerialT m a -> WSerialT m a -> WSerialT m a
166wSerialMin (WSerialT m1) (WSerialT m2) = WSerialT $ K.interleaveMin m1 m2
168instance Semigroup (WSerialT m a) where
169 (<>) = wSerial
171------------------------------------------------------------------------------
172-- Monoid
173------------------------------------------------------------------------------
175instance Monoid (WSerialT m a) where
176 mempty = WSerialT K.nil
177 mappend = (<>)
179{-# INLINE apWSerial #-}
180apWSerial :: Monad m => WSerialT m (a -> b) -> WSerialT m a -> WSerialT m b
181apWSerial (WSerialT m1) (WSerialT m2) =
182 let f x1 = K.concatMapWith K.interleave (pure . x1) m2
183 in WSerialT $ K.concatMapWith K.interleave f m1
185instance Monad m => Applicative (WSerialT m) where
186 {-# INLINE pure #-}
187 pure = WSerialT . K.fromPure
188 {-# INLINE (<*>) #-}
189 (<*>) = apWSerial
191------------------------------------------------------------------------------
192-- Monad
193------------------------------------------------------------------------------
195instance Monad m => Monad (WSerialT m) where
196 return = pure
197 {-# INLINE (>>=) #-}
198 (>>=) (WSerialT m) f = WSerialT $ K.bindWith K.interleave m (getWSerialT . f)
200------------------------------------------------------------------------------
201-- Other instances
202------------------------------------------------------------------------------
204MONAD_COMMON_INSTANCES(WSerialT,)
205LIST_INSTANCES(WSerialT)
206NFDATA1_INSTANCE(WSerialT)
207FOLDABLE_INSTANCE(WSerialT)
208TRAVERSABLE_INSTANCE(WSerialT)