c++ multithreaded message broadcaster with link lifetime management v2

Question

note: This is v2 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?). Code is multithreaded in that any thread can broadcast to the listeners, and any thread can add or remove listeners.

I have two concerns:

1. I am not sure if this method for lifetime management works, or if there is a race condition. Is the following a possible race condition?

   1. broadcaster obtains shared pointer;
   2. before it can invoke the callback, the listener destructs (imaging its a ref to a class member function);
   3. Listener is invoked, but points to an object that no longer exists.

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?

Implementation

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

// 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.
// 2. Listeners added from inside listener callback
//    will participate starting from the next message.
// 3. This class guarantees message ordering with
//    multiple producer threads.
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() const noexcept
    {
        auto l = std::unique_lock(_mut);
        return std::accumulate(_targets.cbegin(), _targets.cend(), size_t{ 0 }, [](size_t cur, const auto& m) { return cur + !m.expired(); });
    }

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

    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;
    }

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

        // remove dead listeners
        _targets.erase(
            std::remove_if(_targets.begin(), _targets.end(),
                [](const auto& ptr) { return ptr.expired(); }
            ),
            _targets.end()
        );

        // take copy
        auto tmp = _targets;
        lt.unlock();

        // 1. mutex unlocked, so new listeners can be added
        // 2. sending mutex remains held so that message
        //    ordering is guaranteed.
            
        // invoke all listeners with message
        for (auto&& f : tmp)
        {
            if (auto targetPtr = f.lock(); targetPtr)
                targetPtr->operator()(msg_...);
        }
    }

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

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

Use the right algorithms

Don't use std::accumulate() to count the number of elements that match a predicate, there is std::count_if() which is a much better match for what you are trying to do:

size_t num_listeners() const noexcept
{
    auto l = std::unique_lock(_mut);
    return std::count_if(_targets.cbegin(), _targets.cend(),
                         [](const auto& m) { return !m.expired(); });
}

Or even better, with C++20 ranges you can use:

    return std::ranges::count_if(_targets, [](const auto& m) { return !m.expired(); });

The erase-remove idiom is no longer necessary in C++20 thanks to the much simpler std::erase_if:

// remove dead listeners
std::erase_if(_targets, [](const auto& ptr) { return ptr.expired(); });

Be careful when adding noexcept

It is great that you add noexcept to functions where possible. But be aware that exceptions come from a lot of places.

Theoretically, std::mutex::lock() might throw. However, this is extremely unlikely, and if it does, you probably won't be able to recover anyway. So for practical purposes, you can leave functions that take a lock noexcept. This means that if an exception is raised, the program will immediately terminate instead of unwinding the stack until the exception is caught.

Lock ordering

Never let locking of multiple mutexes have partial overlaps. In more complex situations this has a high chance of causing deadlocks. Avoid calling lock() and unlock() manually, and always use RAII lock guards. Use braces to limit the scope of a lock where necessary. For example:

void notify_all(const Message&... msg_)
{
    auto ls = std::unique_lock(_sendMut);
    decltype(_targets) tmp;

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

        // take copy
        tmp = _targets;
    }

    ...
}

Cost of making copies

Making copies is a good way to avoid issues with multiple threads trying to modify _targets while you are still broadcasting a message. However, there is of course a cost. Copying a std::weak_ptr is not very cheap, because it needs to atomically increment the reference count of the control block associated with the std::shared_ptr it refers to. So with just a few listeners I think copying _targets will be more expensive than using a recursive mutex and a flag to detect listeners being added during the loop.

On the other hand, you are already calling lock() on every std::weak_ptr in tmp. That just creates a bunch of std::shared_ptrs. So you could do the lock()ing while making a copy:

void notify_all(const Message&... msg_)
{
    auto ls = std::unique_lock(_sendMut);
    std::vector<std::shared_ptr<listener>> tmp;

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

        // take copy
        for (auto& target: _targets)
            tmp.push_back(target.lock());
    }

    for (auto& targetPtr: tmp)
        if (targetPtr)
            targetPtr->operator()(msg_...);
}

Alternatives

I think there is a big overhead coming from that fact that you use a std::shared_ptr for each individual listener. You could avoid that by having a std::shared_ptr for the whole vector of listeners. You could make your own RAII class that replaces the current cookieType, and which contains a std::weak_ptr to the vector of listeners, and a destructor which will erase the listener from that vector if it is still alive.