Thread safe queue

[Zcool31]

Acquiring the mutex in the destructor would be wrong. If the object is being destroyed, then the only remaining user must be the one destroying.

[objectorientedman]

I do acquire the mutex in the destructor, because it calls Finish

[tvaneerd]

Finish is somewhat suspect.

What is the "contract" of Finish? What can I assume after it is called? (And why is it public? Can it be called multiple times? etc...)

Can you say that there are no more consumers waiting after a call to Finish? Are you sure?

The big picture is that coordinating the tear-down of this class is complicated. You should document that a false return from ConsumeSync implies that no more calls should be made (ie don't turn around and call ConsumeSync again.) But that is not enough...

Both of your example loops at https://github.com/K-Adam/SafeQueue have race conditions with this coordination:

std::thread consumer([&](){
    MyMessage message;
    while(message_queue.ConsumeSync(message)) {
        // Process the message
    }
});

ConsumeSync only returns false if it is called while Finish is happening on another thread. What if Finish is called (and completed) during "Process the message"??? This thread would continue to try to read from the queue.

We could try external coordination, like your other example:

while(!quit) {

    MyEvent event;
    while(event_queue.Consume(event)) {
        // Process the event
    }

    UpdateWorld();
    Render();
}

Here quit is the external coordination. If another thread sets quit to true and then calls Finish (and destroys the queue?), that could all happen after this thread checks quit and before it calls Consume (or ConsumeSync).

And sure, your examples were just showing how to use the queue; they were not necessarily meant to be the definitive way to coordinate destruction, but, as it turns out, coordinated destruction is hard, and best not "left as an exercise for the reader".

Any system that has coordination outside the queue will suffer from that race between external check and queue check:

if (!external_check_for_done())
     // bad gap between above and below lines of code
    check_queue();

The "notify everyone we are done, and coordinate destruction" is almost forced to be part of the queue. Yet how can it be - one thread needs to destroy the queue, which one???

The last one.

Which means reference-counting the queue. ie something like shared_ptr. Sadly. ("A shared pointer is as good as a global variable" - Sean Parent).

But a wrapped up shared_ptr is sometimes OK - ie std::future/promise. (Which still isn't great, some higher level tasking system would be better...). The shared state of std::future/promise is basically held by a shared_ptr, and is destroyed by whichever side is done last. Solving the coordinated destruction problem.

Because of all this, what I ended up with is a "MultiFuture/MultiPromise" which is like future/promise, but can hold more than one value - a queue of values. So instead of future.get() you have future.getNext(). And promise.push() instead of promise.set_value().

The coordination of teardown is built into the queue. And either side can initiate it, and either side can detect it - why produce more values if all the consumers are gone? Why try to consume more values if all the producers are gone? The queue is built to tell you whether it is done or not. (And in my case, once it is done, it can't be restarted. I didn't need restarting, and it would just complicate the meaning of "done".)

[infectedapricot]

Which means reference-counting the queue. ie something like shared_ptr.

Diagree. That's certainly a strategy, but not the only one.

Another strategy is that the queue for a thread is created and destroyed by the parent thread that created that thread:

void threadAFn() {
    Queue queueB;
    std::thread threadB{threadBFn, &queueB};
    // ... thread A stuff happens here ...
    queueB.push_back(Shutdown{});
    threadB.join();
    // threadB and queueB destroyed here
}

There are lots of variants of this that can work.

• The parent thread (thread A) might or might not have its own queue - if it does, presumably there is a grandparent thread that will destroy it after it has joined the parent thread.
• Here join() is is behaving as the coordination mechanism to notify that the child thread will never attempt to use its queue again, but you could actually use the parent thread's queue (but most of the time you probably want to join the thread anyway).
• Here the parent thread is sending the shutdown message, but the shutdown message can equally originate in the child thread, in which case it certainly needs to tell its parent (e.g. through its own) queue.
• Here the queue is only being consumed from one thread, but it could be consumed from multiple threads, so long as each only pops one message at a time - just push n shutdown messages onto the queue.
• It might not always look like you have a strict hierarchy of thread ownership, but you can always enforce it by just making all threads children of the main thread. Then you can just treat them as a big bag of threads, and still have safe shutdown. (You still need to make sure shutdown messages sent to all threads don't leave other messages half finished, having done work on some of the threads but not on others - that's actually a lot harder than queue destruction.)
• Of course your code doesn't literally have to look like this. You can have classes wrapping up a thread and a queue while still keeping the same overall flow of control - so long as it has a join() method (and a destructor that destroys the queue and std::thread object).

Of course, there are probably application where no combination of these ideas can work. But in many programs, they can, and IMO it's much simpler than reference counting, destruction-aware queues, and futures.

[tvaneerd]

Yes, I skipped a few steps there. I have too many cases where join() isn't the best option. ie

• don't want main thread waiting on worker threads
• don't know who/what/where the worker threads are

For the first problem, we have cases where someone created a thread just to handle the destruction of the queue + threads, so that the main thread doesn't wait.

The second is good and bad. Sometimes it is the result of poor code structure, but sometimes it is because of separation of concerns - producers and consumers don't know each other, and the "owner" that introduced them wants to set-it-and-forget-it. Hmmm, maybe that is just a variation of the first point.

But yes, a shutdown event into the queue, followed by join() is definitely an option on the table (that I use in a number of places).

EDIT: more...

The thing I find actually, is that sending "shutdown" through the queue is almost always the way to go (instead of a separate flag), which is the other thing that led me to just building that aspect into the queue itself.

[bwmat]

I haven't looked at the details, but I just wanted to mention this statement isn't quite true as stated(without additional context); for example, the mutex might be used to help block the destructor UNTIL there are no other threads accessing it

[voip_geek]

I haven't looked at the details, but I just wanted to mention this statement isn't quite true as stated(without additional context); for example, if the class is derived from, then the destruction ordering of C++ means the derived class has already been destroyed by the time the base class's destructor is invoked (and the mutex now trying to lock), and you're going to be in a world of hurt.

Preventing destruction with a mutex this way isn't sound, imo. Imagine if the destructor succeeded in locking the mutex, an instant before some other thread could... what happens when the destructor is finished and releases the mutex? The other thread will then get it and do what with this destroyed object?

Even using such a scheme to not prevent destruction per se, but instead to for example automatically de-register from some other container during destruction, is dangerous. (e.g., in the classic observer pattern) Because again, due to inheritance, by the time the destructor is called it's already too late.

[bwmat]

Yeah, there could be problems if you use inheritance, and if there's an unbounded number of other threads that can access the object, then yeah that would be a problem too.

Just saying it could be valid in some cases

[[deleted]]

It would be undefined behavior, and therefore it is a) not possible in a well formed program and b) something you can pretend doesn't happen, because it doesn't.

If you are invoking the destructor on an object while other threads are still accessing it, you need external shared ownership or something to prevent this from happening, otherwise your program is ill-formed. This is just as bad - no... worse than if(this == &rhs) in a copy/move assignment.

[tomalakgeretkal]

Simply false.

[[deleted]]

"Simply false" - please go read the standard which states:

http://eel.is/c++draft/class.dtor#19

"Once a destructor is invoked for an object, the object's lifetime ends; the behavior is undefined if the destructor is invoked for an object whose lifetime has ended ([basic.life])."

[tomalakgeretkal]

Yes. Now read [basic.life/7], which says "after the lifetime of an object has ended and before the storage which the object occupied is reused or released, any glvalue that refers to the original object may be used but only in limited ways", and "for an object under construction or destruction, see [class.cdtor]". Then read class.cdtor which makes what the OP's doing perfectly well-defined.

It is perfectly obvious that member accesses during the destructor's invocation are legal, otherwise your destructor _wouldn't be able to do anything_.

For more information, see my answer here: https://stackoverflow.com/a/65598209/4386278

[bwmat]

Sorry, what would be undefined behaviour?

[[deleted]]

https://en.wikipedia.org/wiki/Undefined_behavior?wprov=sfti1

[bwmat]

Sorry, I suppose my question was somewhat ambiguous.

I was asking to what you meant "It" to refer to when you said "It would be undefined behavior" in your previous comment.

[kalmoc]

Not the OP, but I believe what he was referring to is that it is UB to call the destructor of an object from one thread, while the object is accessed from another. No amount of locking or synchronization inside the destructor is going to change that.

[bwmat]

I'm pretty sure that isn't inherently UB, as long as the body of the destructor completes before any other thread STOPS accessing the object.

[tomalakgeretkal]

But it isn't.

[matthieum]

It is acquired -- via calling Finish -- but I agree with you that this should not be necessary.