@@ -28,8 +28,8 @@ import Control.Applicative (Alternative (..))
2828import Control.Concurrent (ThreadId, forkIO, killThread,
2929 myThreadId, newQSem, QSem,
3030 signalQSem, waitQSem)
31−import Control.Concurrent.STM (TChan, atomically, newTChan,
32− tryReadTChan, writeTChan)
31+import Control.Concurrent.STM (TBQueue, atomically, newTBQueue,
32+ tryReadTBQueue, writeTBQueue)
3333import Control.Exception (SomeException (..))
3434import qualified Control.Exception.Lifted as EL
3535import Control.Monad (ap, liftM, MonadPlus(..), mzero)
@@ -85,7 +85,7 @@ data ChildEvent a =
8585------------------------------------------------------------------------------
8686
8787data Context a =
88− Context { childChannel :: TChan (ChildEvent a)
88+ Context { childChannel :: TBQueue (ChildEvent a)
8989 , dispatchReq :: QSem
9090 , runningThreads :: IORef (Set ThreadId)
9191 , doneThreads :: IORef (Set ThreadId)
@@ -193,13 +193,13 @@ push context action = do
193193 -- XXX make this a bounded channel so that we block if the previous value
194194 -- is not consumed yet.
195195 chan = childChannel context
196− channelYield a = liftIO $ atomically $ writeTChan chan (ChildYield a)
196+ channelYield a = liftIO $ atomically $ writeTBQueue chan (ChildYield a)
197197 channelDone a = do
198198 tid <- liftIO myThreadId
199− liftIO $ atomically $ writeTChan chan (ChildDone tid a)
199+ liftIO $ atomically $ writeTBQueue chan (ChildDone tid a)
200200 channelStop = do
201201 tid <- liftIO myThreadId
202− liftIO $ atomically $ writeTChan chan (ChildStop tid Nothing)
202+ liftIO $ atomically $ writeTBQueue chan (ChildStop tid Nothing)
203203
204204 done a = channelDone a
205205 continue a ctx m = channelYield a >> run ctx m
@@ -249,14 +249,13 @@ handleException e ctx tid = do
249249 liftIO $ readIORef (runningThreads ctx) >>= mapM_ killThread
250250 throwM e
251251
252−-- XXX the TBQueue size should be proportional to the pending threads.
253252-- We re-raise any exceptions received from the child threads, that way
254253-- exceptions get propagated to the top level computation and can be handled
255254-- there.
256255{-# NOINLINE pullDrain #-}
257256pullDrain :: (MonadIO m, MonadThrow m) => Context a -> AsyncT m a
258257pullDrain ctx = AsyncT $ \_ stp yld -> do
259− res <- liftIO $ atomically $ tryReadTChan (childChannel ctx)
258+ res <- liftIO $ atomically $ tryReadTBQueue (childChannel ctx)
260259 maybe (dispatch >> continue stp yld)
261260 (\ev -> handleEvent ev stp yld) res
262261
@@ -292,10 +291,10 @@ pullDrain ctx = AsyncT $ \_ stp yld -> do
292291-- the parent. 'Paused' exceptions are to be collected at the top level.
293292-- XXX Paused exceptions should only bubble up to the runRecorder computation
294293{-# NOINLINE handleChildException #-}
295−handleChildException :: TChan (ChildEvent a) -> SomeException -> IO ()
294+handleChildException :: TBQueue (ChildEvent a) -> SomeException -> IO ()
296295handleChildException pchan e = do
297296 tid <- myThreadId
298− atomically $ writeTChan pchan (ChildStop tid (Just e))
297+ atomically $ writeTBQueue pchan (ChildStop tid (Just e))
299298
300299-- | Split the original computation in a pull-push pair. The original
301300-- computation pulls from a Channel while m1 and m2 push to the channel.
@@ -319,10 +318,10 @@ pullFork m1 m2 = AsyncT $ \_ stp yld -> do
319318 liftIO $ modifyIORef (runningThreads ctx) $ (\s -> S.insert tid s)
320319
321320 newContext = do
322− channel <- atomically newTChan
323− running <- liftIO $ newIORef S.empty
324− done <- liftIO $ newIORef S.empty
325− count <- liftIO $ newQSem 1
321+ channel <- atomically $ newTBQueue 16
322+ running <- newIORef S.empty
323+ done <- newIORef S.empty
324+ count <- newQSem 1
326325 return $ Context { childChannel = channel
327326 , dispatchReq = count
328327 , runningThreads = running
@@ -365,7 +364,7 @@ fork :: MonadAsync m => Context a -> AsyncT m a -> m ()
365364fork ctx m = do
366365 let chan = childChannel ctx
367366 tid <- doFork (push ctx m) (handleChildException chan)
368− liftIO $ atomically $ writeTChan chan (ChildCreate tid)
367+ liftIO $ atomically $ writeTBQueue chan (ChildCreate tid)
369368
370369instance MonadAsync m => Alternative (AsyncT m) where
371370 empty = mempty