@@ -383,10 +383,10 @@ sendWorkerWait ctx = dispatch >> void (liftIO $ takeMVar (doorBell ctx))
383383 when (not done) $ (pushWorker ctx) >> dispatch
384384
385385
386−-- Note: This is performance sensitive code.
387−{-# NOINLINE pullWorker #-}
388−pullWorker :: MonadAsync m => Context m a -> AsyncT m a
389−pullWorker ctx = AsyncT $ \_ stp yld -> do
386+-- | Pull an AsyncT stream from a context
387+{-# NOINLINE pullFromCtx #-}
388+pullFromCtx :: MonadAsync m => Context m a -> AsyncT m a
389+pullFromCtx ctx = AsyncT $ \_ stp yld -> do
390390 res <- liftIO $ tryTakeMVar (doorBell ctx)
391391 when (isNothing res) $ sendWorkerWait ctx
392392 list <- liftIO $ atomicModifyIORefCAS (outputQueue ctx) $ \x -> ([], x)
@@ -407,7 +407,7 @@ pullWorker ctx = AsyncT $ \_ stp yld -> do
407407 processEvents [] = AsyncT $ \_ stp yld -> do
408408 done <- allThreadsDone ctx
409409 if not done
410− then (runAsyncT (pullWorker ctx)) Nothing stp yld
410+ then (runAsyncT (pullFromCtx ctx)) Nothing stp yld
411411 else stp
412412
413413 processEvents (ev : es) = AsyncT $ \_ stp yld -> do
@@ -421,61 +421,68 @@ pullWorker ctx = AsyncT $ \_ stp yld -> do
421421 Nothing -> delThread ctx tid >> continue
422422 Just ex -> handleException ex tid
423423
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+
424459-- | Split the original computation in a pull-push pair. The original
425460-- computation pulls from a Channel while m1 and m2 push to the channel.
426−pullForkPush :: MonadAsync m
461+pushPairToCtx :: MonadAsync m
427462 => CtxType -> AsyncT m a -> AsyncT m a -> m (Context m a)
428−pullForkPush ct m1 m2 = do
429− ctx <- liftIO $ newContext
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.
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
430477 pushWorker ctx
431478 return ctx
432479
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 #-}
474−pullFork :: MonadAsync m
480+{-# NOINLINE pushPullFork #-}
481+pushPullFork :: MonadAsync m
475482 => CtxType -> AsyncT m a -> AsyncT m a -> AsyncT m a
476−pullFork ct m1 m2 = AsyncT $ \_ stp yld -> do
477− ctx <- pullForkPush ct m1 m2
478− (runAsyncT (pullForkPull ctx)) Nothing stp yld
483+pushPullFork ct m1 m2 = AsyncT $ \_ stp yld -> do
484+ ctx <- pushPairToCtx ct m1 m2
485+ (runAsyncT (pullFromCtx ctx)) Nothing stp yld
479486
480487-- Concurrency rate control. Our objective is to create more threads on
481488-- demand if the consumer is running faster than us. As soon as we
@@ -514,9 +521,9 @@ pullFork ct m1 m2 = AsyncT $ \_ stp yld -> do
514521parallel :: MonadAsync m => CtxType -> AsyncT m a -> AsyncT m a -> AsyncT m a
515522parallel ct m1 m2 = AsyncT $ \ctx stp yld -> do
516523 case ctx of
517− Nothing -> (runAsyncT (pullFork ct m1 m2)) Nothing stp yld
524+ Nothing -> (runAsyncT (pushPullFork ct m1 m2)) Nothing stp yld
518525 Just c | ctxType c /= ct ->
519− (runAsyncT (pullFork ct m1 m2)) Nothing stp yld
526+ (runAsyncT (pushPullFork ct m1 m2)) Nothing stp yld
520527 Just c -> liftIO ((enqueue c) m2) >> (runAsyncT m1) ctx stp yld
521528
522529-- | `empty` represents an action that takes non-zero time to complete. Since