55356f6 · streamly · git.iko.soy

113newtype AsyncT m a =

114 AsyncT {

115 runAsyncT :: forall r.

116− Maybe (Context m a) -- state

116+ Maybe (Context m a) -- local state

117− -> m r -- stop

117+ -> m r -- stop

118− -> (a -> Maybe (Context m a) -> Maybe (AsyncT m a) -> m r) -- yield

118+ -> (a -> Maybe (AsyncT m a) -> m r) -- yield

119 -> m r

120 }

121

133 where

134 go (AsyncT m) = AsyncT $ \ctx stp yld ->

135 let stop = (runAsyncT m2) ctx stp yld

136− yield a c Nothing = yld a c (Just m2)

136+ yield a Nothing = yld a (Just m2)

137− yield a c (Just r) = yld a c (Just (go r))

137+ yield a (Just r) = yld a (Just (go r))

138 in m ctx stop yield

139

140instance Monoid (AsyncT m a) where

145interleave :: AsyncT m a -> AsyncT m a -> AsyncT m a

146interleave m1 m2 = AsyncT $ \ctx stp yld -> do

147 let stop = (runAsyncT m2) ctx stp yld

148− yield a c Nothing = yld a c (Just m2)

148+ yield a Nothing = yld a (Just m2)

149− yield a c (Just r) = yld a c (Just (interleave m2 r))

149+ yield a (Just r) = yld a (Just (interleave m2 r))

150 (runAsyncT m1) ctx stop yield

151

152infixr 5 <=>

176-- lift the result into AsyncT.

177

178instance Monad m => Applicative (AsyncT m) where

179− pure a = AsyncT $ \ctx _ yld -> yld a ctx Nothing

179+ pure a = AsyncT $ \_ _ yld -> yld a Nothing

180 (<*>) = ap

181

182------------------------------------------------------------------------------

196 go (AsyncT g) =

197 AsyncT $ \_ stp yld ->

198 let run x = (runAsyncT x) Nothing stp yld

199− yield a _ Nothing = run $ f a

199+ yield a Nothing = run $ f a

200− yield a _ (Just r) = run $ f a `k` (go r)

200+ yield a (Just r) = run $ f a `k` (go r)

201 in g Nothing stp yield

202

203-- | '>>=' execute a monadic action for each element generated by the 'AsyncT'

209

210 AsyncT m >>= f = AsyncT $ \_ stp yld ->

211 let run x = (runAsyncT x) Nothing stp yld

212− yield a _ Nothing = run $ f a

212+ yield a Nothing = run $ f a

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

213+ yield a (Just r) = run $ f a <> (r >>= f)

214 in m Nothing stp yield

215

216------------------------------------------------------------------------------

291 Just m -> (runAsyncT m) (Just ctx) run yield

292

293 sendit a = send ctx (ChildYield a)

294− yield a _ Nothing = sendit a >> run

294+ yield a Nothing = sendit a >> run

295− yield a c (Just r) = sendit a >> (runAsyncT r) c run yield

295+ yield a (Just r) = sendit a >> (runAsyncT r) (Just ctx) run yield

296

297 dequeue = liftIO $ atomicModifyIORefCAS q $ \ ms ->

298 case ms of

316

317 dequeue = liftIO $ tryPopR q

318 sendit a = send ctx (ChildYield a)

319− yield a _ Nothing = sendit a >> run

319+ yield a Nothing = sendit a >> run

320− -- XXX do we need to save the context as well?

321− -- we can enqueue "(runAsyncT m) ctx run yield" instead

322− yield a _ (Just r) = sendit a >> liftIO (enqueueFIFO q r) >> run

320+ yield a (Just r) = sendit a >> liftIO (enqueueFIFO q r) >> run

323

321

324-- Thread tracking is needed for two reasons:

322-- Thread tracking is needed for two reasons:

325--

323--

376 res <- liftIO $ tryTakeMVar (doorBell ctx)

374 res <- liftIO $ tryTakeMVar (doorBell ctx)

377 when (isNothing res) $ sendWorkerWait ctx

375 when (isNothing res) $ sendWorkerWait ctx

378 list <- liftIO $ atomicModifyIORefCAS (outputQueue ctx) $ \x -> ([], x)

376 list <- liftIO $ atomicModifyIORefCAS (outputQueue ctx) $ \x -> ([], x)

379− (runAsyncT $ processEvents list) pctx stp yld

377+ (runAsyncT $ processEvents pctx list) Nothing stp yld

380

378

381 where

379 where

382

380

387 throwM e

385 throwM e

388

386

389 {-# INLINE processEvents #-}

387 {-# INLINE processEvents #-}

390− processEvents [] = AsyncT $ \pctx stp yld -> do

388+ processEvents pctx [] = AsyncT $ \_ stp yld -> do

391 done <- allThreadsDone ctx

389 done <- allThreadsDone ctx

392 if not done

390 if not done

393 then (runAsyncT (pullWorker ctx)) pctx stp yld

391 then (runAsyncT (pullWorker ctx)) pctx stp yld

394 else stp

392 else stp

395

393

396− processEvents (ev : es) = AsyncT $ \pctx stp yld -> do

394+ processEvents pctx (ev : es) = AsyncT $ \_ stp yld -> do

397− let continue = (runAsyncT (processEvents es)) pctx stp yld

395+ let continue = (runAsyncT (processEvents pctx es)) Nothing stp yld

398− yield a = yld a pctx (Just (processEvents es))

396+ yield a = yld a (Just (processEvents pctx es))

399

397

400 case ev of

398 case ev of

401 ChildYield a -> yield a

399 ChildYield a -> yield a

578-------------------------------------------------------------------------------

576-------------------------------------------------------------------------------

579

577

580instance MonadTrans AsyncT where

578instance MonadTrans AsyncT where

581− lift mx = AsyncT $ \ctx _ yld -> mx >>= (\a -> (yld a ctx Nothing))

579+ lift mx = AsyncT $ \_ _ yld -> mx >>= (\a -> (yld a Nothing))

582

580

583instance (MonadBase b m, Monad m) => MonadBase b (AsyncT m) where

581instance (MonadBase b m, Monad m) => MonadBase b (AsyncT m) where

584 liftBase = liftBaseDefault

582 liftBase = liftBaseDefault

598 throwError = lift . throwError

596 throwError = lift . throwError

599 catchError m h = AsyncT $ \ctx stp yld ->

597 catchError m h = AsyncT $ \ctx stp yld ->

600 let handle r = r `catchError` \e -> (runAsyncT (h e)) ctx stp yld

598 let handle r = r `catchError` \e -> (runAsyncT (h e)) ctx stp yld

601− yield a c Nothing = yld a c Nothing

599+ yield a Nothing = yld a Nothing

602− yield a c (Just r) = yld a c (Just (catchError r h))

600+ yield a (Just r) = yld a (Just (catchError r h))

603 in handle $ (runAsyncT m) ctx stp yield

601 in handle $ (runAsyncT m) ctx stp yield

604

602

605instance MonadReader r m => MonadReader r (AsyncT m) where

603instance MonadReader r m => MonadReader r (AsyncT m) where

606 ask = lift ask

604 ask = lift ask

607 local f m = AsyncT $ \ctx stp yld ->

605 local f m = AsyncT $ \ctx stp yld ->

608− let yield a c Nothing = local f $ yld a c Nothing

606+ let yield a Nothing = local f $ yld a Nothing

609− yield a c (Just r) = local f $ yld a c (Just (local f r))

607+ yield a (Just r) = local f $ yld a (Just (local f r))

610 in (runAsyncT m) ctx (local f stp) yield

608 in (runAsyncT m) ctx (local f stp) yield

611

609

612instance MonadState s m => MonadState s (AsyncT m) where

610instance MonadState s m => MonadState s (AsyncT m) where

621-- | Run an 'AsyncT' computation, wait for it to finish and discard the

619-- | Run an 'AsyncT' computation, wait for it to finish and discard the

622-- results.

620-- results.

623runAsyncly :: MonadAsync m => AsyncT m a -> m ()

621runAsyncly :: MonadAsync m => AsyncT m a -> m ()

624−runAsyncly m = run Nothing m

622+runAsyncly m = (runAsyncT m) Nothing stop yield

625

623

626 where

624 where

627

625

628 stop = return ()

626 stop = return ()

629

627

630 {-# INLINE yield #-}

628 {-# INLINE yield #-}

631− yield _ _ Nothing = stop

629+ yield _ Nothing = stop

632− yield _ c (Just x) = run c x

630+ yield _ (Just x) = runAsyncly x

633

631

634− run ct mx = (runAsyncT mx) ct stop yield

635

632

636-- | Collect the results of an 'AsyncT' stream into a list.

633-- | Collect the results of an 'AsyncT' stream into a list.

637{-# INLINABLE toList #-}

634{-# INLINABLE toList #-}

638toList :: MonadAsync m => AsyncT m a -> m [a]

635toList :: MonadAsync m => AsyncT m a -> m [a]

639−toList m = run Nothing m

636+toList m = (runAsyncT m) Nothing stop yield

640

637

641 where

638 where

642

639

643 stop = return []

640 stop = return []

644

641

645 {-# INLINE yield #-}

642 {-# INLINE yield #-}

646− yield a _ Nothing = return [a]

643+ yield a Nothing = return [a]

647− yield a c (Just x) = liftM (a :) (run c x)

644+ yield a (Just x) = liftM (a :) (toList x)

648−

649− {-# INLINE run #-}

650− run ctx mx = (runAsyncT mx) ctx stop yield

651

645

652------------------------------------------------------------------------------

646------------------------------------------------------------------------------

653-- Transformation

647-- Transformation

656-- | Take first 'n' elements from the stream and discard the rest.

650-- | Take first 'n' elements from the stream and discard the rest.

657take :: MonadAsync m => Int -> AsyncT m a -> AsyncT m a

651take :: MonadAsync m => Int -> AsyncT m a -> AsyncT m a

658take n m = AsyncT $ \ctx stp yld -> do

652take n m = AsyncT $ \ctx stp yld -> do

659− let yield a c Nothing = yld a c Nothing

653+ let yield a Nothing = yld a Nothing

660− yield a c (Just x) = yld a c (Just (take (n - 1) x))

654+ yield a (Just x) = yld a (Just (take (n - 1) x))

661 if (n <= 0)

655 if (n <= 0)

662 then stp

656 then stp

663 else (runAsyncT m) ctx stp yield

657 else (runAsyncT m) ctx stp yield

665-- | Discard first 'n' elements from the stream and take the rest.

659-- | Discard first 'n' elements from the stream and take the rest.

666drop :: MonadAsync m => Int -> AsyncT m a -> AsyncT m a

660drop :: MonadAsync m => Int -> AsyncT m a -> AsyncT m a

667drop n m = AsyncT $ \ctx stp yld -> do

661drop n m = AsyncT $ \ctx stp yld -> do

668− let yield _ _ Nothing = stp

662+ let yield _ Nothing = stp

669− yield _ c (Just x) = (runAsyncT $ drop (n - 1) x) c stp yld

663+ yield _ (Just x) = (runAsyncT $ drop (n - 1) x) ctx stp yld

670 if (n <= 0)

664 if (n <= 0)

671 then (runAsyncT m) ctx stp yld

665 then (runAsyncT m) ctx stp yld

672 else (runAsyncT m) ctx stp yield

666 else (runAsyncT m) ctx stp yield

700

694

701-- | Convert a callback into an 'AsyncT' stream.

695-- | Convert a callback into an 'AsyncT' stream.

702fromCallback :: (forall r. (a -> m r) -> m r) -> AsyncT m a

696fromCallback :: (forall r. (a -> m r) -> m r) -> AsyncT m a

703−fromCallback k = AsyncT $ \ctx _ yld -> k (\a -> yld a ctx Nothing)

697+fromCallback k = AsyncT $ \_ _ yld -> k (\a -> yld a Nothing)