Signaling between threads

Question

Application design

I have a c++ application that has a producer thread, multiple queues (created during run time) and consumer thread.

Producer thread gets data via Tcp/Ip and puts into the respective queue (for E.g., if data is type A and put into the Queue A).

The consumer thread currently loops the queues from 1 - n to process the data from each queue.

As per the requirement no need to track the queue that is last updated or least. As long as any of the queue is updated, consumer should process from 1 - n queues.

If any of the queues' size is more than the defined limit, producer thread will pop the first item before it inserts the new item (to manage the queue size).

Resource synchronization and signaling between threads:

In this implementation, consumer thread should sleep until there is no queue has data from the listener. consumer thread should wake up only if producer puts data into any one of the queues.

Multiple queues are synchronized between 2 threads using mutex. Event signaling is implemented between threads to wake up consumer thread whenever producer puts data into any of the queues.

However this way of signaling to wake up consumer thread, it is possible for the consumer to sleep although there is a data in any of the queues.

Issue:

Lets take this scenario, consider the consumer is processing n-th queue's data; at the same time it is possible for the producer to put data into the n-1, n-2 queue and signaling is not effective since the consumer is awake and processing n-th data. Once the consumer completes processing the n-th queue data, it will sleep and the data in n-1, n-2 will not be processes until any further signal is given by the listener.

How we can address this scenario? People are also advising to use semophore. Is semaphore relavant to this scenario?

Thanks in advance.

Answer 1

You can use select and wait with it on file descriptor made from signal -> so it can wait on timeout(select has them) and wake up when signal is received (signal must be masked & blocked). When signalfd (look man signalfd) is readable you can read from it a struct signalfd_siginfo and check ssi_signo for signal number (if it's the one you are using for communication).

Answer 2

A semaphore would work, yes.

But I'm not entirely certain if it's even necessary. It sounds like your problem is caused purely because the consumer thread fails to loop back after processing queue N. It should go to sleep only after it has seen N empty queues in succession, while holding a mutex to ensure that no entries were added in the mean time.

Of course, holding that mutex all the time is overkill. Instead, you should just keep looping, emptying queues one by one and counting how many empty queues you've seen. Once you've seen N empty queues in a row, take the mutex so you know no new entries can be added, and now recheck.

It does depend on your signalling mechanism. Robust signalling mechanisms allow you to signal a thread before it enters the check for that signal. This is necessary because you otherwise have a race condition.

Answer 3

This is the classical example for a C++11 std::condition_variable.

The condition in this case is the availability of consumable resources. If a consumer thread runs out of work, he waits on the condition variable which effectively puts him to sleep. The producer notifys after each insert to a queue. Care must be taken to arrange locking in a way that the contention on the queues is kept minimal, while still avoiding the scenario that a consumer misses a notify and goes to sleep although work is available.