@@ -78,9 +78,11 @@ data ChildEvent a = ChildYield a | ChildDone ThreadId (Maybe SomeException)
7878
7979newtype Context a = Context { childChannel :: TChan (ChildEvent a) }
8080
81−-- The 'Maybe' is redundant as we can use 'stop' value in the Nothing case,
82−-- but it makes the fold using '<|>' 25% faster. Need to try again as the code
83−-- has gone through many changes after this was tested.
81+-- The 'Maybe (AsyncT m a)' is redundant as we can use 'stop' value for the
82+-- Nothing case, but it makes the fold using '<|>' 25% faster. Need to try
83+-- again as the code has gone through many changes after this was tested.
84+-- With the Maybe, 'stop' is required only to represent 'empty' in an
85+-- Alternative composition.
8486--
8587-- Currently the only state we need is the thread context, For generality we
8688-- can parameterize the type with a state type 's'.
@@ -214,15 +216,27 @@ handlePushException pchan res = do
214216canFork :: IO Bool
215217canFork = return True
216218
217−-- XXX do not fork if the channel is never empty
218−-- We can fork if the channel goes empty sometimes
219+{-# NOINLINE fork #-}
219220fork :: (MonadIO m, MonadBaseControl IO m) => AsyncT m a -> AsyncT m a
220221fork m = AsyncT $ \ctx stp yld -> do
221− doAsync <- liftIO $ canFork
222222 let c = fromJust ctx -- XXX partial
223+ _ <- doFork (push c m) (handlePushException (childChannel c))
224+ stp
225+
226+fork1 :: (MonadIO m, MonadBaseControl IO m) => AsyncT m a -> AsyncT m a -> AsyncT m a
227+fork1 m m1 = AsyncT $ \ctx stp yld -> do
228+ let c = fromJust ctx -- XXX partial
229+ _ <- doFork (push c m) (handlePushException (childChannel c))
230+ (runAsyncT m1) ctx stp yld
231+
232+-- XXX do not fork if the channel is never empty
233+-- We can fork if the channel goes empty sometimes
234+{-# INLINE tryFork #-}
235+tryFork :: (MonadIO m, MonadBaseControl IO m) => AsyncT m a -> AsyncT m a
236+tryFork m = AsyncT $ \ctx stp yld -> do
237+ doAsync <- liftIO $ canFork
223238 if doAsync then do
224− _ <- doFork (push c m) (handlePushException (childChannel c))
225− stp
239+ (runAsyncT $ fork m) ctx stp yld
226240 else (runAsyncT m) ctx stp yld
227241
228242-- We re-raise any exceptions received from the child threads, that way
@@ -244,15 +258,23 @@ pull ctx = AsyncT $ \_ stp yld -> do
244258 let continue = (runAsyncT (pull ctx)) (Just ctx) stp yld
245259 maybe continue throwM e
246260
247−pushpull :: MonadAsync m => AsyncT m a -> AsyncT m a
248−pushpull m = AsyncT $ \ctx stp yld ->
261+{-# INLINE pushpull #-}
262+pushpull :: MonadAsync m => AsyncT m a -> AsyncT m a -> AsyncT m a
263+pushpull m1 m2 = AsyncT $ \ctx stp yld ->
249264 case ctx of
250265 Nothing -> do
266+ -- liftIO $ putStrLn "creating context"
251267 c <- liftIO $ newContext
268+
269+ -- Not sure why the following show a quadratic complexity
270+ -- 1) let comp = m1 <> (m2 <> empty)
271+ -- 2) let comp = (m1 <> m2) <> empty
272+ -- (runAsyncT comp) (Just c) stp yld
273+
252274 -- XXX m must be a push type computation
253− let comp = serially (fork m) (pull c)
254− (runAsyncT comp) (Just c) stp yld
255− Just _ -> (runAsyncT m) ctx stp yld
275+ -- XXX Create context only if we are really forking
276+ (runAsyncT (fork1 (serially m1 m2) (pull c))) (Just c) stp yld
277+ Just _ -> (runAsyncT (serially m1 m2)) ctx stp yld
256278
257279-- Child threads yield values to the parent's channel. The parent reads them
258280-- from there and yields to its consumer.
@@ -273,9 +295,7 @@ instance MonadAsync m => Alternative (AsyncT m) where
273295 -- a serially running computation. For that we can use a state flag to fork
274296 -- the rest of the computation at any point of time inside the Monad bind
275297 -- operation if the consumer is running at a faster speed.
276− m1 <|> m2 = AsyncT $ \ctx stp yld ->
277− let comp = pushpull $ serially (fork m1) m2
278− in (runAsyncT comp) ctx stp yld
298+ m1 <|> m2 = pushpull (tryFork m1) m2
279299
280300instance MonadAsync m => MonadPlus (AsyncT m) where
281301 mzero = empty