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:
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
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.
- broadcaster obtains shared pointer (
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?As G. Sliepen notes in his suggestions on v2 of this code, it is rather expensive that
std::shared_ptr
s 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;
}