@@ -291,81 +291,88 @@ sendWorkerWait
291291 -> (Channel m a -> m Bool)
292292 -> Channel m a
293293 -> m ()
294−sendWorkerWait eager delay dispatch sv = do
295− -- Note that we are guaranteed to have at least one outstanding worker when
296− -- we enter this function. So if we sleep we are guaranteed to be woken up
297− -- by an outputDoorBell, when the worker exits.
298−
299− liftIO $ delay sv
300− (_, n) <- liftIO $ readIORef (outputQueue sv)
301− when (n <= 0 || eager) $ do
302− -- The queue may be empty temporarily if the worker has dequeued the
303− -- work item but has not enqueued the remaining part yet. For the same
304− -- reason, a worker may come back if it tries to dequeue and finds the
305− -- queue empty, even though the whole work has not finished yet.
306−
307− -- If we find that the queue is empty, but it may be empty
308− -- temporarily, when we checked it. If that's the case we might
309− -- sleep indefinitely unless the active workers produce some
310− -- output. We may deadlock specially if the otuput from the active
311− -- workers depends on the future workers that we may never send.
312− -- So in case the queue was temporarily empty set a flag to inform
313− -- the enqueue to send us a doorbell.
314−
315− -- Note that this is just a best effort mechanism to avoid a
316− -- deadlock. Deadlocks may still happen if for some weird reason
317− -- the consuming computation shares an MVar or some other resource
318− -- with the producing computation and gets blocked on that resource
319− -- and therefore cannot do any pushworker to add more threads to
320− -- the producer. In such cases the programmer should use a parallel
321− -- style so that all the producers are scheduled immediately and
322− -- unconditionally. We can also use a separate monitor thread to
323− -- push workers instead of pushing them from the consumer, but then
324− -- we are no longer using pull based concurrency rate adaptation.
325− --
326− -- XXX update this in the tutorial.
327− --
328− -- Having pending active workers does not mean that we are guaranteed
329− -- to be woken up if we sleep. In case of Ahead streams, there may be
330− -- queued items in the heap even though the outputQueue is empty, and
331− -- we may have active workers which are deadlocked on those items to be
332− -- processed by the consumer. We should either guarantee that any
333− -- worker, before returning, clears the heap or we send a worker to
334− -- clear it. Normally we always send a worker if no output is seen, but
335− -- if the thread limit is reached or we are using pacing then we may
336− -- not send a worker. See the concurrentApplication test in the tests,
337− -- that test case requires at least one yield from the producer to not
338− -- deadlock, if the last workers output is stuck in the heap then this
339− -- test fails. This problem can be extended to n threads when the
340− -- consumer may depend on the evaluation of next n items in the
341− -- producer stream.
342−
343− -- register for the outputDoorBell before we check the queue so that if
344− -- we sleep because the queue was empty we are guaranteed to get a
345− -- doorbell on the next enqueue.
346−
347− liftIO $ atomicModifyIORefCAS_ (needDoorBell sv) $ const True
348− liftIO storeLoadBarrier
349− canDoMore <- dispatch sv
350−
351− -- XXX test for the case when we miss sending a worker when the worker
352− -- count is more than 1500.
353− --
354− -- XXX Assert here that if the heap is not empty then there is at
355− -- least one outstanding worker. Otherwise we could be sleeping
356− -- forever.
357−
358− if canDoMore
359− then sendWorkerWait eager delay dispatch sv
360− else do
361− liftIO
362− $ withDiagMVar
363− (svarInspectMode sv)
364− (dumpSVar sv)
365− "sendWorkerWait: nothing to do"
366− $ takeMVar (outputDoorBell sv)
367− (_, len) <- liftIO $ readIORef (outputQueue sv)
368− when (len <= 0) $ sendWorkerWait eager delay dispatch sv
294+sendWorkerWait eager delay dispatch sv = go
295+
296+ where
297+
298+ go = do
299+
300+ -- Note that we are guaranteed to have at least one outstanding worker
301+ -- when we enter this function. So if we sleep we are guaranteed to be
302+ -- woken up by an outputDoorBell, when the worker exits.
303+
304+ liftIO $ delay sv
305+ (_, n) <- liftIO $ readIORef (outputQueue sv)
306+ when (n <= 0 || eager) $ do
307+ -- The queue may be empty temporarily if the worker has dequeued
308+ -- the work item but has not enqueued the remaining part yet. For
309+ -- the same reason, a worker may come back if it tries to dequeue
310+ -- and finds the queue empty, even though the whole work has not
311+ -- finished yet.
312+
313+ -- If we find that the queue is empty, but it may be empty
314+ -- temporarily, when we checked it. If that's the case we might
315+ -- sleep indefinitely unless the active workers produce some
316+ -- output. We may deadlock specially if the otuput from the active
317+ -- workers depends on the future workers that we may never send.
318+ -- So in case the queue was temporarily empty set a flag to inform
319+ -- the enqueue to send us a doorbell.
320+
321+ -- Note that this is just a best effort mechanism to avoid a
322+ -- deadlock. Deadlocks may still happen if for some weird reason
323+ -- the consuming computation shares an MVar or some other resource
324+ -- with the producing computation and gets blocked on that resource
325+ -- and therefore cannot do any pushworker to add more threads to
326+ -- the producer. In such cases the programmer should use a parallel
327+ -- style so that all the producers are scheduled immediately and
328+ -- unconditionally. We can also use a separate monitor thread to
329+ -- push workers instead of pushing them from the consumer, but then
330+ -- we are no longer using pull based concurrency rate adaptation.
331+ --
332+ -- XXX update this in the tutorial.
333+ --
334+ -- Having pending active workers does not mean that we are
335+ -- guaranteed to be woken up if we sleep. In case of Ahead streams,
336+ -- there may be queued items in the heap even though the
337+ -- outputQueue is empty, and we may have active workers which are
338+ -- deadlocked on those items to be processed by the consumer. We
339+ -- should either guarantee that any worker, before returning,
340+ -- clears the heap or we send a worker to clear it. Normally we
341+ -- always send a worker if no output is seen, but if the thread
342+ -- limit is reached or we are using pacing then we may not send a
343+ -- worker. See the concurrentApplication test in the tests, that
344+ -- test case requires at least one yield from the producer to not
345+ -- deadlock, if the last workers output is stuck in the heap then
346+ -- this test fails. This problem can be extended to n threads when
347+ -- the consumer may depend on the evaluation of next n items in the
348+ -- producer stream.
349+
350+ -- register for the outputDoorBell before we check the queue so
351+ -- that if we sleep because the queue was empty we are guaranteed
352+ -- to get a doorbell on the next enqueue.
353+
354+ liftIO $ atomicModifyIORefCAS_ (needDoorBell sv) $ const True
355+ liftIO storeLoadBarrier
356+ canDoMore <- dispatch sv
357+
358+ -- XXX test for the case when we miss sending a worker when the
359+ -- worker count is more than 1500.
360+ --
361+ -- XXX Assert here that if the heap is not empty then there is at
362+ -- least one outstanding worker. Otherwise we could be sleeping
363+ -- forever.
364+
365+ if canDoMore
366+ then go
367+ else do
368+ liftIO
369+ $ withDiagMVar
370+ (svarInspectMode sv)
371+ (dumpSVar sv)
372+ "sendWorkerWait: nothing to do"
373+ $ takeMVar (outputDoorBell sv)
374+ (_, len) <- liftIO $ readIORef (outputQueue sv)
375+ when (len <= 0) go
369376
370377-- | Start the evaluation of the channel's work queue by kicking off a worker.
371378-- Note: Work queue must not be empty otherwise the worker will exit without