@@ -43,7 +43,8 @@ import Control.Monad.Trans.Control (ComposeSt, MonadBaseControl (..),
4343 defaultLiftBaseWith,
4444 defaultRestoreM, liftBaseWith)
4545 -}
46−import Data.Maybe (fromJust, maybe)
46+import Data.Maybe (maybe)
47+import Data.Monoid ((<>))
4748
4849import Control.Monad.Trans.Recorder (MonadRecorder(..))
4950
@@ -150,39 +151,34 @@ instance Monad m => Applicative (AsyncT m) where
150151-- Alternative
151152------------------------------------------------------------------------------
152153
153−newContext :: IO (Context a)
154−newContext = do
155− childChan <- atomically newTChan
156− return $ Context { childChannel = childChan }
157−
154+{-# NOINLINE doFork #-}
158155doFork :: (MonadIO m, MonadBaseControl IO m)
159156 => m ()
160− -> (Either SomeException () -> IO ())
157+ -> (SomeException -> IO ())
161158 -> m ThreadId
162159doFork action preExit =
163160 EL.mask $ \restore ->
164161 liftBaseWith $ \runInIO -> forkIO $ do
165− _ <- runInIO $ EL.try (restore action) >>= liftIO . preExit
166− -- XXX restore state here
162+ _ <- runInIO $ EL.catch (restore action) (liftIO . preExit)
163+ -- XXX restore state here?
167164 return ()
168165
169−-- XXX make this a bounded channel so that we block if the previous value is
170−-- not consumed yet.
171−{-# INLINE channelDone #-}
172−channelDone :: MonadIO m => TChan (ChildEvent a) -> m ()
173−channelDone chan = do
174− tid <- liftIO myThreadId
175− liftIO $ atomically $ writeTChan chan (ChildDone tid Nothing)
176−
177−{-# INLINE channelYield #-}
178−channelYield :: MonadIO m => TChan (ChildEvent a) -> a -> m ()
179−channelYield chan a = liftIO $ atomically $ writeTChan chan (ChildYield a)
180−
166+{-# NOINLINE push #-}
181167push :: MonadIO m => Context a -> AsyncT m a -> m ()
182168push ctx m = go (Just ctx) m
183169
184170 where
185171
172+ -- XXX make this a bounded channel so that we block if the previous value
173+ -- is not consumed yet.
174+ {-# INLINE channelDone #-}
175+ channelDone chan = do
176+ tid <- liftIO myThreadId
177+ liftIO $ atomically $ writeTChan chan (ChildDone tid Nothing)
178+
179+ {-# INLINE channelYield #-}
180+ channelYield chan a = liftIO $ atomically $ writeTChan chan (ChildYield a)
181+
186182 go c mx = (runAsyncT mx) c (channelDone pchan) yield
187183 pchan = childChannel ctx
188184
@@ -204,53 +200,35 @@ push ctx m = go (Just ctx) m
204200-- If an exception occurs we push it to the channel so that it can handled by
205201-- the parent. 'Paused' exceptions are to be collected at the top level.
206202-- XXX Paused exceptions should only bubble up to the runRecorder computation
207−handlePushException :: TChan (ChildEvent a) -> Either SomeException () -> IO ()
208−handlePushException pchan res = do
209− let r = case res of
210− Left e -> Just e
211− Right _ -> Nothing
212−
203+{-# NOINLINE handlePushException #-}
204+handlePushException :: TChan (ChildEvent a) -> SomeException -> IO ()
205+handlePushException pchan e = do
213206 tid <- myThreadId
214− atomically $ writeTChan pchan (ChildDone tid r)
215−
216−canFork :: IO Bool
217−canFork = return True
218−
219−{-# NOINLINE fork #-}
220−fork :: (MonadIO m, MonadBaseControl IO m) => AsyncT m a -> AsyncT m a
221−fork m = AsyncT $ \ctx stp yld -> do
222− 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
238− if doAsync then do
239− (runAsyncT $ fork m) ctx stp yld
240− else (runAsyncT m) ctx stp yld
207+ atomically $ writeTChan pchan (ChildDone tid (Just e))
208+
209+-- | run m1 in a new thread, pushing its results to a pull channel and then run
210+-- m2 in the parent thread. Any exceptions are also pushed to the channel.
211+{-# INLINE pushFork #-}
212+pushFork :: (MonadIO m, MonadBaseControl IO m)
213+ => Context a -> AsyncT m a -> AsyncT m a -> AsyncT m a
214+pushFork c m1 m2 = AsyncT $ \ctx stp yld -> do
215+ _ <- doFork (push c m1) (handlePushException (childChannel c))
216+ -- XXX run m2 only when the channel becomes empty
217+ (runAsyncT m2) ctx stp yld
241218
242219-- We re-raise any exceptions received from the child threads, that way
243220-- exceptions get propagated to the top level computation and can be handled
244221-- there.
222+{-# NOINLINE pull #-}
245223pull :: (MonadIO m, MonadThrow m) => Context a -> AsyncT m a
246224pull ctx = AsyncT $ \_ stp yld -> do
247− ethr <- liftIO $ try $ atomically $ readTChan (childChannel ctx)
248− case ethr of
249− -- XXX for immediate cleanup and for killing any blocked threads we
250− -- need to track the children
251− Left e -> case fromException e of
252− Just BlockedIndefinitelyOnSTM -> stp
253− Nothing -> throwM e
225+ -- XXX push the try out of the pull loop
226+ res <- liftIO $ try (atomically $ readTChan (childChannel ctx))
227+ case res of
228+ Left e ->
229+ case fromException e of
230+ Just BlockedIndefinitelyOnSTM -> stp
231+ Nothing -> throwM e
254232 Right ev ->
255233 case ev of
256234 ChildYield a -> yld a Nothing (Just (pull ctx))
@@ -258,49 +236,70 @@ pull ctx = AsyncT $ \_ stp yld -> do
258236 let continue = (runAsyncT (pull ctx)) (Just ctx) stp yld
259237 maybe continue throwM e
260238
261−{-# INLINE pushpull #-}
262−pushpull :: MonadAsync m => AsyncT m a -> AsyncT m a -> AsyncT m a
263−pushpull m1 m2 = AsyncT $ \ctx stp yld ->
239+-- | Split the original computation in a pull-push pair. The original
240+-- computation pulls from a Channel while m1 and m2 push to the channel.
241+{-# INLINE pullFork #-}
242+pullFork :: MonadAsync m => AsyncT m a -> AsyncT m a -> AsyncT m a
243+pullFork m1 m2 = AsyncT $ \_ stp yld -> do
244+ -- XXX assert ctx is Nothing
245+ c <- liftIO $ newContext
246+ _ <- doFork (pushBoth c m1 m2) (handlePushException (childChannel c))
247+ (runAsyncT (pull c)) (Just c) stp yld
248+
249+ where
250+
251+ newContext = do
252+ childChan <- atomically newTChan
253+ return $ Context { childChannel = childChan }
254+
255+ pushBoth c ma mb = do
256+ _ <- doFork (push c ma) (handlePushException (childChannel c))
257+ -- XXX push from m2 only when the channel becomes empty
258+ push c mb
259+
260+-- If threads suspend due to IO we can start more of them to utilize concurrent
261+-- IO, but only if we have more IO bandwidth. If we have more parallel CPU
262+-- bandwidth we can start more threads appropriately, but we have to see that
263+-- the thread creation overhead is less than the actual work to be done.
264+canFork :: IO Bool
265+canFork = return True
266+
267+-- Concurrency rate control. Our objective is to create more threads on
268+-- demand if the consumer is running faster than us. As soon as we
269+-- encounter an Alternative composition we create a push pull pair of
270+-- threads. We use a channel for communication between the consumer that
271+-- pulls from the channel and the producer that pushes to the channel. The
272+-- producer creates more threads if the channel becomes empty at times,
273+-- that is the consumer is running faster. However this mechanism can be
274+-- problematic if the initial production latency is high, we may end up
275+-- creating too many threads. So we need some way to monitor and use the
276+-- latency as well.
277+-- TBD For quick response we may have to increase the rate in the middle of
278+-- a serially running computation. For that we can use a state flag to fork
279+-- the rest of the computation at any point of time inside the Monad bind
280+-- operation if the consumer is running at a faster speed.
281+--
282+{-# NOINLINE fork #-}
283+fork :: MonadAsync m => AsyncT m a -> AsyncT m a -> AsyncT m a
284+fork m1 m2 = AsyncT $ \ctx stp yld -> do
264285 case ctx of
265− Nothing -> do
266− -- liftIO $ putStrLn "creating context"
267− 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−
274− -- XXX m must be a push type computation
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
278−
279−-- Child threads yield values to the parent's channel. The parent reads them
280−-- from there and yields to its consumer.
286+ Nothing -> (runAsyncT (pullFork m1 m2)) ctx stp yld
287+ Just c -> (runAsyncT (pushFork c m1 m2)) ctx stp yld
288+
281289instance MonadAsync m => Alternative (AsyncT m) where
282290 empty = AsyncT $ \_ stp _ -> stp
283291
284− -- Concurrency rate control. Our objective is to create more threads on
285− -- demand if the consumer is running faster than us. As soon as we
286− -- encounter an Alternative composition we create a push pull pair of
287− -- threads. We use a channel for communication between the consumer that
288− -- pulls from the channel and the producer that pushes to the channel. The
289− -- producer creates more threads if the channel becomes empty at times,
290− -- that is the consumer is running faster. However this mechanism can be
291− -- problematic if the initial production latency is high, we may end up
292− -- creating too many threads. So we need some way to monitor and use the
293− -- latency as well.
294− -- TBD For quick response we may have to increase the rate in the middle of
295− -- a serially running computation. For that we can use a state flag to fork
296− -- the rest of the computation at any point of time inside the Monad bind
297− -- operation if the consumer is running at a faster speed.
298− m1 <|> m2 = pushpull (tryFork m1) m2
292+ m1 <|> m2 = AsyncT $ \ctx stp yld -> do
293+ doAsync <- liftIO $ canFork
294+ if doAsync then
295+ (runAsyncT $ fork m1 m2) ctx stp yld
296+ else (runAsyncT (m1 <> m2)) ctx stp yld
299297
300298instance MonadAsync m => MonadPlus (AsyncT m) where
301299 mzero = empty
302300 mplus = (<|>)
303301
302+-- XXX Move it in the beginning
304303-- | Appends the results of two AsyncT computations in order.
305304instance MonadAsync m => Monoid (AsyncT m a) where
306305 mempty = empty