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