c++ multithreaded message broadcaster with link lifetime management

Question

note: A version of this code without link lifetime management was previously reviewed. Adding lifetime management actually made the code simpler, and the class simpler to use.

I have written a class that handles listeners registering callbacks to receive messages. Link lifetime is managed. Code is multithreaded in that any thread can broadcast to the listeners, and any thread can add or remove listeners.

I have made a list of requirements for the class:

1. there may be zero, one or multiple listeners
2. listeners can be registered or deregistered from different threads
3. each listener should be called exactly once for each message
4. all listeners should receive all messages
5. messages may originate from multiple sources in different threads
6. each message should be forwarded to all listeners before the next message is processed (messages are not sent in parallel)
7. each listener should receive the messages in the same order, i.e., the order in which they are provided to the broadcaster (from the view of the broadcaster)
8. listeners can be removed
9. listeners are identified by a cookie whose validity state indicates if the listener is still listening
10. anyone with the cookie should be able to deregister the listener
11. listeners should never be called again when deregistration completes
12. listeners should be able to add a new listener when invoked, this listener will participate starting from the next message
13. listeners are noexcept, so code shouldn’t deal with exceptions thrown by listeners
14. adding and removing listeners happens much less frequently than broadcasting.

A further consideration that is not enforced in code:

• Users are expected to read documentation that states which functions are safe to call from inside a listener, and which must not.

Link lifetime management is performed by wrapper each registered listener function in a shared_ptr that is held by the listener, and references as a weak pointer in the broadcaster class. Just before the listener is invoked, the weak pointer is converted to a shared pointer, which would fail if the listener released its end.

I am not sure if this method for lifetime management works, or if there is a race condition: broadcaster obtains shared pointer, but before it can invoke the callback, the listener just destructs. Link remains valid as shared pointer is acquired by broadcaster, but the listener that is about to be invoked no longer exists. That is a possible race, right? How to solve it?

A listener who owns a dangling link (no more broadcaster) is not a problem, as they can't do anything unsafe with their end of the link.

Implementation

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

// based on https://stackoverflow.com/a/47872677
// usage notes:
// 1. None of the member functions of this class may be called from
//    inside an executing listener, except
//    add_listener_from_inside_callback(). Calling them anyway
//    will result in deadlock.
// 2. add_listener_from_inside_callback() may _only_ be called from
//    inside an executing listener.
template <class... Message>
class Broadcaster
{
public:
    using listener = std::function<void(Message...)>;
    using cookieType = std::shared_ptr<void>;

    // returns number of registered listeners
    size_t num_listeners() noexcept
    {
        auto l = std::unique_lock(_mut);
        // prune to reduce chance that we return
        // having more listeners than we do
        for (auto it = _targets.cbegin(); it != _targets.cend(); )
        {
            if (it->expired())
                it = _targets.erase(it);
            else
                ++it;
        }
        return _targets.size();
    }

    // clears all listeners
    void clear() noexcept
    {
        auto l = std::unique_lock(_mut);
        _targets.clear();
    }


    // should never be called from inside a running listener
    template <class F> requires (std::is_nothrow_invocable_r_v<void, F, Message...>)
        [[nodiscard]] cookieType add_listener(F&& r_)
    {
        auto l = std::unique_lock(_mut);
        auto listenFunc = std::make_shared<listener>(std::forward<F>(r_));
        _targets.push_back(listenFunc);
        return listenFunc;
    }

    template <class F> requires (std::is_nothrow_invocable_r_v<void, F, Message...>)
        [[nodiscard]] cookieType add_listener_from_inside_callback(F&& r_)
    {
        auto listenFunc = std::make_shared<listener>(std::forward<F>(r_));
        // place into a holding pen, will be added to active listeners
        // as soon as all currently registered callbacks have received
        // the current message
        _newTargets.push_back(listenFunc);
        return listenFunc;
    }

    void notify_all(const Message&... msg_) noexcept
    {
        auto l = std::unique_lock(_mut);

        // invoke all listeners with message, if still
        // alive. else remove
        for (auto it = _targets.cbegin(); it != _targets.cend(); )
        {
            auto targetPtr = it->lock();
            if (!targetPtr)
                it = _targets.erase(it);
            else
            {
                targetPtr->operator()(msg_...);
                ++it;
            }
        }

        // if any new listeners, move them into active listeners
        // so they'll receive the next message
        if (!_newTargets.empty())
        {
            _targets.insert(_targets.end(), std::make_move_iterator(_newTargets.begin()),
                                            std::make_move_iterator(_newTargets.end()));
            _newTargets.clear();
        }
    }

private:
    mutable std::mutex                      _mut;
    std::vector<std::weak_ptr<listener>>    _targets;
    std::vector<std::weak_ptr<listener>>    _newTargets;
};

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_from_inside_callback([&](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

Unnecessary use of mutable

Since none of the member functions are const, you can't call them on a const Broadcaster object, so making the mutex mutable does not do anything.

Making adding from inside the callback easier

The caller must remember to call add_listener_from_inside_callback() if it is inside the callback, otherwise a deadlock will occur. Also, calling add_listener_from_inside_callback() outside a callback might invalidate iterators. You can avoid the need for this function, and have add_listener() safely detect if it is called from inside a callback.

The first requirement is to use std::recursive_mutex instead of a regular mutex. This ensures calling add_listener() from inside the callback won't deadlock.

The remaining issue is to avoid iterator invalidation. You could consider adding a simple boolean flag _in_callback, which is set at the start of notify_all and cleared at the end, and if set when add_listener() is called, append to _newTargets instead of _targets. However, there are two other options. The first is to always add to _newTargets, and to move the new targets into _targets at the start of notify_all(). The second is to use std::deque instead of std::vector, as it won't invalidate iterators if you insert at the front or back. You can't erase elements in the same loop anymore though, but you can defer that to after the main loop. It could look like:

template <class... Message>
class Broadcaster
{
    ...
    void notify_all(const Message&... msg_) noexcept
    {
        auto l = std::unique_lock(_mut);
        bool flush = false;

        for (auto& target: _targets) {
            if (auto targetPtr = target.lock())
                targetPtr->operator()(msg_...);
            else
                flush = true;
        }

        if (flush)
            std::erase_if(_targets, [](auto& target){return target.expired();});
    }
    ...
private:
    std::recursive_mutex _mut;
    std::deque<std::weak_ptr<listener>> _targets;
};

A big drawback of the std::recursive_mutex is that it will now allow callbacks to also call num_listeners() and clear() without deadlocking, but instead invalidating iterators. You could avoid that again by adding an in_callback flag to detect recursion.