21 AsyncRepeatedTask

class AsyncRepeatedTask is defined in AsyncRepeatedTask.h in the cxThread library.

This is the public interface:

class AsyncRepeatedTask
{
public:
    using ExecuteTaskFn = std::function<void(std::atomic<bool>& abort)>;
    explicit AsyncRepeatedTask(boost::asio::io_context& context);
    void SetDelay(int delayMilliSecs);
    void Start(int delayMilliSecs, ExecuteTaskFn taskFn);
    void Stop();
};

std::shared_ptr<AsyncRepeatedTask> MakeAsyncRepeatedTask(boost::asio::io_context& context);

Instances of AsyncRepeatedTask are reference counted using std::shared_ptr.

Start(delayMilliSecs, taskFn) causes the given task to be called asynchronously at the rate determined by delayMilliSecs.

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.

Note well that Stop() synchronously ensures the io_context has stopped executing the task.

AsyncRepeatedTask has rather limited resolution. If a nonzero time period is passed to Start(), it seems that the minimum timeout is actually about 15 milliseconds.

For Windows running on an x86 processor, the default interval between system clock ticks is typically about 15 milliseconds, and the minimum interval between system clock ticks is about 1 millisecond. See: this article about High-Resolution Timers for kernel drivers in the Microsoft Docs.

Example usage:

// #include "Ceda/cxThread/AsyncRepeatedTask.h"

auto r = ceda::MakeAsyncRepeatedTask(io_context);
int delayMilliSecs = 50;
r->Start(delayMilliSecs, [](std::atomic<bool>& abort)
    {
        std::cout << "Executing task" << std::endl;
    });
std::this_thread::sleep_for(std::chrono::milliseconds(300));
r->Stop();

AsyncRepeatedTask.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 AsyncRepeatedTask, 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:

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 AsyncRepeatedTask still provides a Stop() method that makes the same guarantee - i.e. Stop() synchronously ensures no thread is (or will be) executing the task.

AsyncRepeatedTask is a final class. It is passed the task to be executed repeatedly at regular intervals in the Start() method using a std::function<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 AsyncRepeatedTask are heap allocated and reference counted using std::shared_ptr.

AsyncRepeatedTask is a subclass of std::enable_shared_from_this<AsyncRepeatedTask> so that shared_from_this() can be called within methods of AsyncRepeatedTask to obtain a std::shared_ptr which points at itself.

The lamda passed to async_wait() captures a std::shared_ptr which points at the AsyncRepeatedTask. This means the lambda prevents the AsyncRepeatedTask from being deleted.

Only the AsyncRepeatedTask needs to be reference counted - this requirement is not pushed back into applications. A lambda in turn passed to AsyncRepeatedTask::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 AsyncRepeatedTask.

AsyncRepeatedTask.h

// AsyncRepeatedTask.h
//
// Author David Barrett-Lennard
// (C)opyright Cedanet Pty Ltd 2021

#pragma once
#ifndef Ceda_cxThread_AsyncRepeatedTask_H
#define Ceda_cxThread_AsyncRepeatedTask_H

#include "Ceda/cxUtils/CedaAssert.h"
#include <boost/asio.hpp>
#include <memory>
#include <mutex>
#include <atomic>
#include <functional>

namespace ceda
{
class AsyncRepeatedTask final :  public std::enable_shared_from_this<AsyncRepeatedTask>
{
public:
    using ExecuteTaskFn = std::function<void(std::atomic<bool>& abort)>;
    explicit AsyncRepeatedTask(boost::asio::io_context& context);
    void SetDelay(int delayMilliSecs);
    void Start(int delayMilliSecs, ExecuteTaskFn taskFn);
    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 delayMilliSecs_ = 1;
    ExecuteTaskFn taskFn_;
    boost::asio::system_timer timer_;
};

inline std::shared_ptr<AsyncRepeatedTask> MakeAsyncRepeatedTask(boost::asio::io_context& context)
{
    return std::make_shared<AsyncRepeatedTask>(context);
}

} // namespace ceda

#endif // include guard

AsyncRepeatedTask.cpp

// AsyncRepeatedTask.cpp
//
// Author David Barrett-Lennard
// (C)opyright Cedanet Pty Ltd 2021

#include "AsyncRepeatedTask.h"

namespace ceda
{

AsyncRepeatedTask::AsyncRepeatedTask(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 AsyncRepeatedTask::SetDelay(int delayMilliSecs)
{
    std::lock_guard<std::mutex> lock(mutex_);
    delayMilliSecs_ = delayMilliSecs;
}

void AsyncRepeatedTask::Start(int delayMilliSecs, ExecuteTaskFn taskFn)
{
    abort_ = false;

    std::lock_guard<std::mutex> lock(mutex_);
    delayMilliSecs_ = delayMilliSecs;
    cxAssert(!taskFn_);
    taskFn_ = std::move(taskFn);
    DoAsync();
}

void AsyncRepeatedTask::Stop()
{
    // The async job should check this atomic bool regularly to see whether it should abort
    abort_ = true;
        
    // Put this AsyncRepeatedTask 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 AsyncRepeatedTask::DoAsync()
{
    cxAssert(taskFn_);
        
    // Set an expiry time relative to now.
    timer_.expires_after(std::chrono::milliseconds(delayMilliSecs_));
        
    // asynchronous wait for timer to expire
    auto self(shared_from_this());
    timer_.async_wait(
        [this, self](const boost::system::error_code&)
        {
            std::lock_guard<std::mutex> lock(mutex_);
            if (taskFn_)
            {
                taskFn_(abort_);
                if (!abort_) DoAsync();  // Run again
            }
        });
}

} // namespace ceda

Testing

UnitTestAsyncRepeatedTask() can be run for hours to check for stability of the code.

The test creates a std::thread which calls run() on a single io_context which is used for this test.

A executor_work_guard instance is used to ensure run() on the io_context doesn't exit prematurely - i.e. until the end of the test, just before we call join() on the std::thread.

RepeatedTaskCounter counts the number of times the task has been called.

The test allocates 10000 instances of AsyncRepeatedTask with calls to MakeAsyncRepeatedTask(). In each iteration of the test the main thread does the following:

Start each AsyncRepeatedTask with a task that increments a counter
Sleep for 100 milliseconds
Stop all the AsyncRepeatedTask objects
Check that the count doesn't continue to change

If a nonzero time period is passed to Start(), it seems that the minimum timeout is actually about 16 milliseconds. This is presumably the OS thread scheduling timeslice. Unfortunately this means it's probably impossible to write a test that has a high chance of detecting race conditions

AsyncRepeatedTaskTest.cpp

// AsyncRepeatedTaskTest.cpp

#include "Ceda/cxThread/AsyncRepeatedTask.h"
#include "Ceda/cxThread/IoContextWithThread.h"
#include "Ceda/cxUtils/Tracer.h"
#include "Ceda/cxUtils/TestTimer.h"
#include <thread>

struct RepeatedTaskCounter
{
    void Init(boost::asio::io_context& io_context)
    {
        art = ceda::MakeAsyncRepeatedTask(io_context);
    }

    void Start()
    {
        count = 0;
        int delayMilliSecs = 1;
        art->Start(delayMilliSecs, [this](std::atomic<bool>& abort)
            {
                ++count;
            });
    }

    void Stop()
    {
        art->Stop();
        countJustAfterStop = count;
    }

    std::shared_ptr<ceda::AsyncRepeatedTask> art;
    int count = 0;
    int countJustAfterStop = 0;
};

void UnitTestAsyncRepeatedTask(double timeForTestInSecs)
{
    Tracer() << "Unit test AsyncRepeatedTask\n";
    ceda::TraceIndenter indent(4);
    
    ceda::IoContextWithThread s;

    //////////// Demonstrate usage
    {
        auto r = ceda::MakeAsyncRepeatedTask(s.io_context);
        int delayMilliSecs = 50;
        r->Start(delayMilliSecs, [](std::atomic<bool>& abort)
            {
                Tracer() << "Executing task\n";
            });
        std::this_thread::sleep_for(std::chrono::milliseconds(300));
        r->Stop();
    }

    //////////// Stress test
    int n = 10000;
    std::vector<RepeatedTaskCounter> repeatedTasks(n);
    for (auto& r : repeatedTasks)
        r.Init(s.io_context);

    ceda::HPTimer timer;
    while(timer.GetElapsedTimeInSeconds() < timeForTestInSecs)
    {
        for (auto& r : repeatedTasks)
            r.Start();
        
        std::this_thread::sleep_for(std::chrono::milliseconds(100));
        
        for (auto& r : repeatedTasks)
            r.Stop();

        int total = 0;
        for (auto& r : repeatedTasks)
            total += r.count;
        Tracer() << total << '\n';

        // 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);
        }
    }
}

Typical output:

Unit test AsyncRepeatedTask
Executing task
Executing task
Executing task
Executing task
    20764
    21438
    21885
    22572
    22274
    20943
    23554

Previous20 AsyncTimeOut Next22 AsyncPeriodicTimer