@@ -106,8 +106,8 @@ type MonadAsync m = (MonadIO m, MonadBaseControl IO m, MonadThrow m)
106106instance Monad m => Monoid (AsyncT m a) where
107107 mempty = AsyncT $ \_ stp _ -> stp
108108 mappend (AsyncT m1) m2 = AsyncT $ \ctx stp yld ->
109− m1 ctx ((runAsyncT m2) ctx stp yld) $ \a c r ->
110− let yield x = yld a c (Just x)
109+ m1 ctx ((runAsyncT m2) ctx stp yld) $ \a _ r ->
110+ let yield x = yld a ctx (Just x)
111111 in maybe (yield m2) (\rx -> yield $ mappend rx m2) r
112112
113113-- We do not use bind for parallelism. That is, we do not start each iteration
@@ -152,7 +152,7 @@ instance Monad m => Applicative (AsyncT m) where
152152-- Alternative
153153------------------------------------------------------------------------------
154154
155−{-# NOINLINE doFork #-}
155+{-# INLINE doFork #-}
156156doFork :: (MonadIO m, MonadBaseControl IO m)
157157 => m ()
158158 -> (SomeException -> IO ())
@@ -160,6 +160,7 @@ doFork :: (MonadIO m, MonadBaseControl IO m)
160160doFork action preExit =
161161 EL.mask $ \restore ->
162162 liftBaseWith $ \runInIO -> forkIO $ do
163+ -- XXX test the exception handling
163164 _ <- runInIO $ EL.catch (restore action) (liftIO . preExit)
164165 -- XXX restore state here?
165166 return ()
@@ -219,20 +220,21 @@ pull ctx = AsyncT $ \_ stp yld -> do
219220 Just BlockedIndefinitelyOnSTM -> stp
220221 Nothing -> throwM e
221222 Right ev ->
222− case ev of
223− ChildYield a -> yld a Nothing (Just (pull ctx))
224− ChildDone _ a -> yld a Nothing (Just (pull ctx))
223+ let yield a = yld a Nothing (Just (pull ctx))
224+ in case ev of
225+ ChildYield a -> yield a
226+ ChildDone _ a -> yield a
225227 ChildStop _ e -> do
226228 let continue = (runAsyncT (pull ctx)) (Just ctx) stp yld
227− maybe continue throwM e
229+ in maybe continue throwM e
228230
229231-- | Split the original computation in a pull-push pair. The original
230232-- computation pulls from a Channel while m1 and m2 push to the channel.
231−{-# INLINE pullFork #-}
233+{-# NOINLINE pullFork #-}
232234pullFork :: MonadAsync m => AsyncT m a -> AsyncT m a -> AsyncT m a
233235pullFork m1 m2 = AsyncT $ \_ stp yld -> do
234236 c <- liftIO $ newContext
235− _ <- doFork (pushBoth c m1 m2) (handlePushException (childChannel c))
237+ _ <- doFork (push c (m1 <|> m2)) (handlePushException (childChannel c))
236238 (runAsyncT (pull c)) (Just c) stp yld
237239
238240 where
@@ -241,13 +243,8 @@ pullFork m1 m2 = AsyncT $ \_ stp yld -> do
241243 childChan <- atomically newTChan
242244 return $ Context { childChannel = childChan }
243245
244− pushBoth c ma mb = do
245− _ <- doFork (push c ma) (handlePushException (childChannel c))
246− -- XXX push from m2 only when the channel becomes empty
247− push c mb
248−
249246canFork :: IO Bool
250−canFork = return True
247+canFork = return False
251248
252249-- Concurrency rate control. Our objective is to create more threads on
253250-- demand if the consumer is running faster than us. As soon as we
@@ -264,21 +261,20 @@ canFork = return True
264261-- the rest of the computation at any point of time inside the Monad bind
265262-- operation if the consumer is running at a faster speed.
266263--
267−{-# NOINLINE fork #-}
268−fork :: MonadAsync m => AsyncT m a -> AsyncT m a -> AsyncT m a
269−fork m1 m2 = AsyncT $ \ctx stp yld -> do
270− case ctx of
271− Nothing -> (runAsyncT (pullFork m1 m2)) ctx stp yld
272− Just c -> (runAsyncT (pushFork c m1 m2)) ctx stp yld
273−
274264instance MonadAsync m => Alternative (AsyncT m) where
275265 empty = mempty
276266
267+ -- XXX pullFork can have more than 2x penalty for smallish tasks. Can we
268+ -- avoid this automatically if the task is small?
269+ {-# INLINE (<|>) #-}
277270 m1 <|> m2 = AsyncT $ \ctx stp yld -> do
278− doAsync <- liftIO $ canFork
279− if doAsync then
280− (runAsyncT $ fork m1 m2) ctx stp yld
281− else (runAsyncT (m1 <> m2)) ctx stp yld
271+ case ctx of
272+ Nothing -> (runAsyncT (pullFork m1 m2)) ctx stp yld
273+ Just c -> do
274+ doAsync <- liftIO $ canFork
275+ if doAsync then
276+ (runAsyncT (pushFork c m1 m2)) ctx stp yld
277+ else (runAsyncT (m1 <> m2)) ctx stp yld
282278
283279instance MonadAsync m => MonadPlus (AsyncT m) where
284280 mzero = empty