@@ -78,9 +78,11 @@ data ChildEvent a = ChildYield a | ChildDone ThreadId (Maybe SomeException)
79newtype Context a = Context { childChannel :: TChan (ChildEvent a) }
79newtype Context a = Context { childChannel :: TChan (ChildEvent a) }
81−-- The 'Maybe' is redundant as we can use 'stop' value in the Nothing case,
81+-- The 'Maybe (AsyncT m a)' is redundant as we can use 'stop' value for the
82−-- but it makes the fold using '<|>' 25% faster. Need to try again as the code
82+-- Nothing case, but it makes the fold using '<|>' 25% faster. Need to try
83−-- has gone through many changes after this was tested.
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.
85-- Currently the only state we need is the thread context, For generality we
87-- Currently the only state we need is the thread context, For generality we
86-- can parameterize the type with a state type 's'.
88-- can parameterize the type with a state type 's'.
@@ -214,15 +216,27 @@ handlePushException pchan res = do
214canFork :: IO Bool
216canFork :: IO Bool
215canFork = return True
217canFork = return True
217−-- XXX do not fork if the channel is never empty
219+{-# NOINLINE fork #-}
218−-- We can fork if the channel goes empty sometimes
219fork :: (MonadIO m, MonadBaseControl IO m) => AsyncT m a -> AsyncT m a
220fork :: (MonadIO m, MonadBaseControl IO m) => AsyncT m a -> AsyncT m a
220fork m = AsyncT $ \ctx stp yld -> do
221fork m = AsyncT $ \ctx stp yld -> do
221− doAsync <- liftIO $ canFork
222 let c = fromJust ctx -- XXX partial
222 let c = fromJust ctx -- XXX partial
223+ _ <- doFork (push c m) (handlePushException (childChannel c))
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
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
223 if doAsync then do
238 if doAsync then do
224− _ <- doFork (push c m) (handlePushException (childChannel c))
239+ (runAsyncT $ fork m) ctx stp yld
226 else (runAsyncT m) ctx stp yld
240 else (runAsyncT m) ctx stp yld
228-- We re-raise any exceptions received from the child threads, that way
242-- We re-raise any exceptions received from the child threads, that way
@@ -244,15 +258,23 @@ pull ctx = AsyncT $ \_ stp yld -> do
244 let continue = (runAsyncT (pull ctx)) (Just ctx) stp yld
258 let continue = (runAsyncT (pull ctx)) (Just ctx) stp yld
245 maybe continue throwM e
259 maybe continue throwM e
261+{-# INLINE pushpull #-}
247−pushpull :: MonadAsync m => AsyncT m a -> AsyncT m a
262+pushpull :: MonadAsync m => AsyncT m a -> AsyncT m a -> AsyncT m a
248−pushpull m = AsyncT $ \ctx stp yld ->
263+pushpull m1 m2 = AsyncT $ \ctx stp yld ->
249 case ctx of
264 case ctx of
250 Nothing -> do
265 Nothing -> do
266+ -- liftIO $ putStrLn "creating context"
251 c <- liftIO $ newContext
267 c <- liftIO $ newContext
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
252 -- XXX m must be a push type computation
274 -- XXX m must be a push type computation
253− let comp = serially (fork m) (pull c)
275+ -- XXX Create context only if we are really forking
254− (runAsyncT comp) (Just c) stp yld
276+ (runAsyncT (fork1 (serially m1 m2) (pull c))) (Just c) stp yld
255− Just _ -> (runAsyncT m) ctx stp yld
277+ Just _ -> (runAsyncT (serially m1 m2)) ctx stp yld
257-- Child threads yield values to the parent's channel. The parent reads them
279-- Child threads yield values to the parent's channel. The parent reads them
258-- from there and yields to its consumer.
280-- from there and yields to its consumer.
@@ -273,9 +295,7 @@ instance MonadAsync m => Alternative (AsyncT m) where
273 -- a serially running computation. For that we can use a state flag to fork
295 -- a serially running computation. For that we can use a state flag to fork
274 -- the rest of the computation at any point of time inside the Monad bind
296 -- the rest of the computation at any point of time inside the Monad bind
275 -- operation if the consumer is running at a faster speed.
297 -- 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
298+ m1 <|> m2 = pushpull (tryFork m1) m2
278− in (runAsyncT comp) ctx stp yld
280instance MonadAsync m => MonadPlus (AsyncT m) where
300instance MonadAsync m => MonadPlus (AsyncT m) where
281 mzero = empty
301 mzero = empty