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