22 AsyncPeriodicTimer
class AsyncPeriodicTimer is defined in AsyncPeriodicTimer.h in the cxThread library.
This is the public interface:
class AsyncPeriodicTimer
{
public:
using ExecuteTaskFn = std::function<void(bool expired, std::atomic<bool>& abort)>;
explicit AsyncPeriodicTimer(boost::asio::io_context& context);
void Start(int timeoutMilliSecs, ExecuteTaskFn taskFn);
void Reset(int timeoutMilliSecs = -1);
void SynchronousStop();
};
std::shared_ptr<AsyncPeriodicTimer> MakeAsyncPeriodicTimer(boost::asio::io_context& context);
Instances of AsyncPeriodicTimer are reference counted using std::shared_ptr.
Start(timeoutMilliSecs, taskFn) causes the given task to be called asynchronously
and repeatedly at the rate determined by timeoutMilliSecs.
If the timer is not reset with calls to the Reset method then the task is called with
expired equal to true.
A call to the Reset(timeoutMilliSecs) method does two things:
-
The task function will typically be called very soon afterwards (perhaps within about 10 microseconds)
with
expiredequal to false, indicating that the timer was reset rather than given an opportunity to time out. Note that multiple calls toResetat close to the same time may result in just a single call to the task function withexpiredequal to false. -
The timer restarts, either with a new timeout period if
timeoutMilliSecs >= 0otherwise with the existing timeout period.
SynchronousStop stops further calls to the task that was passed to Start. It causes the
atomic bool abort flag passed to the task to become set.
A long running task should poll the abort flag to see whether it should abort what it's doing.
This can for example help to allow applications to shutdown quickly.
Note well that SynchronousStop() synchronously ensures the io_context has stopped executing
the task.
The implementation waits on a ManualResetEvent until the timer has been stopped.
If Start is called on a timer which is already running then the time out and task to be executed
is updated and the timer is restarted as though there was a call to Reset.
Calling SynchronousStop() on a timer that isn't running is permitted.
It is safe to call Start and Reset from the task being executed, but attempting to call
SynchronousStop from the task produces a dead-lock.
AsyncPeriodicTimer has rather limited resolution.
If a nonzero time period is passed to Start or Reset, the
minimum timeout is typically about 15 milliseconds on Windows running on an x86 processor.
This is the default interval between system clock ticks.
See the article High-Resolution Timers
for kernel drivers in the Microsoft Docs.
Example usage:
// #include "Ceda/cxThread/AsyncPeriodicTimer.h"
auto r = ceda::MakeAsyncPeriodicTimer(io_context);
int timeoutMilliSecs = 50;
r->Start(timeoutMilliSecs, [](bool expired, std::atomic<bool>& abort)
{
std::cout << "Executing task" << std::endl;
});
std::this_thread::sleep_for(std::chrono::milliseconds(300));
r->SynchronousStop();AsyncPeriodicTimer.h includes boost headers, so we are including boost in a public header of cxThread. However most ceda based applications have no need to use AsyncPeriodicTimer, so it's a more restricted dependency on boost.
Putting it another way, the rest of ceda-core tends to use this header internally, for example
to help implement the LSS or a CSpace, so this header is typically only included from cpp files inside
ceda-core libraries.
Purpose
There are some classes in ceda-core which have a concept of being started and stopped and which perform some task at a regular rate (e.g. once every 1000 milliseconds). For example:
- The LSS LazyCleaner
- The LSS LazyFlusher
- The CSpace garbage collector
As of July 2021 these classes use their own thread. In some cases this is through a TimeOutTask.
We want to instead use a boost asio io_context and deadline_timer.
That means we have no counterpart to a Stop() method which calls join() on a
std::thread to wait until the thread exits, ensuring it is no longer executing the task.
Similarly we don't have a cxThread ITaskExecutor to wait on.
Nevertheless AsyncPeriodicTimer still provides a SynchronousStop() method that makes the
same guarantee - i.e. SynchronousStop() synchronously ensures no thread is (or will be) executing
the task.
AsyncPeriodicTimer is a final class.
It is passed the task to be executed repeatedly at regular intervals in the Start() method
using a std::function<bool expired, void(std::atomic<bool>& abort)>.
Implementation
boost::asio::deadline_timer has method async_wait() to start an asynchronous wait
on the dead-line timer.
Instances of AsyncPeriodicTimer are heap allocated and reference counted using std::shared_ptr.
AsyncPeriodicTimer is a subclass of std::enable_shared_from_this<AsyncPeriodicTimer> so
that shared_from_this() can be called within methods of AsyncPeriodicTimer to obtain a
std::shared_ptr which points at itself.
async_wait() captures a std::shared_ptr which points at the AsyncPeriodicTimer.
This means the lambda prevents the AsyncPeriodicTimer from being deleted.
Only the AsyncPeriodicTimer needs to be reference counted - this requirement is not pushed back into applications.
A lambda in turn passed to AsyncPeriodicTimer::Start() doesn't need to hold a strong reference to the objects
which it accesses, because we have implemented a SynchronousStop() function which guarantees the lambda is synchronously no
longer being called.
In order for an application to shut down quickly it needs to be able to abort async waits on boost asio
deadline_timer objects.
deadline_timer::cancel() can be called for this purpose.
The boost asio deadline_timer is not thread-safe.
We ensure all access is by the context, and we assume the context is only running a single thread.
AsyncPeriodicTimer.h
// AsyncPeriodicTimer.h
//
// Author David Barrett-Lennard
// (C)opyright Cedanet Pty Ltd 2021
#pragma once
#ifndef Ceda_cxThread_AsyncPeriodicTimer_H
#define Ceda_cxThread_AsyncPeriodicTimer_H
#include "Ceda/cxUtils/CedaAssert.h"
#include <boost/asio.hpp>
#include <memory>
#include <atomic>
#include <functional>
namespace ceda
{
class AsyncPeriodicTimer final : public std::enable_shared_from_this<AsyncPeriodicTimer>
{
public:
using ExecuteTaskFn = std::function<void(bool expired, std::atomic<bool>& abort)>;
explicit AsyncPeriodicTimer(boost::asio::io_context& context);
void Start(int timeoutMilliSecs, ExecuteTaskFn taskFn);
void Reset(int timeoutMilliSecs = -1);
void SynchronousStop();
private:
void DoAsync();
std::atomic<bool> abort_ = true; // For aborting a task that is already executing
/////////////// only accessed by the single-threaded context passed into the constructor
int timeoutMilliSecs_ = 1;
ExecuteTaskFn taskFn_;
boost::asio::system_timer timer_;
};
inline std::shared_ptr<AsyncPeriodicTimer> MakeAsyncPeriodicTimer(boost::asio::io_context& context)
{
return std::make_shared<AsyncPeriodicTimer>(context);
}
} // namespace ceda
#endif // include guardAsyncPeriodicTimer.cpp
// AsyncPeriodicTimer.cpp
//
// Author David Barrett-Lennard
// (C)opyright Cedanet Pty Ltd 2021
#include "AsyncPeriodicTimer.h"
#include "ManualResetEvent.h"
namespace ceda
{
AsyncPeriodicTimer::AsyncPeriodicTimer(boost::asio::io_context& context) :
timer_(context) // Construct a timer without setting an expiry time.
{
cxAssert(!taskFn_); // No task function is set, indicates not started
}
void AsyncPeriodicTimer::Start(int timeoutMilliSecs, ExecuteTaskFn taskFn)
{
abort_ = false;
auto self(shared_from_this());
boost::asio::post(timer_.get_executor(),
[this, self, timeoutMilliSecs, taskFn]()
{
timeoutMilliSecs_ = timeoutMilliSecs;
bool alreadyRunning = (bool) taskFn_;
taskFn_ = std::move(taskFn);
if (alreadyRunning)
{
timer_.expires_after(std::chrono::milliseconds(timeoutMilliSecs_));
}
else
{
DoAsync();
}
});
}
void AsyncPeriodicTimer::Reset(int timeoutMilliSecs)
{
auto self(shared_from_this());
boost::asio::post(timer_.get_executor(),
[this, self, timeoutMilliSecs]()
{
if (taskFn_)
{
if (timeoutMilliSecs >= 0) timeoutMilliSecs_ = timeoutMilliSecs;
// Changing the expiry time of a timer while there are pending asynchronous waits
// causes those wait operations to be cancelled.
timer_.expires_after(std::chrono::milliseconds(timeoutMilliSecs_));
}
});
}
void AsyncPeriodicTimer::SynchronousStop()
{
// The async job should check this atomic bool regularly to see whether it should abort
abort_ = true;
if (ThreadsAreEnabled())
{
ManualResetEvent event;
auto self(shared_from_this());
boost::asio::post(timer_.get_executor(),
[this, self, &event]()
{
taskFn_ = nullptr;
timer_.cancel();
event.Signal();
});
// Put this AsyncPeriodicTimer synchronously into a state where the taskFn which was passed
// to Start() is not being called and won't be called again.
event.Wait();
}
else
{
taskFn_ = nullptr;
timer_.cancel();
}
}
// Always called by a thread that has locked the mutex
void AsyncPeriodicTimer::DoAsync()
{
// Set an expiry time relative to now.
timer_.expires_after(std::chrono::milliseconds(timeoutMilliSecs_));
// asynchronous wait for timer to expire
auto self(shared_from_this());
timer_.async_wait(
[this, self](const boost::system::error_code& ec)
{
if (taskFn_)
{
// Reset() causes pending asynchronous waits to be cancelled.
// i.e. ec equals boost::asio::error::operation_aborted.
bool expired = (ec != boost::asio::error::operation_aborted);
taskFn_(expired, abort_);
if (!abort_) DoAsync(); // Run again
}
});
}
} // namespace cedaTesting
AsyncPeriodicTimerTest.cpp
// AsyncPeriodicTimerTest.cpp
//
// Author David Barrett-Lennard
// (C)opyright Cedanet Pty Ltd 2021
#include "Ceda/cxThread/AsyncPeriodicTimer.h"
#include "Ceda/cxThread/IoContextPool.h"
#include "Ceda/cxUtils/Tracer.h"
#include "Ceda/cxUtils/HPTime.h"
#include <iostream>
namespace AsyncPeriodicTimerTest
{
void SimpleDemo()
{
Tracer() << "SimpleDemo\n";
auto& io_context = ceda::GetDefaultIoContext();
auto r = ceda::MakeAsyncPeriodicTimer(io_context);
int timeoutMilliSecs = 50;
r->Start(timeoutMilliSecs, [](bool expired, std::atomic<bool>& abort)
{
std::cout << "Executing task" << std::endl;
});
std::this_thread::sleep_for(std::chrono::milliseconds(300));
r->SynchronousStop();
}
/*
If the timeout is 0 then the task can be called at a high rate
(e.g. 100000 calls per second), invariably with expired=true.
*/
void ZeroTimeOutExample()
{
Tracer() << "ZeroTimeOutExample\n";
ceda::TraceIndenter indent(4);
auto& io_context = ceda::GetDefaultIoContext();
auto t = ceda::MakeAsyncPeriodicTimer(io_context);
int numCancelled = 0;
int numExpired = 0;
t->Start(0,
[&](bool expired, std::atomic<bool>& abort)
{
if (expired) ++numExpired; else ++numCancelled;
});
Sleep(100);
t->SynchronousStop();
Tracer() << "numCancelled = " << numCancelled << " numExpired = " << numExpired << '\n';
}
/*
If the timeout is large and Reset() is called repeatedly then the task can be called at a high rate
(e.g. 100000 calls per second), invariably with expired=false.
*/
void AbortExample()
{
Tracer() << "AbortExample\n";
ceda::TraceIndenter indent(4);
auto& io_context = ceda::GetDefaultIoContext();
auto t = ceda::MakeAsyncPeriodicTimer(io_context);
int numCancelled = 0;
int numExpired = 0;
t->Start(1000,
[&](bool expired, std::atomic<bool>& abort)
{
if (expired) ++numExpired; else ++numCancelled;
});
ceda::HPTimer timer;
while(timer.GetElapsedTimeInSeconds() < 0.1)
{
t->Reset();
}
t->SynchronousStop();
Tracer() << "numCancelled = " << numCancelled << " numExpired = " << numExpired << '\n';
}
struct RepeatedTaskCounter
{
void Init(boost::asio::io_context& io_context)
{
art = ceda::MakeAsyncPeriodicTimer(io_context);
}
void Start()
{
count = 0;
int timeoutMilliSecs = 1;
art->Start(timeoutMilliSecs, [this](bool expired, std::atomic<bool>& abort)
{
++count;
});
}
void Stop()
{
art->SynchronousStop();
countJustAfterStop = count;
}
std::shared_ptr<ceda::AsyncPeriodicTimer> art;
int count = 0;
int countJustAfterStop = 0;
};
void StressTest(double timeForTestInSecs)
{
Tracer() << "StressTest\n";
int numThreads = 4;
std::vector<std::thread> threads(numThreads);
for (int i=0 ; i < numThreads ; ++i)
{
threads[i] = std::thread([=]()
{
ceda::TraceIndenter indent(4 + 4*i);
auto& io_context = ceda::GetDefaultIoContext();
int n = 1000;
std::vector<RepeatedTaskCounter> repeatedTasks(n);
for (auto& r : repeatedTasks)
r.Init(io_context);
ceda::HPTimer timer;
while(timer.GetElapsedTimeInSeconds() < timeForTestInSecs)
{
int total = 0;
for (int j=0 ; j < 10 ; ++j)
{
for (auto& r : repeatedTasks)
r.Start();
std::this_thread::sleep_for(std::chrono::milliseconds(100));
for (auto& r : repeatedTasks)
r.Stop();
for (auto& r : repeatedTasks)
total += r.count;
// Wait a bit longer and check that the atomic counts are no longer changing
std::this_thread::sleep_for(std::chrono::milliseconds(1));
for (auto& r : repeatedTasks)
{
cxAlwaysAssert(r.count == r.countJustAfterStop);
}
}
Tracer() << total << '\n';
}
});
}
for (auto& t : threads)
t.join();
}
void WithAndWithoutReset()
{
Tracer() << "WithAndWithoutReset\n";
ceda::TraceIndenter indent(4);
auto& io_context = ceda::GetDefaultIoContext();
auto t = ceda::MakeAsyncPeriodicTimer(io_context);
ceda::HPTimer timer;
Tracer() << "Start with timeout of 25ms\n";
t->Start(20,
[&](bool expired, std::atomic<bool>& abort)
{
ceda::TraceIndenter indent(8);
Tracer() << "Time: " << (int) (1000*timer.GetElapsedTimeInSeconds()) << "ms task( expired=" << (expired?"true":"false") << " ) called\n";
timer.Reset();
});
Tracer() << "Begin sleep 100ms\n";
Sleep(100);
Tracer() << "End sleep\n";
Tracer() << "Calling Reset() repeatedly, preventing the timer from expiring\n";
for (int i=0 ; i < 5 ; ++i)
{
timer.Reset();
t->Reset();
t->Reset();
t->Reset();
Sleep(10);
t->Reset();
Sleep(10);
t->Reset();
Sleep(10);
t->Reset();
Sleep(10);
t->Reset();
Sleep(10);
t->Reset();
t->Reset();
Sleep(10);
}
Tracer() << "Reset() with a new timeout of 100ms\n";
timer.Reset(); t->Reset(100);
Sleep(10);
Tracer() << "Begin sleep 250ms\n";
Sleep(250);
Tracer() << "End sleep\n";
t->SynchronousStop();
Tracer() << "SynchronousStop() called\n";
}
} // namespace AsyncPeriodicTimerTest
void UnitTestAsyncPeriodicTimer(double timeForTestInSecs)
{
Tracer() << "UnitTestAsyncPeriodicTimer\n";
ceda::TraceIndenter indent(4);
AsyncPeriodicTimerTest::SimpleDemo();
AsyncPeriodicTimerTest::ZeroTimeOutExample();
AsyncPeriodicTimerTest::AbortExample();
AsyncPeriodicTimerTest::WithAndWithoutReset();
AsyncPeriodicTimerTest::StressTest(timeForTestInSecs);
}Typical output:
UnitTestAsyncPeriodicTimer
SimpleDemo
Executing task
Executing task
Executing task
Executing task
Executing task
ZeroTimeOutExample
numCancelled = 0 numExpired = 9157
AbortExample
numCancelled = 5969 numExpired = 0
WithAndWithoutReset
Start with timeout of 25ms
Begin sleep 100ms
Time: 26ms task( expired=true ) called
Time: 31ms task( expired=true ) called
Time: 31ms task( expired=true ) called
End sleep
Calling Reset() repeatedly, preventing the timer from expiring
Time: 0ms task( expired=false ) called
Time: 15ms task( expired=false ) called
Time: 14ms task( expired=false ) called
Time: 15ms task( expired=false ) called
Time: 15ms task( expired=false ) called
Time: 15ms task( expired=false ) called
Time: 0ms task( expired=false ) called
Time: 14ms task( expired=false ) called
Time: 15ms task( expired=false ) called
Time: 15ms task( expired=false ) called
Time: 15ms task( expired=false ) called
Time: 14ms task( expired=false ) called
Time: 0ms task( expired=false ) called
Time: 15ms task( expired=false ) called
Time: 14ms task( expired=false ) called
Time: 14ms task( expired=false ) called
Time: 14ms task( expired=false ) called
Time: 14ms task( expired=false ) called
Time: 0ms task( expired=false ) called
Time: 15ms task( expired=false ) called
Time: 14ms task( expired=false ) called
Time: 15ms task( expired=false ) called
Time: 14ms task( expired=false ) called
Time: 14ms task( expired=false ) called
Time: 0ms task( expired=false ) called
Time: 15ms task( expired=false ) called
Time: 15ms task( expired=false ) called
Time: 14ms task( expired=false ) called
Time: 15ms task( expired=false ) called
Time: 15ms task( expired=false ) called
Reset() with a new timeout of 100ms
Time: 0ms task( expired=false ) called
Begin sleep 250ms
Time: 107ms task( expired=true ) called
Time: 110ms task( expired=true ) called
End sleep
SynchronousStop() called
StressTest
73491
73838
75244
75927