754b95b · streamly · git.iko.soy

git.iko.soy/streamly

Update doc for Streamly module

commit754b95b745c37c06f97295f94c75f37fb4d26ac4 authorHarendra Kumar <harendra.kumar@gmail.com> dateWed 22 Nov 2017, 00:09 (9 years ago) parentca79a5f

3 files changed+181−135

compare with master

unifiedsplit

src/Streamly.hs+137−56 src/Streamly/Core.hs+3−1 src/Streamly/Streams.hs+41−78

src/Streamly.hs+137−56

@@ -6,64 +6,40 @@

66-- Maintainer : harendra.kumar@gmail.com

77-- Stability : experimental

88-- Portability : GHC

9−--

10−-- 'AsyncT' is a monad transformer that extends the product style composition

11−-- of monads to streams; it is a functional porgramming equivalent to nested

12−-- loops in imperative programming. This style of composition allows composing

13−-- each element in one stream with each element in the other stream,

14−-- generalizing the monadic product of single elements. For example, you can

15−-- think of the IO Monad as a special case of the more general @AsyncT IO@

16−-- monad; with single element streams. List transformers and logic programming

17−-- monads also provide a similar product style composition of streams, however

18−-- 'AsyncT' generalizes it with the time dimension; allowing streams to be

19−-- composed in an asynchronous and concurrent fashion in many different ways.

20−--

21−-- The power of this seemingly simple addition of asynchronicity and

22−-- concurrency to product style streaming composition should not be

23−-- underestimated. It unifies a number of disparate abstractions into one

24−-- powerful and elegant abstraction. A wide variety of programming problems

25−-- can be solved elegantly with this abstraction. In particular, it unifies

26−-- three major programming domains namely non-deterministic (logic)

27−-- programming, concurrent programming and functional reactive programming. In

28−-- other words, you can do everything with this one abstraction that you could

29−-- with list transformers (e.g.

30−-- <https://hackage.haskell.org/package/list-t list-t>), logic programming

31−-- monads (e.g. <https://hackage.haskell.org/package/logict logict>),

32−-- streaming libraries (a lot of what

33−-- <https://hackage.haskell.org/package/conduit conduit> or

34−-- <https://hackage.haskell.org/package/pipes pipes> can do), concurrency

35−-- libraries (e.g. <https://hackage.haskell.org/package/async async>) and FRP

36−-- libraries (e.g. <https://hackage.haskell.org/package/Yampa Yampa> or

37−-- <https://hackage.haskell.org/package/reflex reflex>).

38−--

39−-- When it comes to streaming composition, if @conduit@ or @pipes@ are arrows

40−-- then streamly is monad. Streaming libraries like conduit or pipes provide

41−-- a categorical composition (sum) of streams whereas streamly provides a more

42−-- general monadic composition (product) of streams, with concurrency.

43−-- Streamly interworks with major streaming libraries providing a way to use

44−-- both types of compositions and using the right tool at the right place.

45−-- When it comes to concurrency, streamly may be comared with imperative style

46−-- concurrency frameworks like

47−-- <https://en.wikipedia.org/wiki/OpenMP OpenMP> or

48−-- <https://en.wikipedia.org/wiki/Cilk Cilk>.

49−--

50−-- For more details please see the 'Streamly.Tutorial' and 'Streamly.Examples'.

519

52−-- A simple inline example here illustrating applicative, monad and alternative

53−-- compositions.

5410module Streamly

55− ( MonadAsync

11+ (

12+ -- * Background

13+ -- $background

14+

15+ -- * Overview

16+ -- $overview

17+

18+ MonadAsync

5619 , Streaming

5720

5821 -- * Product Style Composition

22+ -- $product

5923 , StreamT

6024 , InterleavedT

6125 , AsyncT

6226 , ParallelT

27+

28+ -- * Zip Style Composition

29+ -- $zipping

6330 , ZipStream

6431 , ZipAsync

6532

33+ -- * Sum Style Composition

34+ -- $sum

35+ , (<=>)

36+ , (<|)

37+

38+ -- * Transformation

39+ , async

40+

6641 -- * Stream Type Adapters

42+ -- $adapters

6743 , serially

6844 , interleaving

6945 , asyncly

@@ -81,15 +57,8 @@ module Streamly

8157 , runZipStream

8258 , runZipAsync

8359

84− -- * Transformation

85− , async

86−

87− -- * Sum Style Composition

88− -- $monoidal

89− , (<=>)

90− , (<|)

91−

9260 -- * Fold Utilities

61+ -- $foldutils

9362 , foldWith

9463 , foldMapWith

9564 , forEachWith

@@ -111,7 +80,119 @@ import Control.Monad (MonadPlus(..))

11180import Control.Monad.IO.Class (MonadIO (..))

11281import Control.Monad.Trans.Class (MonadTrans (..))

11382

114−-- $monoidal

83+-- $background

84+--

85+-- Streamly provides a monad transformer that extends the product style

86+-- composition of monads to streams of many elements of the same type; it is a

87+-- functional programming equivalent of nested loops from imperative

88+-- programming. Composing each element in one stream with each element in the

89+-- other stream generalizes the monadic product of single elements. You can

90+-- think of the IO monad as a special case of the more general @StreamT IO@

91+-- monad; with single element streams. List transformers and logic programming

92+-- monads also provide a similar product style composition of streams, however

93+-- streamly generalizes it with the time dimension; allowing streams to be

94+-- composed in an asynchronous and concurrent fashion in many different ways.

95+-- It also provides multiple alternative ways of composing streams e.g.

96+-- serial, interleaved or concurrent.

97+--

98+-- The seemingly simple addition of asynchronicity and concurrency to product

99+-- style streaming composition unifies a number of disparate abstractions into

100+-- one powerful and elegant abstraction. A wide variety of programming

101+-- problems can be solved elegantly with this abstraction. In particular, it

102+-- unifies three major programming domains namely non-deterministic (logic)

103+-- programming, concurrent programming and functional reactive programming. In

104+-- other words, you can do everything with this one abstraction that you could

105+-- with list transformers (e.g.

106+-- <https://hackage.haskell.org/package/list-t list-t>), logic programming

107+-- monads (e.g. <https://hackage.haskell.org/package/logict logict>),

108+-- streaming libraries (a lot of what

109+-- <https://hackage.haskell.org/package/conduit conduit> or

110+-- <https://hackage.haskell.org/package/pipes pipes> can do), concurrency

111+-- libraries (e.g. <https://hackage.haskell.org/package/async async>) and FRP

112+-- libraries (e.g. <https://hackage.haskell.org/package/Yampa Yampa> or

113+-- <https://hackage.haskell.org/package/reflex reflex>).

114+

115+-- $overview

116+--

117+-- Streamly provides six distinct stream types i.e. 'StreamT', 'InterleavedT',

118+-- 'AsyncT' and 'ParallelT', 'ZipStream' and 'ZipAsync', each representing a

119+-- stream of elements. All these types have the same underlying representation

120+-- and can be adapted from one to another using type adaptor combinators

121+-- described later. Each of these types belongs to the 'Streaming' type class

122+-- which helps converting the specific type to and from the underlying generic

123+-- stream type.

124+--

125+-- The types 'StreamT', 'InterleavedT', 'AsyncT' and 'ParallelT' are 'Monad'

126+-- transformers with the monadic bind operation combining streams in a product

127+-- style in much the same way as a list monad or a list transformer i.e. each

128+-- element from one stream is combined with every element of the other stream.

129+-- However, the applicative and monadic composition of these types differ in

130+-- terms of the ordering and time sequence in which the elements from two

131+-- streams are combined. 'StreamT' and 'InterleavedT' compose streams serially

132+-- whereas 'AsyncT' and 'ParallelT' are their concurrent counterparts. See the

133+-- documentation of the respective types for more details.

134+--

135+-- The types 'ZipStream' and 'ZipAsync' provide 'Applicative' instances to zip

136+-- two streams together i.e. each element in one stream is combined with the

137+-- corresponding element in the other stream. 'ZipStream' generates the streams

138+-- being zipped serially whereas 'ZipAsync' produces both the elements being

139+-- zipped concurrently.

140+--

141+-- Two streams of the same type can be combined using a sum style composition

142+-- to generate a stream of the same type where the output stream would contain

143+-- all elements of both the streams. However, the sequence in which the

144+-- elements in the resulting stream are produced depends on the combining

145+-- operator. Four distinct sum style operators, '<>', '<=>', '<|' and '<|>'

146+-- combine two streams in different ways, each corresponding to the one of the

147+-- four ways of combining monadically. See the respective section below for

148+-- more details.

149+--

150+-- Concurrent composition types 'AsyncT', 'ParallelT', 'ZipAsync' and

151+-- concurrent composition operators '<|' and '<|>' require the underlying monad

152+-- of the streaming monad transformer to be 'MonadAsync'.

153+--

154+-- For more details please see the 'Streamly.Tutorial' and 'Streamly.Examples'.

155+

156+-- A simple inline example here illustrating applicative, monad and alternative

157+-- compositions.

158+

159+-- $product

160+--

161+-- Streams that compose serially or non-concurrently come in two flavors i.e.

162+-- 'StreamT' and 'InterleavedT'. Both of these serial flavors have

163+-- corresponding concurrent equivalents, those are 'AsyncT' and 'ParallelT'

164+-- respectively.

165+

166+-- $zipping

167+--

168+-- 'ZipStream' and 'ZipAsync', provide 'Applicative' instances for zipping the

169+-- corresponding elements of two streams together. Note that these types do not

170+-- provide 'Monad' instances.

171+

172+-- $sum

173+--

174+-- Just like product style composition there are four distinct ways to combine

175+-- streams in sum style each directly corresponding to one of the product style

176+-- composition. The standard semigroup append '<>' operator appends two

177+-- streams serially, this style corresponds to the 'StreamT' style of monadic

178+-- composition. The standard 'Alternative' operator '<|>' fairly interleaves

179+-- two streams in parallel, this corresponds to the 'ParallelT' style. Two

180+-- additional sum style composition operators that streamly introduces are

181+-- described below.

182+

183+-- $adapters

184+--

185+-- Code using streamly is written such that it is agnostic of any specific

186+-- streaming type. We use a polymorphic type with the 'Streaming' class

187+-- constraint. Finally when running the monad we can specify the actual type

188+-- that we want to use to interpret the code. However, in certain cases we may

189+-- want to use a specific type to force a certain type of composition. Since

190+-- they have the same underlying type, streams can be converted from one type

191+-- to another freely. These combinators can be used to convert the streaming

192+-- types at will. If you get an ambiguous type error you perhaps forgot to

193+-- use a sepcific type before running the monad.

194+

195+-- $foldutils

115196--

116−-- These combinators can be used in place of 'Monoid' ('<>') or 'Alternative'

117−-- ('<|>') composition to fold streams in alternate ways.

197+-- These are some convenience functions to fold any 'Foldable' container using

198+-- one of the sum composition operators to convert it into a streamly stream.

src/Streamly/Core.hs+3−1

@@ -156,7 +156,9 @@ newtype Stream m a =

156156 -> m r

157157 }

158158

159−-- | A monad that can perform asynchronous/concurrent IO operations.

159+-- | A monad that can perform asynchronous/concurrent IO operations. Streams

160+-- that can be composed concurrently require the underlying monad to be

161+-- 'MonadAsync'.

160162type MonadAsync m = (MonadIO m, MonadBaseControl IO m, MonadThrow m)

161163

162164scons :: a -> Maybe (Stream m a) -> Stream m a

src/Streamly/Streams.hs+41−78

@@ -18,8 +18,6 @@

1818--

1919module Streamly.Streams

2020 (

21− -- * Product Style Composition

22− -- $product

2321 Streaming (..)

2422 , MonadAsync

2523

@@ -99,48 +97,6 @@ import Data.Semigroup (Semigroup(..))

9997import Prelude hiding (drop, take, zipWith)

10098import Streamly.Core

10199

102−-- $product

103−--

104−-- Different styles of streams represented by different types dictate the

105−-- manner in which a stream is processed in monadic or applicative composition.

106−-- Each type differs in the processing order of the elements or processing

107−-- parallelism. The types have the same underlying representation and

108−-- therefore can be converted back and forth easily to switch the composition

109−-- style at any point.

110−--

111−-- 'StreamT' and 'InterleavedT' process the streams serially but differ in the

112−-- order in which the elements of the streams are processed. 'StreamT' goes

113−-- depth first that is it combine one element of the first stream to every

114−-- element of the second stream before it comes back to process the second

115−-- element of the first stream. 'InterleavedT' takes turns when processing

116−-- elements from each stream and therefore all iterations finish at the same

117−-- time rather than one iteration at a time as in the case of 'StreamT'.

118−-- Depending on the cache behavior or IO behavior 'InterleavedT' might be

119−-- slightly more efficient than 'StreamT' or vice-versa.

120−--

121−-- 'AsyncT' and 'ParallelT' can process the individual elements of the streams

122−-- in parallel. 'AsyncT' is the parallel version of 'StreamT', it follows the

123−-- order of 'StreamT', processing multiple elements of a stream concurrently.

124−-- 'ParallelT' follows the order of 'InterleavedT', processing and interleaving

125−-- both streams concurrently. Also, note that 'fmap' on these streams would

126−-- also work in a concurrent fashion i.e. multiple elements in a stream or

127−-- multiple streams can be mapped concurrently.

128−--

129−-- Zipping types have a different style of composition altogether, they zip the

130−-- streams rather than multiplying them. These types provide only Applicative

131−-- composition and no Monad instances. 'ZipStream' zips serially i.e. it

132−-- produces one element from each stream in a serial manner and then zips the

133−-- two items. 'ZipAsync' on the other hand zips in parallel, it produces one

134−-- element from each stream concurrently and then zips the two.

135−--

136−-- It may help develop a better understanding if we think in terms of stream

137−-- generation (production) side parallelism versus stream processing

138−-- (consumption) side parallelism. The sum style operators (e.g. '<|' or '<|>')

139−-- generate streams in parallel and then combine the results into a unified

140−-- stream. On the other hand the product style composition (e.g. 'AsyncT' or

141−-- 'ParallelT' or 'ZipAsync') dictate how the elements of a stream are

142−-- processed when it is consumed.

143−

144100------------------------------------------------------------------------------

145101-- Types that can behave as a Stream

146102------------------------------------------------------------------------------

@@ -231,12 +187,19 @@ async m = do

231187 return $ fromSVar sv

232188

233189------------------------------------------------------------------------------

234−-- StreamT iterates serially in the Monad and Applicative compositions

190+-- StreamT

235191------------------------------------------------------------------------------

236192

237−-- | Note that 'StreamT' style composition works well operationally for

238−-- infinite streams. However if the second stream is infinite we will never

239−-- come back to the second iteration of the first stream.

193+-- | The 'Monad' instance of 'StreamT' combines two streams serially in the

194+-- most straightforward manner, just like the nested 'for' loops in imperative

195+-- paradigm. It moves to the next element in the first stream only after the

196+-- first element has been combined with every element in the second stream. In

197+-- other words, 'StreamT' nests loops in a depth first manner.

198+--

199+-- Note that 'StreamT' style composition works well operationally even when the

200+-- streams being composed are infinite. However, if the second stream is

201+-- infinite we will never come back to the second iteration of the first

202+-- stream.

240203

241204newtype StreamT m a = StreamT {getStreamT :: Stream m a}

242205 deriving (Semigroup, Monoid, MonadTrans, MonadIO, MonadThrow)

@@ -261,8 +224,6 @@ instance Streaming StreamT where

261224-- Monad

262225------------------------------------------------------------------------------

263226

264−-- | Execute a monadic action sequentially for each element in the 'AsyncT'

265−-- stream, i.e. an iteration finishes completely before the next one starts.

266227instance Monad m => Monad (StreamT m) where

267228 return = pure

268229 (StreamT (Stream m)) >>= f = StreamT $ Stream $ \_ stp yld ->

@@ -336,13 +297,16 @@ instance (Monad m, Floating a) => Floating (StreamT m a) where

336297 logBase = liftA2 logBase

337298

338299------------------------------------------------------------------------------

339−-- InterleavedT interleaves the iterations in the Monad and Applicative

340−-- compositions.

300+-- InterleavedT

341301------------------------------------------------------------------------------

342302

343−-- | Note that 'InterleavedT' style composition cannot work well operationally

344−-- on infinite streams. It will keep accumulating the elements of the stream

345−-- resulting in continuous increase in space consumption.

303+-- | The 'Monad' instance of 'InterleavedT' fairly interleaves the iterations

304+-- of the inner and the outer loop, nesting loops in a breadth first manner.

305+--

306+-- Note that 'InterleavedT' style composition cannot work well operationally on

307+-- infinite streams. Since the next iteration of the outer loop is started even

308+-- before the first one has finsihed, it will keep accumulating space resulting

309+-- in a continuous increase in space consumption.

346310--

347311newtype InterleavedT m a = InterleavedT {getInterleavedT :: Stream m a}

348312 deriving (Semigroup, Monoid, MonadTrans, MonadIO, MonadThrow)

@@ -358,8 +322,6 @@ instance Streaming InterleavedT where

358322 toStream = getInterleavedT

359323 fromStream = InterleavedT

360324

361−-- | Execute a monadic action for each element in the stream, in a fairly

362−-- interleaved manner i.e. iterations yield alternately.

363325instance Monad m => Monad (InterleavedT m) where

364326 return = pure

365327 (InterleavedT (Stream m)) >>= f = InterleavedT $ Stream $ \_ stp yld ->

@@ -434,16 +396,18 @@ instance (Monad m, Floating a) => Floating (InterleavedT m a) where

434396 logBase = liftA2 logBase

435397

436398------------------------------------------------------------------------------

437−-- AsyncT runs the iterations asynchronously in the Monad and Applicative

438−-- compositions. More iterations are started in parallel on demand. It can work

439−-- with infinite streams.

399+-- AsyncT

440400------------------------------------------------------------------------------

441401

442−-- | Just like 'StreamT', 'AsyncT' style composition works well operationally

443−-- for infinite streams. However, if the second stream is infinite we will

444−-- never be able to finish the first stream.

402+-- | 'AsyncT' is the concurrent equivalent of 'StreamT'. The type 'AsyncT'

403+-- follows the same composition scheme as 'StreamT' except that it can run

404+-- multiple iterations concurrently. More concurrent iterations are started

405+-- only if the previous iterations are not able to produce enough output for

406+-- the consumer.

445407--

446−-- Note that 'fmap' on 'AsyncT' would map on multiple elements concurrently.

408+-- Just like 'StreamT', 'AsyncT' style composition works well operationally

409+-- for infinite streams. However, if the second stream is infinite we will

410+-- never be able to start the second iteration of the first stream.

447411

448412newtype AsyncT m a = AsyncT {getAsyncT :: Stream m a}

449413 deriving (Semigroup, Monoid, MonadTrans)

@@ -476,9 +440,6 @@ parbind par m f = go m

476440 yield a (Just r) = run $ f a `par` (go r)

477441 in g Nothing stp yield

478442

479−-- | Execute a monadic action for each element in the stream, running

480−-- iterations in parallel, but giving preference to iterations started

481−-- earlier.

482443instance MonadAsync m => Monad (AsyncT m) where

483444 return = pure

484445 (AsyncT m) >>= f = AsyncT $ parbind par m g

@@ -548,17 +509,20 @@ instance (MonadAsync m, Floating a) => Floating (AsyncT m a) where

548509 logBase = liftA2 logBase

549510

550511------------------------------------------------------------------------------

551−-- ParallelT runs all iterations in parallel in the Monad and Applicative

552−-- compositions. Iterations are scheduled in a round-robin fashion.

512+-- ParallelT

553513------------------------------------------------------------------------------

554514

555−-- | Note that, just like 'InterleavedT', 'ParallelT' style composition cannot

556−-- work well operationally on infinite streams. It will keep accumulating the

515+-- | 'ParallelT' is the concurrent equivalent of 'InterleavedT'. It follows the

516+-- same composition scheme as 'InterleavedT' i.e. fairly interleaving the inner

517+-- and outer loops in a round robin fashion except that it can run multiple

518+-- iterations concurrently.

519+--

520+-- Note that, just like 'InterleavedT', 'ParallelT' style composition cannot

521+-- work well operationally on infinite streams. Because earlier iterations do

522+-- not finish before the new ones are started, it will keep on accumulating the

557523-- elements of the stream resulting in continuous increase in space

558524-- consumption.

559525--

560−-- Note that 'fmap' on 'ParallelT' would map on both streams concurrently.

561−--

562526newtype ParallelT m a = ParallelT {getParallelT :: Stream m a}

563527 deriving (Semigroup, Monoid, MonadTrans)

564528

@@ -575,9 +539,6 @@ instance Streaming ParallelT where

575539 toStream = getParallelT

576540 fromStream = ParallelT

577541

578−-- | Execute a monadic action for each element in the stream, running

579−-- iterations in a fairly parallel manner, i.e. all iterations are equally

580−-- likely to run.

581542instance MonadAsync m => Monad (ParallelT m) where

582543 return = pure

583544 (ParallelT m) >>= f = ParallelT $ parbind par m g

@@ -870,7 +831,8 @@ infixr 5 <=>

870831-- | Sequential interleaved composition, in contrast to '<>' this operator

871832-- fairly interleaves the two streams instead of appending them; yielding one

872833-- element from each stream alternately. Unlike '<>' it cannot be used to fold

873−-- an infinite container of streams.

834+-- an infinite container of streams. This operator corresponds to the

835+-- 'InterleavedT' type for product style composition.

874836{-# INLINE (<=>) #-}

875837(<=>) :: Streaming t => t m a -> t m a -> t m a

876838m1 <=> m2 = fromStream $ interleave (toStream m1) (toStream m2)

@@ -880,7 +842,8 @@ m1 <=> m2 = fromStream $ interleave (toStream m1) (toStream m2)

880842-- them. It keeps yielding from the stream on the left as long as it can. If

881843-- the left stream blocks or cannot keep up with the pace of the consumer it

882844-- can yield from the stream on the right in parallel. Unlike '<|>' it can be

883−-- used to fold infinite containers of streams.

845+-- used to fold infinite containers of streams. This operator corresponds to

846+-- the 'AsyncT' type for product style composition.

884847{-# INLINE (<|) #-}

885848(<|) :: (Streaming t, MonadAsync m) => t m a -> t m a -> t m a

886849m1 <| m2 = fromStream $ parLeft (toStream m1) (toStream m2)