echomap::Worker Class Reference

A Worker provides an encapsulated thread-safe despatch model for submitting work and reviewing results. More...

#include <Worker.hpp>

Public Types
using	ResultCallback = std::function<void()>
	The type of callback to indicate new results.

Public Member Functions
	Worker (ResultCallback result_callback={})
	Create a new Worker with an optional callback.
void	submit (std::unique_ptr< ITask > &&task)
	Submit some work to the scheduler for execution on the computation thread.
bool	is_result_available () const
	Checks the state of the result queue.
std::optional< WorkerResult >	try_get_result ()
	Attempt to retrieve the latest WorkerResult object from the computation thread.
void	clear ()
	Clears any scheduled jobs or pending results.

Private Member Functions
void	execute (const std::stop_token &stop_token) noexcept
	Executor running on the computation thread to receive work from the task queue and synchronously execute.

Private Attributes
ThreadSafeQueue< std::unique_ptr< ITask > >	task_queue
ThreadSafeQueue< WorkerResult >	result_queue
ResultCallback	result_callback
	Callable to inform clients of new results.
std::jthread	worker_thread
	RAII computation thread to handle ITask work pieces.

Detailed Description

A Worker provides an encapsulated thread-safe despatch model for submitting work and reviewing results.

Clients may make use of a Worker to carry out thread-safe computation:

1. Clients subscribe to the relevant ResultChannel via the WorkerResultDespatcher.
2. Optionally, a single client is nominated as the consumer of the relevant ResultChannel messages.
3. Client constructs and populates an ITask object with a description of a computation task.
4. Client submits the ITask to the scheduler with Worker::submit.
5. Work is undertaken on a dedicated computation thread maintained by the Worker.
6. Once complete, Worker passes the result onto a ThreadSafeQueue.
7. The owner of Worker publishes the result onto a ResultChannel with WorkerResultDespatcher.
8. All subscribed clients are notified of the WorkerResult and provided with an observing reference.
9. If applicable, the nominated consumer client receives ownership of the WorkerResult.

Clients may receive results of work by periodically polling the Worker (such as checking Worker::try_get_result in a game loop) or through the provided asynchronous callback invoked when a new WorkerResult is available. All Worker operations are atomic, so a single Worker instance may have clients on multiple threads.

Definition at line 44 of file Worker.hpp.

Member Typedef Documentation

ResultCallback

using echomap::Worker::ResultCallback = std::function<void()>

The type of callback to indicate new results.

Definition at line 47 of file Worker.hpp.

Constructor & Destructor Documentation

Worker()

echomap::Worker::Worker ( ResultCallback result_callback = {} )

explicit

Create a new Worker with an optional callback.

Parameters

result_callback

Client callback invoked to indicate new WorkerResult objects available for consumption.

Definition at line 17 of file Worker.cpp.

  :
    result_callback(std::move(result_callback)),
    worker_thread{[this](const std::stop_token& stop_token) {
        execute(stop_token);
    }}
{
}

Member Function Documentation

clear()

void echomap::Worker::clear

(

)

Clears any scheduled jobs or pending results.

Definition at line 45 of file Worker.cpp.

{
    task_queue.clear();
    result_queue.clear();
}

execute()

void echomap::Worker::execute ( const std::stop_token & stop_token )

privatenoexcept

Executor running on the computation thread to receive work from the task queue and synchronously execute.

Parameters

stop_token

Cancellation token for the thread, provided by the thread interface.

Definition at line 51 of file Worker.cpp.

{
    while (!stop_token.stop_requested())
        // ThreadSafeQueue::wait_consume will block the computation thread until some work is available.
        if (auto job = task_queue.wait_consume(stop_token); job.has_value()) {
            // Likewise, ITask::execute runs the work synchronously on our computation thread.
            auto& task = *job;
            LOG_F_DEBUG("Executing {}.", task->get_name());
 
            try {
                auto result = task->execute(stop_token);
                LOG_F_DEBUG("Finished {}.", task->get_name());
                result_queue.produce(std::move(result));
            } catch (const std::exception& exception) {
                LOG_F_ERROR("{} failed with message: {}", task->get_name(), exception.what());
                result_queue.produce(ErrorResult(exception.what(), std::source_location::current(), std::move(task)));
            } catch (...) {
                LOG_F_ERROR("{} failed with a system error. This is bug.", task->get_name());
                result_queue.produce(ErrorResult("System error", std::source_location::current(), std::move(task)));
            }
 
            if (result_callback)
                result_callback();
        }
}

is_result_available()

bool echomap::Worker::is_result_available ( ) const

nodiscard

Checks the state of the result queue.

Returns

Is there a new result available from the Worker?

Definition at line 35 of file Worker.cpp.

{
    return !result_queue.empty();
}

submit()

void echomap::Worker::submit

(

std::unique_ptr< ITask > &&

task

)

Submit some work to the scheduler for execution on the computation thread.

Parameters

task	A description of the task, detained within an owning container transferred to the Worker.

Definition at line 27 of file Worker.cpp.

{
    LOG_F_DEBUG("Scheduling {} {}: {}.", task->get_class_name(), task->get_id(), task->get_name());
    task_queue.produce(std::move(task));
}

try_get_result()

std::optional< WorkerResult > echomap::Worker::try_get_result

(

)

Attempt to retrieve the latest WorkerResult object from the computation thread.

This function does not block.

Returns

The WorkerResult posted by the latest job, or the empty optional if no WorkerResult was available.

Definition at line 40 of file Worker.cpp.

{
    return result_queue.try_consume();
}

Member Data Documentation

result_callback

ResultCallback echomap::Worker::result_callback

private

Callable to inform clients of new results.

Definition at line 95 of file Worker.hpp.

result_queue

ThreadSafeQueue<WorkerResult> echomap::Worker::result_queue

private

Definition at line 93 of file Worker.hpp.

task_queue

ThreadSafeQueue<std::unique_ptr<ITask> > echomap::Worker::task_queue

private

Definition at line 92 of file Worker.hpp.

worker_thread

std::jthread echomap::Worker::worker_thread

private

RAII computation thread to handle ITask work pieces.

Definition at line 96 of file Worker.hpp.

---

The documentation for this class was generated from the following files:

• src/async/Worker.hpp
• src/async/Worker.cpp