0

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?

10
  • Have you heard of the Producer Consumer Pattern? Apr 26, 2023 at 20:06
  • @candied_orange, I hadn't. Now I read about it, and that's what I'm trying to implement. And yet, I couldn't find the answer to my question.
    – g00dds
    Apr 26, 2023 at 20:34
  • Does it meet all your needs or is another requirement still outstanding? Apr 26, 2023 at 20:36
  • @candied_orange, the pattern is what I'm trying to implement, but as I read it defines a way to communicate and synchronize many threads, but it doesn't define a way to terminate that communication, and that is what I don't know how to implement better in my situation, so I still don't understand what solution you propose
    – g00dds
    Apr 26, 2023 at 20:54
  • 1
    How about after the consumer takes the sentinel value out, it puts another one back in?
    – user253751
    Apr 26, 2023 at 23:27

2 Answers 2

0
  • 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.

1

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.

2
  • Thank you for the answer. In my case, I want not just parallel execution, but efficient use of multiple CPU cores, that's why I use multiple threads and not asyncio. As for your approach, you are right, technically I can keep track of how many consumers there are, but it would require me do to more synchronization stuff because new threads may be created or some of them deleted just while the termination of the queue occurs. It also compicates code a little bit. At the end, if I don't find a better solution, I would use this.
    – g00dds
    Apr 26, 2023 at 21:13
  • @g00dds Note that you don't need to know the exact number of consumers. You just need the maximum that could exist. You could track a high-water mark for example. If there some sentinel messages left on the queue after all the consumers exit, that shouldn't be an issue. And, realistically, there's a maximum number of threads you can run before the overhead outweighs the performance.
    – JimmyJames
    Apr 27, 2023 at 14:39

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.