-
Notifications
You must be signed in to change notification settings - Fork 5
TutorialIntermediate
- Connecting from another thread
- Guarding an external state with a signal mutex
- The external state and the populators
- Async handlers and task_executors
- Listenable and listeners
- Signal and handler attributes
The default specialization of signal template is thread-safe.
You may invoke a signal and connect to it from different threads without any problems if you have no external state to guard.
class StupidTimer
{
std::atomic<bool> _alive;
std::thread _thread;
public:
StupidTimer()
: _alive(true)
{
_thread = std::thread(&StupidTimer::ThreadFunc, this);
}
~StupidTimer()
{
_alive = false;
_thread.join();
}
wigwag::signal<void()> OnTick;
private:
void ThreadFunc()
{
while (_alive)
{
std::this_thread::sleep(std::chrono::seconds(1));
OnTick();
}
}
};
// ...
{
StupidTimer timer;
wigwag::token t = timer.OnTick.connect([] { std::cout << "tick" << std::endl; });
std::this_thread::sleep(std::chrono::milliseconds(3500));
}Output:
tick
tick
tick
However, usually you do have a state associated with a signal. In such a case, if you want to modify the state and invoke the signal atomically, you should use the signal::lock_primitive() getter.
In this sample code, the signal internal mutex is locked when the associated state is modified and the signal is invoked and also when the associated state is obtained.
class StupidTimer
{
std::atomic<bool> _alive;
int _tickCount;
std::thread _thread;
public:
StupidTimer()
: _alive(true), _tickCount(0)
{
_thread = std::thread(&StupidTimer::ThreadFunc, this);
}
~StupidTimer()
{
_alive = false;
_thread.join();
}
wigwag::signal<void(int)> OnTick;
int GetTickCount() const
{
std::lock_guard<std::recursive_mutex> l(OnTick.lock_primitive());
return _tickCount;
}
private:
void ThreadFunc()
{
while (_alive)
{
std::this_thread::sleep(std::chrono::seconds(1));
std::lock_guard<std::recursive_mutex> l(OnTick.lock_primitive());
++_tickCount;
OnTick(_tickCount);
}
}
};
// ...
{
StupidTimer timer;
std::cout << "ticks before connect: " << timer.GetTickCount() << std::end;
wigwag::token t = timer.OnTick.connect([] (int n) { std::cout << "tick " << n << std::endl; });
std::this_thread::sleep(std::chrono::milliseconds(3500));
}Output:
ticks before connect: 0
tick 1
tick 2
tick 3
The previous code snippet has an important design flaw:
- There is a race condition between connecting to a signal and obtaining the initial object state.
wigwag::token t = timer.OnTick.connect([] (int n) { std::cout << "tick " << n << std::endl; });
// If a StupidTimer ticks before the following line, we process the new _tickCount value twice
std::cout << "ticks before connect: " << timer.GetTickCount() << std::end;std::cout << "ticks before connect: " << timer.GetTickCount() << std::end;
// If a StupidTimer ticks before we connect to a signal, we skip that _tickCount value
wigwag::token t = timer.OnTick.connect([] (int n) { std::cout << "tick " << n << std::endl; });This may not look dangerous in this particular example. But when the state is more complicated than an integer _tickCount value, this becomes really dangerous.
The naive solution here is to lock the mutex in the code that connects a handler.
{
StupidTimer timer;
std::unique_lock<std::recursive_mutex> l(timer.OnTick.lock_primitive());
std::cout << "ticks before connect: " << timer.GetTickCount() << std::end;
wigwag::token t = timer.OnTick.connect([] (int n) { std::cout << "tick " << n << std::endl; });
l.unlock();
std::this_thread::sleep(std::chrono::milliseconds(3500));
}Luckily, the wigwag library has a better solution for this problem: the populators.
A populator is a user-defined function that is invoked when a handler is connected to a signal. The handler is passed to this function, and the function should invoke it to populate the associated state.
The signal mutex is locked while connecting to a signal and populating the state, so there is no more race condition there.
For a single-object state, this function would just pass the state to the handler:
void OnStateChangedPopulator(const std::function<void(State)>& handler) const
{
handler(_state);
}For a container state, this function would invoke the handler multiple times, as if all the elements were added right at the moment of connecting:
void OnContainerChangedPopulator(const std::function<void(ContainerOp, ElementType)>& handler) const
{
for (const auto& e : _container)
handler(ContainerOp::ItemAdded, e);
}So, here's the StupidTimer code with a populator:
class StupidTimer
{
std::atomic<bool> _alive;
int _tickCount;
std::thread _thread;
public:
StupidTimer()
: _alive(true),
OnTick(std::bind(&StupidTimer::OnTickPopulator, this, std::placeholders::_1))
{
_thread = std::thread(&StupidTimer::ThreadFunc, this);
}
~StupidTimer()
{
_alive = false;
_thread.join();
}
wigwag::signal<void(int)> OnTick;
private:
// This function is invoked when a handler is being connected to a signal
void OnTickPopulator(const std::function<void(int)>& handler) const
{
// The signal mutex is already locked here!
handler(_tickCount);
}
void ThreadFunc()
{
while (_alive)
{
std::this_thread::sleep(std::chrono::seconds(1));
std::lock_guard<std::recursive_mutex> l(OnTick.lock_primitive());
++_tickCount;
OnTick(_tickCount);
}
}
};
// ...
{
StupidTimer timer;
wigwag::token t = timer.OnTick.connect([] (int n) { std::cout << "tick " << n << std::endl; });
std::this_thread::sleep(std::chrono::milliseconds(3500));
}Output:
tick 0
tick 1
tick 2
tick 3
If you want shorter code, you may use a labmda function as a populator:
StupidTimer()
: _alive(true), OnTick([&](const std::function<void(int)>& h) { h(_tickCount); })
{
_thread = std::thread(&StupidTimer::ThreadFunc, this);
}
Or a lambda function with an auto argument type if you use C++14:
StupidTimer()
: _alive(true), OnTick([&](auto&& h) { h(_tickCount); })
{
_thread = std::thread(&StupidTimer::ThreadFunc, this);
}Please note: populators are about connecting to a signal, not invoking it, so you still have to lock the mutex when you modify the associated state and invoke a signal!
Sometimes you want signal handler to be executed not in the same thread where the signal is invoked. A good example is a UI component connecting to some core subsystem.
To connect an asynchronous handler, you should use signal::connect method with a task_executor parameter:
class MediaScannerStateLabel : public UiLabel
{
private:
wigwag::token _token;
public:
MediaScannerStateLabel(const std::shared_ptr<wigwag::task_executor>& uiThread, const MediaScannerPtr& scanner)
{ _token = scanner->OnStateChanged.connect(uiThread, [&](MediaScannerState s) { this->StateChangedHandler(s); }); }
~MediaScannerStateLabel()
{ _token.reset(); }
private:
void StateChangedHandler(MediaScannerState s)
{
switch (s)
{
case MediaScannerState::Scanning:
SetText("Scanning...");
break;
case MediaScannerState::Idle:
default:
SetText("");
break;
}
}
};Here, when a MediaScanner invokes the OnStateChanged signal from its own thread, the MediaScannerStateLabel::StateChangedHandler is executed in the uiThread task_executor.
task_executor is an abstract class. You may use thread_task_executor and threadless_task_executor subclasses from the wigwag library, or implement your own (e.g. a wrapper for a similar entity in you project).
Along with a signal, wigwag library has a similar listenable class. Unfortunately, listenables do not support asynchronous handlers, but still may be preferred in the following situations:
- The state is very complicated and its modifications are hard to express via a single function
- You want emphasize the passing of ownership by using more explicit AddListener semantics instead of signal::connect
Here you can see two identical versions of ObservableSet, that use signal and listenable accordingly to notify the client code about the set modifications. If you don't need asynchronous handlers, you may choose any approach that you like.
You should note that the signal version requres a CollectionOp enum and a switch in any handler, and the listenable version has slightly different invokation syntax.
enum class CollectionOp { ItemAdded, ItemRemoved };
template < typename T_ >
class ObservableSet
{
private:
std::set<T_> _set;
public:
ObservableSet()
: OnChanged([&](const typename decltype(OnChanged)::handler_type& h) { for (auto&& e : _set) h(CollectionOp::ItemAdded, e); })
{ }
wigwag::signal<void(CollectionOp, const T_&)> OnChanged;
void Add(const T_& obj)
{
std::lock_guard<std::recursive_mutex> l(OnChanged.lock_primitive());
auto it = _set.find(obj);
if (it != _set.end())
{
OnChanged(CollectionOp::ItemRemoved, *it);
_set.erase(it);
}
OnChanged(CollectionOp::ItemAdded, obj);
_set.insert(obj);
}
// ...
};
template < typename T_ >
void SetChangedHandler(CollectionOp op, const T_& val)
{
switch (op)
{
case CollectionOp::ItemAdded: std::cout << "Item added: " << val << std::endl; break;
case CollectionOp::ItemRemoved: std::cout << "Item removed: " << val << std::endl; break;
}
}template < typename T_ >
struct IObservableSetListener
{
virtual ~IObservableSetListener() { }
virtual void ItemAdded(const T& obj) = 0;
virtual void ItemRemoved(const T& obj) = 0;
};
template < typename T_ >
using IObservableSetListenerPtr = std::shared_ptr<IObservableSetListener<T_>>;
template < typename T_ >
class ObservableSet
{
private:
std::set<T_> _set;
wigwag::listenable<IObservableSetListenerPtr<T_>> _listenable;
public:
ObservableSet()
: _listenable([&](const IObservableSetListenerPtr<T_>& l) { for (auto&& e : _set) l->ItemAdded(e); })
{ }
wigwag::token AddListener(const IObservableSetListenerPtr<T_>& l)
{ _listenable.connect(l); }
void Add(const T_& obj)
{
std::lock_guard<std::recursive_mutex> l(_listenable.lock_primitive());
auto it = _set.find(obj);
if (it != _set.end())
{
_listenable.invoke([&](const IObservableSetListenerPtr<T_>& l) { l->ItemRemoved(*it); });
_set.erase(it);
}
_listenable.invoke([&](const IObservableSetListenerPtr<T_>& l) { l->ItemAdded(obj); });
_set.insert(obj);
}
// ...
};
template < typename T_ >
struct SetChangedHandler : public IObservableSetListener<T_>
{
void ItemAdded(const T_& obj) { std::cout << "Item added: " << val << std::endl; }
void ItemRemoved(const T_& obj) { std::cout << "Item removed: " << val << std::endl; }
};Here's a simple Logger implementation that uses listenable for managing its sinks:
struct ILoggerSink
{
virtual ~ILoggerSink() { }
virtual Log(LogLevel logLevel, const LoggerMessage& msg) = 0;
};
using ILoggerSinkPtr = std::shared_ptr<ILoggerSink>;
class Logger
{
private:
static wigwag::listenable<ILoggerSinkPtr> s_sinks;
static std::atomic<LogLevel> s_logLevel { LogLevel::Info };
public:
static void SetLogLevel(LogLevel logLevel)
{ s_logLevel = logLevel; }
static wigwag::token RegisterSink(const ILoggerSinkPtr& sink)
{ s_sinks.connect(sink); }
static void Log(LogLevel logLevel, const std::string& msg)
{
if (logLevel > s_logLevel)
s_sinks.invoke([&](const ILoggerSinkPtr& s) { s->Log(logLevel, msg); });
}
};You can also adjust some behavior of signals and handler by setting signal_attributes and handler_attributes flags.
If you want to forbid connecting synchronous handlers because they might block the signalling thread for a long time, you should pass the signal_attributes::connect_async_only value to the appropriate signal constructor:
signal<void()> s(signal_attributes::connect_async_only);You may use handler attributes to suppress populator for this particular handler. This may be useful in UI code, when you want to show notifications for all new events (removable storage inserted, etc) but do not want to do that for the populated ones (10 more storages that were already there when your application started). To do that, pass the handler_attributes::suppress_populator value to the signal::connect method.
token t = s.connect(&HandlerFunc, handler_attributes::suppress_populator);