Use case
This is an event loop and signaling system I created for a piece of software which will have multiple asynchronous server/clients/event-emitters/ui, some of those components will have their own event loop running on a separate thread, some will just post events in the main thread event loop.
I used to work with Qt and I'm kinda new to the stdlib so I'm wondering if there are obvious errors in the thread synchronization stuff and in the use of the std::*. Also I don't know if there are existing solutions for this in the stdlib which can replace part of the solution.
The code seems to work fine and I think it is fairly clean and remarkably simple but any suggestion is welcome and I don't exclude to have made macroscopic errors in the synchronization (you just can't be sure).
Description
The core of the system is the EventLoop class which keeps the queue of events to handle, there are no dispatchers and handlers because I think that part is responsibility of something else and is also case-specific. So the base event class is a simple Callable with a virtual () operator which is called upon event execution in the loop: the event handles himself.
The EventLoop can't be created directly, it is only possible to get the thread_local instance of it (EventLoop::threadInstance()) to be called, for example, in the main thread; or create a new instance in a new thread (EventLoop::newThreadInstance()), so for each thread there can be only one event loop and vice versa.
The system is meant to use (abuse?) lambdas so instead of creating many classes for each type of event we can just create a lambda wrapped in a callable, the possibility of pushing custom events to the loop remains by simply extending Callable.
To be able to directly push lambdas or function pointers to the queue the class Task has been created, A Task wraps an std::function, provides the implicit conversion from lambdas and also adds the functionality of waiting for completion of event in the client thread, see wait() in the examples.
Examples
Create a new loop thread and start adding tasks to it from different threads:
EventLoop & el = EventLoop::newThreadInstance();
el.postTask([](){
//Do stuff on another thread
});
Wait for a task to be completed (like futures)
EventLoop & el = EventLoop::newThreadInstance();
auto task = el.postTask([](){
//Do stuff on another thread
});
task.wait();
2 event loops: Signaling completion with callback:
void callback(){
cout << "Task complete" << endl;
}
int main()
{
EventLoop & el1 = EventLoop::threadInstance();
EventLoop & el2 = EventLoop::newThreadInstance();
el2.postTask([&el1](){
cout << "Start slow task on second thread" << endl;
this_thread::sleep_for(chrono::milliseconds(3000) );
el1.postTask(callback);
});
el1.execute();
}
With lambdas and the event loop we can do quite horrible/wonderful things, we can generically pass input and outputs between threads:
void mainThreadCallback( int result) {
cout << "1st thread: Result is: " << result << endl;
}
int main()
{
EventLoop & el1 = EventLoop::threadInstance();
EventLoop & el2 = EventLoop::newThreadInstance();
EventLoop & el3 = EventLoop::newThreadInstance();
el2.postTask([&el3,&el1](){
while (true) {
this_thread::sleep_for(chrono::milliseconds(rand() % 5000) );
int input = rand() % 100;
cout << "2nd thread: Adding request for " << input << " to 3rd thread" << endl;
el3.postTask([input,&el1](){
this_thread::sleep_for(chrono::milliseconds(rand() % 2000) );
int out = input*2;
cout << "3rd thread: Operation completed, calling callback on first thread" << endl;
el1.postTask([out](){mainThreadCallback(out);});
});
}
});
el1.execute();
}
Code
The source code is less than 200 lines, requires C++11, the git repo also contains the examples.
eventloop.h
#pragma once
#include <queue>
#include <mutex>
#include <atomic>
#include <functional>
#include <condition_variable>
/// @brief Basic callable class interface.
class Callable {
public:
virtual void operator() () = 0;
};
typedef std::shared_ptr<Callable> CallableSharedPtr;
typedef std::function<void(void)> Function;
/// @brief A concrete but generic std::function-based Callable
class Task : public Callable {
public:
Task(Function);
virtual void operator() ();
void wait();
private:
Function _task;
std::mutex _waitMutex;
std::atomic_bool _executed;
std::condition_variable _waitContidition;
};
typedef std::shared_ptr<Task> TaskSharedPtr;
/**
* @brief Keeps a collection of Callables and executes them in a FIFO order.
* Adding callables is thread safe. Can be created by calling the
* thread_local singleton instance threadInstance() or by creating a new
* thread whit its own event loop with newThreadInstance()
*/
class EventLoop {
EventLoop(const EventLoop&) = delete;
EventLoop& operator=(const EventLoop&) = delete;
protected:
EventLoop() = default;
public:
/// @brief Returns the thread-local EventLoop instance of the current thread
static EventLoop & threadInstance();
/// @brief Create a new thread and returns the event loop running on that thread.
static EventLoop & newThreadInstance();
/// @brief Start the event loop, to be used in conjunction with threadInstance()
void execute();
/// @brief Add a Callable to the queue
CallableSharedPtr postCall( CallableSharedPtr event );
/// @brief Add a Task to the queue
TaskSharedPtr postTask( TaskSharedPtr task );
/// @brief Creates an new Task from a function/lambda and add it to the queue, returns the shared ptr to the new task
TaskSharedPtr postTask( Function );
/// @brief Exit from the event loop but finishes the current task if one is being processed
void exit();
private:
std::condition_variable _messagePresent;
std::queue<CallableSharedPtr> _queue;
std::mutex _queueAccessMutex;
std::atomic_bool _running;
};
/// @brief Class to extend to let any object know the event loop it was created in
class EventLoopAware {
public:
EventLoopAware() : eventLoop(EventLoop::threadInstance()){};
EventLoop & eventLoop;
};
eventloop.cpp
#include "eventloop.h"
#include <thread>
using namespace std;
CallableSharedPtr EventLoop::postCall( CallableSharedPtr event ) {
lock_guard<mutex> lock(_queueAccessMutex);
_queue.push(event);
_messagePresent.notify_one();
return event;
}
TaskSharedPtr EventLoop::postTask( TaskSharedPtr taskSharedPointer ) {
postCall(taskSharedPointer);
return taskSharedPointer;
}
TaskSharedPtr EventLoop::postTask( Function f ) {
TaskSharedPtr t = make_shared<Task>(f);
postCall(t);
return t;
}
EventLoop & EventLoop::newThreadInstance() {
EventLoop * el = nullptr;
atomic_bool loopCreated(false);
condition_variable loopCreatedVar;
mutex loopCreatedMutex;
//Start a new thread and initialize the event loop in it
thread thread([&el, &loopCreated, &loopCreatedVar, &loopCreatedMutex](){
// BEWARE: Apparently in some conditions this will not initialize the thread local object but return the reference to an uninitialized instance!
el = &EventLoop::threadInstance();
// ..So we post an empty task to make sure that the thread local Event Loop constructor has been called.
EventLoop::threadInstance().postTask([](){});
// And then notify the outer method that it can return the ready Event Loop instance
unique_lock<mutex> lock(loopCreatedMutex);
loopCreated = true;
loopCreatedVar.notify_all();
lock.unlock();
// Finally we execute the loop itself
EventLoop::threadInstance().execute();
});
thread.detach();
//Wait for the new thread to initialize its eventloop
unique_lock<mutex> lock(loopCreatedMutex);
while (!loopCreated)
loopCreatedVar.wait(lock);
return *el;
}
void EventLoop::execute() {
if (_running)
return;
_running = true;
while (_running)
{
// Wait with a condition variable until someone pushes an event.
unique_lock<mutex> lk(_queueAccessMutex);
while (_queue.empty())
_messagePresent.wait(lk); //Remember: this unlocks the mutex until it returns
// Pop an event from the queue & then unlock the mutex
CallableSharedPtr event = _queue.front();
_queue.pop();
lk.unlock();
//Execute the event in the loop thread
event.get()->operator()();
}
}
void EventLoop::exit(){
_running = false;
}
EventLoop & EventLoop::threadInstance() {
thread_local EventLoop ev;
return ev;
}
Task::Task(Function t)
: _task(t)
, _executed(false)
{}
void Task::operator()() {
_task();
unique_lock<mutex> lk(_waitMutex);
_executed = true;
_waitContidition.notify_all();
}
void Task::wait() {
unique_lock<mutex> lk(_waitMutex);
while (!_executed)
_waitContidition.wait(lk);
}
The code is also on github
