Asynqro

Asynqro is a small library with purpose to make C++ programming easier by giving developers rich monadic Future API (mostly inspired by Future API in Scala language). This library is another implementation of ideas in https://github.com/opensoft/proofseed (now moved to asynqro usage, for historic purposes check tags before 02/25/19), but has much cleaner API, refined task scheduling logic and is not tied to any framework.

Dependencies

• C++17: Should work with Clang >=6, GCC >=7 and MSVC >=15. Tested with Clang9 (travis), GCC9 (travis) and MSVC15 (VS2017, appveyor). Also tested for compatibility with clang 6.0 and gcc 7.4 on Ubuntu Bionic image ((travis)). Conan packages exist for GCC 7/8/9@Linux, Clang 6/7/8@Linux, Apple-Clang 10/11@MacOSX, MSVC15@Windows.
• CMake >= 3.12.0
• GoogleTest. Will be automatically downloaded during cmake phase.
• lcov >= 1.14. Used for code coverage calculation, not needed for regular build.
• Optional Qt5 >= 5.10. It is not required though and by default asynqro is built without Qt support. There is no Qt dependency in library itself, but enabling it brings support for Qt containers, adds Future::fromQtSignal() and Future::fromQtFuture(). Also Future::wait() becomes guithread-aware.

Asynqro has two main parts:

• Future/Promise
• Tasks scheduling

Future/Promise

There are already a lot of implementations of Future mechanism in C++:

• std::future - no API at all, except very basic operations like retrieve value and wait for it. There is a Concurrency TS with .then, but it is still not in the standard.
• boost::future - almost the same as std::future.
• QFuture - Mostly unusable outside of QtConcurrent without Qt private headers because there is no way to fill it from user code.
• Folly - Folly futures are also inspired by Scala ones (but different ones, from Twitter framework) and have good API but are a part of huge framework which is too hard to use partially.
• ProofSeed - this one is also a part of rather big framework and mostly unusable outside of it due to reasons like non-portable Failure type. Now is ported to asynqro usage.
• Many others not mentioned here

So why not to create another one?

Future-related part of asynqro contains:

• Promise
• Future
• CancelableFuture
• WithFailure
• Trampoline
• repeat() and repeatForSequence() helpers

All classes are reentrant and thread-safe.

All higher-order methods are exception-safe. If any exception happens inside function passed to such method, then Future will fail (or task will gracefully stop if it is runAndForget).

It is possible to use Future with movable-only classes (except sequence()). In this case resultRef() should be used instead of result().

Asynqro is intended to be used by including asynqro/asynqro header that includes asynqro/future.h and asynqro/tasks.h. It is also possible to include only asynqro/futures.h if task scheduling is not needed. simplefuture.h provides simple wrapper with std::any as failure type. All other headers except these three are considered as implementation and should not be included directly.

Good example of customizing Future to specific needs can be found in https://github.com/opensoft/proofseed/blob/develop/include/proofseed/asynqro_extra.h .

Promise

There is not a lot of methods in this class and its main usage is to generate Future object which later will be filled by this Promise at most one time. All subsequent fills will be ignored.

Future

This class shouldn't be instantiated directly in users code, but rather is obtained either from Promise or from tasks scheduling part. Also new Future object is returned from all transformation Future methods. It complies with functor and monad laws from FP and provides all operators required by them (successful/failed, map, flatMap).

Future is also sort of left-biased EitherT with result type as left side and failure type as right value (to provide failure reason). Sides were chosen non-canonical way (typical Either usually has right side as result and left as error) for compatibility purposes: std::expected type in C++ is sided the same way.

Almost all Future methods returns Future object, so they can be effectively chained. Futures are almost immutable. Almost because they will change there state at most one time when they are filled. Adding callbacks doesn't change behavior of other already applied callbacks (i.e. it will not change state of Future and/or its value).

if higher-order method is called on already filled Future it will be called (in case of matching status) immediately in the same thread. If it is not yet filled, it will be put to queue, which will be called (in non-specified order) on Future filling in thread that filled the Future.

Future API

• successful - T->Future<T, FailureType> creates new Future object filled as successful with provided value
• failed - FailureType->Future<T, FailureType> creates new Future object filled as failed with provided reason
• fromQtSignal - (QObject, Signal)->Future<T, FailureType> creates new Future object that will be filled when signal emits. T here should be either bool or same type as signal parameter (if signal has more than one parameter T should be std::tuple)
• fromQtFuture - QFuture<OtherT>->Future<T, FailureType> creates new Future object that will be filled with QFuture result. OtherT and T must follow next rules:
  • if OtherT is void then T must be bool
  • if T is bool and OtherT is not convertible to bool then result will always be true
  • if T is a container of OtherT all results from QFuture will be used
  • if nothing above is true then OtherT must be convertible to T and in this case first result from QFuture will be used
• wait - waits for Future to be filled (either as successful or as failed) if it is not yet filled with optional timeout
• isCompleted/isFailed/isSucceeded - returns current state of Future
• result/resultRef/failureReason - returns result of Future or failure reason. Will wait for Future to be filled if it isn't already.
• onSuccess - (T->void)->Future<T, FailureType> adds a callback for successful case.
• onFailure - (FailureType->void)->Future<T, FailureType> adds a callback for failure case.
• filter - (T->bool, FailureType)->Future<T, FailureType> fails Future if function returns false for filled value.
• map - (T->U)->Future<U, FailureType> transforms Future inner type. Also available as >> operator.
• mapFailure - (FailureType->OtherFailureType)->Future<T, OtherFailureType> transforms Future failure type.
• flatMap - (T->Future<U, FailureType>)->Future<U, FailureType> transforms Future inner type. Also available as >> operator.
• andThen - (void->Future<U, FailureType>)->Future<U, FailureType> shortcut for flatMap if value of previous Future doesn't matter. Also available as >> operator.
• andThenValue - U->Future<U, FailureType> shortcut for andThen if all we need is to replace value of successful Future with some already known value.
• innerReduce/innerMap/innerFilter/innerFlatten - applicable only for Future with sequence inner type. Allows to modify sequence by reducing, mapping, filtering or flattening it.
• recover - (FailureType->T)->Future<T, FailureType> transform failed Future to successful
• recoverWith - (FailureType->Future<T, FailureType>)->Future<T, FailureType> the same as recover, but allows to return Future in callback
• recoverValue - T->Future<T, FailureType> shortcut for recover when we just need to replace with some already known value
• zip - (Future<U, FailureType>, ...) -> Future<std::tuple<T, U, ...>, FailureType> combines values from different Futures. If any of the Futures already have tuple as inner type, then U will be list of types from this std::tuple (so resulting tuple will be a flattened one). If zipped Futures have different FailureTypes then they will be combined in std::variant (with flattening if some of FailureTypes are already variants). Also available as + operator.
• zipValue - U->Future<std::tuple<T, U>, FailureType> - shortcut for zip with already known value.
• sequence - Sequence<Future<T, FailureType>> -> Future<Sequence<T>, FailureType> transformation from sequence of Futures to single Future.
• sequenceWithFailures - Sequence<Future<T, FailureType>> -> Future<std::pair<AssocSequence<Sequence::size_type, T>, AssocSequence<Sequence::size_type, FailureType>>, FailureType> transformation from sequence of Futures to single Future with separate containers for successful Futures and failed ones. AssocSequence can be set as optional type parameter.

CancelableFuture

API of this class is the same as Future API plus cancel method, that immediately fills this Future. CancelableFuture can be created only from Promise so it is up to providing side to decide if return value should be cancelable or not. Returning CancelableFuture however doesn't bind to follow cancelation as order, it can be considered as a hint. For example, Network API can return CancelableFuture and cancelation will be provided only for requests that are still in queue.

Providing side can check if Future was canceled by checking if Promise was already filled.

All CancelableFuture methods return simple Future to prevent possible cancelation of original Promise somewhere in downstream.

WithFailure

It is possible to fail any transformation by using WithFailure helper struct.

Future<int, std::string> f = /*...*/;
f.flatMap([](int x) -> Future<int, std::string> {
  if (shouldNotPass(x))
    return WithFailure<std::string>("You shall not pass!");
  else
    return asyncCalculation(x);
})
.map([](int x) -> int {
  if (mayItPass(x))
    return 42;
  return WithFailure<std::string>("You shall not pass!");
})
.recover([](const std::string &reason) -> int {
  if (reason.empty())
    return -1;
  return WithFailure<std::string>("You shall not pass, I said.");
});

This structure SHOULD NOT be saved anyhow and should be used only as a helper to return failure. Implicit casting operator will move from stored failure.

Trampoline

Using map() and other blocking transformations is something where we expect that stack can overflow, because we know that it will be called immediately each after another.

Although, for flatMap() or andThen() it is definitely not something one can expect due to its pseudo-asynchronous nature. But, in case of lots of flatMaps, it will still overflow on backward filling when last Future is filled.

To avoid such behavior Trampoline struct can be used anywhere where Future return is expected. It wraps a Future with extra transformation which will make sure that stack will be reset by moving it to another thread from Intensive thread pool.

Future<int, std::string> f = /*...*/;
f.flatMap([](int x) -> Future<int, std::string> {
    return Trampoline(asyncCalculation(x));
});

Repeat Helpers

Header asynqro/repeat.h contains asynqro::repeat() function that allows to do while-loop-styled calls to user function that can return either data or future with this data. User function should return either Continue with new set of arguments or Finish with final result.

Typical use case for repeat() is when it is needed to process something in serial manner, but do so using Future mechanism.

repeat() signature can be:

• ((Args...->RepeaterResult<T, Args...>), Args...) -> Future<T, FailureT> - this form passes Args to function while function returns Continue with new set of arguments. When function returns Finish invocation stops and repeat() itself returns Future filled with result. This form is blocking. It doesn't do any extra copies of arguments or result if function properly moves everything.
• ((Args...->RepeaterFutureResult<T, FailureT, Args...>), Args...) -> Future<T, FailureT> - this form passes Args to function and expects Future in return. This Future can be filled either with Finish or Continue or TrampolinedContinue. Third option is the same as regular Continue but uses Trampoline inside to ensure that stack wouldn't be overflown. This form is non-blocking if function is non-blocking. It doesn't do any extra copies on top of what is done by flatMap().

In case when there is a container with data we need to pass to our function one by one in serial manner, it is better to use repeatForSequence(). It accepts container, initial value and (Data, T)->Future<T, FailureT> function, where first argument is element from container and second is previous result (or initial value in case of first element). repeatForSequence() function returns Future<T, FailureT> with either final result or first occurred failure (and will not proceed forward with container values after failed one).

Tasks scheduling

The same as with futures, there are lots of implementations of task scheduling:

• Boost.Asio - Asio is much bigger than just scheduling, but it also provides thread pool with some API for running jobs in it
• QtConcurrent - part of Qt that uses QFuture and provides run() for scheduling task somewhere in QThreadPool.
• Thread-Pool-Cpp - most simplest possible implementation of thread pool that is extremely fast. It is as basic as it can be though - round-robin on threads with ability to pass queued jobs between them.
• Folly - Folly task scheduling using two different executors and MPMC queue is, again as Futures, an awesome solution, but it is a huge framework and can't be easily added only for task scheduling.
• many others not listed here.

Asynqro's task scheduling provides next functionality:

• Priorities. Tasks can be prioritized or de-prioritized to control when they should be executed
• Subpools. By default all tasks are running in subpool named Intensive, it is non-configurable and depends on number of cores in current system. It is, however, only a subpool of whole threads pool available in asynqro and it is possible to create Custom subpools with specified size to schedule other tasks (like IO or other mostly waiting operations). Custom subpools can also be paused (i.e. moved to 0 capacity with restoring old value on resuming).
• Thread binding. It is possible to assign subset of jobs to specific thread so they could use some shared resource that is not thread-safe (like QSqlDatabase for example).
• Future as return type. by default task scheduling returns CancelableFuture object that can be used for further work on task result. It also provides ability to cancel task if it is not yet started. It is also possible to specify what failure type should be in this Future by passing TaskRunner specialization to run (example can be found in https://github.com/opensoft/proofseed/blob/develop/include/proofseed/asynqro_extra.h).
• Sequence scheduling. Asynqro allows to run the same task on sequence of data in specified subpool.
• Clustering. Similar to sequence scheduling, but doesn't run each task in new thread. Instead of that divides sequence in clusters and iterates through each cluster in its own thread.
• Task continuation. It is possible to return Future<T> from task. It will still give Future<T> as scheduling result but will fulfill it only when inner Future is filled (without keeping thread occupied of course).
• Fine tuning. Some scheduling parameters can be tuned:
  • Idle amount - specifies how much empty loops worker should do in case of no tasks available for it before going to wait mode. More idle loops uses more CPU after tasks are done (so it is not really efficient in case of rare tasks) but in case when tasks are scheduled frequently it can be feasible to use bigger idle amount to not let workers sleep. 1024 by default.
  • Pool capacity - qMax(64, INTENSIVE_CAPACITY * 8) by default.
  • Thread binding amount - max amount of threads to be used for thread bound tasks. 1/4 of total pool size by default.
  • Preheating - it is possible to preheat (i.e. create worker threads) pool in advance. Either whole pool can be preheated or fraction of it.

Task scheduling performance

Task scheduling engine should be not only rich in its API but also has good performance in scheduling itself. benchmarks directory contains 4 synthetic benchmarks that can show at least some hints about how big the overhead of Asynqro is.

Tests were run few times for each solution on i7 with 4 cores+HT. Smallest one was chosen for each case.

Intensive and ThreadBound mean what type of scheduling was used in this suite. In ThreadBound tasks were assigned to amount of cores not bigger than number of logic cores. All asynqro benchmarks in this section use runAndForget function because we don't really need resulting Future. For Future usage overhead please see Futures usage overhead section below.

These benchmarks are synthetical and it is not an easy thing to properly benchmark such thing as task scheduling especially due to non-exclusive owning of CPU, non-deterministic nature of spinlocks and other stuff, but at least it can be used to say with some approximation how big overhead is gonna be under different amount of load.

Benchmarks listed above were collected with 0.1.0 version (QVariant-based one), but numbers are pretty the same for current generic version (probably a bit better for futures usage if some small failure types are used).

empty-avalanche

Big for loop that sends a lot of tasks without any payload (except filling current time of execution) to thread pool. It produces only one result - how many time it took to go through whole list.

timed-avalanche

The same as empty-avalanche, but in this case tasks are with some payload that tracks time. Each task should be ~0.1ms of payload. Result in this benchmark is difference between total time and summary of payload time divided by number of cores.

empty-repost

This benchmark was originally taken from thread-pool-cpp and adapted to qtconcurrent and asynqro usage. It starts C tasks, each of them counts how many times it was sent and if not enough yet (1kk) - sends itself again. Otherwise it reports time spent. It produces C different results. For each run we take highest one as a result (which actually means how much time it took to run all of them).

timed-repost

Almost the same as empty-repost, but tasks are filled with payload (the same way as timed-avalanche). Number of task runs for each task is reduced to 100k. Result of benchmark is again difference between total time and summary of payload time divided by number of cores.

payload of ~0.1ms

payload of ~1ms

Futures usage overhead

We can also measure overhead of using futures in task scheduling by running the same benchmarks for asynqro with run and runAndForget and compare them. Idle amount is 1000 in all tests.

empty-avalanche

timed-avalanche with 0.1ms

empty-repost with 100k tasks

empty-repost with 1kk tasks

timed-repost with 10k tasks of 0.1ms each

timed-repost with 100k tasks of 0.1ms each

Examples

Student in library

Let's say we need to authenticate student in library system, and after that fetch list of books she loaned with extra info about each of them. We also will need to fetch personalized suggestions and show them with list of books to return. However we know that there is a bug in suggestions and sometimes it can return book with age restriction higher than users age, so we need to filter them out.

We already have library system API designed as class that returns Future for each request.

We need to emit a signal loanedBooksFetched with loaned books list and suggestionsFetched with suggestions list. We can't, however send list of Book objects directly to QML, we need to transform it to QVariantList using static Book method.

We need to return resulting Future<bool> to know when everything is loaded or if any error occurred.

Future<bool, MyFailure> Worker::fetchData(QString username, QString password)
{
  return api->authenticate(username, password).flatMap([this](const User &userInfo) {
    auto taken = api->fetchTakenBooks().flatMap([this](const QVector<QString> &bookIds) {
      QVector<Future<Book>> result;
      result.reserve(bookIds.count());
      for (const QString &id : bookIds)
        result << api->fetchBook(id);
      return Future<Book>::sequence(result);
    })
    .map([this](const QVector<Book> &books) { return Book::qmled(books); })
    .onSuccess([this](const QVariantList &books) { emit loanedBooksFetched(books); });

    auto suggestions = api->fetchSuggestions().innerFilter([userInfo](const Book &book) {
      return book->ageRestriction < userInfo.age;
    })
    .map([](const QVector<Book> &books) { return Book::qmled(books); })
    .onSuccess([this](const QVariantList &books) { emit suggestionsFetched(books); });

    return taken.zip(suggestions).andThenValue(true);
  });
}

Repeat

We have some data that we need to send to our API and we want to do few retries before we decide that it is not possible to send it now.

using DataSendRepeater = RepeaterFutureResult<bool, std::string, int>;
Future<bool, std::string> Worker::sendData(Data data, int retries)
{
  return repeat<bool, int>([data](int retriesLeft) -> DataSendRepeater {
    if (retries < 0)
      return Future<bool, MyFailure>::failed("Too many retries");
    return api->sendData(data)
      .map([](bool result) -> DataSendRepeater::Value { return Finish(result); }
      .recover([](const auto &) -> DataSendRepeater::Value { return TrampolinedContinue(retries - 1); };
  }, retries);
}

Repeat for known sequence

We have input sequence that we need to process serially in determined order. If error occurs during calculation - it will fastfail.

Future<double, std::string> Worker::blackBox(int value, double accumulator);

Future<double, std::string> Worker::calculate(std::vector<int> data)
{
  return repeatForSequence(data, 0.0, [](int x, double result) -> Future<double, std::string> {
    return blackBox(x, result) >> [](double result){ return result < 0.0 ? 0.0 : result; };
  });
}