18 AsyncSignaledTask

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

This is the public interface:

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

std::shared_ptr<AsyncSignaledTask> MakeAsyncSignaledTask(boost::asio::io_context& context);

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

Calls to Signal cause the task passed to Start to be executed asynchronously by the context which must be run by a single thread. Note however that by design there is no requirement that there will be one execution of the task for each call to Signal (i.e. that the calls will be paired). Rather, the implementation only records the signaled state with a boolean, not a counter for the number of calls to Signal. If while a task is executing there are multiple calls to Signal then the task will only be made to execute one more time by those calls to Signal.

Stop stops further calls to the task. 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 task is no longer executing. The implementation waits on a ManualResetEvent until the task has been stopped.

Calling Stop from the task results in a dead-lock. It is otherwise safe to call Start, Signal and Stop from any thread any number of times in any order. Note that it is safe to call Start and Signal from the task.

AsyncSignaledTask.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 AsyncSignaledTask, 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, so this header is typically only included from cpp files inside ceda-core libraries.

Implementation

Instances of AsyncSignaledTask are heap allocated and reference counted using std::shared_ptr.

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

AsyncSignaledTask.h

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

#pragma once
#ifndef Ceda_cxThread_AsyncSignaledTask_H
#define Ceda_cxThread_AsyncSignaledTask_H

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

namespace ceda
{
class AsyncSignaledTask final :  public std::enable_shared_from_this<AsyncSignaledTask>
{
public:
    using ExecuteTaskFn = std::function<void(std::atomic<bool>& abort)>;
    explicit AsyncSignaledTask(boost::asio::io_context& context);
    void Start(ExecuteTaskFn taskFn);
    void Signal();
    void Stop();
private:
    std::atomic<bool> abort_ = true;        // For aborting a task that is already executing
    std::atomic<bool> signaled = false;
    boost::asio::io_context& context_;
    ExecuteTaskFn taskFn_;
};

inline std::shared_ptr<AsyncSignaledTask> MakeAsyncSignaledTask(boost::asio::io_context& context)
{
    return std::make_shared<AsyncSignaledTask>(context);
}
} // namespace ceda
#endif // include guard

AsyncSignaledTask.cpp

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

#include "AsyncSignaledTask.h"
#include "ManualResetEvent.h"

namespace ceda
{
AsyncSignaledTask::AsyncSignaledTask(boost::asio::io_context& context) : 
    context_(context)
{
}

void AsyncSignaledTask::Start(ExecuteTaskFn taskFn)
{
    abort_ = false;
    auto self(shared_from_this());
    boost::asio::post(context_,
        [this, self, taskFn]()
        {
            taskFn_ = taskFn;
        });
}

void AsyncSignaledTask::Signal()
{
    if (std::atomic_exchange(&signaled, true) == false)
    {
        // The signaled flag transitioned from false to true so execute the task (asynchronously)
        auto self(shared_from_this());
        boost::asio::post(context_,
            [this, self]()
            {
                if (taskFn_)
                {
                    // Before running the task reset the signal to false so the task can be signaled 
                    // to run again while the task is running
                    signaled = false;
                    taskFn_(abort_);
                }
            });
    }
}

void AsyncSignaledTask::Stop()
{
    abort_ = true;      // Make long running task abort

    if (ThreadsAreEnabled())
    {
        ManualResetEvent event;
        auto self(shared_from_this());
        boost::asio::post(context_,
            [this, self, &event]()
            {
                taskFn_ = nullptr;
                event.Signal();
            });
        // Wait until the task is not being called and won't be called again.
        event.Wait();
    }
    else
    {
        taskFn_ = nullptr;
    }
}

} // namespace ceda

Testing

AsyncSignaledTaskTest.cpp

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

#include "Ceda/cxThread/AsyncSignaledTask.h"
#include "Ceda/cxThread/IoContextPool.h"
#include "Ceda/cxUtils/Tracer.h"
#include "Ceda/cxUtils/HPTime.h"
#include <map>

void SelfSignalingAsyncSignaledTask(double timeForTestInSecs)
{
    Tracer() << "SelfSignalingAsyncSignaledTask\n";
    ceda::TraceIndenter indent(4);
    auto& io_context = ceda::GetDefaultIoContext();

    int count = 0;
    auto st = ceda::MakeAsyncSignaledTask(io_context);
    st->Start([&](std::atomic<bool>& abort)
        {
            ++count;
            st->Signal();
        });
    st->Signal();
    std::this_thread::sleep_for(std::chrono::milliseconds((int)(1000*timeForTestInSecs)));
    st->Stop();
    Tracer() << "count rate = " << count/(1000000*timeForTestInSecs) << " MHz in " << timeForTestInSecs << "s\n";
}

void CountTasksOnAsyncSignaledTask(double timeForTestInSecs)
{
    Tracer() << "CountTasksOnAsyncSignaledTask\n";
    ceda::TraceIndenter indent(4);

    auto& io_context = ceda::GetDefaultIoContext();
    
    std::map<int,int> counts;
    ceda::HPTimer timer;    
    while(timer.GetElapsedTimeInSeconds() < timeForTestInSecs)
    {
        int count = 0;
        auto st = ceda::MakeAsyncSignaledTask(io_context);
        st->Start([&](std::atomic<bool>& abort)
            {
                cxAlwaysAssert(!abort);
                ++count;
            });

        st->Signal();
        st->Signal();
        st->Signal();
        Sleep(1);

        st->Signal();
        Sleep(1);

        st->Signal();
        st->Signal();
        Sleep(1);

        st->Stop();
        counts[count]++;
    }

    for (auto& c : counts)
    {
        Tracer() << "count=" << c.first << " occurred " << c.second << " times\n";
    }
    Tracer() << "Time taken = " << timer.GetElapsedTimeInSeconds() << "s\n";
}

void UnitTestAsyncSignaledTask(double timeForTestInSecs)
{
    Tracer() << "UnitTestAsyncSignaledTask\n";
    ceda::TraceIndenter indent(4);

    timeForTestInSecs /= 2;
    SelfSignalingAsyncSignaledTask(timeForTestInSecs);
    CountTasksOnAsyncSignaledTask(timeForTestInSecs);
}

Example output, windows-x64 on Ryzen 9 machine:

UnitTestAsyncSignaledTask
    SelfSignalingAsyncSignaledTask
        count rate = 2.67453 MHz in 60s
    CountTasksOnAsyncSignaledTask
        count=3 occurred 1274 times
        count=4 occurred 2 times
        Time taken = 60.0342s