34cbc34 · streamly · git.iko.soy

git.iko.soy/streamly

Define functor instances for Zip types instead of deriving

commit34cbc3476b4a35ebd936568194960edc21e958b5 authorHarendra Kumar <harendra.kumar@gmail.com> dateThu 26 Oct 2017, 08:22 (9 years ago) parent8f89d92

3 files changed+221−106

compare with master

unifiedsplit

src/Asyncly/AsyncT.hs+60−62 src/Asyncly/Prelude.hs+148−44 src/Asyncly/Tutorial.hs+13

src/Asyncly/AsyncT.hs+60−62

@@ -32,6 +32,7 @@ module Asyncly.AsyncT

3232 , yielding

3333 , interleave

3434 , parallel

35+ , parAlt

3536 , parLeft

3637 , CtxType (..)

3738 , SchedPolicy (..)

@@ -157,37 +158,7 @@ interleave m1 m2 = Stream $ \_ stp yld -> do

157158 (runStream m1) Nothing stop yield

158159

159160------------------------------------------------------------------------------

160−-- Functor

161−------------------------------------------------------------------------------

162−

163−instance Monad m => Functor (Stream m) where

164− fmap = liftM

165−

166−------------------------------------------------------------------------------

167−-- Applicative

168−------------------------------------------------------------------------------

169−

170−instance Monad m => Applicative (Stream m) where

171− pure = yielding

172− (<*>) = ap

173−

174−------------------------------------------------------------------------------

175−-- Monad

176−------------------------------------------------------------------------------

177−

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

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

180−instance Monad m => Monad (Stream m) where

181− return = pure

182−

183− Stream m >>= f = Stream $ \_ stp yld ->

184− let run x = (runStream x) Nothing stp yld

185− yield a Nothing = run $ f a

186− yield a (Just r) = run $ f a <> (r >>= f)

187− in m Nothing stp yield

188−

189−------------------------------------------------------------------------------

190−-- Alternative

161+-- Streaming threads for parallel composition

191162------------------------------------------------------------------------------

192163

193164{-# INLINE doFork #-}

@@ -468,19 +439,28 @@ parallel ct m1 m2 = Stream $ \ctx stp yld -> do

468439 (runStream (pushPullFork ct m1 m2)) Nothing stp yld

469440 Just c -> liftIO ((enqueue c) m2) >> (runStream m1) ctx stp yld

470441

471−-- | `empty` represents an action that takes non-zero time to complete. Since

472−-- all actions take non-zero time, an `Alternative` composition ('<|>') is a

473−-- monoidal composition executing all actions in parallel, it is similar to

474−-- '<>' except that it runs all the actions in parallel and interleaves their

475−-- results fairly.

476−instance MonadAsync m => Alternative (Stream m) where

477− empty = mempty

478−

479− {-# INLINE (<|>) #-}

480− (<|>) = parallel (CtxType Disjunction FIFO)

442+-- | The computation specified in the argument is pushed to a new thread and

443+-- the context is returned. The returned context can be used to pull the output

444+-- of the computation.

445+pushOneToCtx :: MonadAsync m => CtxType -> Stream m a -> m (Context m a)

446+pushOneToCtx ctype m = do

447+ ctx <- liftIO $

448+ case ctype of

449+ CtxType _ FIFO -> do

450+ c <- getFifoCtx ctype

451+ return c

452+ CtxType _ LIFO -> do

453+ c <- getLifoCtx ctype

454+ return c

455+ -- Note: We must have all the work on the queue before sending the

456+ -- pushworker, otherwise the pushworker may exit before we even get a

457+ -- chance to push.

458+ liftIO $ (enqueue ctx) m

459+ pushWorker ctx

460+ return ctx

481461

482462------------------------------------------------------------------------------

483−-- Other monoidal compositions for parallel actions

463+-- Semigroup and Monoid style compositions for parallel actions

484464------------------------------------------------------------------------------

485465

486466{-

@@ -495,36 +475,54 @@ parAhead = undefined

495475(<>|) = parAhead

496476-}

497477

478+-- | Same as '<|>'.

479+{-# INLINE parAlt #-}

480+parAlt :: MonadAsync m => Stream m a -> Stream m a -> Stream m a

481+parAlt = parallel (CtxType Disjunction FIFO)

482+

498483-- | Same as '<|'.

499484{-# INLINE parLeft #-}

500485parLeft :: MonadAsync m => Stream m a -> Stream m a -> Stream m a

501486parLeft = parallel (CtxType Disjunction LIFO)

502487

488+-------------------------------------------------------------------------------

489+-- Stream type is not exposed, these instances are only for deriving instances

490+-- for the newtype wrappers based on Stream.

491+-------------------------------------------------------------------------------

492+

493+-- Dummy Instances, defined to enable the definition of other instances that

494+-- require a Monad constraint. Must be defined by the newtypes.

495+

496+instance Monad m => Functor (Stream m) where

497+ fmap = undefined

498+

499+instance Monad m => Applicative (Stream m) where

500+ pure = undefined

501+ (<*>) = undefined

502+

503+instance Monad m => Monad (Stream m) where

504+ return = pure

505+ (>>=) = undefined

506+

507+------------------------------------------------------------------------------

508+-- Alternative & MonadPlus

509+------------------------------------------------------------------------------

510+

511+-- | `empty` represents an action that takes non-zero time to complete. Since

512+-- all actions take non-zero time, an `Alternative` composition ('<|>') is a

513+-- monoidal composition executing all actions in parallel, it is similar to

514+-- '<>' except that it runs all the actions in parallel and interleaves their

515+-- results fairly.

516+instance MonadAsync m => Alternative (Stream m) where

517+ empty = mempty

518+ (<|>) = parAlt

519+

503520instance MonadAsync m => MonadPlus (Stream m) where

504521 mzero = empty

505522 mplus = (<|>)

506523

507−-- | The computation specified in the argument is pushed to a new thread and

508−-- the context is computation pulls output from the thread.

509−pushOneToCtx :: MonadAsync m => CtxType -> Stream m a -> m (Context m a)

510−pushOneToCtx ctype m = do

511− ctx <- liftIO $

512− case ctype of

513− CtxType _ FIFO -> do

514− c <- getFifoCtx ctype

515− return c

516− CtxType _ LIFO -> do

517− c <- getLifoCtx ctype

518− return c

519− -- Note: We must have all the work on the queue before sending the

520− -- pushworker, otherwise the pushworker may exit before we even get a

521− -- chance to push.

522− liftIO $ (enqueue ctx) m

523− pushWorker ctx

524− return ctx

525−

526524-------------------------------------------------------------------------------

527−-- AsyncT transformer

525+-- Transformer

528526-------------------------------------------------------------------------------

529527

530528instance MonadTrans Stream where

src/Asyncly/Prelude.hs+148−44

@@ -19,6 +19,7 @@

1919module Asyncly.Prelude

2020 (

2121 -- * Product Style Composition

22+ -- $product

2223 StreamT

2324 , InterleavedT

2425 , AsyncT

@@ -66,13 +67,58 @@ import Data.Semigroup (Semigroup(..), cycle1)

6667import Prelude hiding (drop, take, zipWith)

6768import Asyncly.AsyncT

6869

70+-- $product

71+--

72+-- Different styles of streams dictate the manner in which a stream is

73+-- processed in monadic or applicative composition. Each style is represented

74+-- by a different type. The types have the same underlying representation and

75+-- therefore can be converted back and forth easily to switch the composition

76+-- style at any point. In other words, processing order of the elements or

77+-- processing parallelism is decided by the type of the stream.

78+--

79+-- 'StreamT' and 'InterleavedT' process the streams serially but differ in the

80+-- order in which the elements of the streams are processed. 'StreamT' goes

81+-- depth first that is it combine one element of the first stream to every

82+-- element of the second stream before it comes back to process th second

83+-- element of the first stream. 'InterleavedT' takes turns when processing

84+-- elements from each stream and therefore all iterations finish at the same

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

86+-- Depending on the cache behavior or IO behavior 'InterleavedT' might be

87+-- slightly more efficient than 'StreamT' or vice-versa.

88+--

89+-- 'AsyncT' and 'ParallelT' can process the individual elements of the streams

90+-- in parallel. 'AsyncT' is the parallel version of 'StreamT', it follows the

91+-- order of 'StreamT' processing multiple elements of a stream concurrently.

92+-- 'ParallelT' follows the order of 'InterleavedT' processing and interleaving

93+-- both streams concurrently. Also, note that 'fmap' on these streams would

94+-- also work in a concurrent fashion i.e. multiple elements in a stream or

95+-- multiple streams can be mapped concurrently.

96+--

97+-- Zipping types have a different style of composition altogether, they zip the

98+-- streams rather than multiplying them. These types provide only Applicative

99+-- composition and no Monad instances. 'ZipStream' zips serially i.e. it

100+-- produces one element from each stream in a serial manner and then zips the

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

102+-- element from each stream concurrently and then zips the two.

103+--

104+-- It may help develop a better understanding if we think in terms of stream

105+-- generation (production) side parallelism versus stream processing

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

107+-- generate streams in parallel and then combine the results into a unified

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

109+-- 'ParallelT' or 'ZipAsync') dictate how the elements of a stream are

110+-- processed when it is consumed.

111+

69112------------------------------------------------------------------------------

70113-- StreamT iterates serially in the Monad and Applicative compositions

71114------------------------------------------------------------------------------

72115

116+-- | Note that 'StreamT' style composition works well operationally for

117+-- infinite streams. However if the second stream is infinite we will never

118+-- come back to the second iteration of the first stream.

119+

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

74− deriving (Functor, Applicative, Monad, Semigroup, Monoid,

75− MonadTrans, MonadIO, MonadThrow)

121+ deriving (Semigroup, Monoid, MonadTrans, MonadIO, MonadThrow)

76122

77123deriving instance MonadAsync m => Alternative (StreamT m)

78124deriving instance MonadAsync m => MonadPlus (StreamT m)

@@ -90,6 +136,36 @@ instance Streaming StreamT where

90136-- from a single base type because they depend on the Monad instance which is

91137-- different for each type.

92138

139+------------------------------------------------------------------------------

140+-- Monad

141+------------------------------------------------------------------------------

142+

143+-- | Execute a monadic action sequentially for each element in the 'AsyncT'

144+-- stream, i.e. an iteration finishes completely before the next one starts.

145+instance Monad m => Monad (StreamT m) where

146+ return = pure

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

148+ let run x = (runStream x) Nothing stp yld

149+ yield a Nothing = run $ getStreamT (f a)

150+ yield a (Just r) = run $ getStreamT (f a)

151+ <> getStreamT (StreamT r >>= f)

152+ in m Nothing stp yield

153+

154+------------------------------------------------------------------------------

155+-- Applicative

156+------------------------------------------------------------------------------

157+

158+instance Monad m => Applicative (StreamT m) where

159+ pure = StreamT . yielding

160+ (<*>) = ap

161+

162+------------------------------------------------------------------------------

163+-- Functor

164+------------------------------------------------------------------------------

165+

166+instance Monad m => Functor (StreamT m) where

167+ fmap = liftM

168+

93169------------------------------------------------------------------------------

94170-- Num

95171------------------------------------------------------------------------------

@@ -136,9 +212,13 @@ instance (Monad m, Floating a) => Floating (StreamT m a) where

136212

137213------------------------------------------------------------------------------

138214-- InterleavedT interleaves the iterations in the Monad and Applicative

139−-- compositions. It does not make much sense for infinite streams.

215+-- compositions.

140216------------------------------------------------------------------------------

141217

218+-- | Note that 'InterleavedT' style composition cannot work well operationally

219+-- on infinite streams. It will keep accumulating the elements of the stream

220+-- resulting in continuous increase in space consumption.

221+--

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

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

144224

@@ -162,16 +242,9 @@ instance Monad m => Monad (InterleavedT m) where

162242 yield a Nothing = run $ getInterleavedT (f a)

163243 yield a (Just r) = run $ getInterleavedT (f a)

164244 `interleave`

165− getInterleavedT ((InterleavedT r) >>= f)

245+ getInterleavedT (InterleavedT r >>= f)

166246 in m Nothing stp yield

167247

168−------------------------------------------------------------------------------

169−-- Functor

170−------------------------------------------------------------------------------

171−

172−instance Monad m => Functor (InterleavedT m) where

173− fmap = liftM

174−

175248------------------------------------------------------------------------------

176249-- Applicative

177250------------------------------------------------------------------------------

@@ -180,6 +253,13 @@ instance Monad m => Applicative (InterleavedT m) where

180253 pure = InterleavedT . yielding

181254 (<*>) = ap

182255

256+------------------------------------------------------------------------------

257+-- Functor

258+------------------------------------------------------------------------------

259+

260+instance Monad m => Functor (InterleavedT m) where

261+ fmap = liftM

262+

183263------------------------------------------------------------------------------

184264-- Num

185265------------------------------------------------------------------------------

@@ -230,14 +310,20 @@ instance (Monad m, Floating a) => Floating (InterleavedT m a) where

230310-- with infinite streams.

231311------------------------------------------------------------------------------

232312

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

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

315+-- never be able to finish the first stream.

316+--

317+-- Note that 'fmap' on 'AsyncT' would map on multiple elements concurrently.

318+

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

234320 deriving (Semigroup, Monoid, MonadTrans)

235321

322+deriving instance MonadAsync m => Alternative (AsyncT m)

323+deriving instance MonadAsync m => MonadPlus (AsyncT m)

236324deriving instance MonadAsync m => MonadIO (AsyncT m)

237325deriving instance MonadAsync m => MonadThrow (AsyncT m)

238326deriving instance (MonadBase b m, MonadAsync m) => MonadBase b (AsyncT m)

239−deriving instance MonadAsync m => Alternative (AsyncT m)

240−deriving instance MonadAsync m => MonadPlus (AsyncT m)

241327deriving instance (MonadError e m, MonadAsync m) => MonadError e (AsyncT m)

242328deriving instance (MonadReader r m, MonadAsync m) => MonadReader r (AsyncT m)

243329deriving instance (MonadState s m, MonadAsync m) => MonadState s (AsyncT m)

@@ -270,13 +356,6 @@ instance MonadAsync m => Monad (AsyncT m) where

270356 where g x = getAsyncT (f x)

271357 par = parallel (CtxType Conjunction LIFO)

272358

273−------------------------------------------------------------------------------

274−-- Functor

275−------------------------------------------------------------------------------

276−

277−instance MonadAsync m => Functor (AsyncT m) where

278− fmap = liftM

279−

280359------------------------------------------------------------------------------

281360-- Applicative

282361------------------------------------------------------------------------------

@@ -285,6 +364,13 @@ instance MonadAsync m => Applicative (AsyncT m) where

285364 pure = AsyncT . yielding

286365 (<*>) = ap

287366

367+------------------------------------------------------------------------------

368+-- Functor

369+------------------------------------------------------------------------------

370+

371+instance MonadAsync m => Functor (AsyncT m) where

372+ fmap = liftM

373+

288374------------------------------------------------------------------------------

289375-- Num

290376------------------------------------------------------------------------------

@@ -331,18 +417,24 @@ instance (MonadAsync m, Floating a) => Floating (AsyncT m a) where

331417

332418------------------------------------------------------------------------------

333419-- ParallelT runs all iterations in parallel in the Monad and Applicative

334−-- compositions. Iterations are scheduled in a round-robin fashion. It cannot

335−-- work well when the streams being composed are infinite.

420+-- compositions. Iterations are scheduled in a round-robin fashion.

336421------------------------------------------------------------------------------

337422

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

424+-- work well operationally on infinite streams. It will keep accumulating the

425+-- elements of the stream resulting in continuous increase in space

426+-- consumption.

427+--

428+-- Note that 'fmap' on 'ParallelT' would map on both streams concurrently.

429+--

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

339431 deriving (Semigroup, Monoid, MonadTrans)

340432

433+deriving instance MonadAsync m => Alternative (ParallelT m)

434+deriving instance MonadAsync m => MonadPlus (ParallelT m)

341435deriving instance MonadAsync m => MonadIO (ParallelT m)

342436deriving instance MonadAsync m => MonadThrow (ParallelT m)

343437deriving instance (MonadBase b m, MonadAsync m) => MonadBase b (ParallelT m)

344−deriving instance MonadAsync m => Alternative (ParallelT m)

345−deriving instance MonadAsync m => MonadPlus (ParallelT m)

346438deriving instance (MonadError e m, MonadAsync m) => MonadError e (ParallelT m)

347439deriving instance (MonadReader r m, MonadAsync m) => MonadReader r (ParallelT m)

348440deriving instance (MonadState s m, MonadAsync m) => MonadState s (ParallelT m)

@@ -360,13 +452,6 @@ instance MonadAsync m => Monad (ParallelT m) where

360452 where g x = getParallelT (f x)

361453 par = parallel (CtxType Conjunction FIFO)

362454

363−------------------------------------------------------------------------------

364−-- Functor

365−------------------------------------------------------------------------------

366−

367−instance MonadAsync m => Functor (ParallelT m) where

368− fmap = liftM

369−

370455------------------------------------------------------------------------------

371456-- Applicative

372457------------------------------------------------------------------------------

@@ -375,6 +460,13 @@ instance MonadAsync m => Applicative (ParallelT m) where

375460 pure = ParallelT . yielding

376461 (<*>) = ap

377462

463+------------------------------------------------------------------------------

464+-- Functor

465+------------------------------------------------------------------------------

466+

467+instance MonadAsync m => Functor (ParallelT m) where

468+ fmap = liftM

469+

378470------------------------------------------------------------------------------

379471-- Num

380472------------------------------------------------------------------------------

@@ -451,13 +543,19 @@ zipWith f m1 m2 = fromStream $ go (toStream m1) (toStream m2)

451543 yield1 a (Just ra) = merge a ra

452544 (runStream mx) Nothing stp yield1

453545

454−-- | Wrapper around AsyncT type with a serial zipping Applicative instance.

455−-- Note that the binary function interleave (\<=>) is a special case of

456−-- ZipStream Applicative.

457−--

458−-- > f <$> ZipStream xs1 <*> ... <*> ZipStream xsN

546+-- | 'ZipStream' zips serially i.e. it produces one element from each stream in

547+-- a serial manner and then zips the two elements.

459548newtype ZipStream m a = ZipStream {getZipStream :: Stream m a}

460− deriving (Functor)

549+ deriving (Semigroup, Monoid)

550+

551+deriving instance MonadAsync m => Alternative (ZipStream m)

552+

553+instance Monad m => Functor (ZipStream m) where

554+ fmap f (ZipStream (Stream m)) = ZipStream $ Stream $ \_ stp yld ->

555+ let yield a Nothing = yld (f a) Nothing

556+ yield a (Just r) = yld (f a)

557+ (Just (getZipStream (fmap f (ZipStream r))))

558+ in m Nothing stp yield

461559

462560instance Monad m => Applicative (ZipStream m) where

463561 pure a = ZipStream $ cycle1 (pure a)

@@ -508,7 +606,7 @@ instance (Monad m, Floating a) => Floating (ZipStream m a) where

508606 logBase = liftA2 logBase

509607

510608------------------------------------------------------------------------------

511−-- Asyncly Zipping Streams

609+-- Parallely Zipping Streams

512610------------------------------------------------------------------------------

513611

514612-- | Zip two AsyncT streams asyncly (i.e. both the streams are generated

@@ -529,13 +627,19 @@ zipAsyncWith f m1 m2 = fromStream $ Stream $ \_ stp yld -> do

529627 mb <- async m2

530628 (runStream (toStream (zipWith f ma mb))) Nothing stp yld

531629

532−-- | Wrapper around AsyncT type with a parallel zipping Applicative instance.

533−-- Note that the binary operator (\<|>) from the Alternative instance of AsyncT

534−-- is a special case of ZipAsync Applicative.

535−--

536−-- > f <$> ZipAsync xs1 <*> ... <*> ZipAsync xsN

630+-- | 'ZipAsync' zips in parallel, it produces one element from each stream

631+-- concurrently and then zips the two.

537632newtype ZipAsync m a = ZipAsync {getZipAsync :: Stream m a}

538− deriving (Functor)

633+ deriving (Semigroup, Monoid)

634+

635+deriving instance MonadAsync m => Alternative (ZipAsync m)

636+

637+instance Monad m => Functor (ZipAsync m) where

638+ fmap f (ZipAsync (Stream m)) = ZipAsync $ Stream $ \_ stp yld ->

639+ let yield a Nothing = yld (f a) Nothing

640+ yield a (Just r) = yld (f a)

641+ (Just (getZipAsync (fmap f (ZipAsync r))))

642+ in m Nothing stp yield

539643

540644instance MonadAsync m => Applicative (ZipAsync m) where

541645 pure a = ZipAsync $ cycle1 (pure a)

src/Asyncly/Tutorial.hs+13

@@ -585,3 +585,16 @@ import Control.Monad.Trans.Class (MonadTrans (lift))

585585

586586-- $statemachine

587587-- State machine stuff

588+

589+-- $comparison

590+--

591+-- Even though asyncly covers all that is provided by the 'async' package or

592+-- most of what is provided by the 'streaming', 'pipes' or 'conduit' packages,

593+-- I would not say that it renders those useless. Asyncly is like monad if

594+-- 'async' is applicative and monads and applicatives both have their use

595+-- cases. It can completely repalce 'async', the ZipAsync type is equivalent to

596+-- the functionality provided by 'async'.

597+--

598+-- pipes and conduit are like Arrows and asyncly is like monad to them. You

599+-- would use pipes and conduit when you do not need the product style

600+-- composition and the implicit concurrency.