@@ -28,8 +28,8 @@ import Control.Applicative (Alternative (..))
28import Control.Concurrent (ThreadId, forkIO, killThread,
28import Control.Concurrent (ThreadId, forkIO, killThread,
29 myThreadId, newQSem, QSem,
29 myThreadId, newQSem, QSem,
30 signalQSem, waitQSem)
30 signalQSem, waitQSem)
31−import Control.Concurrent.STM (TChan, atomically, newTChan,
31+import Control.Concurrent.STM (TBQueue, atomically, newTBQueue,
32− tryReadTChan, writeTChan)
32+ tryReadTBQueue, writeTBQueue)
33import Control.Exception (SomeException (..))
33import Control.Exception (SomeException (..))
34import qualified Control.Exception.Lifted as EL
34import qualified Control.Exception.Lifted as EL
35import Control.Monad (ap, liftM, MonadPlus(..), mzero)
35import Control.Monad (ap, liftM, MonadPlus(..), mzero)
@@ -85,7 +85,7 @@ data ChildEvent a =
85------------------------------------------------------------------------------
85------------------------------------------------------------------------------
87data Context a =
87data Context a =
88− Context { childChannel :: TChan (ChildEvent a)
88+ Context { childChannel :: TBQueue (ChildEvent a)
89 , dispatchReq :: QSem
89 , dispatchReq :: QSem
90 , runningThreads :: IORef (Set ThreadId)
90 , runningThreads :: IORef (Set ThreadId)
91 , doneThreads :: IORef (Set ThreadId)
91 , doneThreads :: IORef (Set ThreadId)
@@ -193,13 +193,13 @@ push context action = do
193 -- XXX make this a bounded channel so that we block if the previous value
193 -- XXX make this a bounded channel so that we block if the previous value
194 -- is not consumed yet.
194 -- is not consumed yet.
195 chan = childChannel context
195 chan = childChannel context
196− channelYield a = liftIO $ atomically $ writeTChan chan (ChildYield a)
196+ channelYield a = liftIO $ atomically $ writeTBQueue chan (ChildYield a)
197 channelDone a = do
197 channelDone a = do
198 tid <- liftIO myThreadId
198 tid <- liftIO myThreadId
199− liftIO $ atomically $ writeTChan chan (ChildDone tid a)
199+ liftIO $ atomically $ writeTBQueue chan (ChildDone tid a)
200 channelStop = do
200 channelStop = do
201 tid <- liftIO myThreadId
201 tid <- liftIO myThreadId
202− liftIO $ atomically $ writeTChan chan (ChildStop tid Nothing)
202+ liftIO $ atomically $ writeTBQueue chan (ChildStop tid Nothing)
204 done a = channelDone a
204 done a = channelDone a
205 continue a ctx m = channelYield a >> run ctx m
205 continue a ctx m = channelYield a >> run ctx m
@@ -249,14 +249,13 @@ handleException e ctx tid = do
249 liftIO $ readIORef (runningThreads ctx) >>= mapM_ killThread
249 liftIO $ readIORef (runningThreads ctx) >>= mapM_ killThread
252−-- XXX the TBQueue size should be proportional to the pending threads.
253-- We re-raise any exceptions received from the child threads, that way
252-- We re-raise any exceptions received from the child threads, that way
254-- exceptions get propagated to the top level computation and can be handled
253-- exceptions get propagated to the top level computation and can be handled
256{-# NOINLINE pullDrain #-}
255{-# NOINLINE pullDrain #-}
257pullDrain :: (MonadIO m, MonadThrow m) => Context a -> AsyncT m a
256pullDrain :: (MonadIO m, MonadThrow m) => Context a -> AsyncT m a
258pullDrain ctx = AsyncT $ \_ stp yld -> do
257pullDrain ctx = AsyncT $ \_ stp yld -> do
259− res <- liftIO $ atomically $ tryReadTChan (childChannel ctx)
258+ res <- liftIO $ atomically $ tryReadTBQueue (childChannel ctx)
260 maybe (dispatch >> continue stp yld)
259 maybe (dispatch >> continue stp yld)
261 (\ev -> handleEvent ev stp yld) res
260 (\ev -> handleEvent ev stp yld) res
@@ -292,10 +291,10 @@ pullDrain ctx = AsyncT $ \_ stp yld -> do
292-- the parent. 'Paused' exceptions are to be collected at the top level.
291-- the parent. 'Paused' exceptions are to be collected at the top level.
293-- XXX Paused exceptions should only bubble up to the runRecorder computation
292-- XXX Paused exceptions should only bubble up to the runRecorder computation
294{-# NOINLINE handleChildException #-}
293{-# NOINLINE handleChildException #-}
295−handleChildException :: TChan (ChildEvent a) -> SomeException -> IO ()
294+handleChildException :: TBQueue (ChildEvent a) -> SomeException -> IO ()
296handleChildException pchan e = do
295handleChildException pchan e = do
297 tid <- myThreadId
296 tid <- myThreadId
298− atomically $ writeTChan pchan (ChildStop tid (Just e))
297+ atomically $ writeTBQueue pchan (ChildStop tid (Just e))
300-- | Split the original computation in a pull-push pair. The original
299-- | Split the original computation in a pull-push pair. The original
301-- computation pulls from a Channel while m1 and m2 push to the channel.
300-- computation pulls from a Channel while m1 and m2 push to the channel.
@@ -319,10 +318,10 @@ pullFork m1 m2 = AsyncT $ \_ stp yld -> do
319 liftIO $ modifyIORef (runningThreads ctx) $ (\s -> S.insert tid s)
318 liftIO $ modifyIORef (runningThreads ctx) $ (\s -> S.insert tid s)
321 newContext = do
320 newContext = do
322− channel <- atomically newTChan
321+ channel <- atomically $ newTBQueue 16
323− running <- liftIO $ newIORef S.empty
322+ running <- newIORef S.empty
324− done <- liftIO $ newIORef S.empty
323+ done <- newIORef S.empty
325− count <- liftIO $ newQSem 1
324+ count <- newQSem 1
326 return $ Context { childChannel = channel
325 return $ Context { childChannel = channel
327 , dispatchReq = count
326 , dispatchReq = count
328 , runningThreads = running
327 , runningThreads = running
@@ -365,7 +364,7 @@ fork :: MonadAsync m => Context a -> AsyncT m a -> m ()
365fork ctx m = do
364fork ctx m = do
366 let chan = childChannel ctx
365 let chan = childChannel ctx
367 tid <- doFork (push ctx m) (handleChildException chan)
366 tid <- doFork (push ctx m) (handleChildException chan)
368− liftIO $ atomically $ writeTChan chan (ChildCreate tid)
367+ liftIO $ atomically $ writeTBQueue chan (ChildCreate tid)
370instance MonadAsync m => Alternative (AsyncT m) where
369instance MonadAsync m => Alternative (AsyncT m) where
371 empty = mempty
370 empty = mempty