C++ 异步编程笔记 (ISO C++ 17)

Posted 2021-05-29 麦仲肥

#include <chrono>
#include <thread>
#include <future>
#include <numeric>
#include <vector>
#include <iostream>

using namespace std;

namespace std_async_demo {
	using vec_itr_t = vector<int>::iterator;

	void promise_future() {
		std::vector nums = { 1, 2, 3, 4, 5 };

		auto f = [](auto first, auto last, auto p) { // (1) define async function
			p.set_value(std::accumulate(first, last, 0)); // (2) set return value to PROMISE 
		};
		
		std::promise<int> p;
		std::future<int> r = p.get_future();
		std::thread t(f, nums.begin(), nums.end(), std::move(p)); // (3) new thread with async function 

		std::cout << "result = " << r.get() << endl; // (4) wait & get result 
		t.join(); // (5) join 
	}

	void package_task() {
		std::vector nums = { 1, 2, 3, 4 };

		auto f = [](vec_itr_t first, vec_itr_t last) { // (1) define async function
			return std::accumulate(first, last, 0);
		};

		std::packaged_task<int(vec_itr_t, vec_itr_t)> task(f); // (2) package function call into a task

		auto r = task.get_future();
		task(nums.begin(), nums.end()); // (3) activate the task 

		std::cout << "result = " << r.get() << endl; // (4) wait & get result
	}
	
	void async_get() {
		cout << "async call & wait infinitely..." << endl;

		std::vector nums = { 1, 2, 3 };
		auto f = std::accumulate<vec_itr_t, int>; // (1) define async function
		auto r = std::async(std::launch::async, f, nums.begin(), nums.end(), 0); // (2) async call 

		std::cout << "result = " << r.get() << endl; // (3) waint infinitely & get result
	}

	void async_wait() {
		cout << "async call & wait & check periodically for a certain time --" << endl;

		std::vector nums = { 1, 2 };
		
		auto f = [&](vec_itr_t first, vec_itr_t last) { // (1) define async function
			std::this_thread::sleep_for(std::chrono::seconds(5));
			return std::accumulate(first, last, 0);
		};

		auto r = std::async(std::launch::async, f, nums.begin(), nums.end()); // (2) async call

		std::future_status status;
		do
		{
			status = r.wait_for(std::chrono::seconds(1)); // (3) waint & check status periodically
			if (status == std::future_status::timeout || status == std::future_status::deferred) {
				cout << "not ready, do something else ..." << endl;
			}
		} while (status != std::future_status::ready);

		std::cout << "result = " << r.get() << endl; // (4) get result
	}
}

int main() {
	using namespace std_async_demo;
	promise_future();
	package_task();
	async_get();
	async_wait();
	return 0;
}

运行结果：

result = 15
result = 10
async call & wait infinitely...
result = 6
async call & wait & check periodically for a certain time --
not ready, do something else ...
not ready, do something else ...
not ready, do something else ...
not ready, do something else ...
result = 3

　　

小结：

promise_future() 展示的是C++异步基本机制：引入promise用于线程函数里存结果(set)；引入跟promise关联的future用于在调用线程取结果(get)；引入thread执行线程函数(run)；主线程启动线程，并等待通过future获取结果(call & wait)。

便利之处：通过promise和future封装好了线程同步和数据的线程安全，省心了很多这方面的工作。

package_task() 展示基本机制上的进一步封装：引入task包装了promise和thread，把普通函数或函数式交给task(package)，task则执行时自动起线程执行，并自动把结果存好在promise里，在task关联的fucture里直接获取结果即可(call & wait)。

便利之处：不需要自建promise和thread实例及执行相关的调用，又省心了一些。

async_get() 展示task基础上的进一步封装：连task都干掉了，定义好普通函数或函数式，直接通过std::async() 异步调用，然后通过所关联的future取结果(call & wait)

便利之处：到这一步已经是直接call就好了，这已经是省心至极了吧？是，但也不是，因为其实里面还隐藏了一个wait动作，即线程的同步与等待，主线程必须等在那里取结果，而且，是永久地等，干不了其他；另外，取结果也是手动显式地get一下。能不能把wait这一步也省了？能不能不用手动get结果？这个就是ISO C++ 20 引入Coroutine要解决的问题。另开笔记说。

async_wait() 展示是对上一步的完善：让主线程干等着、死等着子线程结束和返回结果在大多生产环境下恐怕是不能接受的，在还没有Coroutine机制帮忙时，在future里加了个wait_for函数，用来每等一会就看看结果，还没好就干点别的再回来等等看看。

便利之处：比较好的折中了启用新线程做异步调用的性能加分和主线程无奈得等待结果的性能减分，符合生产环境需要。

 

延申：

C++的多线程 -> 异步调用 --> 协程 的演进线路，乃至实现形式，看上去都很像C#，猜测是有所借鉴。C#更早地实现了这一切，而且非常好用。（当然C#也可能有借鉴其他语言的演进和实现）

多线程以及相关的线程安全、线程同步机制是基础性的。对于比较大的、比较复杂的任务，个人觉得理当还是新开一个线程处理，老老实实按经典的多线程编程模型实现好。异步调用是针对功能单一的任务，即执行时只需要关心有限数据的线程安全，执行完后只需要把结果存好以备查看即好，全程不需要线程间多次同步或互动，能提高程序性能又具开发效率，并且代码可读性强，是非常好的一个工具。协程是让主线程和子线程间透明、无缝协作，线程切换的上下文恢复，数据的线程安全，线程间的同步等待都透明化，有更好的运行效率和响应速度，并且有最直观自然的编程模型，个人非常喜欢，能用的情况都推荐优先使用。

// 以上均个人理解。水平所限，如有错漏，望指正。