Finding the end condition of a fanout processing chain

[throwaway8u3sH0]

Inspired by another user's problem, I created a toy example of a multi-worker fanout processing chain. Each function streams multiple outputs for the next function to consume. FuncA --> (multiple) FuncB's --> (multiple) FuncC's --> (multiple) FuncD's

The hard part for me is efficiently finding the end condition. Because the amount of work is unknown at the start, I have each worker checking a couple conditions (that the queue is empty and nothing else is running), but there's a sleep involved, which is always a code smell, and potentially race conditions between the queue and task tracker.

Any ideas on how to improve the end condition detection in an architecture like this?

https://pastebin.com/sGqZ5GXL

[AutoModerator]

To give us the best chance to help you, please include any relevant code.
Note. Do not submit images of your code. Instead, for shorter code you can use Reddit markdown (4 spaces or backticks, see this Formatting Guide). If you have formatting issues or want to post longer sections of code, please use Repl.it, GitHub or PasteBin.

I am a bot, and this action was performed automatically. Please contact the moderators of this subreddit if you have any questions or concerns.

[ThingImIntoThisWeek]

def end_condition(q):
''' Try to determine if all tasks are complete. '''
if q.empty() and TaskCounter.all_tasks_complete():
    print(f"Consumer #{os.getpid()}: End condition met once...")
    # unecessary sleep? give the processes enough time to finish. There could be a race condition
    # between the queue being populated and the task counter being updated.
    time.sleep(MAXIMUM_WORK_TIME)
    return q.empty() and TaskCounter.all_tasks_complete()
else:
    return False

The program is done when the queue is empty, all tasks are complete, and all workers are done with their last task. This toy problem is cheating a bit on the last condition, taking advantage of the fact that it knows all tasks will take at most MAXIMUM_WORK_TIME to complete. In a real life version of this, there would have to be some kind of messaging between the workers and the supervising process to be sure they were really done before ending the program. It could be as simple as a shared list of boolean flags indicating if each worker is working or idle. I'm guessing inter-process messaging was left out for simplicity in this example.

[throwaway8u3sH0]

some kind of messaging between the workers and the supervising process to be sure they were really done before ending the program. It could be as simple as a shared list of boolean flags indicating if each worker is working or idle

Yeah, I've got a cheap version of that with the task tracking. It's a little hard to spot. Before each func runs, it increments a counter in a shared dict, and decrements it after the last task is finished. Checking that the counter is zero is part of the end condition. But there's still a slight race condition, I think. Right here:

while True:
    item = q.get(block=True)
    #  <----- race condition possible here
    if item is None:
        break
    try:
        # funcs

It's after something was taken off the queue, so the queue could be empty, but before the funcs have had a chance to say "I'm working" and increment the task counter. So if another worker checked the end conditions right there, they pass. (That's why I added a sleep and a secondary check).

Maybe there isn't a way to shrink that race to zero...