20 AsyncTimeOut
class AsyncTimeOut is defined in AsyncTimeOut.h in the cxThread library.
This is the public interface:
class AsyncTimeOut
{
public:
using ExecuteTaskFn = std::function<void(bool expired, std::atomic<bool>& abort)>;
explicit AsyncTimeOut(boost::asio::io_context& context);
void Start(int timeoutMilliSecs, ExecuteTaskFn taskFn);
void Reset();
void Reset(int timeoutMilliSecs);
void Stop();
};
std::shared_ptr<AsyncTimeOut> MakeAsyncTimeOut(boost::asio::io_context& context);
Instances of AsyncTimeOut 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 either Reset method then the task is called with
expired equal to true.
Reset() causes the timer to restart "from zero" using the existing timeout period.
Reset(int timeoutMilliSecs) restarts the timer with a new timeout period.
Both of these calls typically cause the task to be invoked very soon afterwards (perhaps within about
10 microseconds) with expired equal to false, indicating that the timer was reset
rather than given an opportunity to time out.
Stop 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 Stop() synchronously ensures the io_context has stopped executing
the task.
AsyncTimeOut 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 because that is the default interval between system clock ticks.
See: this article about High-Resolution Timers
for kernel drivers in the Microsoft Docs.
Example usage:
// #include "Ceda/cxThread/AsyncTimeOut.h"
auto r = ceda::MakeAsyncTimeOut(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->Stop();AsyncTimeOut.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 AsyncTimeOut, 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 AsyncTimeOut still provides a Stop() method that makes the
same guarantee - i.e. Stop() synchronously ensures no thread is (or will be) executing
the task.
AsyncTimeOut 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 AsyncTimeOut are heap allocated and reference counted using std::shared_ptr.
AsyncTimeOut is a subclass of std::enable_shared_from_this<AsyncTimeOut> so
that shared_from_this() can be called within methods of AsyncTimeOut to obtain a
std::shared_ptr which points at itself.
async_wait() captures a std::shared_ptr which points at the AsyncTimeOut.
This means the lambda prevents the AsyncTimeOut from being deleted.
Only the AsyncTimeOut needs to be reference counted - this requirement is not pushed back into applications.
A lambda in turn passed to AsyncTimeOut::Start() doesn't need to hold a strong reference to the objects
which it accesses, because we have implemented a Stop() 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 protected by a std::mutex member of the AsyncTimeOut.
AsyncTimeOut.h
// AsyncTimeOut.h
//
// Author David Barrett-Lennard
// (C)opyright Cedanet Pty Ltd 2021
#pragma once
#ifndef Ceda_cxThread_AsyncTimeOut_H
#define Ceda_cxThread_AsyncTimeOut_H
#include "Ceda/cxUtils/CedaAssert.h"
#include <boost/asio.hpp>
#include <memory>
#include <mutex>
#include <atomic>
#include <functional>
namespace ceda
{
class AsyncTimeOut final : public std::enable_shared_from_this<AsyncTimeOut>
{
public:
using ExecuteTaskFn = std::function<void(bool expired, std::atomic<bool>& abort)>;
explicit AsyncTimeOut(boost::asio::io_context& context);
void Start(int timeoutMilliSecs, ExecuteTaskFn taskFn);
void Reset();
void Reset(int timeoutMilliSecs);
void AsyncReset(int timeoutMilliSecs = -1);
void Stop();
private:
void DoAsync();
std::atomic<bool> abort_ = true; // For aborting a task that is already executing
std::mutex mutex_;
/////// protected by mutex
int timeoutMilliSecs_ = 1;
ExecuteTaskFn taskFn_;
boost::asio::system_timer timer_;
};
inline std::shared_ptr<AsyncTimeOut> MakeAsyncTimeOut(boost::asio::io_context& context)
{
return std::make_shared<AsyncTimeOut>(context);
}
} // namespace ceda
#endif // include guardAsyncTimeOut.cpp
// AsyncTimeOut.cpp
//
// Author David Barrett-Lennard
// (C)opyright Cedanet Pty Ltd 2021
#include "AsyncTimeOut.h"
namespace ceda
{
AsyncTimeOut::AsyncTimeOut(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 AsyncTimeOut::Start(int timeoutMilliSecs, ExecuteTaskFn taskFn)
{
abort_ = false;
std::lock_guard<std::mutex> lock(mutex_);
timeoutMilliSecs_ = timeoutMilliSecs;
cxAssert(!taskFn_);
taskFn_ = std::move(taskFn);
DoAsync();
}
void AsyncTimeOut::AsyncReset(int timeoutMilliSecs)
{
auto self(shared_from_this());
boost::asio::post(timer_.get_executor(),
[this, self, timeoutMilliSecs]()
{
std::lock_guard<std::mutex> lock(mutex_);
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 AsyncTimeOut::Reset(int timeoutMilliSecs)
{
std::lock_guard<std::mutex> lock(mutex_);
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 AsyncTimeOut::Reset()
{
std::lock_guard<std::mutex> lock(mutex_);
// 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 AsyncTimeOut::Stop()
{
// The async job should check this atomic bool regularly to see whether it should abort
abort_ = true;
// Put this AsyncTimeOut synchronously into a state where the taskFn which was passed
// to Start() is not being called and won't be called again.
std::lock_guard<std::mutex> lock(mutex_);
cxAssert(taskFn_);
taskFn_ = nullptr;
timer_.cancel();
}
// Always called by a thread that has locked the mutex
void AsyncTimeOut::DoAsync()
{
cxAssert(taskFn_);
// 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)
{
std::lock_guard<std::mutex> lock(mutex_);
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
AsyncTimeOutTest.cpp
// AsyncTimeOutTest.cpp
//
// Author David Barrett-Lennard
// (C)opyright Cedanet Pty Ltd 2021
#include "Ceda/cxThread/AsyncTimeOut.h"
#include "Ceda/cxThread/IoContextPool.h"
#include "Ceda/cxUtils/Tracer.h"
#include "Ceda/cxUtils/HPTime.h"
#include <iostream>
/*
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 t = ceda::MakeAsyncTimeOut(ceda::GetDefaultIoContext());
int numCancelled = 0;
int numExpired = 0;
t->Start(0,
[&](bool expired, std::atomic<bool>& abort)
{
if (expired) ++numExpired; else ++numCancelled;
});
Sleep(100);
t->Stop();
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 t = ceda::MakeAsyncTimeOut(ceda::GetDefaultIoContext());
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->Stop();
Tracer() << "numCancelled = " << numCancelled << " numExpired = " << numExpired << '\n';
}
void UnitTestAsyncTimeOut()
{
Tracer() << "UnitTestAsyncTimeOut\n";
ceda::TraceIndenter indent(4);
// Demo
{
auto& io_context = ceda::GetDefaultIoContext();
auto r = ceda::MakeAsyncTimeOut(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->Stop();
}
ZeroTimeOutExample();
AbortExample();
auto t = ceda::MakeAsyncTimeOut(ceda::GetDefaultIoContext());
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->Stop();
Tracer() << "Stop() called\n";
}Typical output:
UnitTestAsyncTimeOut
Executing task
Executing task
Executing task
Executing task
ZeroTimeOutExample
numCancelled = 0 numExpired = 9535
AbortExample
numCancelled = 13177 numExpired = 0
Start with timeout of 25ms
Begin sleep 100ms
Time: 24ms task( expired=true ) called
Time: 30ms 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: 14ms 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: 15ms task( expired=false ) called
Time: 15ms task( expired=false ) called
Time: 15ms task( expired=false ) called
Time: 0ms task( expired=false ) called
Time: 15ms 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: 15ms task( expired=false ) called
Time: 14ms task( expired=false ) called
Time: 14ms 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: 14ms task( expired=false ) called
Time: 14ms task( expired=false ) called
Reset() with a new timeout of 100ms
Time: 0ms task( expired=false ) called
Begin sleep 250ms
Time: 110ms task( expired=true ) called
Time: 107ms task( expired=true ) called
End sleep
Stop() called