c++ multithreaded message broadcaster with link lifetime management v3

Question

note: This is v3 of code that was previously reviewed.

I have written a class that handles listeners registering callbacks to receive messages. Link lifetime is managed (or is it? See below). Code is thread-safe in that any thread can broadcast to the listeners, and any thread can add or remove listeners. There is one (documented) exception to this safety in that calling add_listener from inside an invoked listener would deadlock. A live example is on godbolt.

I have three concerns:

1. Despite my attempt at lifetime management, i think there is a race condition in notify_all() in the following case:

   1. broadcaster obtains shared pointer (tmp.push_back(target.lock()))
   2. before it can invoke the callback, the listener destructs (imagine its a ref to a class member function);
   3. Listener is invoked (targetPtr->operator()(msg_...)), but points to an object that no longer exists -> boom

   I'm pretty sure that is an issue, but am not sure how to avoid it. Perhaps each listener could come with a mutex on the other side of the link that is locked each time a listener is invoked and that the listening side of the class tries to acquire upon destruction? Or some other kind of double locking pattern? Thats very hairy and I guess i run a big risk of shooting myself in the foot. In any case, i can't quite wrap my head around how it would look.

2. It is possible that the broadcaster is destructed while another thread has invoked one of the member functions, and thus holds (or is trying to acquire) the mutex. That would crash, and i have tried to minimize the problem by acquiring both mutexes in the Broadcaster destructor. Yet, as discussed here, that is not a great solution nor the only problem. If another thread is waiting to acquire the mutex as the destructor is run, we're still dead. As the answeres noted there, it is an issue of resource management. I have seen this as a potential solution, but find it hard to oversee if it is. Those of you who know better, is this road worth exploring?

3. As G. Sliepen notes in his suggestions on v2 of this code, it is rather expensive that std::shared_ptrs are being acquired and released each time a listener is invoked. He suggests an alternative implementation in which the vector of listeners is guarded by a shared pointer, and with a separate cookie class containing a weak_ptr to the whole vector of listeners. If the cookie gets destructed/reset, it deletes itself from the listeners if the vector of listeners is still alive. With suitable locking. I cannot wrap my head around how this would look however, so have not implemented it. The above two are the more important issues anyway, as they cause crashes or deadlocks of the hard to debug kind.

Implementation

#include <vector>
#include <mutex>
#include <memory>
#include <functional>
#include <type_traits>
#include <algorithm>

// based on https://stackoverflow.com/a/47872677
// usage notes:
// 1. It is safe to call any of the member functions
//    from inside a listener invoked by this class,
//    except notify_all(). Calling notify_all() from
//    inside a listener will deadlock.
// 2. Listeners added from inside listener callback
//    will participate starting from the next message.
// 3. This class guarantees message ordering with
//    multiple producer threads.
// 4. add_listener() returns a cookie that you need to
//    hold on to for as long as you want to receive
//    messages. Call reset() on the cookie to
//    deregister and no longer receive messages.
template <class... Message>
class Broadcaster
{
public:
    using listener = std::function<void(Message...)>;
    using cookieType = std::shared_ptr<void>;

    ~Broadcaster()
    {
        // acquire locks to guarantee we do not destruct while they're held by other threads
        std::scoped_lock l{ _listenMut,_senderMut };
    }

    // returns number of registered listeners
    size_t num_listeners() const noexcept
    {
        auto ll = std::unique_lock(_listenMut);
        return std::ranges::count_if(_listeners, [](const auto& m) { return !m.expired(); });
    }

    // clears all listeners
    void clear() noexcept
    {
        auto ll = std::unique_lock(_listenMut);
        _listeners.clear();
    }

    template <class F>
    requires (std::is_nothrow_invocable_r_v<void, F, Message...>)
    [[nodiscard]]
    cookieType add_listener(F&& r_)
    {
        auto ll = std::unique_lock(_listenMut);
        auto listenFunc = std::make_shared<listener>(std::forward<F>(r_));
        _listeners.push_back(listenFunc);
        return listenFunc;
    }

    void notify_all(const Message&... msg_) noexcept
    {
        auto ls = std::unique_lock(_senderMut);   // to guarantee message ordering
        std::vector<std::shared_ptr<listener>> tmp;

        {
            // remove dead listeners
            auto ll = std::unique_lock(_listenMut);
            std::erase_if(_listeners, [](const auto& ptr) { return ptr.expired(); });

            // take copy. This may yield a handle to nothing if other side
            // stopped listening between above pruning and this call. That
            // is ok, handled below
            for (auto& target : _listeners)
                tmp.push_back(target.lock());

            // listener mutex will now be unlocked, so new listeners can be added
        }

        for (auto&& targetPtr : tmp)
            if (targetPtr)  // check that we have a handle to a listener
                targetPtr->operator()(msg_...);
    }

private:
    mutable std::mutex                      _listenMut;
    std::mutex                              _senderMut;
    std::vector<std::weak_ptr<listener>>    _listeners;
};

Example usage

#include <iostream>
#include <string>
#include <functional>

void freeTestFunction(std::string msg_) noexcept
{
    std::cout << "from freeTestFunction: " << msg_ << std::endl;
}
struct test
{
    using StringBroadcaster = Broadcaster<std::string>;

    void simpleCallback(std::string msg_) noexcept
    {
        std::cout << "from simpleCallback: " << msg_ << std::endl;
    }
    void oneShotCallback(std::string msg_) noexcept
    {
        std::cout << "from oneShotCallback: " << msg_ << std::endl;
        _cb_oneShot_cookie.reset();
    }
    void twoStepCallback_step1(std::string msg_) noexcept
    {
        std::cout << "from twoStepCallback_step1: " << msg_ << std::endl;

        // replace callback
        _cb_twostep_cookie = _broadcast.add_listener([&](auto fr_) noexcept { twoStepCallback_step2(fr_); });
    }
    void twoStepCallback_step2(std::string msg_) noexcept
    {
        std::cout << "from twoStepCallback_step2: " << msg_ << std::endl;
    }

    void runExample()
    {
        auto cb_simple_cookie = _broadcast.add_listener([&](auto fr_) noexcept { simpleCallback(fr_); });
        _cb_oneShot_cookie = _broadcast.add_listener([&](auto fr_) noexcept { oneShotCallback(fr_); });
        _cb_twostep_cookie = _broadcast.add_listener([&](auto fr_) noexcept { twoStepCallback_step1(fr_); });
        auto free_func_cookie = _broadcast.add_listener(&freeTestFunction);

        _broadcast.notify_all("message 1");  // should be received by simpleCallback, oneShotCallback, twoStepCallback_step1, freeTestFunction
        free_func_cookie.reset();
        _broadcast.notify_all("message 2");  // should be received by simpleCallback and twoStepCallback_step2
        cb_simple_cookie.reset();
        _broadcast.notify_all("message 3");  // should be received by twoStepCallback_step2
        _cb_twostep_cookie.reset();
        _broadcast.notify_all("message 4");  // should be received by none
    }

    StringBroadcaster _broadcast;
    StringBroadcaster::cookieType _cb_oneShot_cookie;
    StringBroadcaster::cookieType _cb_twostep_cookie;
};

int main(int argc, char **argv)
{
    test t;
    t.runExample();

    return 0;
}

Answer 1

        auto ls = std::unique_lock(_senderMut);   // to guarantee message ordering

this is toxic.

Multi-threaded code using locks does not compose. Code that is locally correct using locks, composed with other code that is locally correct, can be together incorrect.

And here we are holding a lock while we invoke an arbitrary number of callbacks. Those callbacks cannot be locally verified to compose properly with broadcaster; so this code is by design broken.

You cannot hold a lock while sending out messages from broadcaster. Whatever guarantee you are trying to get here, either don't have the guarantee or build a much more complex system to make it work (like a message queue?).

If your approach to multithreading is "I will provide as many guarantees as I can", your system will fall apart. Your approach should be "I will make a few guarantees as I can that still solve the problem and I can show to be correct".

Later in broadcast method:

        auto ll = std::unique_lock(_listenMut);

this is toxic. Don't acquire a mutex while you hold another, not without an extensive proof that it is safe, and an extensive justification why you should do it.

size_t num_listeners() const noexcept

Why in the world would you want to know the number of listeners that this broadcaster once had? Adding stuff to a multithreaded API should only be done when there is a significant reason for it, because multithreaded code is more than linearly hard, and writing less of it has high value.

void clear() noexcept

another operation that seems like a bad plan. The code registered for messages expect to get messages, and you are pulling the rug out from under them. The broadcaster still exists and is sending messages with the same rules, but it now has a different meaning?

I've used broadcaster heavily, and never had a use for this. Of course, I've been willing to attach broadcasters to specific contexts, so I 'clear' it by destroying the broadcaster.

In the destructor:

    std::scoped_lock l{ _listenMut,_senderMut };

if we are destroying a broadcaster unsequenced relative to code that has acquired these locks, we are already doomed. Getting these locks in the destructor is at best a false sense of security.

This also involves locking 2 mutexes. This is usually a flag something is going wrong.

        for (auto& target : _listeners)
            tmp.push_back(target.lock());

this means that if you unregister a listener in a previous listener, your request is ignored. Which is bad.

Instead, copy the weak pointers over, and lock before calling each one. Then if the previous listener caused the next listener to be unregistered, the next listener won't be called.

The only remaining case of danger is if the target listener is destroyed in another thread than the broadcaster broadcasts in. We'll still have a valid shared ptr to the listener callback, but that callback might end up using resources inside itself and go boom.

This specific case can be handled by the callback having internal lifetime management, or adding a add_listener method that takes a shared_ptr<reciever> and using the shared ptr aliasing ctor to extend the lifetime of the target's dependencies.

Often, in my experience, you'll know what thread objects die in, and can avoid this annoyance. If you don't, you have to be extra careful.

(This is one of the reasons why widespread use of shared_ptr is dangerous, because it makes lifetime extremely hard to understand.)

...

Despite my attempt at lifetime management, i think there is a race condition in notify_all() in the following case: broadcaster obtains shared pointer (tmp.push_back(target.lock())) before it can invoke the callback, the listener destructs (imagine its a ref to a class member function); Listener is invoked (targetPtr->operator()(msg_...)), but points to an object that no longer exists -> boom

In the multi-threaded destruction case, you the listener invocation has to remain valid so long as it is alive.

This can involve work. You can add a register function that takes a shared ptr, and use the aliasing ctor to extend the life of the invoked object while passing a std::function through. Or you can use a small lambda that handles such problems.

If you know about when the broadcasts happen and where the listened objects are destroyed, you can relax here.

In general, while the code is robust in simple multithreaded contexts, in complex ones using something as simple as broadcaster won't work. In complex contexts, listeners should generally know which thread they'll be getting messages in; such as the executor concept.

Having messages sent from random threads to objects executing in other random threads will not work in practice. Nothing you do will fix this, as even if the first order problems are patched (like crashes), second order problems remain.

This broadcaster permits you to have registration happen in thread 1 and broadcasting happen in thread 2, but it doesn't make all threaded interaction with it safe. It is possible to use it in that context, but there is no way lock-based code can make all other code that interacts with it safe. At best it can provide specific guarantees against an increasing number of cases.

This is because lock-based code fundamentally doesn't compose.

So it maintains its own local state, and provides a minimal number of guarantees.

If you want multi threaded code that composes, you'll want something like immutable shared objects passing messages through executors and locally pseudo-SIMD kernels run on uniform data. Which could mutexes under the hood as implementation details, but fundamentally works differently than this stuff.

It is possible that the broadcaster is destructed while another thread has invoked one of the member functions, and thus holds (or is trying to acquire) the mutex. That would crash, and i have tried to minimize the problem by acquiring both mutexes in the Broadcaster destructor. Yet, as discussed here, that is not a great solution nor the only problem. If another thread is waiting to acquire the mutex as the destructor is run, we're still dead. As the answeres noted there, it is an issue of resource management. I have seen this as a potential solution, but find it hard to oversee if it is. Those of you who know better, is this road worth exploring?

Simple: Never hold a mutex on anything when invoking callbacks. Never assume the broadcaster exists when returning from any callbacks.

Any time you leave local scope, you have to surrender your existence.

As G. Sliepen notes in his suggestions on v2 of this code, it is rather expensive that std::shared_ptrs are being acquired and released each time a listener is invoked. He suggests an alternative implementation in which the vector of listeners is guarded by a shared pointer, and with a separate cookie class containing a weak_ptr to the whole vector of listeners. If the cookie gets destructed/reset, it deletes itself from the listeners if the vector of listeners is still alive. With suitable locking. I cannot wrap my head around how this would look however, so have not implemented it. The above two are the more important issues anyway, as they cause crashes or deadlocks of the hard to debug kind.

An atomic increment is a medium cost operation. If you are using this at the scale of per-pixel times per-frame operations, you are utterly screwed. If you are using this on the scale of per-user-input or per-frame an atomic increment is very very cheap.

Answer 2

Lifetime of Broadcaster

~Broadcaster()
{
    // acquire locks to guarantee we do not destruct while they're held by other threads
    std::scoped_lock l{ _listenMut,_senderMut };
}

As to note 2, locking inside the destructor is a problem. I think you are trying to solve a problem at the wrong place. Think about it like this, you have two or more threads and to interact with the broadcaster they need a reference to the broadcaster. So as long as long as each thread has a strong reference to the broadcaster it will not be destructed. Your code should be written it a way that references to things should not "disapear" from under running code.

I would remove the entire thing, as this should not be an issue; and if it is the code using the Broadcaster class is not managing lifetime properly.

Cookies

(I love cookies, but never heard them used in this context.)

Your approach to auto deregistration is an interesting one. But I am not sure if that is a required feature. This is not in scope, I would think about using a add/remove semantics, then you can use a cheaper token type.

The shared_ptr/weak_ptr semantics are quite expensive. The good news you can be certain that it works. Keeping in the vain of the cookie auto derigistration, I would consider using a unitue_ptr to a struct containing the function and an atomic counter. Then wrapping a raw pointer into a Cookie class that increments and decrements the coutner.

This removes the double indirection in shared_ptr and the entire shared state. The downside is that you need to ensure Broadcaster will live longer than all Cookie instances.

You solution is not bad, it's expensive but I think anything similar will have a similar cost. You might want think about alternate ways of achieving that goal and see what the pros and cons are.

Sanity Checks at the Boundary

void notify_all(const Message&... msg_) noexcept
{
  // ...
  for (auto&& targetPtr : tmp)
      if (targetPtr)  // check that we have a handle to a listener
          targetPtr->operator()(msg_...);
}

You should not check the validity of the function here. This should have happened in add_listener. You can have an assert here if you like, but add_listener should check the validity and throw std::invalid_argument if the function is invalid.

Sender Mutex

As I understand _senderMut's job is to make sure that when sending an event will call all listeners before a different thread can send a new event.

There are are two options to consider:

1. the listeners run for a long time:

   Then you should not have _senderMut, since that would create unnecessary contention on the notify_all.

2. the listeners run really quickly:

   Then you should use only one mutex and remove the vector tmp. If locking the senders is ok, then making threads wait on add_listener is ok. I expect notify_all to be called at least an oder of magnitude more often than add_listener anyway. This removes the creation and copy of all the shared_ptr for only one call.

   This would also allow you to remove the stale listeners during the iteration of all the listeners. (Use the return value from vector::erase).

Unnecessary Functions

Now I don't know your exact use case, but as I think about the auto cleanup means you do not really intend to explicitly track things. You probably should take a good hard look if you actually need num_listeners and clear. I can not see any good uses for these functions. (If you have clear, you probably also need remove_listener.)

Autoderigistration is Hard

About the race condition in note 1. Yes, that is a race condition. This is a hard case to handle. If you look at sigc++'s trackable or lsignal's connection they maintain a reference to the signal (in you case Broadcaster) and then call the "noraml" deregistration function that then will wait until done; since only once mutex per signal.

Consider using a Library

This obviously is not part of the code review, if you are writing this code to learn, that is fine. If you are considering to actually use the code in production consider using one of the many existing libraries. sigc++, lsignal or rsig come to mind.