@@ -97,7 +97,7 @@ data ChildEvent a =
97 -- exceptions or using atomicModify
97 -- exceptions or using atomicModify
98data Context m a =
98data Context m a =
99 Context { outputQueue :: IORef [ChildEvent a]
99 Context { outputQueue :: IORef [ChildEvent a]
100− , synchOutQ :: MVar Bool -- wakeup mechanism for outQ
100+ , doorBell :: MVar Bool -- wakeup mechanism for outQ
101 , enqueue :: AsyncT m a -> IO ()
101 , enqueue :: AsyncT m a -> IO ()
102 , runqueue :: m ()
102 , runqueue :: m ()
103 , runningThreads :: IORef (Set ThreadId)
103 , runningThreads :: IORef (Set ThreadId)
@@ -219,7 +219,7 @@ send ctx msg = liftIO $ do
219 atomicModifyIORefCAS_ (outputQueue ctx) $ \es -> msg : es
219 atomicModifyIORefCAS_ (outputQueue ctx) $ \es -> msg : es
220 -- XXX need a memory barrier? The wake up must happen only after the
220 -- XXX need a memory barrier? The wake up must happen only after the
221 -- store has finished otherwise we can have lost wakeup problems.
221 -- store has finished otherwise we can have lost wakeup problems.
222− void $ tryPutMVar (synchOutQ ctx) True
222+ void $ tryPutMVar (doorBell ctx) True
224{-# INLINE sendStop #-}
224{-# INLINE sendStop #-}
225sendStop :: MonadIO m => Context m a -> m ()
225sendStop :: MonadIO m => Context m a -> m ()
@@ -319,13 +319,13 @@ sendWorkerWait ctx = do
319 -- sending a worker in a loop running into a livelock
319 -- sending a worker in a loop running into a livelock
320 done <- queueEmpty ctx
320 done <- queueEmpty ctx
321 when (not done) (pushWorker ctx)
321 when (not done) (pushWorker ctx)
322− void $ liftIO $ takeMVar (synchOutQ ctx)
322+ void $ liftIO $ takeMVar (doorBell ctx)
324-- Note: This is performance sensitive code.
324-- Note: This is performance sensitive code.
325{-# NOINLINE pullWorker #-}
325{-# NOINLINE pullWorker #-}
326pullWorker :: MonadAsync m => Context m a -> AsyncT m a
326pullWorker :: MonadAsync m => Context m a -> AsyncT m a
327pullWorker ctx = AsyncT $ \pctx stp yld -> do
327pullWorker ctx = AsyncT $ \pctx stp yld -> do
328− res <- liftIO $ tryTakeMVar (synchOutQ ctx)
328+ res <- liftIO $ tryTakeMVar (doorBell ctx)
329 when (isNothing res) $ sendWorkerWait ctx
329 when (isNothing res) $ sendWorkerWait ctx
330 list <- liftIO $ atomicModifyIORefCAS (outputQueue ctx) $ \x -> ([], x)
330 list <- liftIO $ atomicModifyIORefCAS (outputQueue ctx) $ \x -> ([], x)
331 (runAsyncT $ processEvents list) pctx stp yld
331 (runAsyncT $ processEvents list) pctx stp yld
@@ -377,7 +377,7 @@ pullFork m1 m2 fifo = AsyncT $ \_ stp yld -> do
377 pushL q m1 >> pushL q m2
377 pushL q m1 >> pushL q m2
378 let ctx =
378 let ctx =
379 Context { outputQueue = outQ
379 Context { outputQueue = outQ
380− , synchOutQ = outQMv
380+ , doorBell = outQMv
381 , runningThreads = running
381 , runningThreads = running
382 , runqueue = runqueueFIFO ctx q
382 , runqueue = runqueueFIFO ctx q
383 , enqueue = pushL q
383 , enqueue = pushL q
@@ -390,7 +390,7 @@ pullFork m1 m2 fifo = AsyncT $ \_ stp yld -> do
390 let checkEmpty = liftIO (readIORef q) >>= return . null
390 let checkEmpty = liftIO (readIORef q) >>= return . null
391 let ctx =
391 let ctx =
392 Context { outputQueue = outQ
392 Context { outputQueue = outQ
393− , synchOutQ = outQMv
393+ , doorBell = outQMv
394 , runningThreads = running
394 , runningThreads = running
395 , runqueue = runqueueLIFO ctx q
395 , runqueue = runqueueLIFO ctx q
396 , enqueue = enqueueLIFO q
396 , enqueue = enqueueLIFO q