{"id":387029,"date":"2024-06-29T07:04:42","date_gmt":"2024-06-29T07:04:42","guid":{"rendered":"http:\/\/savepearlharbor.com\/?p=387029"},"modified":"-0001-11-30T00:00:00","modified_gmt":"-0001-11-29T21:00:00","slug":"","status":"publish","type":"post","link":"https:\/\/savepearlharbor.com\/?p=387029","title":{"rendered":"<span>Idiomatic Event Loop in C++<\/span>"},"content":{"rendered":"<div><!--[--><!--]--><\/div>\n<div id=\"post-content-body\">\n<div>\n<div class=\"article-formatted-body article-formatted-body article-formatted-body_version-2\">\n<div xmlns=\"http:\/\/www.w3.org\/1999\/xhtml\">\n<ul>\n<li>\n<p><a href=\"#Introduction\" rel=\"noopener noreferrer nofollow\">Introduction<\/a><\/p>\n<\/li>\n<li>\n<p><a href=\"#Event%20Loop%20as%20a%20Tool\" rel=\"noopener noreferrer nofollow\">Event Loop as a Tool<\/a><\/p>\n<\/li>\n<li>\n<p><a href=\"#TL;DR;%20Show%20Me%20The%20Code!\" rel=\"noopener noreferrer nofollow\">TL;DR; Show Me The Code!<\/a><\/p>\n<\/li>\n<li>\n<p><a href=\"#The%20Basic%20Implementation\" rel=\"noopener noreferrer nofollow\">The Basic Implementation<\/a><\/p>\n<ul>\n<li>\n<p><a href=\"#The%20Power%20of%20stdfunction\" rel=\"noopener noreferrer nofollow\">The Power of std::function<\/a><\/p>\n<\/li>\n<li>\n<p><a href=\"#The%20Power%20of%20stdcondition_variable\" rel=\"noopener noreferrer nofollow\">The Power of std::condition_variable<\/a><\/p>\n<\/li>\n<li>\n<p><a href=\"#The%20Power%20of%20Double%20Buffering\" rel=\"noopener noreferrer nofollow\">The Power of Double Buffering<\/a><\/p>\n<\/li>\n<li>\n<p><a href=\"#A%20Couple%20of%20Remarks%20Regarding%20noexcept\" rel=\"noopener noreferrer nofollow\">A Couple of Remarks Regarding noexcept<\/a><\/p>\n<\/li>\n<\/ul>\n<\/li>\n<li>\n<p><a href=\"#Just%20a%20Bit%20Extra\" rel=\"noopener noreferrer nofollow\">Just a Bit Extra<\/a><\/p>\n<ul>\n<li>\n<p><a href=\"#enqueueSync()\" rel=\"noopener noreferrer nofollow\">enqueueSync()<\/a><\/p>\n<\/li>\n<li>\n<p><a href=\"#enqueueAsync()\" rel=\"noopener noreferrer nofollow\">enqueueAsync()<\/a><\/p>\n<\/li>\n<\/ul>\n<\/li>\n<li>\n<p><a href=\"#Examples\" rel=\"noopener noreferrer nofollow\">Examples<\/a><\/p>\n<ul>\n<li>\n<p><a href=\"#Access%20Serialization\" rel=\"noopener noreferrer nofollow\">Access Serialization<\/a><\/p>\n<\/li>\n<li>\n<p><a href=\"#Event%20Handlers\" rel=\"noopener noreferrer nofollow\">Event Handlers<\/a><\/p>\n<\/li>\n<\/ul>\n<\/li>\n<li>\n<p><a href=\"#Conclusion\" rel=\"noopener noreferrer nofollow\">Conclusion<\/a><\/p>\n<\/li>\n<\/ul>\n<p><a class=\"anchor\" name=\"Introduction\" id=\"Introduction\"><\/a><\/p>\n<h2>Introduction<\/h2>\n<p>Today I want to show you a simple but at the same time a very efficient implementation of a well-known concurrency pattern called Event Loop. There are very good libraries out there implementing this pattern but there are lots of cases when using them is an overcomplication. Sometimes it\u2019s enough to have something small and idiomatic, made of C++11 standard library elements, rather than a universal multitool.<\/p>\n<p><a class=\"anchor\" name=\"Event%20Loop%20as%20a%20Tool\" id=\"Event Loop as a Tool\"><\/a><\/p>\n<h2>Event Loop as a Tool<\/h2>\n<p>As a first case, just imagine you\u2019ve got a bunch of classes that are not designed to work in a multi-threaded environment. Maybe because they are inherited from a legacy part of the system, or maybe you are designing new classes right now and you don\u2019t want to overengineer them with a bunch of mutexes inside. But you need them to get accessed from different threads keeping things as simple as possible.<\/p>\n<p>The second case when the pattern can be really helpful is when you have a global object which is either impractical or even impossible to guard with a mutex. As an example, let&#8217;s take the OpenGL context. Actually, there is no such thing as a \u201ccontext\u201d class\/structure\/type in OpenGL like <em>ID3D11Device<\/em> in DirectX 11 or <em>VkInstance<\/em> in Vulkan. The context is inherently global, but for the current thread. It is exclusively owned by the thread through <em>Thread Local Storage<\/em>. So each time you try to change your OpenGL state it really does matter which thread you make OpenGL function calls from. In this case, things get problematic if you have a dedicated thread that loads assets (images, geometry, etc) and you want to transfer the loaded data from the loading thread to the context-owning thread.<\/p>\n<p>In both cases, it would be useful to let other threads \u201csee\u201d a shareable object. However, the real access to that object would go through a dedicated thread.<\/p>\n<p>To summarize, Event Loop can be considered as an alternative for the mutex. Both of them serialize accesses to the guarded object, but in a slightly different manner.<\/p>\n<p><a class=\"anchor\" name=\"TL;DR;%20Show%20Me%20The%20Code!\" id=\"TL;DR; Show Me The Code!\"><\/a><\/p>\n<h2>TL;DR; Show Me The Code!<\/h2>\n<p>An example of how to use it.<\/p>\n<pre><code class=\"cpp\">#include &lt;iostream>  int main() { { EventLoop eventLoop;  eventLoop.enqueue([] { std::cout &lt;&lt; \"message from a different thread\\n\"; });  std::cout &lt;&lt; \"prints before or after the message above\\n\"; }  std::cout &lt;&lt; \"guaranteed to be printed the last\\n\"; } <\/code><\/pre>\n<p>And the implementation itself.<\/p>\n<pre><code class=\"cpp\">#include &lt;condition_variable> #include &lt;functional> #include &lt;future> #include &lt;thread> #include &lt;vector>  class EventLoop { public: using callable_t = std::function&lt;void()>;  EventLoop() = default; EventLoop(const EventLoop&amp;) = delete; EventLoop(EventLoop&amp;&amp;) noexcept = delete; ~EventLoop() noexcept { enqueue([this] { m_running = false; }); m_thread.join(); }  EventLoop&amp; operator= (const EventLoop&amp;) = delete; EventLoop&amp; operator= (EventLoop&amp;&amp;) noexcept = delete;  void enqueue(callable_t&amp;&amp; callable) noexcept { { std::lock_guard&lt;std::mutex> guard(m_mutex); m_writeBuffer.emplace_back(std::move(callable)); } m_condVar.notify_one(); }  private: std::vector&lt;callable_t> m_writeBuffer; std::mutex m_mutex; std::condition_variable m_condVar; bool m_running{ true }; std::thread m_thread{ &amp;EventLoop::threadFunc, this };  void threadFunc() noexcept { std::vector&lt;callable_t> readBuffer;  while (m_running) { { std::unique_lock&lt;std::mutex> lock(m_mutex); m_condVar.wait(lock, [this] { return !m_writeBuffer.empty(); }); std::swap(readBuffer, m_writeBuffer); }  for (callable_t&amp; func : readBuffer) { func(); }  readBuffer.clear(); } } }; <\/code><\/pre>\n<p><a class=\"anchor\" name=\"The%20Basic%20Implementation\" id=\"The Basic Implementation\"><\/a><\/p>\n<h2>The Basic Implementation<\/h2>\n<p>The implementation is quite compact. Feels idiomatic, doesn\u2019t it? But what is so special about it and what makes it efficient?<\/p>\n<p><a class=\"anchor\" name=\"The%20Power%20of%20stdfunction\" id=\"The Power of stdfunction\"><\/a><\/p>\n<h3>The Power of std::function<\/h3>\n<p><em>std::function&lt;R(Args\u2026)><\/em> is a really interesting thing. It uses two very important idioms that make it so useful for us \u2013 <em>Type Erasure<\/em> and <em>Small-Object Optimization<\/em>.<\/p>\n<p><em>Type Erasure<\/em> idiom allows us to store anything that we can apply the <em>call operator <\/em>to. I will refer to it as <em>callable_t<\/em>. It can be a C-like function, it can be a functor. It can also be a lambda, including a generic lambda.<\/p>\n<p>Since our <em>callable_t<\/em> can have internal data, such as the functor\u2019s members or the lambda\u2019s capture, <em>std::function<\/em> also has to store all this data inside. In order to avoid or at least minimize heap allocations, <em>std::function<\/em> can store <em>callable_t<\/em> in place if it\u2019s small enough. What is \u201csmall enough\u201d depends on the implementation. If it doesn\u2019t fit into the internal storage, then heap allocation happens as a fallback. This is what <em>Small-Object Optimization<\/em> essentially is.<\/p>\n<p>Knowing all of that, you can use a <em>std::vector<\/em> of <em>std::function<\/em> to keep both data and pointers to <em>vtables <\/em>in a single chunk of memory for most cases. And this is the first member of our class \u2013 <em>std::vector&lt;callable_t> m_writeBuffer<\/em>;<\/p>\n<p><a class=\"anchor\" name=\"The%20Power%20of%20stdcondition_variable\" id=\"The Power of stdcondition_variable\"><\/a><\/p>\n<h3>The Power of std::condition_variable<\/h3>\n<p>A condition variable is a synchronization primitive that makes one thread postpone its execution via <em>wait()<\/em> until another thread wakes it up via <em>notify_one()<\/em>. But what if the second thread has called <em>notify_one()<\/em> just before the first thread calls <em>wait()<\/em>? If you used a simpler synchronization primitive, such as <a href=\"https:\/\/docs.microsoft.com\/en-us\/windows\/win32\/sync\/event-objects\" rel=\"noopener noreferrer nofollow\"><u>Win32 Event Object<\/u><\/a>, the first thread would not see the notification at all. In this case, it would fall asleep ending you up having your program deadlocked. Fortunately, <em>std::condition_variable<\/em> is a bit smarter. It deliberately wants you to lock the mutex which guards some state. While you\u2019re holding the locked mutex <em>std::condition_variable<\/em> wants you to check the state if the first thread has to fall asleep or the condition has already been satisfied and the thread just needs to keep going. If it turns out that the thread has to be postponed something interesting happens next. Have you noticed that the wait<em>()<\/em> function accepts that locked mutex? This is because the <em>wait()<\/em> function asks your operating system to do the following:<\/p>\n<ol>\n<li>\n<p>Atomically unlock the mutex and postpone the execution.<\/p>\n<\/li>\n<li>\n<p>Atomically lock the mutex and resume the execution when <em>notify_one()<\/em> \/ <em>notify_all()<\/em> call occurs.<\/p>\n<\/li>\n<\/ol>\n<p>I\u2019m saying \u201catomically\u201d here, but maybe it\u2019s not what you\u2019re thinking about. It\u2019s a feature of the operating system thread scheduler, not the processors\u2019 hardware. You cannot emulate that behavior using atomic variables.<\/p>\n<p>Sometimes the OS may wake the waiting thread up spontaneously. It\u2019s called \u201cspurious wakeups\u201d. There are reasons for that, but I won&#8217;t cover them here. But you should be prepared for them. So if the thread has been woken up, you need to check your condition again. That\u2019s why I\u2019m passing the predicate to the <em>wait()<\/em> function here. That version of <em>wait()<\/em> tests the condition before falling asleep and immediately after. You can read more <a href=\"https:\/\/en.cppreference.com\/w\/cpp\/thread\/condition_variable\/wait\" rel=\"noopener noreferrer nofollow\"><u>about it here<\/u><\/a>.<\/p>\n<p>As you can see we have to have a protected state guarded by the mutex. Therefore, the second notifying thread should do the following:<\/p>\n<ol>\n<li>\n<p>Lock the mutex, change the shared state, and unlock the mutex.<\/p>\n<\/li>\n<li>\n<p>Notify the first thread.<\/p>\n<\/li>\n<\/ol>\n<p>This is essentially what happens in the <em>enqueue()<\/em> function. Some developers call <em>notify_one()<\/em> while holding the mutex. It\u2019s not wrong, but it makes the scheme inefficient. In order to avoid extra synchronizations, just make sure you call the <em>notify_one() <\/em><strong>after<\/strong> you release the mutex. You can read <a href=\"https:\/\/en.cppreference.com\/w\/cpp\/thread\/condition_variable\/notify_one\" rel=\"noopener noreferrer nofollow\"><u>the Notes section<\/u><\/a> to learn more about this problem.<\/p>\n<p><a class=\"anchor\" name=\"The%20Power%20of%20Double%20Buffering\" id=\"The Power of Double Buffering\"><\/a><\/p>\n<h3>The Power of Double Buffering<\/h3>\n<p>What really makes this implementation especially efficient is that we have two buffers here. You may have noticed that we are swapping <em>readBuffer<\/em> and <em>m_writeBuffer<\/em>. And we are doing this while the mutex is being locked. <em>std::swap<\/em> simply swaps the pointers inside those two vectors, which is an extremely fast operation. So we are leaving <em>m_writeBuffer<\/em> empty, ready to be filled again. Next to the <em>std::swap<\/em> the scope ends unlocking the mutex.<\/p>\n<p>Now we have a situation where the write buffer is getting filled while the read buffer is being processed. Now, these two processes can go simultaneously without any intersection! When the processing is over, we clear the read buffer. Clearing of <em>std::vector<\/em> does not cause the underlying storage deallocation. So when we quickly swap those two buffers again this underlying storage is going to be filled up again as the write buffer.<\/p>\n<p>When you\u2019re calling <em>enqueue()<\/em> and your <em>callable_t<\/em> is small enough you won\u2019t even touch the heap while constructing it. Inserting this <em>callable_t <\/em>into the vector that has already got some storage left after the processing step takes almost nothing! What about locking the mutex? It turns out that modern mutexes use atomic spin-locks as the first step and after several iterations, they ask the operating system to postpone the thread. So if you do something really quick between <em>lock()<\/em> and <em>unlock()<\/em> you won\u2019t even disturb the operating system. <em>notify_one()<\/em> is also designed to be fast in case you call it while the mutex is unlocked. So you shouldn\u2019t be bothered with it here. It makes <em>enqueue()<\/em> extremely fast on average. It also makes <em>wait()<\/em> fast as well, because we simply check if <em>m_writeBuffer <\/em>is not empty and swap it.<\/p>\n<p><a class=\"anchor\" name=\"A%20Couple%20of%20Remarks%20Regarding%20noexcept\" id=\"A Couple of Remarks Regarding noexcept\"><\/a><\/p>\n<h3>A Couple of Remarks Regarding &#171;noexcept&#187;<\/h3>\n<p>You may be wondering, why am I using <em>noexcept<\/em> keywords all over the place. And there is a really good explanation in <a href=\"https:\/\/isocpp.github.io\/CppCoreGuidelines\/CppCoreGuidelines#Rf-noexcept\" rel=\"noopener noreferrer nofollow\"><u>ISO C++ Core Guidelines<\/u><\/a> why you should at least consider using it. Let\u2019s take the <em>enqueue()<\/em> function. Even though we do have vector insertion here, which may throw <em>std::bad_alloc<\/em>, this scenario is actually disastrous. Your system either is lacking memory, which can cause a crash somewhere else, or you pushed too many tasks that your working thread cannot handle, which is in fact a poor application design. The lack of the system memory may also prevent throwing the exception about well\u2026 the lack of the system memory since <em>throw<\/em> uses heap allocation. <a href=\"https:\/\/isocpp.github.io\/CppCoreGuidelines\/CppCoreGuidelines#Rc-dtor-noexcept\" rel=\"noopener noreferrer nofollow\"><u>The same logic is applied to the destructor<\/u><\/a>. I\u2019m also enforcing <em>callable_t<\/em> to be passed by r-value reference to the <em>enqueue()<\/em> function. In this case, if the <em>callable_t<\/em> constructor fails it happens outside the Event Loop class.<\/p>\n<p>The situation with the thread function is different. Unfortunately, we cannot declare <em>callable_t<\/em> as <em>std::function&lt;void()<\/em><strong><em>noexcept<\/em><\/strong><em>>()<\/em> enforcing the client to catch all the exceptions. So if the user\u2019s code throws, we cannot properly handle it. I\u2019m not sure that catching all exceptions by the event loop is a suitable strategy. I\u2019d prefer to just automatically <em>std::terminate()<\/em> here. But it\u2019s up to you to decide.<\/p>\n<p><a class=\"anchor\" name=\"Just%20a%20Bit%20Extra\" id=\"Just a Bit Extra\"><\/a><\/p>\n<h2>Just a Bit Extra<\/h2>\n<p>If having the <em>enqueue()<\/em> function is not enough for you and you want to wait for the result I\u2019ve got a couple of solutions for you.<\/p>\n<p><a class=\"anchor\" name=\"enqueueSync()\" id=\"enqueueSync()\"><\/a><\/p>\n<h3>enqueueSync()<\/h3>\n<p>Just an example of what it does and how to use <em>enqueueSync()<\/em>:<\/p>\n<pre><code class=\"cpp\">std::cout &lt;&lt; eventLoop.enqueueSync([](const int&amp; x, int&amp;&amp; y, int z) { return x + y + z; }, 1, 2, 3); <\/code><\/pre>\n<p>And the implementation:<\/p>\n<pre><code class=\"cpp\">template&lt;typename Func, typename... Args> auto enqueueSync(Func&amp;&amp; callable, Args&amp;&amp; ...args) { if (std::this_thread::get_id() == m_thread.get_id()) { return std::invoke( std::forward&lt;Func>(callable), std::forward&lt;Args>(args)...); }  using return_type = std::invoke_result_t&lt;Func, Args...>; using packaged_task_type = std::packaged_task&lt;return_type(Args&amp;&amp;...)>;  packaged_task_type task(std::forward&lt;Func>(callable));  enqueue([&amp;] { task(std::forward&lt;Args>(args)...); });  return task.get_future().get(); } <\/code><\/pre>\n<p>The first <em>if<\/em>-condition is a protection from a deadlock. Sometimes you may discover a situation when some synchronous task is trying to schedule another synchronous task, leading to a deadlock.<\/p>\n<p>I\u2019m also using here<em> std::packaged_task<\/em> in conjunction with the <em>std::future<\/em>. This is a nice way to transfer the function invocation result across the threads via <em>std::future<\/em>, including all of the exceptions that occurred. Please note, that the <em>enqueueSync() <\/em>function is not declared as <em>noexcept<\/em> for this purpose.<\/p>\n<p><a class=\"anchor\" name=\"enqueueAsync()\" id=\"enqueueAsync()\"><\/a><\/p>\n<h3>enqueueAsync()<\/h3>\n<p>The previous example uses <em>std::future<\/em>. Sometimes you may find it useful to obtain it for further usage instead of waiting for it immediately. This is an example of how to use <em>enqueueAsync()<\/em>.<\/p>\n<pre><code class=\"cpp\">std::future&lt;int> result = eventLoop.enqueueAsync([](int x, int y) {   return x + y; }, 1, 2); \/\/ \/\/do some heavy work here \/\/ std::cout &lt;&lt; result.get(); <\/code><\/pre>\n<p>And this is the implementation.<\/p>\n<pre><code class=\"cpp\">template&lt;typename Func, typename... Args> [[nodiscard]] auto enqueueAsync(Func&amp;&amp; callable, Args&amp;&amp; ...args) { using return_type = std::invoke_result_t&lt;Func, Args...>; using packaged_task_type = std::packaged_task&lt;return_type()>;  auto taskPtr = std::make_shared&lt;packaged_task_type>(std::bind( std::forward&lt;Func>(callable), std::forward&lt;Args>(args)...));  enqueue(std::bind(&amp;packaged_task_type::operator(), taskPtr));  return taskPtr->get_future(); } <\/code><\/pre>\n<p>Several remarks here.<\/p>\n<p>Firstly, as you can see, there is no deadlock protection here, since it\u2019s impossible to detect when and where the result is going to be used. So it\u2019s up to the user to call the method properly.<\/p>\n<p>Secondly, we are using <em>std::shared_ptr<\/em> here. This is because we are bypassing the limitation of <em>std::packaged_task<\/em>, which is movable only. However, <em>std::function<\/em> requires the underlying object to be copyable.<\/p>\n<p>And finally, we\u2019re using here <em>std::bind<\/em> to <strong>copy or move<\/strong> all the arguments, because we are not aware of their lifetime. It\u2019s a protection from dangling references. If you really want to pass an object by reference to <em>enqueueAsync()<\/em>, you can either capture it as [&amp;] while defining lambda or using <em>std::ref()<\/em> or <em>std::cref()<\/em>.<\/p>\n<p><a class=\"anchor\" name=\"Examples\" id=\"Examples\"><\/a><\/p>\n<h2>Examples<\/h2>\n<p><a class=\"anchor\" name=\"Access%20Serialization\" id=\"Access Serialization\"><\/a><\/p>\n<h2>Access Serialization<\/h2>\n<p>Let\u2019s just imagine you have a bank account that is not thread-safe.<\/p>\n<pre><code class=\"cpp\">struct IBankAccount { virtual ~IBankAccount() = default; virtual void pay(unsigned amount) noexcept = 0; virtual void acquire(unsigned amount) noexcept = 0; virtual long long balance() const noexcept = 0; };  class ThreadUnsafeAccount : public IBankAccount { public: ThreadUnsafeAccount(long long balance) : m_balance(balance) { } void pay(unsigned amount) noexcept override { m_balance -= amount; } void acquire(unsigned amount) noexcept override { m_balance += amount; } long long balance() const noexcept override { return m_balance; } private: long long m_balance; }; <\/code><\/pre>\n<p>If we wrap it around a proxy object like as follows we can start using it in a multithreaded environment.<\/p>\n<pre><code class=\"cpp\">class ThreadSafeAccount : public IBankAccount { public: ThreadSafeAccount( std::shared_ptr&lt;EventLoop> eventLoop, std::shared_ptr&lt;IBankAccount> unknownBankAccount) :  m_eventLoop(std::move(eventLoop)), m_unknownBankAccount(std::move(unknownBankAccount)) { }  void pay(unsigned amount) noexcept override { \/\/don't use this alternative because [=] or [&amp;] captures this, \/\/but not std::shared_ptr. \/\/m_eventLoop->enqueue([=]() \/\/{ \/\/m_unknownBankAccount->pay(amount); \/\/});  \/\/use this alternative instead m_eventLoop->enqueue(std::bind( &amp;IBankAccount::pay, m_unknownBankAccount, amount)); } void acquire(unsigned amount) noexcept override { m_eventLoop->enqueue(std::bind( &amp;IBankAccount::acquire, m_unknownBankAccount, amount)); } long long balance() const noexcept override { \/\/capturing via [&amp;] is perfectly valid here return m_eventLoop->enqueueSync([&amp;] { return m_unknownBankAccount->balance(); });  \/\/or you can use this variant for consistency \/\/return m_eventLoop->enqueueSync( \/\/&amp;IBankAccount::balance, m_unknownBankAccount); } private: std::shared_ptr&lt;EventLoop> m_eventLoop; std::shared_ptr&lt;IBankAccount> m_unknownBankAccount; }; <\/code><\/pre>\n<p>Now you can start using your initially thread-unsafe bank account from various threads via proxies without any risk of a race condition.<\/p>\n<pre><code class=\"cpp\">int main() { auto eventLoop = std::make_shared&lt;EventLoop>(); auto bankAccount = std::make_shared&lt;ThreadUnsafeAccount>(100'000);  std::thread buy = std::thread([](std::unique_ptr&lt;IBankAccount> account) { for (int i = 1; i &lt;= 10; ++i) { account->pay(i); } }, std::make_unique&lt;ThreadSafeAccount>(eventLoop, bankAccount));  std::thread sell = std::thread([](std::unique_ptr&lt;IBankAccount> account) { for (int i = 1; i &lt;= 10; ++i) { account->acquire(i); } }, std::make_unique&lt;ThreadSafeAccount>(eventLoop, bankAccount));  buy.join(); sell.join();  std::cout &lt;&lt; bankAccount->balance() &lt;&lt; '\\n'; } <\/code><\/pre>\n<p>Interestingly, the proxy object itself is inherently thread-safe, so you can safely share it between multiple threads.<\/p>\n<pre><code class=\"cpp\">int main() { ThreadSafeAccount safeAccount( std::make_shared&lt;EventLoop>(), std::make_shared&lt;ThreadUnsafeAccount>(100'000));   std::thread buy = std::thread([&amp;]() { for (int i = 1; i &lt;= 10; ++i) { safeAccount.pay(i); } });  std::thread sell = std::thread([&amp;] { for (int i = 1; i &lt;= 10; ++i) { safeAccount.acquire(i); } });  buy.join(); sell.join();  std::cout &lt;&lt; safeAccount.balance() &lt;&lt; '\\n'; } <\/code><\/pre>\n<p>If you\u2019re old enough and it all seems familiar to you, you\u2019re right. This is a manually-crafted <a href=\"https:\/\/docs.microsoft.com\/en-us\/windows\/win32\/com\/single-threaded-apartments\" rel=\"noopener noreferrer nofollow\"><u>Single-Threaded Apartment<\/u><\/a>. Or oversimplified <a href=\"https:\/\/doc.qt.io\/qt-5\/qeventloop.html\" rel=\"noopener noreferrer nofollow\"><u>QEventLoop<\/u><\/a>.<\/p>\n<p><a class=\"anchor\" name=\"Event%20Handlers\" id=\"Event Handlers\"><\/a><\/p>\n<h2>Event Handlers<\/h2>\n<p>Let\u2019s imagine you need to build a message based-system that supports dedicated handlers for each message type. Each type of message should be processed by a specific handler. However, it is possible that a message can be sent from an arbitrary thread. A canonical example is GUI \u2013 mouse clicks, button clicks, and all that jazz. This is how you can implement it with the basic event loop.<\/p>\n<p>First, you declare the event and its trigger.<\/p>\n<pre><code class=\"cpp\">std::function&lt;void(std::vector&lt;char>)> OnNetworkEvent;  void emitNetworkEvent(EventLoop&amp; loop, std::vector&lt;char> data) { if (!OnNetworkEvent) return;  loop.enqueue(std::bind(std::ref(OnNetworkEvent), std::move(data))); } <\/code><\/pre>\n<p>And then you just register your handler and start triggering it from various threads.<\/p>\n<pre><code class=\"cpp\">int main() { \/\/registering event handler OnNetworkEvent = [](std::vector&lt;char>&amp; message) { std::cout &lt;&lt; message.size() &lt;&lt; ' '; };  EventLoop loop;  \/\/let's trigger the event from different threads std::thread t1 = std::thread([](EventLoop&amp; loop) { for (std::size_t i = 0; i &lt; 10; ++i) { emitNetworkEvent(loop, std::vector&lt;char>(i)); } }, std::ref(loop));  std::thread t2 = std::thread([](EventLoop&amp; loop) { for (int i = 10; i &lt; 20; ++i) { emitNetworkEvent(loop, std::vector&lt;char>(i)); } }, std::ref(loop));  for (int i = 20; i &lt; 30; ++i) { emitNetworkEvent(loop, std::vector&lt;char>(i)); }  t1.join(); t2.join();  loop.enqueue([] { std::cout &lt;&lt; std::endl; }); } <\/code><\/pre>\n<p><a class=\"anchor\" name=\"Conclusion\" id=\"Conclusion\"><\/a><\/p>\n<h2>Conclusion<\/h2>\n<p>I hope I managed to give you a new tool that could make your life easier by adding just a few lines of code. Of course, it\u2019s not a universal tool, but in most cases, you simply just don\u2019t need anything more than <em>enqueue()<\/em> or <em>enqueueSync()<\/em>.<\/p>\n<p>For more complicated use-cases such as adding priority to messages maybe it\u2019s better to consider a different implementation. You could actually just replace <em>std::vector<\/em> with <em>std::priority_queue<\/em> but it would also mean that you should keep only the <em>enqueue()<\/em> function.<\/p>\n<p>Thanks for reading.<\/p>\n<\/p>\n<\/div>\n<\/div>\n<\/div>\n<p><!----><!----><\/div>\n<p><!----><!----><br \/> \u0441\u0441\u044b\u043b\u043a\u0430 \u043d\u0430 \u043e\u0440\u0438\u0433\u0438\u043d\u0430\u043b \u0441\u0442\u0430\u0442\u044c\u0438 <a href=\"https:\/\/habr.com\/ru\/articles\/665730\/\"> https:\/\/habr.com\/ru\/articles\/665730\/<\/a><\/p>\n","protected":false},"excerpt":{"rendered":"<div><!--[--><!--]--><\/div>\n<div id=\"post-content-body\">\n<div>\n<div class=\"article-formatted-body article-formatted-body article-formatted-body_version-2\">\n<div xmlns=\"http:\/\/www.w3.org\/1999\/xhtml\">\n<ul>\n<li>\n<p><a href=\"#Introduction\" rel=\"noopener noreferrer nofollow\">Introduction<\/a><\/p>\n<\/li>\n<li>\n<p><a href=\"#Event%20Loop%20as%20a%20Tool\" rel=\"noopener noreferrer nofollow\">Event Loop as a Tool<\/a><\/p>\n<\/li>\n<li>\n<p><a href=\"#TL;DR;%20Show%20Me%20The%20Code!\" rel=\"noopener noreferrer nofollow\">TL;DR; Show Me The Code!<\/a><\/p>\n<\/li>\n<li>\n<p><a href=\"#The%20Basic%20Implementation\" rel=\"noopener noreferrer nofollow\">The Basic Implementation<\/a><\/p>\n<ul>\n<li>\n<p><a href=\"#The%20Power%20of%20stdfunction\" rel=\"noopener noreferrer nofollow\">The Power of std::function<\/a><\/p>\n<\/li>\n<li>\n<p><a href=\"#The%20Power%20of%20stdcondition_variable\" rel=\"noopener noreferrer nofollow\">The Power of std::condition_variable<\/a><\/p>\n<\/li>\n<li>\n<p><a href=\"#The%20Power%20of%20Double%20Buffering\" rel=\"noopener noreferrer nofollow\">The Power of Double Buffering<\/a><\/p>\n<\/li>\n<li>\n<p><a href=\"#A%20Couple%20of%20Remarks%20Regarding%20noexcept\" rel=\"noopener noreferrer nofollow\">A Couple of Remarks Regarding noexcept<\/a><\/p>\n<\/li>\n<\/ul>\n<\/li>\n<li>\n<p><a href=\"#Just%20a%20Bit%20Extra\" rel=\"noopener noreferrer nofollow\">Just a Bit Extra<\/a><\/p>\n<ul>\n<li>\n<p><a href=\"#enqueueSync()\" rel=\"noopener noreferrer nofollow\">enqueueSync()<\/a><\/p>\n<\/li>\n<li>\n<p><a href=\"#enqueueAsync()\" rel=\"noopener noreferrer nofollow\">enqueueAsync()<\/a><\/p>\n<\/li>\n<\/ul>\n<\/li>\n<li>\n<p><a href=\"#Examples\" rel=\"noopener noreferrer nofollow\">Examples<\/a><\/p>\n<ul>\n<li>\n<p><a href=\"#Access%20Serialization\" rel=\"noopener noreferrer nofollow\">Access Serialization<\/a><\/p>\n<\/li>\n<li>\n<p><a href=\"#Event%20Handlers\" rel=\"noopener noreferrer nofollow\">Event Handlers<\/a><\/p>\n<\/li>\n<\/ul>\n<\/li>\n<li>\n<p><a href=\"#Conclusion\" rel=\"noopener noreferrer nofollow\">Conclusion<\/a><\/p>\n<\/li>\n<\/ul>\n<p><a class=\"anchor\" name=\"Introduction\" id=\"Introduction\"><\/a><\/p>\n<h2>Introduction<\/h2>\n<p>Today I want to show you a simple but at the same time a very efficient implementation of a well-known concurrency pattern called Event Loop. There are very good libraries out there implementing this pattern but there are lots of cases when using them is an overcomplication. Sometimes it\u2019s enough to have something small and idiomatic, made of C++11 standard library elements, rather than a universal multitool.<\/p>\n<p><a class=\"anchor\" name=\"Event%20Loop%20as%20a%20Tool\" id=\"Event Loop as a Tool\"><\/a><\/p>\n<h2>Event Loop as a Tool<\/h2>\n<p>As a first case, just imagine you\u2019ve got a bunch of classes that are not designed to work in a multi-threaded environment. Maybe because they are inherited from a legacy part of the system, or maybe you are designing new classes right now and you don\u2019t want to overengineer them with a bunch of mutexes inside. But you need them to get accessed from different threads keeping things as simple as possible.<\/p>\n<p>The second case when the pattern can be really helpful is when you have a global object which is either impractical or even impossible to guard with a mutex. As an example, let&#8217;s take the OpenGL context. Actually, there is no such thing as a \u201ccontext\u201d class\/structure\/type in OpenGL like <em>ID3D11Device<\/em> in DirectX 11 or <em>VkInstance<\/em> in Vulkan. The context is inherently global, but for the current thread. It is exclusively owned by the thread through <em>Thread Local Storage<\/em>. So each time you try to change your OpenGL state it really does matter which thread you make OpenGL function calls from. In this case, things get problematic if you have a dedicated thread that loads assets (images, geometry, etc) and you want to transfer the loaded data from the loading thread to the context-owning thread.<\/p>\n<p>In both cases, it would be useful to let other threads \u201csee\u201d a shareable object. However, the real access to that object would go through a dedicated thread.<\/p>\n<p>To summarize, Event Loop can be considered as an alternative for the mutex. Both of them serialize accesses to the guarded object, but in a slightly different manner.<\/p>\n<p><a class=\"anchor\" name=\"TL;DR;%20Show%20Me%20The%20Code!\" id=\"TL;DR; Show Me The Code!\"><\/a><\/p>\n<h2>TL;DR; Show Me The Code!<\/h2>\n<p>An example of how to use it.<\/p>\n<pre><code class=\"cpp\">#include &lt;iostream>  int main() { { EventLoop eventLoop;  eventLoop.enqueue([] { std::cout &lt;&lt; \"message from a different thread\\n\"; });  std::cout &lt;&lt; \"prints before or after the message above\\n\"; }  std::cout &lt;&lt; \"guaranteed to be printed the last\\n\"; } <\/code><\/pre>\n<p>And the implementation itself.<\/p>\n<pre><code class=\"cpp\">#include &lt;condition_variable> #include &lt;functional> #include &lt;future> #include &lt;thread> #include &lt;vector>  class EventLoop { public: using callable_t = std::function&lt;void()>;  EventLoop() = default; EventLoop(const EventLoop&amp;) = delete; EventLoop(EventLoop&amp;&amp;) noexcept = delete; ~EventLoop() noexcept { enqueue([this] { m_running = false; }); m_thread.join(); }  EventLoop&amp; operator= (const EventLoop&amp;) = delete; EventLoop&amp; operator= (EventLoop&amp;&amp;) noexcept = delete;  void enqueue(callable_t&amp;&amp; callable) noexcept { { std::lock_guard&lt;std::mutex> guard(m_mutex); m_writeBuffer.emplace_back(std::move(callable)); } m_condVar.notify_one(); }  private: std::vector&lt;callable_t> m_writeBuffer; std::mutex m_mutex; std::condition_variable m_condVar; bool m_running{ true }; std::thread m_thread{ &amp;EventLoop::threadFunc, this };  void threadFunc() noexcept { std::vector&lt;callable_t> readBuffer;  while (m_running) { { std::unique_lock&lt;std::mutex> lock(m_mutex); m_condVar.wait(lock, [this] { return !m_writeBuffer.empty(); }); std::swap(readBuffer, m_writeBuffer); }  for (callable_t&amp; func : readBuffer) { func(); }  readBuffer.clear(); } } }; <\/code><\/pre>\n<p><a class=\"anchor\" name=\"The%20Basic%20Implementation\" id=\"The Basic Implementation\"><\/a><\/p>\n<h2>The Basic Implementation<\/h2>\n<p>The implementation is quite compact. Feels idiomatic, doesn\u2019t it? But what is so special about it and what makes it efficient?<\/p>\n<p><a class=\"anchor\" name=\"The%20Power%20of%20stdfunction\" id=\"The Power of stdfunction\"><\/a><\/p>\n<h3>The Power of std::function<\/h3>\n<p><em>std::function&lt;R(Args\u2026)><\/em> is a really interesting thing. It uses two very important idioms that make it so useful for us \u2013 <em>Type Erasure<\/em> and <em>Small-Object Optimization<\/em>.<\/p>\n<p><em>Type Erasure<\/em> idiom allows us to store anything that we can apply the <em>call operator <\/em>to. I will refer to it as <em>callable_t<\/em>. It can be a C-like function, it can be a functor. It can also be a lambda, including a generic lambda.<\/p>\n<p>Since our <em>callable_t<\/em> can have internal data, such as the functor\u2019s members or the lambda\u2019s capture, <em>std::function<\/em> also has to store all this data inside. In order to avoid or at least minimize heap allocations, <em>std::function<\/em> can store <em>callable_t<\/em> in place if it\u2019s small enough. What is \u201csmall enough\u201d depends on the implementation. If it doesn\u2019t fit into the internal storage, then heap allocation happens as a fallback. This is what <em>Small-Object Optimization<\/em> essentially is.<\/p>\n<p>Knowing all of that, you can use a <em>std::vector<\/em> of <em>std::function<\/em> to keep both data and pointers to <em>vtables <\/em>in a single chunk of memory for most cases. And this is the first member of our class \u2013 <em>std::vector&lt;callable_t> m_writeBuffer<\/em>;<\/p>\n<p><a class=\"anchor\" name=\"The%20Power%20of%20stdcondition_variable\" id=\"The Power of stdcondition_variable\"><\/a><\/p>\n<h3>The Power of std::condition_variable<\/h3>\n<p>A condition variable is a synchronization primitive that makes one thread postpone its execution via <em>wait()<\/em> until another thread wakes it up via <em>notify_one()<\/em>. But what if the second thread has called <em>notify_one()<\/em> just before the first thread calls <em>wait()<\/em>? If you used a simpler synchronization primitive, such as <a href=\"https:\/\/docs.microsoft.com\/en-us\/windows\/win32\/sync\/event-objects\" rel=\"noopener noreferrer nofollow\"><u>Win32 Event Object<\/u><\/a>, the first thread would not see the notification at all. In this case, it would fall asleep ending you up having your program deadlocked. Fortunately, <em>std::condition_variable<\/em> is a bit smarter. It deliberately wants you to lock the mutex which guards some state. While you\u2019re holding the locked mutex <em>std::condition_variable<\/em> wants you to check the state if the first thread has to fall asleep or the condition has already been satisfied and the thread just needs to keep going. If it turns out that the thread has to be postponed something interesting happens next. Have you noticed that the wait<em>()<\/em> function accepts that locked mutex? This is because the <em>wait()<\/em> function asks your operating system to do the following:<\/p>\n<ol>\n<li>\n<p>Atomically unlock the mutex and postpone the execution.<\/p>\n<\/li>\n<li>\n<p>Atomically lock the mutex and resume the execution when <em>notify_one()<\/em> \/ <em>notify_all()<\/em> call occurs.<\/p>\n<\/li>\n<\/ol>\n<p>I\u2019m saying \u201catomically\u201d here, but maybe it\u2019s not what you\u2019re thinking about. It\u2019s a feature of the operating system thread scheduler, not the processors\u2019 hardware. You cannot emulate that behavior using atomic variables.<\/p>\n<p>Sometimes the OS may wake the waiting thread up spontaneously. It\u2019s called \u201cspurious wakeups\u201d. There are reasons for that, but I won&#8217;t cover them here. But you should be prepared for them. So if the thread has been woken up, you need to check your condition again. That\u2019s why I\u2019m passing the predicate to the <em>wait()<\/em> function here. That version of <em>wait()<\/em> tests the condition before falling asleep and immediately after. You can read more <a href=\"https:\/\/en.cppreference.com\/w\/cpp\/thread\/condition_variable\/wait\" rel=\"noopener noreferrer nofollow\"><u>about it here<\/u><\/a>.<\/p>\n<p>As you can see we have to have a protected state guarded by the mutex. Therefore, the second notifying thread should do the following:<\/p>\n<ol>\n<li>\n<p>Lock the mutex, change the shared state, and unlock the mutex.<\/p>\n<\/li>\n<li>\n<p>Notify the first thread.<\/p>\n<\/li>\n<\/ol>\n<p>This is essentially what happens in the <em>enqueue()<\/em> function. Some developers call <em>notify_one()<\/em> while holding the mutex. It\u2019s not wrong, but it makes the scheme inefficient. In order to avoid extra synchronizations, just make sure you call the <em>notify_one() <\/em><strong>after<\/strong> you release the mutex. You can read <a href=\"https:\/\/en.cppreference.com\/w\/cpp\/thread\/condition_variable\/notify_one\" rel=\"noopener noreferrer nofollow\"><u>the Notes section<\/u><\/a> to learn more about this problem.<\/p>\n<p><a class=\"anchor\" name=\"The%20Power%20of%20Double%20Buffering\" id=\"The Power of Double Buffering\"><\/a><\/p>\n<h3>The Power of Double Buffering<\/h3>\n<p>What really makes this implementation especially efficient is that we have two buffers here. You may have noticed that we are swapping <em>readBuffer<\/em> and <em>m_writeBuffer<\/em>. And we are doing this while the mutex is being locked. <em>std::swap<\/em> simply swaps the pointers inside those two vectors, which is an extremely fast operation. So we are leaving <em>m_writeBuffer<\/em> empty, ready to be filled again. Next to the <em>std::swap<\/em> the scope ends unlocking the mutex.<\/p>\n<p>Now we have a situation where the write buffer is getting filled while the read buffer is being processed. Now, these two processes can go simultaneously without any intersection! When the processing is over, we clear the read buffer. Clearing of <em>std::vector<\/em> does not cause the underlying storage deallocation. So when we quickly swap those two buffers again this underlying storage is going to be filled up again as the write buffer.<\/p>\n<p>When you\u2019re calling <em>enqueue()<\/em> and your <em>callable_t<\/em> is small enough you won\u2019t even touch the heap while constructing it. Inserting this <em>callable_t <\/em>into the vector that has already got some storage left after the processing step takes almost nothing! What about locking the mutex? It turns out that modern mutexes use atomic spin-locks as the first step and after several iterations, they ask the operating system to postpone the thread. So if you do something really quick between <em>lock()<\/em> and <em>unlock()<\/em> you won\u2019t even disturb the operating system. <em>notify_one()<\/em> is also designed to be fast in case you call it while the mutex is unlocked. So you shouldn\u2019t be bothered with it here. It makes <em>enqueue()<\/em> extremely fast on average. It also makes <em>wait()<\/em> fast as well, because we simply check if <em>m_writeBuffer <\/em>is not empty and swap it.<\/p>\n<p><a class=\"anchor\" name=\"A%20Couple%20of%20Remarks%20Regarding%20noexcept\" id=\"A Couple of Remarks Regarding noexcept\"><\/a><\/p>\n<h3>A Couple of Remarks Regarding &#171;noexcept&#187;<\/h3>\n<p>You may be wondering, why am I using <em>noexcept<\/em> keywords all over the place. And there is a really good explanation in <a href=\"https:\/\/isocpp.github.io\/CppCoreGuidelines\/CppCoreGuidelines#Rf-noexcept\" rel=\"noopener noreferrer nofollow\"><u>ISO C++ Core Guidelines<\/u><\/a> why you should at least consider using it. Let\u2019s take the <em>enqueue()<\/em> function. Even though we do have vector insertion here, which may throw <em>std::bad_alloc<\/em>, this scenario is actually disastrous. Your system either is lacking memory, which can cause a crash somewhere else, or you pushed too many tasks that your working thread cannot handle, which is in fact a poor application design. The lack of the system memory may also prevent throwing the exception about well\u2026 the lack of the system memory since <em>throw<\/em> uses heap allocation. <a href=\"https:\/\/isocpp.github.io\/CppCoreGuidelines\/CppCoreGuidelines#Rc-dtor-noexcept\" rel=\"noopener noreferrer nofollow\"><u>The same logic is applied to the destructor<\/u><\/a>. I\u2019m also<\/p>\n<\/div>\n<\/div>\n<\/div>\n<\/div>\n","protected":false},"author":1,"featured_media":0,"comment_status":"open","ping_status":"open","sticky":false,"template":"","format":"standard","meta":{"footnotes":""},"categories":[],"tags":[],"class_list":["post-387029","post","type-post","status-publish","format-standard","hentry"],"_links":{"self":[{"href":"https:\/\/savepearlharbor.com\/index.php?rest_route=\/wp\/v2\/posts\/387029","targetHints":{"allow":["GET"]}}],"collection":[{"href":"https:\/\/savepearlharbor.com\/index.php?rest_route=\/wp\/v2\/posts"}],"about":[{"href":"https:\/\/savepearlharbor.com\/index.php?rest_route=\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"https:\/\/savepearlharbor.com\/index.php?rest_route=\/wp\/v2\/users\/1"}],"replies":[{"embeddable":true,"href":"https:\/\/savepearlharbor.com\/index.php?rest_route=%2Fwp%2Fv2%2Fcomments&post=387029"}],"version-history":[{"count":0,"href":"https:\/\/savepearlharbor.com\/index.php?rest_route=\/wp\/v2\/posts\/387029\/revisions"}],"wp:attachment":[{"href":"https:\/\/savepearlharbor.com\/index.php?rest_route=%2Fwp%2Fv2%2Fmedia&parent=387029"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/savepearlharbor.com\/index.php?rest_route=%2Fwp%2Fv2%2Fcategories&post=387029"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/savepearlharbor.com\/index.php?rest_route=%2Fwp%2Fv2%2Ftags&post=387029"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}