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 >=16. Tested with Clang9 (travis), GCC9 (travis) and MSVC16 (VS2019, appveyor). Also tested for compatibility with clang 6.0 and gcc 7.4 on Ubuntu Bionic image (travis). Conan packages available with help of bincrafters.
• 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.
• 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.
• Intel TBB - pretty good in performance (lags on some corner cases, see 8+ concurrency in timed reposts, but it is probably some bug that would be fixed eventually), not that good in API and requires a lot of boilerplate if low-level task scheduling needed. TBB is more oriented towards parallel computations (for which it provides solid set of helpers) than general parallel tasks.
• 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.

Limitations:

• Maximum amount of possible threads is 512. It is an artificial limitation that could be lifted in future versions.

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 (MacBook Pro 15 mid 2014, i7 2.2GHz). Smallest value 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.

If asynqro benchmark is marked with +F then it is using run function (that returns Future). If it isn't mark so - it uses runAndForget. +F mark can indirectly show how much overhead Future usage adds. Keep in mind that this overhead is not only about pure Future versus nothing, but also about run() logic overhead related to Future filling.

For Intel TBB repost benchmarks there are two different modes.

• enqueue means that all tasks are added using enqueue(), which adds them to shared queue. It is comparable to how Intensive tasks scheduling works in asynqro.
• spawn means that all child tasks (i.e. ones that were added from other tasks) are added by spawn(), which adds them to current thread queue, which resembles a bit ThreadBound behavior (but for TBB they also can be stolen by other threads).

Ideas behind what exactly each benchmarks measures:

• timed-repost - the most real life benchmark from the set. It emulates system with a lot of tasks that are being added from multiple threads with some pauses (payload of the task). It allocates these task additions across timeline and shows how system under stress would work.
• timed-avalanche - same tasks as in repost, but ALL tasks are added from one main thread. Tasks don't add new tasks anymore. This benchmark is about working with long tasks queue still keeping these payloads to emulate some useful work done in real systems.
• empty-repost - same as timed-repost, but without any payload. Equals to extra brutality on concurrent access to shared queue. This benchmark is where asynqro lags behind comparing to Boost.Asio and TBB and can have some improvements. But, as mentioned earlier - such case should rarely happen in real projects.
• empty-avalanche - same as timed-avalanche, but without any payload.

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 below were collected with 0.7.0 version.

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.

ThreadBound asynqro and threadpoolcpp behaves poorly on this benchmark on 10, 12, 14 jobs (i.e. more than cores and not the multiplier) due to nature of these schedulers. Intensive asynqro, boostasio and qtconcurrent worked with similar behavior as 8 and 16 results. These odd results are not included in the table for brevity, but it is something reader should be aware of.

payload of ~0.1ms

payload of ~1ms

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