This time around, I’m going to focus on the equivalent of boost::timer::deadline_timer<\/a>. The implementation shown here is custom made, portable C++11 and self contained.<\/p>\n If I had to roughly describe what deadline_timer<\/em> is for while at the same time hinting at the implementation, I would say “deadline_timer allows you group together a bunch of handlers to be called when a timer expires<\/em>“. So, how do we go about implementing timers? The trickiest thing to solve is how to correctly notify the worker thread that something happened that might change the current wait time.<\/p>\n For example:<\/p>\n Now, if you throw in a couple more threads adding or cancelling timers, things might get confusing. Personally, I think it’s a bad sign if to verify correctness of your multithreaded code you have to think too hard. Sure, performance is good, but first get it right. If performance is a problem, profile it, then think as hard as you need to get it right and fast.<\/p>\n My first pass at this was convoluted to say the least. There was a queue for the timers, then communication with the worker thread was done with another queue where I would pass commands such as “recalculate wait time”, or “execute this timer right now”, or “shutdown”. To be honest, in hindsight that first pass was quite bad. Maybe it just got that bad in small increments as I was focusing on the wrong problems.<\/p>\n What was sabotaging my efforts was that I was in the wrong mindset. I kept thinking off the thread-to-thread communication as “work to be done<\/em>“, and thus the obvious choice there was to have a command queue.<\/p>\n After a moment of clarity, I changed my mindset from “work to be done<\/em>” to “something has changed<\/em>“, and a cleaner solution formed:<\/p>\n The thread wakes up when something has changed (e,g: Timer added\/cancelled, or expired), and checks for work (e.g: execute handlers for expired timers). Extraneous notifications to wake up the thread have no side effects, since the thread wakes up, checks for work, recalculates the new wait time and goes back to waiting.<\/p>\n The only helper class needed is a semaphore<\/a>, which can be implemented in portable C++11 with a mutex<\/a>, a condition_variable<\/a> and a counter.<\/p>\n A few more useful methods could be added to Semaphore<\/em>, but were omitted to keep it short.<\/p>\n And the TimerQueue<\/em> implementation, heavily documented to further explain the implementation…<\/p>\n And a small example just adding and cancelling timers…<\/p>\n The output will be this (with varying *Elapsed” values)…<\/p>\n Main objective with this implementation was to have a small, self-contained and portable implementation. Therefore a few things can be improved if performance is a problem:<\/p>\n Although TimerQueue<\/em> is useful as-is, if you are familiar with Boost’s Asio deadline_timer<\/em>, you will spot a couple of differences:<\/p>\n A close match to Boost Asio deadline_timer<\/em> can be implemented on top of TimerQueue<\/em> by having a Timer<\/em> class that aggregates handlers, and puts a one-off timer on TimerQueue<\/em> whenever we set an expiry time. That one-off timer will run any handlers the Timer<\/em> instance owns. Even more, instead of executing the handlers in the TimerQueue<\/em> thread, Timer<\/em> can forward execution of the handlers to another object, thus giving the application control when\/where to execute the handlers.<\/p>\n While writing this, I took a quick look at how Boost’s Asio’s implements deadline timers (on Windows), to see if it was significantly different. Particularly, I was curious how deadline_timer<\/em> is tied to io_service<\/em>. If there is another thread involved for triggering timer related things, or if somehow everything was controlled by Windows Completion Ports.
\nI think the “deadline” in deadline_timer<\/em> is a bit misleading. With the definition of “deadline” being “the latest time or date by which something should be completed<\/em>“, the first thing that crosses my mind is that it is used to cancel some operation that did not complete by the specified time\/date. Such as to cancel a connect attempt. Certainly it can, but can be used for a lot more.
\nFor simplicity, I’ll be calling it “timer(s)”, and not “deadline timer(s)”<\/p>\nA bit of brainstorming<\/h4>\n
\nAt its core, you need a queue for the timers, and a worker thread that dequeues and executes timers according to the specified expiry times\/dates. Lets call it “Timer Queue”. Win32 API uses the same name for such a thing: Timer Queues<\/a>
\nWhat I explore in this post is a Timer Queue implementation. It allows one-shot timers. Higher level things such as grouping handlers in one timer and timers that can be reused (like boost Asio’s own implementation), can be built on top.<\/p>\n\n
\nTo add insult to injury, those two queues were locked independently.
\nSure enough, as I started to write this post to share the code, I’ve spotted another bug. Ever had that feeling of “Surely there must be a cleaner way to do this<\/em>” ?<\/p>\n![\"\"](\"https:\/\/www.crazygaze.com\/blog\/wp-content\/uploads\/2016\/03\/img_56f46caa712ca.png\")
<\/p>\nImplementation<\/h4>\n
\n#pragma once\n#include <mutex>\n#include <condition_variable>\n\nclass Semaphore {\npublic:\nSemaphore(unsigned int count = 0) : m_count(count) {}\n\nvoid notify() {\nstd::unique_lock<std::mutex> lock(m_mtx);\nm_count++;\nm_cv.notify_one();\n}\n\nvoid wait() {\nstd::unique_lock<std::mutex> lock(m_mtx);\nm_cv.wait(lock, [this]() { return m_count > 0; });\nm_count--;\n}\n\ntemplate <class Clock, class Duration>\nbool waitUntil(const std::chrono::time_point<Clock, Duration>& point) {\nstd::unique_lock<std::mutex> lock(m_mtx);\nif (!m_cv.wait_until(lock, point, [this]() { return m_count > 0; }))\nreturn false;\nm_count--;\nreturn true;\n}\n\nprivate:\nstd::mutex m_mtx;\nstd::condition_variable m_cv;\nunsigned int m_count;\n};\n<\/pre>\n\n
\n#pragma once\n#include "Semaphore.h"\n#include <thread>\n#include <queue>\n#include <chrono>\n#include <assert.h>\n\n\/\/ Timer Queue\n\/\/\n\/\/ Allows execution of handlers at a specified time in the future\n\/\/ Guarantees:\n\/\/\t- All handlers are executed ONCE, even if canceled (aborted parameter will\n\/\/be set to true)\n\/\/\t\t- If TimerQueue is destroyed, it will cancel all handlers.\n\/\/\t- Handlers are ALWAYS executed in the Timer Queue worker thread.\n\/\/\t- Handlers execution order is NOT guaranteed\n\/\/\nclass TimerQueue {\npublic:\nTimerQueue() {\nm_th = std::thread([this] { run(); });\n}\n\n~TimerQueue() {\ncancelAll();\n\/\/ Abusing the timer queue to trigger the shutdown.\nadd(0, [this](bool) { m_finish = true; });\nm_th.join();\n}\n\n\/\/! Adds a new timer\n\/\/ return\n\/\/\tReturns the ID of the new timer. You can use this ID to cancel the\n\/\/ timer\nuint64_t add(int64_t milliseconds, std::function<void(bool)> handler) {\nWorkItem item;\nitem.end = Clock::now() + std::chrono::milliseconds(milliseconds);\nitem.handler = std::move(handler);\n\nstd::unique_lock<std::mutex> lk(m_mtx);\nuint64_t id = ++m_idcounter;\nitem.id = id;\nm_items.push(std::move(item));\nlk.unlock();\n\n\/\/ Something changed, so wake up timer thread\nm_checkWork.notify();\nreturn id;\n}\n\n\/\/! Cancels the specified timer\n\/\/ return\n\/\/\t1 if the timer was cancelled.\n\/\/\t0 if you were too late to cancel (or the timer ID was never valid to\n\/\/ start with)\nsize_t cancel(uint64_t id) {\n\/\/ Instead of removing the item from the container (thus breaking the\n\/\/ heap integrity), we set the item as having no handler, and put\n\/\/ that handler on a new item at the top for immediate execution\n\/\/ The timer thread will then ignore the original item, since it has no\n\/\/ handler.\nstd::unique_lock<std::mutex> lk(m_mtx);\nfor (auto&& item : m_items.getContainer()) {\nif (item.id == id && item.handler) {\nWorkItem newItem;\n\/\/ Zero time, so it stays at the top for immediate execution\nnewItem.end = Clock::time_point();\nnewItem.id = 0; \/\/ Means it is a canceled item\n\/\/ Move the handler from item to newitem.\n\/\/ Also, we need to manually set the handler to nullptr, since\n\/\/ the standard does not guarantee moving an std::function will\n\/\/ empty it. Some STL implementation will empty it, others will\n\/\/ not.\nnewItem.handler = std::move(item.handler);\nitem.handler = nullptr;\nm_items.push(std::move(newItem));\n\nlk.unlock();\n\/\/ Something changed, so wake up timer thread\nm_checkWork.notify();\nreturn 1;\n}\n}\nreturn 0;\n}\n\n\/\/! Cancels all timers\n\/\/ return\n\/\/\tThe number of timers cancelled\nsize_t cancelAll() {\n\/\/ Setting all "end" to 0 (for immediate execution) is ok,\n\/\/ since it maintains the heap integrity\nstd::unique_lock<std::mutex> lk(m_mtx);\nfor (auto&& item : m_items.getContainer()) {\nif (item.id) {\nitem.end = Clock::time_point();\nitem.id = 0;\n}\n}\nauto ret = m_items.size();\n\nlk.unlock();\nm_checkWork.notify();\nreturn ret;\n}\n\nprivate:\nusing Clock = std::chrono::steady_clock;\nTimerQueue(const TimerQueue&) = delete;\nTimerQueue& operator=(const TimerQueue&) = delete;\n\nvoid run() {\nwhile (!m_finish) {\nauto end = calcWaitTime();\nif (end.first) {\n\/\/ Timers found, so wait until it expires (or something else\n\/\/ changes)\nm_checkWork.waitUntil(end.second);\n} else {\n\/\/ No timers exist, so wait forever until something changes\nm_checkWork.wait();\n}\n\n\/\/ Check and execute as much work as possible, such as, all expired\n\/\/ timers\ncheckWork();\n}\n\n\/\/ If we are shutting down, we should not have any items left,\n\/\/ since the shutdown cancels all items\nassert(m_items.size() == 0);\n}\n\nstd::pair<bool, Clock::time_point> calcWaitTime() {\nstd::lock_guard<std::mutex> lk(m_mtx);\nwhile (m_items.size()) {\nif (m_items.top().handler) {\n\/\/ Item present, so return the new wait time\nreturn std::make_pair(true, m_items.top().end);\n} else {\n\/\/ Discard empty handlers (they were cancelled)\nm_items.pop();\n}\n}\n\n\/\/ No items found, so return no wait time (causes the thread to wait\n\/\/ indefinitely)\nreturn std::make_pair(false, Clock::time_point());\n}\n\nvoid checkWork() {\nstd::unique_lock<std::mutex> lk(m_mtx);\nwhile (m_items.size() && m_items.top().end <= Clock::now()) {\nWorkItem item(std::move(m_items.top()));\nm_items.pop();\n\nlk.unlock();\nif (item.handler)\nitem.handler(item.id == 0);\nlk.lock();\n}\n}\n\nSemaphore m_checkWork;\nstd::thread m_th;\nbool m_finish = false;\nuint64_t m_idcounter = 0;\n\nstruct WorkItem {\nClock::time_point end;\nuint64_t id; \/\/ id==0 means it was cancelled\nstd::function<void(bool)> handler;\nbool operator>(const WorkItem& other) const {\nreturn end > other.end;\n}\n};\n\nstd::mutex m_mtx;\n\/\/ Inheriting from priority_queue, so we can access the internal container\nclass Queue : public std::priority_queue<WorkItem, std::vector<WorkItem>,\nstd::greater<WorkItem>> {\npublic:\nstd::vector<WorkItem>& getContainer() {\nreturn this->c;\n}\n} m_items;\n};\n<\/pre>\n\n#include "TimerQueue.h"\n#include <future>\n\nnamespace Timing {\n\nusing Clock = std::chrono::high_resolution_clock;\nstatic thread_local Clock::time_point ms_previous;\ndouble now() {\nstatic auto start = Clock::now();\nreturn std::chrono::duration<double, std::milli>(Clock::now() - start)\n.count();\n}\n\nvoid sleep(unsigned ms) {\nstd::this_thread::sleep_for(std::chrono::milliseconds(ms));\n}\n\n} \/\/ namespace Timing\n\nint main() {\nTimerQueue q;\n\n\/\/ Create timer with ID 1\nq.add(10000, [start = Timing::now()](bool aborted) mutable {\nprintf("ID 1: aborted=%s, Elapsed %4.2fmsn",\naborted ? "true " : "false", Timing::now() - start);\n});\n\n\/\/ Create Timer with ID 2\nq.add(10001, [start = Timing::now()](bool aborted) mutable {\nprintf("ID 2: aborted=%s, Elapsed %4.2fmsn",\naborted ? "true " : "false", Timing::now() - start);\n});\n\n\/\/ Should cancel timers with ID 1 and 2\nq.cancelAll();\n\n\/\/ Create timer with ID 3\nq.add(1000, [start = Timing::now()](bool aborted) mutable {\nprintf("ID 3: aborted=%s, Elapsed %4.2fmsn",\naborted ? "true " : "false", Timing::now() - start);\n});\n\n\/\/ Create timer with ID 4\nauto id = q.add(2000, [start = Timing::now()](bool aborted) mutable {\nprintf("ID 4: aborted=%s, Elapsed %4.2fmsn",\naborted ? "true " : "false", Timing::now() - start);\n});\n\n\/\/ Cancel timer with ID 4\nauto ret = q.cancel(id);\nassert(ret == 1);\n\n\/\/ Give just enough time to execute timer with ID 3 before destroying the\n\/\/ TimerQueue\nTiming::sleep(1500);\n\n\/\/ At this point, when destroying TimerQueue, the timer with ID 4 is still\n\/\/ pending and will be cancelled implicitly by the destructor\nreturn 0;\n}\n<\/pre>\n\nID 1: aborted=true , Elapsed 0.01ms\nID 2: aborted=true , Elapsed 0.03ms\nID 4: aborted=true , Elapsed 0.03ms\nID 3: aborted=false, Elapsed 1000.92ms\n<\/pre>\n
Things to improve<\/h5>\n
\n
\n
Conclusion<\/h4>\n
\n
\n
\n
\nFrom my short study of the beast’s guts, it seems it uses a somewhat similar method:<\/p>\n\n