Concurrency

Table of Contents

• Classes
• Typedefs
• Functions
• Detailed Description
• Typedef Documentation
• NonFifoMpmcQueue
• NonFifoMpscQueue
• SpmcQueue
• SpscQueue
• StringStreamQueue
• Function Documentation
• Async() [1/4]
• Async() [2/4]
• Async() [3/4]
• Async() [4/4]
• AsyncBackground()
• Usage example
• Arguments
• AtomicMax()
• AtomicMin()
• AtomicUpdate()
• CriticalAsync() [1/2]
• CriticalAsync() [2/2]
• GetAll()
• SharedAsync() [1/4]
• SharedAsync() [2/4]
• SharedAsync() [3/4]
• SharedAsync() [4/4]
• SharedCriticalAsync() [1/2]
• SharedCriticalAsync() [2/2]
• WaitAllChecked()
• WaitAny()

Task construction and synchronization primitives for coroutines. More...

Collaboration diagram for Concurrency:

Classes
class	utils::TokenBucket
class	concurrent::AsyncEventChannel< Args >
class	concurrent::AsyncEventSource< Args >
	The read-only side of an event channel. Events are delivered to listeners in a strict FIFO order, i.e. only after the event was processed a new event may appear for processing, same listener is never called concurrently. More...
class	concurrent::BackgroundTaskStorageCore
class	concurrent::BackgroundTaskStorage
class	concurrent::ConflatedEventChannel
	A non-blocking version of 'AsyncEventChannel'. More...
class	concurrent::MpscQueue< T >
class	concurrent::MutexSet< Key, Hash, Equal >
	A dynamic set of mutexes. More...
class	concurrent::Variable< Data, Mutex >
class	dist_lock::DistLockStrategyBase
	Interface for distributed lock strategies. More...
class	dist_lock::DistLockedTask
	A task that tries to acquire a distributed lock and runs user callback once while the lock is held. More...
class	engine::ConditionVariable
	std::condition_variable replacement for asynchronous tasks More...
class	engine::Promise< T >
	std::promise replacement for asynchronous tasks that works in pair with engine::Future More...
class	engine::Future< T >
	std::future replacement for asynchronous tasks that works in pair with engine::Promise More...
class	engine::Mutex
	std::mutex replacement for asynchronous tasks More...
class	engine::CancellableSemaphore
	Class that allows up to max_simultaneous_locks concurrent accesses to the critical section. It honours task cancellation, unlike Semaphore. More...
class	engine::Semaphore
	Class that allows up to max_simultaneous_locks concurrent accesses to the critical section. It ignores task cancellation, unlike CancellableSemaphore. More...
class	engine::SharedMutex
	std::shared_mutex replacement for asynchronous tasks More...
class	engine::SingleConsumerEvent
	A multiple-producers, single-consumer event. More...
class	engine::SingleUseEvent
	A single-producer, single-consumer event. More...
class	engine::SingleWaitingTaskMutex
	Lighter version of Mutex with not more than 1 waiting task. More...
class	engine::TaskInheritedVariable< T >
	TaskInheritedVariable is a per-coroutine variable of arbitrary type. More...
class	engine::TaskLocalVariable< T >
	TaskLocalVariable is a per-coroutine variable of arbitrary type. More...
class	rcu::Variable< T, RcuTraits >
	Read-Copy-Update variable. More...
class	rcu::RcuMap< Key, Value, RcuMapTraits >
	Map-like structure allowing RCU keyset updates. More...
class	utils::PeriodicTask
	Task that periodically runs a user callback. Callback is started after the previous callback execution is finished every period + A - B, where: More...

Typedefs
template<typename T >
using	concurrent::NonFifoMpmcQueue = GenericQueue< T, impl::SimpleQueuePolicy< true, true > >
	Non FIFO multiple producers multiple consumers queue.
template<typename T >
using	concurrent::NonFifoMpscQueue = GenericQueue< T, impl::SimpleQueuePolicy< true, false > >
	Non FIFO multiple producers single consumer queue.
template<typename T >
using	concurrent::SpmcQueue = GenericQueue< T, impl::SimpleQueuePolicy< false, true > >
	Single producer multiple consumers queue.
template<typename T >
using	concurrent::SpscQueue = GenericQueue< T, impl::SimpleQueuePolicy< false, false > >
	Single producer single consumer queue.
using	concurrent::StringStreamQueue = GenericQueue< std::string, impl::ContainerQueuePolicy< false, false > >
	Single producer single consumer queue of std::string which is bounded bytes inside.

Functions
template<typename T , typename Func >
T	utils::AtomicUpdate (std::atomic< T > &atomic, Func updater)
	Atomically performs the operation of updater on atomic
template<typename T >
T	utils::AtomicMin (std::atomic< T > &atomic, T value)
	Concurrently safe sets atomic to a value if value is less.
template<typename T >
T	utils::AtomicMax (std::atomic< T > &atomic, T value)
	Concurrently safe sets atomic to a value if value is greater.
template<typename... Tasks>
auto	engine::GetAll (Tasks &... tasks)
	Waits for the successful completion of all of the specified tasks or the cancellation of the caller.
template<typename... Tasks>
void	engine::WaitAllChecked (Tasks &... tasks)
	Waits for the successful completion of all of the specified tasks or for the cancellation of the caller.
template<typename... Tasks>
std::optional< std::size_t >	engine::WaitAny (Tasks &... tasks)
	Waits for the completion of any of the specified tasks or the cancellation of the caller.
template<typename Function , typename... Args>
auto	utils::CriticalAsync (engine::TaskProcessor &task_processor, std::string name, Function &&f, Args &&... args)
template<typename Function , typename... Args>
auto	utils::SharedCriticalAsync (engine::TaskProcessor &task_processor, std::string name, Function &&f, Args &&... args)
template<typename Function , typename... Args>
auto	utils::Async (engine::TaskProcessor &task_processor, std::string name, Function &&f, Args &&... args)
template<typename Function , typename... Args>
auto	utils::SharedAsync (engine::TaskProcessor &task_processor, std::string name, Function &&f, Args &&... args)
template<typename Function , typename... Args>
auto	utils::Async (engine::TaskProcessor &task_processor, std::string name, engine::Deadline deadline, Function &&f, Args &&... args)
template<typename Function , typename... Args>
auto	utils::SharedAsync (engine::TaskProcessor &task_processor, std::string name, engine::Deadline deadline, Function &&f, Args &&... args)
template<typename Function , typename... Args>
auto	utils::CriticalAsync (std::string name, Function &&f, Args &&... args)
template<typename Function , typename... Args>
auto	utils::SharedCriticalAsync (std::string name, Function &&f, Args &&... args)
template<typename Function , typename... Args>
auto	utils::Async (std::string name, Function &&f, Args &&... args)
template<typename Function , typename... Args>
auto	utils::SharedAsync (std::string name, Function &&f, Args &&... args)
template<typename Function , typename... Args>
auto	utils::Async (std::string name, engine::Deadline deadline, Function &&f, Args &&... args)
template<typename Function , typename... Args>
auto	utils::SharedAsync (std::string name, engine::Deadline deadline, Function &&f, Args &&... args)
template<typename Function , typename... Args>
auto	utils::AsyncBackground (std::string name, engine::TaskProcessor &task_processor, Function &&f, Args &&... args)

Detailed Description

Task construction and synchronization primitives for coroutines.

Typedef Documentation

NonFifoMpmcQueue

template<typename T >

using concurrent::NonFifoMpmcQueue = typedef GenericQueue<T, impl::SimpleQueuePolicy<true, true> >

Non FIFO multiple producers multiple consumers queue.

Items from the same producer are always delivered in the production order. Items from different producers (or when using a MultiProducer token) are delivered in an unspecified order. In other words, FIFO order is maintained only within producers, but not between them. This may lead to increased peak latency of item processing.

In exchange for this, the queue has lower contention and increased throughput compared to a conventional lock-free queue.

See also

Synchronization Primitives

Definition at line 621 of file queue.hpp.

NonFifoMpscQueue

template<typename T >

using concurrent::NonFifoMpscQueue = typedef GenericQueue<T, impl::SimpleQueuePolicy<true, false> >

Non FIFO multiple producers single consumer queue.

See also

concurrent::NonFifoMpmcQueue for the description of what NonFifo means.

Synchronization Primitives

Definition at line 630 of file queue.hpp.

SpmcQueue

template<typename T >

using concurrent::SpmcQueue = typedef GenericQueue<T, impl::SimpleQueuePolicy<false, true> >

Single producer multiple consumers queue.

See also

Synchronization Primitives

Definition at line 638 of file queue.hpp.

SpscQueue

template<typename T >

using concurrent::SpscQueue = typedef GenericQueue<T, impl::SimpleQueuePolicy<false, false> >

Single producer single consumer queue.

See also

Synchronization Primitives

Definition at line 646 of file queue.hpp.

StringStreamQueue

using concurrent::StringStreamQueue = typedef GenericQueue<std::string, impl::ContainerQueuePolicy<false, false> >

Single producer single consumer queue of std::string which is bounded bytes inside.

See also

Synchronization Primitives

Definition at line 654 of file queue.hpp.

Function Documentation

Async() [1/4]

template<typename Function , typename... Args>

auto utils::Async	(	engine::TaskProcessor &	task_processor,
		std::string	name,
		engine::Deadline	deadline,
		Function &&	f,
		Args &&...	args
	)

Starts an asynchronous task with deadline, task execution may be cancelled before the function starts execution in case of TaskProcessor overload.

By default, arguments are copied or moved inside the resulting TaskWithResult, like std::thread does. To pass an argument by reference, wrap it in std::ref / std::cref or capture the arguments using a lambda.

Parameters

tasks_processor	Task processor to run on
name	Name of the task to show in logs
f	Function to execute asynchronously
args	Arguments to pass to the function

Returns

engine::TaskWithResult

Definition at line 175 of file async.hpp.

Async() [2/4]

template<typename Function , typename... Args>

auto utils::Async	(	engine::TaskProcessor &	task_processor,
		std::string	name,
		Function &&	f,
		Args &&...	args
	)

Starts an asynchronous task, task execution may be cancelled before the function starts execution in case of TaskProcessor overload.

Use utils::CriticalAsync if the function execution must start and you are absolutely sure that you need it.

By default, arguments are copied or moved inside the resulting TaskWithResult, like std::thread does. To pass an argument by reference, wrap it in std::ref / std::cref or capture the arguments using a lambda.

Parameters

tasks_processor	Task processor to run on
name	Name of the task to show in logs
f	Function to execute asynchronously
args	Arguments to pass to the function

Returns

engine::TaskWithResult

Examples

components/component_sample_test.cpp, and samples/tcp_full_duplex_service/tcp_full_duplex_service.cpp.

Definition at line 127 of file async.hpp.

Async() [3/4]

template<typename Function , typename... Args>

auto utils::Async	(	std::string	name,
		engine::Deadline	deadline,
		Function &&	f,
		Args &&...	args
	)

Starts an asynchronous task with deadline on current task processor, task execution may be cancelled before the function starts execution in case of engine::TaskProcessor overload.

By default, arguments are copied or moved inside the resulting TaskWithResult, like std::thread does. To pass an argument by reference, wrap it in std::ref / std::cref or capture the arguments using a lambda.

Parameters

name	Name of the task to show in logs
f	Function to execute asynchronously
args	Arguments to pass to the function

Returns

engine::TaskWithResult

Definition at line 319 of file async.hpp.

Async() [4/4]

template<typename Function , typename... Args>

auto utils::Async	(	std::string	name,
		Function &&	f,
		Args &&...	args
	)

Starts an asynchronous task on current task processor, task execution may be cancelled before the function starts execution in case of engine::TaskProcessor overload.

Use utils::CriticalAsync if the function execution must start and you are absolutely sure that you need it.

By default, arguments are copied or moved inside the resulting TaskWithResult, like std::thread does. To pass an argument by reference, wrap it in std::ref / std::cref or capture the arguments using a lambda.

Parameters

name	Name of the task to show in logs
f	Function to execute asynchronously
args	Arguments to pass to the function

Returns

engine::TaskWithResult

Definition at line 274 of file async.hpp.

AsyncBackground()

template<typename Function , typename... Args>

auto utils::AsyncBackground	(	std::string	name,
		engine::TaskProcessor &	task_processor,
		Function &&	f,
		Args &&...	args
	)

Starts an asynchronous task without propagating engine::TaskInheritedVariable. tracing::Span is still inherited. Task execution may be cancelled before the function starts execution in case of engine::TaskProcessor overload.

Typically used from a request handler to launch tasks that outlive the request and do not effect its completion.

Usage example

Suppose you have some component that runs asynchronous tasks:

class AsyncRequestProcessor final {
 public:
  AsyncRequestProcessor();
 
  void FooAsync(Request&& request);
 
  Response WaitAndGetAggregate();
 
 private:
  static Response Foo(Request&& request);
 
  engine::TaskProcessor& task_processor_;
  concurrent::Variable<std::vector<engine::TaskWithResult<Response>>> tasks_;
};

  auto handler = [&](Request&& request) {
    async_request_processor.FooAsync(std::move(request));
    return "Please wait, your request is being processed.";
  };

If the tasks logically belong to the component itself (not to the method caller), then they should be launched using utils::AsyncBackground instead of the regular utils::Async

void AsyncRequestProcessor::FooAsync(Request&& request) {
  auto tasks = tasks_.Lock();
  tasks->push_back(
      utils::AsyncBackground("foo", task_processor_, &Foo, std::move(request)));
}

Arguments

By default, arguments are copied or moved inside the resulting TaskWithResult, like std::thread does. To pass an argument by reference, wrap it in std::ref / std::cref or capture the arguments using a lambda.

Parameters

name	Name of the task to show in logs
tasks_processor	Task processor to run on
f	Function to execute asynchronously
args	Arguments to pass to the function

Returns

engine::TaskWithResult

Definition at line 380 of file async.hpp.

AtomicMax()

template<typename T >

T utils::AtomicMax	(	std::atomic< T > &	atomic,
		T	value
	)

Concurrently safe sets atomic to a value if value is greater.

Definition at line 45 of file atomic.hpp.

AtomicMin()

template<typename T >

T utils::AtomicMin	(	std::atomic< T > &	atomic,
		T	value
	)

Concurrently safe sets atomic to a value if value is less.

Definition at line 35 of file atomic.hpp.

AtomicUpdate()

template<typename T , typename Func >

T utils::AtomicUpdate	(	std::atomic< T > &	atomic,
		Func	updater
	)

Atomically performs the operation of updater on atomic

updater may be called multiple times per one call of AtomicUpdate, so it must be idempotent. To ensure that the function does not spin for a long time, updater must be fairly simple and fast.

Parameters

atomic	the variable to update
updater	a lambda that takes the old value and produces the new value

Returns

The updated value

Definition at line 22 of file atomic.hpp.

CriticalAsync() [1/2]

template<typename Function , typename... Args>

auto utils::CriticalAsync	(	engine::TaskProcessor &	task_processor,
		std::string	name,
		Function &&	f,
		Args &&...	args
	)

Starts an asynchronous task, execution of function is guaranteed to start regardless of engine::TaskProcessor load limits.

Prefer using utils::Async if not sure that you need this.

By default, arguments are copied or moved inside the resulting TaskWithResult, like std::thread does. To pass an argument by reference, wrap it in std::ref / std::cref or capture the arguments using a lambda.

Parameters

tasks_processor	Task processor to run on
name	Name for the tracing::Span to use with this task
f	Function to execute asynchronously
args	Arguments to pass to the function

Returns

engine::TaskWithResult

Definition at line 75 of file async.hpp.

CriticalAsync() [2/2]

template<typename Function , typename... Args>

auto utils::CriticalAsync	(	std::string	name,
		Function &&	f,
		Args &&...	args
	)

Starts an asynchronous task on current task processor, execution of function is guaranteed to start regardless of engine::TaskProcessor load limits.

Prefer using utils::Async if not sure that you need this.

By default, arguments are copied or moved inside the resulting TaskWithResult, like std::thread does. To pass an argument by reference, wrap it in std::ref / std::cref or capture the arguments using a lambda.

Parameters

name	Name for the tracing::Span to use with this task
f	Function to execute asynchronously
args	Arguments to pass to the function

Returns

engine::TaskWithResult

Definition at line 224 of file async.hpp.

GetAll()

template<typename... Tasks>

auto engine::GetAll

(

Tasks &...

tasks

)

Waits for the successful completion of all of the specified tasks or the cancellation of the caller.

Effectively performs for (auto& task : tasks) task.Get(); with a twist: task.Get() is called in tasks completion order rather than in provided order, thus exceptions are rethrown ASAP.

After successful return from this method the tasks are invalid, in case of an exception being thrown some of the tasks might be invalid.

Parameters

tasks

either a single container, or a pack of future-like elements.

Returns

std::vector<Result> or void, depending on the tasks result type (which must be the same for all tasks).

Exceptions

WaitInterruptedException	when current_task::IsCancelRequested() and no TaskCancellationBlockers are present.
std::exception	rethrows one of specified tasks exception, if any, in no particular order.

Note

Has overall computational complexity of O(N^2), where N is the number of tasks.

Prefer engine::WaitAllChecked for tasks with a result, unless you specifically need the results stored in a std::vector or when storing the results long-term.

Definition at line 84 of file get_all.hpp.

SharedAsync() [1/4]

template<typename Function , typename... Args>

auto utils::SharedAsync	(	engine::TaskProcessor &	task_processor,
		std::string	name,
		engine::Deadline	deadline,
		Function &&	f,
		Args &&...	args
	)

Starts an asynchronous task with deadline, task execution may be cancelled before the function starts execution in case of TaskProcessor overload.

By default, arguments are copied or moved inside the resulting SharedTaskWithResult, like std::thread does. To pass an argument by reference, wrap it in std::ref / std::cref or capture the arguments using a lambda.

Parameters

tasks_processor	Task processor to run on
name	Name of the task to show in logs
f	Function to execute asynchronously
args	Arguments to pass to the function

Returns

engine::SharedTaskWithResult

Definition at line 199 of file async.hpp.

SharedAsync() [2/4]

template<typename Function , typename... Args>

auto utils::SharedAsync	(	engine::TaskProcessor &	task_processor,
		std::string	name,
		Function &&	f,
		Args &&...	args
	)

Starts an asynchronous task, task execution may be cancelled before the function starts execution in case of TaskProcessor overload.

Use utils::SharedCriticalAsync if the function execution must start and you are absolutely sure that you need it.

By default, arguments are copied or moved inside the resulting SharedTaskWithResult, like std::thread does. To pass an argument by reference, wrap it in std::ref / std::cref or capture the arguments using a lambda.

Parameters

tasks_processor	Task processor to run on
name	Name of the task to show in logs
f	Function to execute asynchronously
args	Arguments to pass to the function

Returns

engine::SharedTaskWithResult

Definition at line 153 of file async.hpp.

SharedAsync() [3/4]

template<typename Function , typename... Args>

auto utils::SharedAsync	(	std::string	name,
		engine::Deadline	deadline,
		Function &&	f,
		Args &&...	args
	)

Starts an asynchronous task with deadline on current task processor, task execution may be cancelled before the function starts execution in case of engine::TaskProcessor overload.

By default, arguments are copied or moved inside the resulting SharedTaskWithResult, like std::thread does. To pass an argument by reference, wrap it in std::ref / std::cref or capture the arguments using a lambda.

Parameters

name	Name of the task to show in logs
f	Function to execute asynchronously
args	Arguments to pass to the function

Returns

engine::SharedTaskWithResult

Definition at line 342 of file async.hpp.

SharedAsync() [4/4]

template<typename Function , typename... Args>

auto utils::SharedAsync	(	std::string	name,
		Function &&	f,
		Args &&...	args
	)

Starts an asynchronous task on current task processor, task execution may be cancelled before the function starts execution in case of engine::TaskProcessor overload.

Use utils::SharedCriticalAsync if the function execution must start and you are absolutely sure that you need it.

By default, arguments are copied or moved inside the resulting SharedTaskWithResult, like std::thread does. To pass an argument by reference, wrap it in std::ref / std::cref or capture the arguments using a lambda.

Parameters

name	Name of the task to show in logs
f	Function to execute asynchronously
args	Arguments to pass to the function

Returns

engine::SharedTaskWithResult

Definition at line 298 of file async.hpp.

SharedCriticalAsync() [1/2]

template<typename Function , typename... Args>

auto utils::SharedCriticalAsync	(	engine::TaskProcessor &	task_processor,
		std::string	name,
		Function &&	f,
		Args &&...	args
	)

Starts an asynchronous task, execution of function is guaranteed to start regardless of engine::TaskProcessor load limits.

Prefer using utils::SharedAsync if not sure that you need this.

By default, arguments are copied or moved inside the resulting SharedTaskWithResult, like std::thread does. To pass an argument by reference, wrap it in std::ref / std::cref or capture the arguments using a lambda.

Parameters

tasks_processor	Task processor to run on
name	Name for the tracing::Span to use with this task
f	Function to execute asynchronously
args	Arguments to pass to the function

Returns

engine::SharedTaskWithResult

Definition at line 101 of file async.hpp.

SharedCriticalAsync() [2/2]

template<typename Function , typename... Args>

auto utils::SharedCriticalAsync	(	std::string	name,
		Function &&	f,
		Args &&...	args
	)

Starts an asynchronous task on current task processor, execution of function is guaranteed to start regardless of engine::TaskProcessor load limits.

Prefer using utils::SharedAsync if not sure that you need this.

By default, arguments are copied or moved inside the resulting SharedTaskWithResult, like std::thread does. To pass an argument by reference, wrap it in std::ref / std::cref or capture the arguments using a lambda.

Parameters

name	Name for the tracing::Span to use with this task
f	Function to execute asynchronously
args	Arguments to pass to the function

Returns

engine::SharedTaskWithResult

Definition at line 249 of file async.hpp.

WaitAllChecked()

template<typename... Tasks>

void engine::WaitAllChecked

(

Tasks &...

tasks

)

Waits for the successful completion of all of the specified tasks or for the cancellation of the caller.

Effectively performs for (auto& task : tasks) task.Wait(); with a twist: if any task completes with an exception, it gets rethrown ASAP.

Invalid tasks are skipped.

Tasks are not invalidated by WaitAllChecked; the result can be retrieved after the call.

Parameters

tasks

either a single container, or a pack of future-like elements.

Exceptions

WaitInterruptedException	when current_task::ShouldCancel() (for WaitAllChecked versions without a deadline)
std::exception	one of specified tasks exception, if any, in no particular order.

Note

Has overall computational complexity of O(N^2), where N is the number of tasks.

Keeping the tasks valid may have a small extra memory impact. Make sure to drop the tasks after reading the results.

Prefer engine::GetAll for tasks without a result, unless you specifically need to keep the tasks alive for some reason.

Definition at line 101 of file wait_all_checked.hpp.

WaitAny()

template<typename... Tasks>

std::optional< std::size_t > engine::WaitAny

(

Tasks &...

tasks

)

Waits for the completion of any of the specified tasks or the cancellation of the caller.

Could be used to get the ready HTTP requests ASAP:

std::size_t ProcessReadyRequests(
    std::vector<clients::http::ResponseFuture>& requests,
    engine::Deadline deadline) {
  std::size_t processed_requests = 0;
 
  while (auto indx = engine::WaitAnyUntil(deadline, requests)) {
    ++processed_requests;
 
    std::shared_ptr<clients::http::Response> response = requests[*indx].Get();
    EXPECT_TRUE(response->IsOk());
  }
 
  return processed_requests;
}

Works with different types of tasks and futures:

  auto task0 = engine::AsyncNoSpan([] { return 1; });
 
  auto task1 = utils::Async("long_task", [] {
    engine::InterruptibleSleepFor(20s);
    return std::string{"abc"};
  });
 
  auto task_idx_opt = engine::WaitAny(task0, task1);
  ASSERT_TRUE(task_idx_opt);
  EXPECT_EQ(*task_idx_opt, 0);

Parameters

tasks

either a single container, or a pack of future-like elements.

Returns

the index of the completed task, or std::nullopt if there are no completed tasks (possible if current task was cancelled).

Definition at line 56 of file wait_any.hpp.