C++ Standard Library: Concurrency (1) -- High-level interfaces async() and futrue

future is a high-level API provided in the c++ standard library for performing asynchronous tasks. Its related API can be introduced via #include <future>.

Create concurrent tasks using future

It is simple to create an asynchronous task using future’s API.

#include <iostream>
#include <thread>
#include <chrono>
#include <random>
#include <exception>
#include <future>

int doSomething(char c)
{

    std::default_random_engine dre(c);
    std::uniform_int_distribution<int> id(10, 1000);
    for (int i = 0; i < 5; i++)
    {
        auto sleep_time = std::chrono::milliseconds(id(dre));
        std::this_thread::sleep_for(sleep_time);
        std::cout.put(c).flush();
    }
    return c;
}

int func1()
{
    return doSomething('.');
}

int func2()
{
    return doSomething('+');
}

// Test normal async asynchronous
// default by the system scheduling appropriate to start asynchronous tasks, any time may be turned on; when calling .get(), forcing the opening of the thread task
// At this time will block the execution of func2, then block the execution of func1, there is actually no asynchronous operation
void TestAsync()
{
    std::cout << "start func1 in background and func2 foreground" << std::endl;
    std::future<int> result1(std::async(func1));
    int result2 = func2();
    int result = result1.get() + result2;
    std::cout << " final result :" << result << std::endl;
}

Here doSomething is a time consuming task, it will sleep 50~5000ms randomly when executed.

In the above example, a concurrent task is created based on func1. The program tries to execute the task of func1 in a separate thread as much as possible, and if it doesn’t work, it executes it in the current thread. In fact, I run this program under gcc4.8 and all the computations are done in the main thread. The key points here are as follows.

When get is called, if it’s finished, you get the result directly; if it’s not finished, it blocks the current thread until it’s finished.
Under gcc4.8, if you don’t call get, the concurrent task will never start, and after actively calling get, the task is still executed serially.

launch policy

In the basic usage above, the task is actually still executed serially, and that’s because we didn’t give the application a launch policy, which is the launch policy for concurrent tasks.

std::launch::async asynchronous launch

//test passes async launch
//If std::launch::async is passed, the asynchronous task will start immediately, and an exception will be thrown if it doesn't. At this point multiple tasks are actually executed in parallel
void TestLaunchAsync()
{
    //launch async forces a direct asynchronous start, and you can see the two symbols typed alternately
    std::cout << "start func1 in background and func2 background, launch by async" << std::endl;
    // Note that if asynchronous is not supported here, a system error will be thrown, e.g. if libpthread.a is not connected at compile time
    std::future<int> result1(std::async(std::launch::async, func1));
    std::future<int> result2(std::async(std::launch::async, func2));
    //int result2 = func2();
    // Note that when using launch async, the asynchronous function will be executed at the end of the life cycle even if you don't get here
    std::cout << " before final " << std::endl;
    int result = result1.get() + result2.get();
    std::cout << " final result :" << result << std::endl;
}

When using the std::launch::async launch policy, it immediately forces concurrent tasks to be launched in a separate thread, when the tasks are truly parallel
If asynchrony is not supported, a system error is thrown

std::launch::deferred Delayed start

// 测试通过deferred launch
// 当用launch deferred的时候，调用get之前绝对不会执行异步任务
// 因为最终还是调用get时启动且阻塞当前线程的，所以这俩任务实际上也是串行，而不是异步并行的。
void TestLaunchDeferred()
{
    std::cout << "start func1 in background and func2 background, launch by deferred" << std::endl;
    std::future<int> result1(std::async(std::launch::deferred, func1));
    std::future<int> result2(std::async(std::launch::deferred, func2));
    //int result2 = func2();
    //std::default_random_engine dre('x');
    std::default_random_engine dre('x');
    std::uniform_int_distribution<int> id(10, 1000);
    int random = id(dre);
    while (random < 500)
    {
        std::cout << " not to call get, and not start deferred task, random:" << random << std::endl;
        random = id(dre);
    }
    std::cout << " random: " << random << std::endl;
    int result = result1.get() + result2.get();
    std::cout << " call get and start deferred task, final result :" << result << std::endl;
}

Absolutely no asynchronous tasks are executed until get is called
This startup strategy under gcc4.8 is the same as the default mode, where all tasks are done serially in the main thread

wait, wait_for, wait_until wait, polling

// 测试wait
void TestWait()
{
    std::future<int> result1 = std::async(func1);
    std::future<int> result2 = std::async(func2);
    result1.wait();
    result2.wait();
    std::cout << " async task done " << std::endl;
    std::cout << " wait会强制启动async任务：" << (result1.get()+result2.get()) << std::endl;
}

wait will force an async task to start, wait will block the current thread, so the two wait tasks are synchronous and serial

std::future<int> f;
void TestWaitFor()
{
    int result = 1;
    //不加launch，因为wait_for不会主动开启任务，这里任务都不会开始执行，怎么等都是timeout
    //std::future<int> f = std::async(func1);

    //加launch::async，这里根据等待时间不同，可能是ready或者timeout，但是有个问题，无论是哪种情况，函数析构的时候都会等待async的任务执行完毕
    //std::future<int> f = std::async(std::launch::async, func1);

    //这种写法，把future对象放外面，在此函数结束的时候，不会等待f执行结束。 不过此demo中，在程序退出前仍然会等待退出
    //f = std::async(std::launch::async, func1);

    //通过deferred，wait_for之后也总是timeout
    //TODO: std::future_status::deferred如何触发？
    f = std::async(std::launch::deferred, func1);

    //等一段时间，若异步的有结果了则返回异步的，否则返回别的
    std::future_status status = f.wait_for(std::chrono::milliseconds(500));
    switch (status)
    {
    case std::future_status::deferred:
        std::cout << "还未开启异步任务" << std::endl;
        break;
    case std::future_status::ready:
        std::cout << "异步任务已完成" << std::endl;
        break;
    case std::future_status::timeout:
        std::cout << "异步任务等待时间到了，还没结束" << std::endl;
        break;
    default:
        break;
    }
}
}

wait_for blocks the current thread and waits for a fixed amount of time without forcing the task to start
wait_until is the same, except that it can wait until a specific point in time
after the wait timeout, the current thread will continue to execute without waiting, and the task in the future thread will not be terminated, but will continue to finish executing

// 测试利用等待0时间来轮询
void TestPolling()
{
    //这样会一直卡住，c++标准库中对不带launch选项的async的解释，是会自动在合适的时机拉起，但是这里gcc4.8中，看起来永远不会开启任务，一直卡在循环中
    // std::future<int> result = std::async(func1);
    // while(result.wait_for(std::chrono::seconds(0)) != std::future_status::ready){
    //     std::cout << "wait for 0, polling..." << (result.wait_for(std::chrono::seconds(0)) == std::future_status::timeout) << std::endl;
    //     std::this_thread::sleep_for(std::chrono::seconds(1));
    // }
    // std::cout << "wait for polling done:" << result.get() << std::endl;

    //这样也会一直卡住，或许是运行环境的原因，这里和《C++标准库》里的描述不同。实际运行时，这个wait_for(0)一直会返回timeout
    //可能在某些编译器或者运行环境下，不带launch的async是会立即启动的
    // std::future<int> result = std::async(func1);
    // if(result.wait_for(std::chrono::seconds(0)) != std::future_status::deferred){
    //     while(result.wait_for(std::chrono::seconds(1)) != std::future_status::ready){
    //         std::cout << "wait for 0, polling..." << std::endl;
    //         std::this_thread::sleep_for(std::chrono::seconds(1));
    //     }
    // }

    // 貌似只有这样，主动用async 来launch，才能polling成功
    std::future<int> result = std::async(std::launch::async, func1);
    if (result.wait_for(std::chrono::seconds(0)) != std::future_status::deferred)
    {
        while (result.wait_for(std::chrono::milliseconds(100)) != std::future_status::ready)
        {
            std::cout << "wait for 100ms, polling..." << std::endl;
            std::this_thread::yield();
        }
    }
    std::cout << "wait for polling done:" << result.get() << std::endl;
}

You can poll and wait until the task is completed by constantly sleeping and wait_for.

std::shared_future fetches results multiple times

By default a std::future can only get once, multiple fetches will throw a std::future_error exception. The standard library provides shared_future for use in scenarios where multiple gets are required.

void TestSharedAsync()
{
    std::cout << "main thread:" << std::this_thread::get_id() << std::endl;
    std::cout << "start func1 in background and func2 foreground" << std::endl;
    //std::future<int> result1(std::async(func1));
    std::shared_future<int> result1(std::async(func1));
    int result2 = func2();
    int result = result1.get() + result1.get() + result2;
    std::cout << " final result :" << result << std::endl;
}

std::future multiple calls to get will throw std::future_error exception, just replace it with std::shared_future.

Parameter passing

When creating concurrent tasks via future, it is sometimes necessary to pass parameters to the called function. The following code tests passing parameters via lamda value passing, function value passing, lamda reference passing, and function reference passing in several ways.

void TestPassingArguments()
{
    char a = 'a', b='b', c='c', d='d';
    //lamda值传递
    std::future<int> res1 = std::async(std::launch::async, [=]{return doSomething(a);});
    //函数值传递
    std::future<int> res2 = std::async(std::launch::async, doSomething, b);

    //lamda引用传递
    std::future<int> res3 = std::async(std::launch::async, [&]{return doSomething(c);});
    //函数引用传递
    std::future<int> res4 = std::async(std::launch::async, doSomething, std::ref(d));

    std::cout << "res1:" << res1.get() << "res2:" << res2.get()<< "res3:" << res3.get()<< "res4:" << res4.get() << std::endl;
}

Exception Handling

If an exception is thrown in an asynchronous task, how should the program handle it?

//测试异常处理
void TestHandleException()
{
    try
    {
        std::future<void> ex(std::async(func_ex));
        ex.get();
    }
    catch (std::exception &e)
    {
        std::cout << "e:" << e.what() << std::endl;
    }
}

Catch like a normal exception
This exception is only available and thrown after the call to get, just like a normal result of get

Reference https://www.zoucz.com/blog/2021/06/08/fbcfc460-c86d-11eb-9fe7-534bbf9f369d/

Table of Contents