@@ -86,8 +86,8 @@ data ChildEvent a =
86------------------------------------------------------------------------------
86------------------------------------------------------------------------------
88data Context m a =
88data Context m a =
89− Context { childChannel :: TBQueue (ChildEvent a)
89+ Context { outputQueue :: TBQueue (ChildEvent a)
90− , pendingWork :: TBQueue (AsyncT m a)
90+ , workQueue :: TBQueue (AsyncT m a)
91 , runningThreads :: IORef (Set ThreadId)
91 , runningThreads :: IORef (Set ThreadId)
92 , doneThreads :: IORef (Set ThreadId)
92 , doneThreads :: IORef (Set ThreadId)
@@ -185,9 +185,9 @@ push ctx = run (Just ctx) (dequeueLoop ctx)
188− send msg = atomically $ writeTBQueue (childChannel ctx) msg
188+ send msg = atomically $ writeTBQueue (outputQueue ctx) msg
189 stop = do
189 stop = do
190− workQEmpty <- liftIO $ atomically $ isEmptyTBQueue (pendingWork ctx)
190+ workQEmpty <- liftIO $ atomically $ isEmptyTBQueue (workQueue ctx)
191 if (not workQEmpty) then push ctx
191 if (not workQEmpty) then push ctx
192 else liftIO $ myThreadId >>= \tid -> send (ChildStop tid Nothing)
192 else liftIO $ myThreadId >>= \tid -> send (ChildStop tid Nothing)
193 yield a _ Nothing = liftIO $ myThreadId >>= \tid -> send (ChildDone tid a)
193 yield a _ Nothing = liftIO $ myThreadId >>= \tid -> send (ChildDone tid a)
@@ -250,16 +250,16 @@ pullWorker ctx = AsyncT $ \pctx stp yld -> do
250 done <- f ctx tid
250 done <- f ctx tid
251 if done then finish else cont
251 if done then finish else cont
253− res <- liftIO $ atomically $ tryReadTBQueue (childChannel ctx)
253+ res <- liftIO $ atomically $ tryReadTBQueue (outputQueue ctx)
254 case res of
254 case res of
255 Nothing -> do
255 Nothing -> do
256 liftIO $ threadDelay 4
256 liftIO $ threadDelay 4
257− let workQ = pendingWork ctx
257+ let workQ = workQueue ctx
258− outQ = childChannel ctx
258+ outQ = outputQueue ctx
259 workQEmpty <- liftIO $ atomically $ isEmptyTBQueue workQ
259 workQEmpty <- liftIO $ atomically $ isEmptyTBQueue workQ
260 outQEmpty <- liftIO $ atomically $ isEmptyTBQueue outQ
260 outQEmpty <- liftIO $ atomically $ isEmptyTBQueue outQ
261 when (not workQEmpty && outQEmpty) $ pushWorker ctx
261 when (not workQEmpty && outQEmpty) $ pushWorker ctx
262− void $ liftIO $ atomically $ peekTBQueue (childChannel ctx)
262+ void $ liftIO $ atomically $ peekTBQueue (outputQueue ctx)
264 Just ev ->
264 Just ev ->
265 case ev of
265 case ev of
@@ -289,8 +289,7 @@ handleChildException pchan e = do
289{-# INLINE pushWorker #-}
289{-# INLINE pushWorker #-}
290pushWorker :: MonadAsync m => Context m a -> m ()
290pushWorker :: MonadAsync m => Context m a -> m ()
291pushWorker ctx = do
291pushWorker ctx = do
292− let chan = childChannel ctx
293− tid <- doFork (push ctx) (handleChildException chan)
292+ tid <- doFork (push ctx) (handleChildException (outputQueue ctx))
294 liftIO $ modifyIORef (runningThreads ctx) $ (\s -> S.insert tid s)
293 liftIO $ modifyIORef (runningThreads ctx) $ (\s -> S.insert tid s)
296-- | Split the original computation in a pull-push pair. The original
295-- | Split the original computation in a pull-push pair. The original
@@ -305,14 +304,14 @@ pullFork m1 m2 = AsyncT $ \_ stp yld -> do
307 newContext = do
306 newContext = do
308− channel <- atomically $ newTBQueue 32
309− work <- atomically $ newTBQueue 32
307+ outQ <- atomically $ newTBQueue 32
308+ workQ <- atomically $ newTBQueue 32
310 running <- newIORef S.empty
309 running <- newIORef S.empty
311 done <- newIORef S.empty
310 done <- newIORef S.empty
312− return $ Context { childChannel = channel
311+ return $ Context { outputQueue = outQ
313 , runningThreads = running
312 , runningThreads = running
314 , doneThreads = done
313 , doneThreads = done
315− , pendingWork = work
314+ , workQueue = workQ
318-- Concurrency rate control. Our objective is to create more threads on
317-- Concurrency rate control. Our objective is to create more threads on
@@ -349,9 +348,9 @@ pullFork m1 m2 = AsyncT $ \_ stp yld -> do
349{-# INLINE forkWorker #-}
348{-# INLINE forkWorker #-}
350forkWorker :: MonadAsync m => Context m a -> m ()
349forkWorker :: MonadAsync m => Context m a -> m ()
351forkWorker ctx = do
350forkWorker ctx = do
352− let chan = childChannel ctx
351+ let q = outputQueue ctx
353− tid <- doFork (push ctx) (handleChildException chan)
352+ tid <- doFork (push ctx) (handleChildException q)
354− liftIO $ atomically $ writeTBQueue chan (ChildCreate tid)
353+ liftIO $ atomically $ writeTBQueue q (ChildCreate tid)
356{-# INLINE queueWork #-}
355{-# INLINE queueWork #-}
357queueWork :: MonadAsync m => Context m a -> AsyncT m a -> m ()
356queueWork :: MonadAsync m => Context m a -> AsyncT m a -> m ()
@@ -363,14 +362,14 @@ queueWork ctx m = do
363 -- TBD If we run out of threads we can also evaluate the action completely
362 -- TBD If we run out of threads we can also evaluate the action completely
364 -- right here, disallowing any further child workers and turning the
363 -- right here, disallowing any further child workers and turning the
365 -- parallel composition into interleaved serial composition.
364 -- parallel composition into interleaved serial composition.
366− workQFull <- liftIO $ atomically $ isFullTBQueue (pendingWork ctx)
365+ workQFull <- liftIO $ atomically $ isFullTBQueue (workQueue ctx)
367 when (workQFull) $ forkWorker ctx
366 when (workQFull) $ forkWorker ctx
368− liftIO $ atomically $ writeTBQueue (pendingWork ctx) m
367+ liftIO $ atomically $ writeTBQueue (workQueue ctx) m
370{-# INLINE dequeueLoop #-}
369{-# INLINE dequeueLoop #-}
371dequeueLoop :: MonadAsync m => Context m a -> AsyncT m a
370dequeueLoop :: MonadAsync m => Context m a -> AsyncT m a
372dequeueLoop ctx = AsyncT $ \_ stp yld -> do
371dequeueLoop ctx = AsyncT $ \_ stp yld -> do
373− work <- liftIO $ atomically $ tryReadTBQueue (pendingWork ctx)
372+ work <- liftIO $ atomically $ tryReadTBQueue (workQueue ctx)
374 case work of
373 case work of
375 Nothing -> stp
374 Nothing -> stp
376 Just m -> do
375 Just m -> do