Что такое сигналы?

struct Button
{
    boost::signals2::signal<void()> OnClick;
};

void ClickHandler()
{ cout << “Button clicked” << endl; }

// ...

boost::signals2::connection c = button->OnClick.connect(&ClickHandler);

// ...

c.disconnect();

struct Button
{
    boost::signals2::signal<void()> OnClick;

private:
    void MouseDownHandler()
    {
        OnClick();
    }
};

Теперь о проблемах

1. Нет способа атомарно подключиться к сигналу и получить связанное состояние
2. Неблокирующее отключение от сигнала
3. Отключение асинхронного обработчика не отменяет вызовы, которые уже попали в очередь его потока

• StorageManager — класс, который реагирует на флэшки, DVD-диски и прочие носители, которые пользователь вставил в приставку
• MediaScanner — класс, который ищет медиа-файлы на каждом из таких устройств
• MediaUiModel — модель для отображения этих медиа-файлов в воображаемом Model-View-что-нибудь фреймворке

Нет способа атомарно подключиться к сигналу и получить связанное состояние

class StorageManager
{
public:
    std::vector<StoragePtr> GetStorages() const;
    boost::signals2::signal<void(const StoragePtr&)> OnStorageAdded;

    // ...
};

storageManager->OnStorageAdded.connect(&StorageHandler);

// Если пользователь вставляет флэшку до цикла, она будет обработана дважды
for (auto&& storage : storageManager->GetStorages())
    StorageHandler(storage);

for (auto&& storage : storageManager->GetStorages())
    StorageHandler(storage);

// Если пользователь вставляет флэшку до подключения к сигналу, она не будет обработана совсем
storageManager->OnStorageAdded.connect(&StorageHandler);

Распространённое решение

class StorageManager
{
    mutable std::recursive_mutex   _mutex;
    std::vector<StoragePtr>        _storages;

public:
    StorageManager()
    { /* ... */ }

    boost::signals2::signal<void(const StoragePtr&)> OnStorageAdded;

    std::recursive_mutex& GetMutex() const
    { return _mutex; }

    std::vector<StoragePtr> GetStorages() const
    {
        std::lock_guard<std::recursive_mutex> l(_mutex);
        return _storages;
    }

private:
    void ReportNewStorage(const StoragePtr& storage)
    {
        std::lock_guard<std::recursive_mutex> l(_mutex);

        _storages.push_back(storage);
        OnStorageAdded(storage);
    }
};

// ...

{
    std::lock_guard<std::recursive_mutex> l(storageManager->GetMutex());

    storageManager->OnStorageAdded.connect(&StorageHandler);
    for (auto&& storage : storageManager->GetStorages())
        StorageHandler(storage);
}

• Если вы хотите использовать std::mutex вместо std::recursive_mutex, вы теряете возможность захватывать его внутри метода GetStorages, что делает класс StorageManager непотокобезопасным
• Вы не можете избавиться от копирования коллекции внутри GetStorages, не потеряв потокобезопасность StorageManager'а
• Вам приходится показывать наружу тип std::vector<StoragePtr>, хотя на деле это всего лишь детали реализации
• Довольно объёмный код подключения к сигналу и получения текущего состояния, который при этом почти никак не отличается для разных сигналов

Как сделать лучше?

template < typename Signature >
class signal
{
    using populator_type = std::function<void(const std::function<Signature>&)>;

    mutable std::mutex                       _mutex;
    std::list<std::function<Signature> >     _handlers;
    populator_type                           _populator;

public:
    signal(populator_type populator)
        : _populator(std::move(populator))
    { }

    std::mutex& get_mutex() const { return _mutex; }

    signal_connection connect(std::function<Signature> handler)
    {
        std::lock_guard<std::mutex> l(_mutex);

        _populator(handler); // Владелец сигнала определяет конкретный алгоритм получения состояния
        _handlers.push_back(std::move(handler));
	
        return signal_connection([&]() { /* удаляем обработчик из _handlers */ } );
    }

    // ...
};

class StorageManager
{
    std::vector<StoragePtr>        _storages;

public:
    StorageManager()
        : _storages([&](const std::function<void(const StoragePtr&)>& h) { for (auto&& s : _storages) h(s); })
    { /* ... */ }

    signal<void(const StoragePtr&)> OnStorageAdded;

private:
    void ReportNewStorage(const StoragePtr& storage)
    {
        // Мы должны захватить мьютекс именно тут, а не внутри вызова сигнала,
        // потому что он защищает в том числе и коллекцию _storages
        std::lock_guard<std::mutex> l(OnStorageAdded.get_mutex());

        _storages.push_back(storage);
        OnStorageAdded(storage);
    }
};

StorageManager()
    : _storages([&](auto&& h) { for (auto&& s : _storages) h(s); })
{ }

storageManager->OnStorageAdded.connect(&StorageHandler);

Неблокирующее отключение от сигнала

class MediaScanner
{
private:
    boost::signals2::connection _connection;

public:
    MediaScanner(const StorageManagerPtr& storageManager)
    { _connection = storageManager->OnStorageAdded.connect([&](const StoragePtr& s) { this->StorageHandler(s); }); }

    ~MediaScanner()
    {
        _connection.disconnect();
        // Обработчик сигнала может всё ещё исполняться в потоке, вызвавшем сигнал.
        // В этом случае, далее он будет обращаться к разрушенному объекту MediaScanner.
    }

private:
    void StorageHandler(const StoragePtr& storage)
    { /* Здесь что-то долгое */ }
};

Решение в Qt

mediaScanner->deleteLater();

Решение в boost::signals2

class MediaScanner
{
public:
    static std::shared_ptr<MediaScanner> Create(const StorageManagerPtr& storageManager)
    {
        std::lock_guard<std::recursive_mutex> l(storageManager->GetMutex());

        MediaScannerPtr result(new MediaScanner);

        boost::signals2::signal<void(const StoragePtr&)>::slot_type
            slot(bind(&MediaScanner::StorageHandler, result.get(), _1));
        slot.track_foreign(result);

        storageManager->OnStorageAdded.connect(slot);
        for (auto&& storage : storageManager->GetStorages())
            result->StorageHandler(storage);

        return result;
    }

private:
    MediaScanner() // приватный конструктор!
    { /* Проинициализировать всё, кроме обработчиков сигналов */ }

    void StorageHandler(const StoragePtr& storage);
    { /* Здесь что-то долгое */ }
};

• О-очень много кода, который вы каждый раз будете копипастить
• Инициализация MediaScanner'а распалась на две части: подписывание на сигналы в методе Create, и всё остальное в конструкторе
• Вы обязаны использовать shared_ptr для хранения MediaScanner
• Вы не уверены, что MediaScanner удалён, когда вы отпустили последнюю внешнюю ссылку на него. Это может быть проблемой, если он использует какой-либо ограниченный ресурс, который вы хотите переиспользовать после освобождения MediaScanner

Как сделать лучше?

• life_token — контролирует время жизни обработчика, позволяет пометить его, как «умирающий», и дождаться окончания исполнения, если необходимо
• life_token::checker — хранится внутри сигнала рядом с обработчиком, ссылается на свой life_token
• life_token::checker::execution_guard — создаётся на стеке на время исполнения обработчика, блокирует соответствующий life_token и позволяет проверить, не «умер» ли обработчик ранее

class signal_connection
{
    life_token               _token;
    std::function<void()>    _eraseHandlerFunc;

public:
    signal_connection(life_token token, std::function<void()> eraseHandlerFunc)
        : _token(token), _eraseHandlerFunc(eraseHandlerFunc)
    { }

    ~signal_connection();
    { disconnect(); }

    void disconnect()
    {
        if (_token.released())
            return;

        _token.release(); // Тут мы ждём, если обработчик сейчас заблокирован (т. е. исполняется)
        _eraseHandler(); // Та самая лямбда-функция, которая удалит обработчик из списка
    }
};

template < typename Signature >
class signal
{
    using populator_type = std::function<void(const std::function<Signature>&)>;

    struct handler
    {
        std::function<Signature>    handler_func;
        life_token::checker         life_checker;
    };

    mutable std::mutex            _mutex;
    std::list<handler>            _handlers;
    populator_type                _populator;

public:
    // ...

    signal_connection connect(std::function<Signature> handler)
    {
        std::lock_guard<std::mutex> l(_mutex);

        life_token token;
        _populator(handler);
        _handlers.push_back(Handler{std::move(handler), life_token::checker(token)});
	
        return signal_connection(token, [&]() { /* удаляем обработчик из _handlers */ } );
    }

    template < typename... Args >
    void operator() (Args&&... args) const
    {
        for (auto&& handler : _handlers)
        {
            life_token::checker::execution_guard g(handler.life_checker);
            if (g.is_alive())
                handler.handler_func(forward<Args>(args)...);
        }
    }
};

class life_token
{
    struct impl
    {
        std::mutex              mutex;
        bool                    alive = true;
    };

    std::shared_ptr<impl>       _impl;

public:
    life_token() : _impl(std::make_shared<impl>()) { }
    ~life_token() { release(); }
    bool released() const { return !_impl; }

    void release()
    {
        if (released())
            return;

        std::lock_guard<std::mutex> l(_impl->mutex);
        _impl->alive = false;
        _impl.reset();
    }

    class checker
    {
        shared_ptr<impl>	_impl;

    public:
        checker(const life_token& t) : _impl(t._impl) { }

        class execution_guard
        {
            shared_ptr<Impl>	_impl;

        public:
            execution_guard(const checker& c) : _impl(c._impl) { _impl->mutex.lock(); }
            ~execution_guard() { _impl->mutex.unlock(); }
            bool is_alive() const { return _impl->alive; }
        };
    };
};

class MediaScanner
{
private:
    signals_connection    _connection;

public:
    MediaScanner(const StorageManagerPtr& storageManager)
    { _connection = storageManager->OnStorageAdded.connect([&](const StoragePtr& s) { this->StorageHandler(s); }); }

    ~MediaScanner()
    { _connection.disconnect(); }

private:
    void StorageHandler(const StoragePtr& storage)
    { /* Здесь что-то долгое */ }
};

Отключение асинхронного обработчика не отменяет вызовы, которые уже попали в очередь его потока

signal<void(const MediaPtr&)> OnMediaFound;

• Модель — это объект UI-библиотеки, поэтому все действия с ней должны производиться из потока UI.
• Часто в UI-библиотеках используется своя иерархия владения, поэтому мы не можем использовать shared_ptr для хранения модели. Соответственно, фокус с track/track_foreign тут не пройдёт, но это сейчас не главное, так что притворимся, что всё хорошо

class MediaUiModel : public UiModel<MediaUiModelRow>
{
private:
    boost::io_service&             _uiThread;
    boost::signals2::connection    _connection;

public:
    MediaUiModel(boost::io_service& uiThread, const MediaScanner& scanner)
        : _uiThread(uiThread)
    {
        std::lock_guard<std::recursive_mutex> l(scanner.GetMutex());
        scanner.OnMediaFound.connect([&](const MediaPtr& m) { this->MediaHandler(m); });
        for (auto&& m : scanner.GetMedia())
            AppendRow(MediaUiModelRow(m))
    }

    ~MediaUiModel()
    { _connection.disconnect(); }

private:
    // Этот метод выполняется в потоке MediaScanner'а, и всю реальную работу перебрасывает в поток UI.
    void MediaHandler(const MediaPtr& m)
    { _uiThread.post([&]() { this->AppendRow(MediaUiModelRow(m)); }); }
};

Решение в Qt

Решение в boost::signals2

template < typename Signature_ >
class AsyncToUiHandlerWrapper
{
private:
    boost::io_service&          _uiThread;
    std::function<Signature_>   _realHandler;
    bool                        _released;
    mutable std::mutex          _mutex;

public:
    AsyncToUiHandlerWrapper(boost::io_service& uiThread, std::function<Signature_> realHandler)
        : _uiThread(uiThread), _realHandler(realHandler), _released(false)
    { }

    void Release()
    {
        std::lock_guard<std::mutex> l(_mutex);
        _released = true;
    }

    template < typename... Args_ >
    static void AsyncHandler(const std::weak_ptr<AsyncToUiHandlerWrapper>& selfWeak, Args_&&... args)
    {
        auto self = selfWeak.lock();

        std::lock_guard<std::mutex> l(self->_mutex);
        if (!self->_released) // AsyncToUiHandlerWrapper не был освобождён, значит _uiThread всё ещё ссылается на живой объект
            self->_uiThread.post(std::bind(&AsyncToUiHandlerWrapper::UiThreadHandler<Args_&...>, selfWeak, std::forward<Args_>(args)...)));
    }

private:
    template < typename... Args_ >
    static void UiThreadHandler(const std::weak_ptr<AsyncToUiHandlerWrapper>& selfWeak, Args_&&... args)
    {
        auto self = selfWeak.lock();
        if (!self)
            return;

        if (!self->_released) // AsyncToUiHandlerWrapper не был освобождён, значит, объекты, доступные _realHandler, ещё живы
            self->_realHandler(std::forward<Args_>(args)...);
    }
};


class MediaUiModel : public UiModel<MediaUiModelRow>
{
private:
    using AsyncMediaHandler = AsyncToUiHandlerWrapper<void(const MediaPtr&)>;

private:
    std::shared_ptr<AsyncMediaHandler>    _asyncHandler;

public:
    MediaUiModel(boost::io_service& uiThread, const MediaScanner& scanner)
    {
        try
        {
            _asyncHandler = std::make_shared<AsyncMediaHandler>(std::ref(uiThread), [&](const MediaPtr& m) { this->AppendRow(MediaUiModelRow(m)); });

            std::lock_guard<std::recursive_mutex> l(scanner.GetMutex());

            boost::signals2::signal<void(const MediaPtr&)>::slot_type
                slot(std::bind(&AsyncMediaHandler::AsyncHandler<const MediaPtr&>, std::weak_ptr<AsyncMediaHandler>(_asyncHandler), std::placeholders::_1));
            slot.track_foreign(_asyncHandler);
            scanner.OnMediaFound.connect(slot);

            for (auto&& m : scanner.GetMedia())
                AppendRow(MediaUiModelRow(m));
        }
        catch (...)
        {
            Destroy();
            throw;
        }
    }

    ~MediaUiModel()
    { Destroy(); }

private:
    void Destroy()
    {
        if (_asyncHandler)
            _asyncHandler->Release(); // Асинхронный код не обращается к MediaUiModel после этой строки, так что можно окончательно разрушать объект
        _asyncHandler.reset();
    }
};

Как сделать лучше?

struct task_executor
{
    virtual ~task_executor() { }
    virtual void add_task(const std::function<void()>& task) = 0;
};

signal_connection connect(const std::shared_ptr<task_executor>& worker, std::function<Signature> handler);

class MediaUiModel : public UiModel<MediaUiModelRow>
{
private:
    signal_connection    _connection;

public:
    MediaUiModel(const std::shared_ptr<task_executor>& uiThread, const MediaScanner& scanner)
    { _connection = scanner.OnMediaFound.connect(uiThread, [&](const MediaPtr& m) { this->AppendRow(MediaUiModelRow(m)); }); }

    ~MediaUiModel()
    { _connection.reset(); }
};

Подводя итоги

1. Популяторы позволяют удобно получать текущее состояние во время подключения к сигналу
2. Блокирующий метод disconnect позволяет отписывать объект в его же деструкторе
3. Чтобы предыдущий пункт был верен и для асинхронных обработчиков, disconnect должен также помечать те вызовы, которые уже лежат в очереди потока, как «неактуальные»

Готовая реализация

Мини-FAQ

class mutex
{
private:
    std::timed_mutex    _m;

public:
    void lock()
    {
        if (_m.try_lock())
            return;

        while (!_m.try_lock_for(std::chrono::seconds(10)))
            Logger::Warning() << "Could not lock mutex " << (void*)this << " for a long time:\n" << get_backtrace_string();
    }

    // ...
};