@@ -97,6 +97,10 @@ data ChildEvent a =
9797-- State threaded around the monad for thread management
9898------------------------------------------------------------------------------
9999
100+data Dimension = Conjunction | Disjunction deriving Eq
101+data SchedPolicy = LIFO | FIFO deriving Eq
102+data CtxType = CtxType Dimension SchedPolicy deriving Eq
103+
100104data Context m a =
101105 Context { outputQueue :: IORef [ChildEvent a]
102106 , doorBell :: MVar Bool -- wakeup mechanism for outQ
@@ -104,6 +108,7 @@ data Context m a =
104108 , runqueue :: m ()
105109 , runningThreads :: IORef (Set ThreadId)
106110 , queueEmpty :: m Bool
111+ , ctxType :: CtxType
107112 }
108113
109114-- | Represents a monadic stream of values of type 'a' resulting from actions
@@ -224,15 +229,15 @@ instance Monad m => Monad (AsyncT m) where
224229(>->) = bindWith (<=>)
225230
226231-- | Execute a monadic action for each element in the stream, running nested
227−-- loops in parallel, but with a depth first preference.
232+-- loops in parallel, but giving preference to inner loops.
228233(>>|) :: MonadAsync m => AsyncT m a -> (a -> AsyncT m b) -> AsyncT m b
229−(>>|) = bindWith (<|)
234+(>>|) = bindWith (parallel (CtxType Conjunction LIFO))
230235
231236-- | Execute a monadic action for each element in the stream, running nested
232237-- loops in parallel, but in a fairly interleaved manner i.e. both inner and
233238-- outer loops yield alternately.
234239(>|>) :: MonadAsync m => AsyncT m a -> (a -> AsyncT m b) -> AsyncT m b
235−(>|>) = bindWith (<|>)
240+(>|>) = bindWith (parallel (CtxType Conjunction FIFO))
236241
237242-- TBD Ideally we would want parallelism to be completely hidden from the user,
238243-- for example we can automatically decide where to start an action in parallel
@@ -408,8 +413,9 @@ pullWorker ctx = AsyncT $ \pctx stp yld -> do
408413-- | Split the original computation in a pull-push pair. The original
409414-- computation pulls from a Channel while m1 and m2 push to the channel.
410415{-# NOINLINE pullFork #-}
411−pullFork :: MonadAsync m => AsyncT m a -> AsyncT m a -> Bool -> AsyncT m a
412−pullFork m1 m2 fifo = AsyncT $ \_ stp yld -> do
416+pullFork :: MonadAsync m
417+ => CtxType -> AsyncT m a -> AsyncT m a -> AsyncT m a
418+pullFork ct m1 m2 = AsyncT $ \_ stp yld -> do
413419 ctx <- liftIO $ newContext
414420 pushWorker ctx >> (runAsyncT (pullWorker ctx)) Nothing stp yld
415421
@@ -420,8 +426,8 @@ pullFork m1 m2 fifo = AsyncT $ \_ stp yld -> do
420426 outQMv <- newEmptyMVar
421427 running <- newIORef S.empty
422428
423− case fifo of
424− True -> do
429+ case ct of
430+ CtxType _ FIFO -> do
425431 q <- newQ
426432 pushL q m1 >> pushL q m2
427433 let ctx =
@@ -431,9 +437,10 @@ pullFork m1 m2 fifo = AsyncT $ \_ stp yld -> do
431437 , runqueue = runqueueFIFO ctx q
432438 , enqueue = pushL q
433439 , queueEmpty = liftIO $ nullQ q
440+ , ctxType = ct
434441 }
435442 in return ctx
436− False -> do
443+ CtxType _ LIFO -> do
437444 q <- newIORef []
438445 enqueueLIFO q m2 >> enqueueLIFO q m1
439446 let checkEmpty = liftIO (readIORef q) >>= return . null
@@ -444,6 +451,7 @@ pullFork m1 m2 fifo = AsyncT $ \_ stp yld -> do
444451 , runqueue = runqueueLIFO ctx q
445452 , enqueue = enqueueLIFO q
446453 , queueEmpty = checkEmpty
454+ , ctxType = ct
447455 }
448456 in return ctx
449457
@@ -481,11 +489,13 @@ pullFork m1 m2 fifo = AsyncT $ \_ stp yld -> do
481489-- Note: This is designed to scale for right associated compositions,
482490-- therefore always use a right fold for folding bigger structures.
483491{-# INLINE parallel #-}
484−parallel :: MonadAsync m => AsyncT m a -> AsyncT m a -> Bool -> AsyncT m a
485−parallel m1 m2 fifo = AsyncT $ \ctx stp yld -> do
492+parallel :: MonadAsync m => CtxType -> AsyncT m a -> AsyncT m a -> AsyncT m a
493+parallel ct m1 m2 = AsyncT $ \ctx stp yld -> do
486494 case ctx of
487− Nothing -> (runAsyncT (pullFork m1 m2 fifo)) Nothing stp yld
488− Just c -> liftIO ((enqueue c) m2) >> (runAsyncT m1) ctx stp yld
495+ Nothing -> (runAsyncT (pullFork ct m1 m2)) Nothing stp yld
496+ Just c | ctxType c /= ct ->
497+ (runAsyncT (pullFork ct m1 m2)) Nothing stp yld
498+ Just c -> liftIO ((enqueue c) m2) >> (runAsyncT m1) ctx stp yld
489499
490500-- | `empty` represents an action that takes non-zero time to complete. Since
491501-- all actions take non-zero time, an `Alternative` composition ('<|>') is a
@@ -496,7 +506,7 @@ instance MonadAsync m => Alternative (AsyncT m) where
496506 empty = mempty
497507
498508 {-# INLINE (<|>) #-}
499− m1 <|> m2 = parallel m1 m2 True
509+ (<|>) = parallel (CtxType Disjunction FIFO)
500510
501511------------------------------------------------------------------------------
502512-- Other monoidal compositions for parallel actions
@@ -515,7 +525,7 @@ parAhead = undefined
515525-- | Same as '<|'.
516526{-# INLINE parLeft #-}
517527parLeft :: MonadAsync m => AsyncT m a -> AsyncT m a -> AsyncT m a
518−parLeft m1 m2 = parallel m1 m2 False
528+parLeft = parallel (CtxType Disjunction LIFO)
519529
520530-- | Parallel composition similar to `<|>` except that it is left-biased
521531-- instead of being fair. Action on the left is likely to be given a chance to