652d466 · streamly · git.iko.soy

383 when (not done) $ (pushWorker ctx) >> dispatch

384

385

386−-- Note: This is performance sensitive code.

386+-- | Pull an AsyncT stream from a context

387−{-# NOINLINE pullWorker #-}

387+{-# NOINLINE pullFromCtx #-}

388−pullWorker :: MonadAsync m => Context m a -> AsyncT m a

388+pullFromCtx :: MonadAsync m => Context m a -> AsyncT m a

389−pullWorker ctx = AsyncT $ \_ stp yld -> do

389+pullFromCtx ctx = AsyncT $ \_ stp yld -> do

390 res <- liftIO $ tryTakeMVar (doorBell ctx)

391 when (isNothing res) $ sendWorkerWait ctx

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

407 processEvents [] = AsyncT $ \_ stp yld -> do

408 done <- allThreadsDone ctx

409 if not done

410− then (runAsyncT (pullWorker ctx)) Nothing stp yld

410+ then (runAsyncT (pullFromCtx ctx)) Nothing stp yld

411 else stp

412

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

421 Nothing -> delThread ctx tid >> continue

422 Just ex -> handleException ex tid

423

424+getFifoCtx :: MonadIO m => CtxType -> IO (Context m a)

425+getFifoCtx ctype = do

426+ outQ <- newIORef []

427+ outQMv <- newEmptyMVar

428+ running <- newIORef S.empty

429+ q <- newQ

430+ let ctx =

431+ Context { outputQueue = outQ

432+ , doorBell = outQMv

433+ , runningThreads = running

434+ , runqueue = runqueueFIFO ctx q

435+ , enqueue = pushL q

436+ , queueEmpty = liftIO $ nullQ q

437+ , ctxType = ctype

438+ }

439+ in return ctx

440+

441+getLifoCtx :: MonadIO m => CtxType -> IO (Context m a)

442+getLifoCtx ctype = do

443+ outQ <- newIORef []

444+ outQMv <- newEmptyMVar

445+ running <- newIORef S.empty

446+ q <- newIORef []

447+ let checkEmpty = liftIO (readIORef q) >>= return . null

448+ let ctx =

449+ Context { outputQueue = outQ

450+ , doorBell = outQMv

451+ , runningThreads = running

452+ , runqueue = runqueueLIFO ctx q

453+ , enqueue = enqueueLIFO q

454+ , queueEmpty = checkEmpty

455+ , ctxType = ctype

456+ }

457+ in return ctx

458+

424-- | Split the original computation in a pull-push pair. The original

459-- | Split the original computation in a pull-push pair. The original

425-- computation pulls from a Channel while m1 and m2 push to the channel.

460-- computation pulls from a Channel while m1 and m2 push to the channel.

426−pullForkPush :: MonadAsync m

461+pushPairToCtx :: MonadAsync m

427 => CtxType -> AsyncT m a -> AsyncT m a -> m (Context m a)

462 => CtxType -> AsyncT m a -> AsyncT m a -> m (Context m a)

428−pullForkPush ct m1 m2 = do

463+pushPairToCtx ctype m1 m2 = do

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

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

466+ -- chance to push.

429− ctx <- liftIO $ newContext

467+ ctx <- liftIO $

468+ case ctype of

469+ CtxType _ FIFO -> do

470+ c <- getFifoCtx ctype

471+ (enqueue c) m1 >> (enqueue c) m2

472+ return c

473+ CtxType _ LIFO -> do

474+ c <- getLifoCtx ctype

475+ (enqueue c) m2 >> (enqueue c) m1

476+ return c

430 pushWorker ctx

477 pushWorker ctx

431 return ctx

478 return ctx

432

479

433− where

434−

435− newContext = do

436− outQ <- newIORef []

437− outQMv <- newEmptyMVar

438− running <- newIORef S.empty

439−

440− case ct of

441− CtxType _ FIFO -> do

442− q <- newQ

443− pushL q m1 >> pushL q m2

444− let ctx =

445− Context { outputQueue = outQ

446− , doorBell = outQMv

447− , runningThreads = running

448− , runqueue = runqueueFIFO ctx q

449− , enqueue = pushL q

450− , queueEmpty = liftIO $ nullQ q

451− , ctxType = ct

452− }

453− in return ctx

454− CtxType _ LIFO -> do

455− q <- newIORef []

456− enqueueLIFO q m2 >> enqueueLIFO q m1

457− let checkEmpty = liftIO (readIORef q) >>= return . null

458− let ctx =

459− Context { outputQueue = outQ

460− , doorBell = outQMv

461− , runningThreads = running

462− , runqueue = runqueueLIFO ctx q

463− , enqueue = enqueueLIFO q

464− , queueEmpty = checkEmpty

465− , ctxType = ct

466− }

467− in return ctx

468−

469−pullForkPull :: MonadAsync m => Context m a -> AsyncT m a

470−pullForkPull ctx = AsyncT $ \_ stp yld ->

471− (runAsyncT (pullWorker ctx)) Nothing stp yld

472−

473−{-# NOINLINE pullFork #-}

480+{-# NOINLINE pushPullFork #-}

474−pullFork :: MonadAsync m

481+pushPullFork :: MonadAsync m

475 => CtxType -> AsyncT m a -> AsyncT m a -> AsyncT m a

482 => CtxType -> AsyncT m a -> AsyncT m a -> AsyncT m a

476−pullFork ct m1 m2 = AsyncT $ \_ stp yld -> do

483+pushPullFork ct m1 m2 = AsyncT $ \_ stp yld -> do

477− ctx <- pullForkPush ct m1 m2

484+ ctx <- pushPairToCtx ct m1 m2

478− (runAsyncT (pullForkPull ctx)) Nothing stp yld

485+ (runAsyncT (pullFromCtx ctx)) Nothing stp yld

479

486

480-- Concurrency rate control. Our objective is to create more threads on

487-- Concurrency rate control. Our objective is to create more threads on

481-- demand if the consumer is running faster than us. As soon as we

488-- demand if the consumer is running faster than us. As soon as we

514parallel :: MonadAsync m => CtxType -> AsyncT m a -> AsyncT m a -> AsyncT m a

521parallel :: MonadAsync m => CtxType -> AsyncT m a -> AsyncT m a -> AsyncT m a

515parallel ct m1 m2 = AsyncT $ \ctx stp yld -> do

522parallel ct m1 m2 = AsyncT $ \ctx stp yld -> do

516 case ctx of

523 case ctx of

517− Nothing -> (runAsyncT (pullFork ct m1 m2)) Nothing stp yld

524+ Nothing -> (runAsyncT (pushPullFork ct m1 m2)) Nothing stp yld

518 Just c | ctxType c /= ct ->

525 Just c | ctxType c /= ct ->

519− (runAsyncT (pullFork ct m1 m2)) Nothing stp yld

526+ (runAsyncT (pushPullFork ct m1 m2)) Nothing stp yld

520 Just c -> liftIO ((enqueue c) m2) >> (runAsyncT m1) ctx stp yld

527 Just c -> liftIO ((enqueue c) m2) >> (runAsyncT m1) ctx stp yld

521

528

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

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