C++ callback multithreaded, can unregister itself

Question

update: a new version of this code is posted here

---

With this post, i would like to 1) ask for feedback on below code as it stands:

• do i apply all best practices for c++20?
• is it safe?
• is my way to update a callback from inside a running callback sensible/best?

Based on code and design advice found on stackoverflow (e.g. here), I have written a class that handles listeners registering callbacks to receive messages in multithreaded code (listeners can be added from different threads than the thread that messages are sent from, only one thread every sends messages). This code also needed multiple ways for listeners to unregister themselves:

1. when the class owning the receiver destructs or is otherwise done receiving: hence add_listener() returns a cookie in the form of a std::list::const_iterator (list because iterators to none of the other elements are invalidated when an element is removed, and other cookies thus stay valid)
2. when the callback during execution determines it no longer needs further messages, this is handled by returning true ("remove myself please").

I now run into the situation where a listener callback, when invoked, needs to replace itself with another callback. I can't call add_listener() as that would deadlock when called from inside an invoked callback. Unless i move to use a std::recursive_mutex instead. That seems heavy.

So instead i have added a replacement list to the broadcaster storing a pair of <listenerCookie, listener> which lists what updates should be done once the current message has been broadcasted to all listeners. The updates are performed at the end of the notify_all function. This guarantees that the old callback is never called again and that the cookie held stays valid but now points to the new callback. But this design seems complicated and unsafe: it required adding an extra member function to the broadcaster class, which should only be called from inside a callback. And that is easy to misuse unless code comments (and the horrible function name :p) saying to only invoke from inside a callback are honored. Or is there a way to ensure a function can only be called by the invoked callback?

Am i missing some simpler solution?

A simpler, but sadly impossible solution i considered was for the callback to return the new callback to register, which would replace the currently registered one (the cookie thus stays valid). But I do not know what my listener's function signature would then be, as you cannot do something CRTP-like with using statements (using listener = std::function<listener(Message...)>; is invalid) (NB: in reality if this option were possible i'd return a variant with bool as well i guess, as i need to also maintain self-deregistration functionality.)

Implementation

#include <list>
#include <mutex>
#include <functional>
#include <vector>
#include <utility>

template <class... Message>
class Broadcaster
{
public:
    using listener       = std::function<bool(Message...)>;
    using listenerCookie = typename std::list<listener>::const_iterator;

    listenerCookie add_listener(listener&& r_)
    {
        auto l = lock();
        return targets.emplace(targets.cend(), std::move(r_));
    }
    void remove_listener(listenerCookie it_)
    {
        auto l = lock();
        remove_listener_impl(it_);
    }

    void notify_all(Message... msg_)
    {
        auto l = lock();

        for (auto it = targets.cbegin(), e = targets.cend(); it != e; )
            if ((*it)(msg_...))
                it = remove_listener_impl(it);
            else
                ++it;

        if (!update_list.empty())
        {
            for (auto&& [c_it, r] : update_list)
            {
                // turn cookie into non-const iterator
                auto it = targets.erase(c_it, c_it);    // NB, [c_it, c_it) is an empty range, so nothing is removed
                // update callback
                *it = std::move(r);
            }
            update_list.clear();
        }
    }

    // NB: should only be called from running callback!
    void update_callback_from_inside_callback(listenerCookie it_, listener&& r_)
    {
        update_list.emplace_back(it_, std::move(r_));
    }

private:
    auto lock()
    {
        return std::unique_lock<std::mutex>(m);
    }
    listenerCookie remove_listener_impl(listenerCookie it_)
    {
        return targets.erase(it_);
    }

private:
    std::mutex m;
    std::list<listener> targets;

    std::vector<std::pair<listenerCookie, listener>> update_list;
};

Example usage

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

struct test
{
    using StringBroadcaster = Broadcaster<std::string>;

    bool simpleCallback(std::string msg_)
    {
        std::cout << "from simpleCallback: " << msg_ << std::endl;
        return false;   // don't remove self: run indefinitely
    }
    bool oneShotCallback(std::string msg_)
    {
        std::cout << "from oneShotCallback: " << msg_ << std::endl;
        return true;    // remove self
    }
    bool twoStepCallback_step1(std::string msg_)
    {
        std::cout << "from twoStepCallback_step1: " << msg_ << std::endl;

        // replace callback (so don't request to delete through return argument!)
        broadcast.update_callback_from_inside_callback(*cb_twostep_cookie, std::bind(&test::twoStepCallback_step2, this, std::placeholders::_1));

        return false;   // don't remove self: we just indicated another callback be placed in our slot: slot needs to remain existent
    }
    bool twoStepCallback_step2(std::string msg_)
    {
        std::cout << "from twoStepCallback_step2: " << msg_ << std::endl;
        return false;   // don't remove self: run indefinitely
    }

    void runExample()
    {
        cb_simple_cookie  = broadcast.add_listener(std::bind(&test::simpleCallback       , this, std::placeholders::_1));
        cb_oneShot_cookie = broadcast.add_listener(std::bind(&test::oneShotCallback      , this, std::placeholders::_1));
        cb_twostep_cookie = broadcast.add_listener(std::bind(&test::twoStepCallback_step1, this, std::placeholders::_1));

        broadcast.notify_all("message 1");  // should be received by simpleCallback, oneShotCallback, twoStepCallback_step1
        // NB: cb_oneShot_cookie is an invalid iterator now, oneShotCallback removed itself through its return argument. clear it
        cb_oneShot_cookie.reset();
        broadcast.notify_all("message 2");  // should be received by simpleCallback and twoStepCallback_step2
        broadcast.remove_listener(*cb_simple_cookie);
        broadcast.notify_all("message 3");  // should be received by twoStepCallback_step2
        broadcast.remove_listener(*cb_twostep_cookie);
        broadcast.notify_all("message 4");  // should be received by none
    }

    StringBroadcaster broadcast;
    std::optional<StringBroadcaster::listenerCookie> cb_simple_cookie;
    std::optional<StringBroadcaster::listenerCookie> cb_oneShot_cookie;
    std::optional<StringBroadcaster::listenerCookie> cb_twostep_cookie;
};

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

Answer 1

Very good design! And very clean code. As-is, there’s very little I can recommend to improve things. In fact, the only code suggestions I can come up with off the top of my head are:

• You should probably take all arguments in the callbacks (and notify function) by const&. That allows you to pass more complicated (and even non-copyable) types to callback functions.

  This may add some overhead if the compiler doesn’t optimize trivial or built-in types like int to by-val parameters, but the overhead should be so small as to be impossible to detect if the callback functions are really small, because then the callback arguments will all be hot in cache. (And if the callback functions are not really small, then you won’t notice the extra indirection anyway.)

• You are not considering exceptions, and that’s usually not a major problem if you just write good code—and your code is well-written—but there are some places where it might be a problem.

  What happens if a callback throws? You could end up with surprising behaviour. Nothing will leak, and no UB will be triggered (at least in the code visible), but what will happen is that all remaining callbacks that were supposed to be notified won’t be. That may or may not be a serious problem, depending on the nature of the message.

  What may be more deadly is that callbacks that have replaced themselves will be surprised to discover… they haven’t been replaced. That will be at least surprising… but if the callbacks have already deleted themselves some way, under the assumption that they have replaced themselves and successfully returned, so they’re done, right?… boom.

  So how to fix this? Well, the easiest way is to require the callbacks to be noexcept. Not like the broadcaster is going to handle any failures on their part, right?

  Another way would be to toss the update_list, and instead only keep track of a single optional<pair<listenerCookie, listener>>. If it has_value(), then right after the callback—before either removing anything, or advancing the iterator—you replace the contents of *it then and there. That way, even if a callback throws, at least all previous callbacks that completed successfully and wanted to be replaced are cool.

  (I’ll also offer another option, later, in the design review.)

  That still leaves the issue that if a callback throws, no other callbacks will be notified… which will lead to non-deterministic behaviour in multi-threaded code, because there doesn’t seem to be a way to sort callbacks. In other words, if you start a notify_all(), and a callback throws, you have no way of knowing which callbacks completed and which didn’t. So you can’t even catch the exception, handle/ignore it, then continue notifying.

  You could catch any exceptions thrown by callbacks in a std::exception_ptr, continue notifying, and then, at the end, throw if the exception pointer tests true. But that would only handle a single exception. What if multiple callbacks throw? Tricky! Not impossible to handle, but damn tricky. (You might need an exception type that holds a list of exceptions!) Making callbacks noexcept is a MUCH easier solution. But that’s a design decision you’ll have to wrestle over.

• You have a serious bug, in that callbacks can deadlock your program if they call any public member functions of the broadcast type.

  The problem is that you are allowing arbitrary code execution while holding a lock within your notify_all() function. Each callback being notified can do… bugger-well anything! There’s nothing to stop one from calling notify_all() again! Or from calling one of the other public member function of the broadcast type.

  Now, you can argue that doing so would be logically stupid, and for the most part that’s true: You already provide means to remove and replace callbacks within the notify loop, so there should never be a need to call remove_listener() from a callback. But a callback might very well reasonably want to add a listener, and there’s no other way to do that other than via add_listener(). It might also be reasonable for a callback to want to trigger a new notification, in which case it will need to call notify_all().

  I can’t tell you the “right” way to fix this problem, because it will depend on your design space. I can offer suggestions.

  1. Within notify_all(), do the lock, COPY THE LIST OF LISTENERS, then release the lock, and start notifying using the copied list. That way, all the callbacks are done without holding the lock. If you need to reacquire the lock—such as to do a remove or replace of a listener—that’s fine.

     This is not a perfect solution, because you could still end up with the situation where a listener has removed itself and returned successfully, then gets called again. (And, of course, you’ve pretty much lost everything you gained by using std::list iterators in the first place.) Fixing that issue will be tricky. (One potential solution would be to require all the callbacks to be std::shared_ptr. That’s an interesting solution in that it allows thread-safe auto-deregistration any time if you only hold a container of std::weak_ptr.)

     However, it would allow you to do the notifies concurrently, which could be a huge gain.

  2. Disallow callbacks to do anything dirty by using std::try_lock(). In each function, replace the lock line with: `auto lock = std::unique_lock{m}; if (auto res = std::try_lock(m); res == -1) { /* do your business */ } else { throw std::logic_error{"recursive lock"}; }

  3. Use a recursive mutex so callbacks won’t actually lock. No, don’t do this. It will only work if the callbacks stay in the same thread (and there’s no way to force that), and even then, it will require herculean efforts to handle race conditions.

There’s one style issue I’d like to critique, then I’d like to discuss your design. So this will be more of an overall design review than a specifically-code review.

Style issue: naming convetions

On the style front, I find your choice to mangle member function arguments… but not data members… a bit odd. Standard practice is to add an underscore to data member names (or to tag them with m_, or something else like that). Why? Because those data members are visible in multiple places (every member function, for example). But member function arguments are only visible in that one, single member function. There doesn’t seem to be any point in mangling them, because no one else will ever see them.

For example, if I see the following member function:

auto class_t::func(int a)
{
    a = 1;
    b = 2;
    _c = 3;
}

… one of the most important things I want to be able to do is see the definitions of those three variables. When I see an undecorated variable name, like a, the first place I look is within the closest scope. Then the next scope, and the next scope, and so on, until I reach function scope, at which point I check the function arguments. In this case, that’s where I’ll find a.

But I won’t find b. So the next step is to check the class scope, for static data members. If I don’t find it there, then I know b is at namespace scope—it’s a “global variable”—and start looking out in the world for it.

All of which is to say is that variables like b are a pain in the ass. Well, if they’re static data members, they’re only mildly annoying for forcing me to leave function I’m investigating and go digging through the class declaration. But if I have to go search the entire friggin’ project for a variable declaration (and even then, never be 100% sure I haven’t missed a declaration in some other namespace that might get picked up first)… yeah, I be peeved. Luckily this isn’t that great a problem in practice.

Which is why variables names like _c (or c_ or m_c; whatever the project convention) are so lovely. When I don’t find the definition locally—and, frankly, I usually don’t even bother to check locally, because, yanno, convention—I know EXACTLY where to find it. I look in the class definition, and, if it’s a const member function, add const, and… done. Simple.

In general, when I see an undecorated name that isn’t local, I know it’s a global variable of some kind—either literally global (that is, at namespace scope) or functionally global (that is, a class static member). In either case, I know without even looking whether this function is one of those functions whose side-effects are going to be a problem. If a function has only undecorated names that are local, and decorated names, then I know at a glance that its effects are all local; either completely internal to the function itself, or completely internal to the object being operated on (that is, *this).

In other words, without even the foggiest clue about the rest of the project, I can look at func() above and know that it has globally-visible side-effects, because of b. I don’t even need to know what b is, or where b is defined, or anything about class_t. Just by the convention, I can tell that this is a function that may cause surprises elsewhere in the code. I can be pretty damn sure that it’s not re-entrant (and thus not thread-safe). On the other hand, if b were removed, then I would be able to assume that func() is safe; that if I call it, it won’t touch anything other than its own internals and *this, which means it’s thread-safe (so long as *this isn’t being shared among threads, in which case, if it were, I would assume it’s being synchronized externally). That’s the power of a good convention.

Now let’s consider your convention:

auto class_t::func(int a_)
{
    a_ = 1;
    b = 2;
    c = 3;
}

Okay, so I see a_, and I know to check the local scope, and eventually find it as a function argument. Cool.

Now what happens with b and c? Where do I even begin looking? Should I assume they are non-static data members? Static data members? Globals? Is this function re-entrant? I mean, if I went hunting around for the definitions of b and c eventually I would find out… but that’s a far cry from being able to know just from a glance.

Design review

First off, I think the idea to take advantage of the invalidation characteristics of std::list’s iterators is brilliant. I admit I rarely think about std::list when reaching for a container (honestly, the only time I think of it is when testing bi-di iterators), but now you’ve inspired me to give it a second look for some design problems I’ve run into.

Now, when I considered your design questions, I have to admit I went down exactly the same trail you did. I first thought, well, let’s use a variant for the return with an optional replacement… oh, no, that won’t work because that would make the function’s type recursive… okay, what if we made the update_callback_from_inside_callback() safer… could use a recursive_mutex, oh you already thought of that…. I ended up stuck exactly where you are.

But then I had one of those face-palming brainstorms, and came up with this (note, code is not tested, probably won’t compile, and is likely broken and incomplete—this is just for illustration of an idea):

struct return_t {};
struct remove_t {};

using replace_t = std::any;

using result_t = std::variant<return_t, remove_t, replace_t>;

template <typename... Args>
class broadcaster
{
    // ... [snip] ...

    using listener = std::function<result_t (Args const&...)>;

    // ... [snip] ...

    void notify_all(Args const&... args)
    {
        auto l = lock();

        for (auto it = _targets.cbegin(); it != _targets.cend(); )
        {
            // "overload" is a pretty common utility type - for example,
            // there's a version on the cppreference page on visit().
            //
            // Until we get pattern-matching, this is the prettiest we can do.

            it = std::visit(
                overload{
                    // Remove listener.
                    [&_targets, it](remove_t) { return _targets.erase(it); },

                    // Replace listener.
                    [&_targets, it](replace_t& r)
                    {
                        auto i = _targets.erase(it, it);
                        *i = std::move(std::any_cast<listener>(r));
                        return ++it;
                    },

                    // Default: just continue notifying.
                    [it](auto&&) { return ++it; },
                },
                (*it)(args...));
        }
    }

    // ... [snip] ...
};

Now, as written, this will work (I presume!), but because std::any is smaller than std::function in every stdlib I’m aware of, it will always trigger an allocation. Now, you may not care, because replacing a listener will be damn rare, so… meh. But if you do care, then you could replace std::any with a custom type-erasing type that holds a properly aligned array of std::byte that is sizeof(std::function) large. Doesn’t even matter which instantiation of std::function you use to determine the type, so you don’t need to know Args.... In fact, rather than std::any, you could just use std::function<void()>, because function pointers can be converted to other function pointers, and back. But then you lose the type check of any_cast. As written, it is type safe because any_cast will detect shenanigans. If you want this check done earlier—like within the callback, or even at compile-time—there are tricks you can use, like making replace_t a type with a private constructor, and broadcast<Args...> is a friend, and it has a make_replace_token(std::function<result_t(Args const&...)>) function, or something like that.

You could even add another behaviour—add_t—which is a type-erased callback that gets added. (Of course, you can’t do it the way notify_all() is currently written, because you can’t modify the list while iterating on it. But it may work with other options!) You may even add add_multiple_t, which allows a callback to register multiple listeners.

But, honestly, it would be better if it were safe for callbacks to modify the broadcaster safely, because then they could add, remove, or even notify-all without a care. Then you wouldn’t even need any of these gymnastics, because a callback could simply call remove_listener() (and you could have a replace_listener(), too!).

The key point to keep in mind is that allowing arbitrary code to execute while holding a lock is like break-dancing at a pyrotechnics display while doused in gasoline.

The best possible fix is to do all the callbacks while not holding the lock. If you can make this work, it’s all wins: callbacks can add new callbacks, remove callbacks (including themselves), replace callbacks, do new notify runs… basically anything. The tricky part is making this work.

The big catch to pulling off the trick is to make sure a callback can know for damn, 100% sure if it’s been removed (or replaced), so it can know cannot be called again (and thus, is safe to destruct). This is not trivial. One solution is to share ownership between the broadcaster and anything else that wants to know if the callback is live; with std::shared_ptr (used properly), there’s no possible way to call or otherwise access an already-deleted callback. There are other, lighter solutions, too, like keeping an atomic “is alive” flag for each callback… but you have to be careful about TOCTOU issues. As I said, tricky!

Extras

Compile-time checking for noexcept functions

Unfortunately std::function doesn’t properly support noexcept functions as of C++20. This has been a known problem since C++17; I don’t know why it hasn’t been fixed yet. (Well, okay, I can guess it’s because it isn’t an easy fix, and because the pandemic did slow things down for the committee. Maybe it just slipped under the priority bar.)

But that’s not that big a deal, because there are only two places where you need to verify the callback: add_listener() and update_callback_from_inside_callback(). Everywhere else you just use the callback, and you can put a noexcept function in a non-noexcept std::function. (You just can’t do the opposite; you can’t put a non-noexcept function in a noexcept std::function.)

So here’s how you could fix add_listener():

template <class... Message>
class Broadcaster
{
private:
    using listener = std::function<bool(Message...)>; // this no longer needs to be public
public:
    using listenerCookie = typename std::list<listener>::const_iterator;

    // ... [snip] ...

    // if you don't have C++20 concepts, that's fine; the concept is just
    // an extra check, not a necessary one
    template <std::invocable<Message...> F>
    listenerCookie add_listener(F&& r_)
    {
        // this is where the real check happens
        static_assert(std::is_nothrow_invocable_r_v<bool, F, Message...>);

        auto l = lock();

        // it's cool to put a noexcept function in a non-noexcept std::function
        // so we just forward what we're given into a function<bool(Message...)>

        return targets.emplace(targets.cend(), listener{std::forward<F>(r_)});
    }

    // ... [snip] ...

Same idea for update_callback_from_inside_callback().

This will bloat the interface a bit, so you might want to do:

    template <std::invocable<Message...> F>
    listenerCookie add_listener(F&& r_)
    {
        static_assert(std::is_nothrow_invocable_r_v<bool, F, Message...>);

        return add_listener_impl_(listener{std::forward<F>(r_)});
    }

    // ... [snip] ...

private:
    listenerCookie add_listener_impl_(listener&& r)
    {
        auto l = lock();
        return targets.emplace(targets.cend(), std::move(r));
    }

    // ... [snip] ...

Either way, it will make no difference at the call site (except, of course, if your callback is not noexcept, it will no longer compile). All these should work:

auto foo_1(int, double) noexcept -> bool;

b.add_listener(foo_1);

auto foo_2 = [=](int, double) noexcept -> bool { return true; };

b.add_listener(foo_2);
// or
b.add_listener(std::move(foo_2));

b.add_listener([](int, double) noexcept -> bool { return true; });

struct func
{
    std::string name = {};
    int call_count = 0;

    auto operator()(int, double) noexcept
    {
        std::cout << name << " has been called " << ++call_count << " times";

        return call_count < 100;
    }
};

b.add_listener(func{.name = "steve"});

auto f = func{.name = "anya"};
b.add_listener(std::move(f));
```