How to terminate python queue and instruct all consumer-threads to finish their tasks?

Question

I have a multi-threaded application. There is 1 thread that produces a resource and puts it into a queue and many (not constant amount) consumer-threads that get the resources from the queue.

When there are no more resources that the first thread procudes, the first thread must terminate the queue, that is, notify all the consumer-threads to finish their tasks and don't wait for the resource any more.

How to terminate the queue?
One approach is to put a sentinel value (like None) into the queue by the first thread so that all the consumer threads after receiving the sentinel value from the queue know that the queue has been terminated and they must finish their tasks. The problem is that the producer thread needs to put as much sentinel values into the queue as many consumer threads there are. If it puts less sentinel values than the amount of the consumer threads, the left consumer threads will wait forever for an item in the queue. Like I said, the amount of consumer-threads is not constant, and for now, the producer thread doesn't know their amount. It feels like a bad and unstable approach, because the producer must know the exact amount of consumers at the specific point in time, which might be changed.

Another approach I can think about is to have a Queue-wrapper class, the only purpose of which is to add a terminated-flag. When the queue is to be terminated, the flag is set, and all further calls to get method of the queue would result in the sentinel value. The problem with this approach is that when setting terminated-flag it doesn't kick the already waiting consumer threads from waiting, so they will still be waiting forever even after setting the terminated-flag.

I can think of some other approaches like creating my own custom queue with deque and threading.Event and the terminated-flag, but I don't feel like it's a proper solution, after all people use the built-in Queue somehow.

What approach is used in practice?

Answer 1

• The suggestion by @user253751 is actually a reasonable approach, and it should work with most queuing systems:

  When a worker receives a sentinel from the queue, it stops consuming that queue (possibly using an unsubscribe operation, but depending on the system just terminating itself after the next step might be enough) and puts the sentinel into the queue for the next worker to process. I've used this approach occasionally, and it works well enough. I don't think there's a race condition unless the worker can die between receiving the sentinel and re-inserting it into the queue (which is always possible but it's hard to handle panic failures robustly).

• Another option would be for workers to receive resources with some timeout if that is supported, and check a global termination flag to see whether they should continue working each time through the loop. Some pseudocode sketch:

  while (!termination_flag) {
    resource = queue.receive_with_timeout(TIMEOUT_DURATION);
    if (resource != TIMED_OUT) {
      process(resource);
    }
  }
  

  This causes a low level of busy-waiting when resources are produced at a much lower rate than workers can process them, but depending on the choice of TIMEOUT_DURATION that should be pretty negligible.

Answer 2

You mention multithreading here which is something I am pretty familiar with in other languages but in Python, I have focused on asyncio. Perhaps this is helpful, I'm not sure. If you haven't considered asyncio, there are a lot of advantages to it.

I've worked on a similar (but perhaps simpler) problem in Python and the questions you are asking are very familiar. In my case I ended up using a combination of the sentinel values (I knew how many consumers there were) and a flag. However, I was never quite satisfied with this. The documentation suggests to me that this kind of issue is handled. I was just too lazy or time-constrained to work through it.

First off, it seems to me that you should be able to determine how many consumers exist at a given point in time. At the very least, I would expect knowledge of an upper-bound. Somewhere in your code, you must be creating them and that's where you could track them. I think this is important w/ regard to using the built in Python library features I discuss below. In any event, as long as you can work out an upper-bound on the number of consumers, you should be able to add that many sentinel values to the queue. As long as each consumer will stop reading once it sees the sentinel value, this should be enough to get them all to stop waiting and exit. I prefer a special object or string that is not None BTW. None could be created in a lot of ways. A unique object that you create and check by instance identity is a lot less likely to get added to the queue due to a bug. If there's some reason this can't work, please let me know in the comments.

That aside, if we look at the asyncio.Queue documentation, there's an interesting example at the end. I'm not going to try to explain all of it but the part that seems relevant here is task.cancel() call. As I understand it, this will cause the task to throw an exception at the next possible moment. I would love to hear what you come up with.