C++并发编程实战读书笔记——同步并发操作

淡淡的烟草味﹌ 2022-03-01 04:06 242阅读 0赞

One thread might need to wait for another thread to complete a task before the first thread can complete its own.

### 等待事件 ###

waiting for an event  
当一个线程等待另一个线程完成任务时，它会有很多选择:  
第一，它可以持续的检查共享数据标志（由锁保护），直到另一个线程完成工作并对这个标志进行重设。线程消耗宝贵的执行时间持续检查对应标志。当互斥量被等待线程上锁后，其他线程没有办法获取锁，这样其他线程会等待，等待线程阻止了工作线程对标识的设置。  
第二：让等待线程在检查标识的间隙休眠。

bool flag;
    std::mutex m;
    void wait_for_flag()
    {
    	std::unique_lock<std::mutex> lk(m);
    	while(!flag)
    	{
    		lk.unlock();
    		std::this_thread::sleep_for(std::chrono::milliseconds(100));
    		lk.lock();
    	}
    }

在这个循环中，线程休眠前，函数对互斥量进行解锁，并且在休眠结束后再对互斥量加锁，所以另外的线程就有机会获取锁并设置标识。  
第三：使用C++标准库提供的工具去等待事件的发生，通过另一线程触发等待事件的机制是最基本的唤醒方式。这种机制就称为“条件变量”。从概念上来说，一个条件变量会与多个事件或其他条件相关，并且一个或多个线程会等待条件的达成。当某些线程直到条件变量达成时，为了唤醒等待线程(允许等待线程继续执行)，它将会向等待着的线程广播“条件达成”的信息。

#### 等待条件达成 ####

C++标准库对条件变量有两套实现： std::condition\_variable 和 std::condition\_variable\_any。这两个实现都包含在<condition\_variable>头文件的声明中。两者都需要与一个互斥量一起才能工作（需要互斥量是为了同步）。前者仅限于与 std::mutex 一起工作，而后者可以和任何满足最低标准的互斥量一起工作，从而加上了\_any的后缀。

std::mutex mut;
    std::queue<data_chunk> data_queue;
    std::condition_variable data_cond;
    
    void data_preparation_thread()
    {
    	while(more_data_to_prepare())
    	{
    		data_chunk const data=prepare_data();
    		std::lock_guard<std::mutex> lk(mut);
    		data_queue.push(data);
    		data_cond.notify_one();
    	}
    }
    void data_processing_thread()
    {
    	while(true)
    	{
    		std::unique_lock<std::mutex> lk(mut);
    		data_cond.wait(lk,[]{return !data_queue.empty();});
    		data_chunk data=data_queue.front();
    		data_queue.pop();
    		lk.unlock();
    		process(data);
    		if(is_last_chunk(data))
    			break;
    	}
    }

有用来在两个线程之间传递数据的队列，当数据准备好时，使用std::lock\_guard对队列上锁，将准备好的数据压入队列中。然后调用std::condition\_variable的notify\_one()成员函数，对等待的线程进行通知。  
处理线程首先加锁，但是是使用std::unique\_lock而不是std::lock\_guard。线程之后调用std::condition\_variable的成员函数wait()，传递一个锁和一个lambda函数表达式（作为等待的条件）。在这个例子中，简单的lambda函数会检查data\_queue是否不为空，当data\_queue不为空就意味着队列中已经准备好数据了。wait()会去检查这些条件（通过调用所提供的lambda函数），当条件满足（lambda函数返回true）时返回。如果条件不满足（lambda函数返回false），wait()函数将解锁互斥量，并将这个线程置于阻塞或等待状态。当准备数据的线程调用notify\_one()通知条件变量时，处理数据的线程从睡眠状态中苏醒，重新获取互斥锁，并且对条件再次检查，在条件满足的情况下，从wait()返回并继续持有锁。当条件不满足时，线程将对互斥量解锁，并且重新开始等待。这就是为什么用std::unique\_lock而不使用std::lock\_guard-等待中的线程必须在等待期间解锁互斥量，并在这之后对互斥量再次上锁，而std::lock\_guard没有这么灵活。在调用wait()的过程中，一个条件变量可能会检查给定条件若干次，然而，它总是在互斥量被锁定时这么做，当提供测试条件的函数返回true时，它就会立即返回。当等待线程重新获取互斥量并检查条件时，如果它并没有直接响应另一个线程的通知，这就是所谓的伪唤醒spurious wake。

### 带有期望的等待一次性事件 ###

waiting for one-off events with futures  
当一个线程需要等待一个特定的一次性事件，C++标准库模型将这种一次性事件称为期望(future)。线程会周期性(较短的周期)的等待或检查，事件是否触发；在检查期间也会执行其他任务。另外，在等待任务期间它可以先执行另外一些任务，直到对应的任务触发，而后等待期望的状态会变为就绪(ready)。一个期望可能是数据相关的，也可能不是。当事件发生时（并且期望状态为就绪），这个期望就不能被重置。  
在C++标准库中，有两种“期望”，使用两种类型模板实现，声明在头文件<future>: 唯一期望(unique futures)( std::future<> )和共享期望(shared futures)( std::shared\_future<> )。std::future的实例只能与一个指定事件相关联，而std::shared\_future的实例就能关联多个事件。后者的实现中，所有实例会在同时变为就绪状态，并且他们可以访问与事件相关的任何数据。虽然，希望用于线程间的通讯，但是“期望”对象本身并不提供同步访问。当多个线程需要访问一个独立“期望”对象时，他们必须使用互斥量或类似同步机制对访问进行保护。

#### 带返回值的后台任务 ####

因为std::thread不提供直接接收返回值的机制。这里需要std::async函数模板（在头文件<future>中声明的）。当任务的结果不着急要时，可以使用std::async启动一个异步任务。与std::thread对象等待的方式不同，std::async会返回一个std::future对象，这个对象持有最终计算出来的结果。当你需要这个值时，你只需要调用这个对象的get()成员函数；线程会阻塞直到期望状态就绪为止，之后，返回计算结果。

#include <future>
    #include <iostream>
    int find_the_answer_to_ltuae();
    void do_other_stuff();
    int main()
    {
    	std::future<int> the_answer=std::async(find_the_answer_to_ltuae);
    	do_other_stuff();
    	std::cout<<"The answer is "<<the_answer.get()<<std::endl;
    }

std::async允许通过添加额外调用参数的方式，向函数传递额外的参数。当第一个参数是一个指向成员函数的指针，第二个参数提供提供这个函数成员类的具体对象（不是直接，就是通过指针，还可以包装在std::ref中），剩余的参数可作为成员函数的参数传入。否则，第二个和随后的参数将作为函数的参数，或作为指定可调用对象的第一个参数。

#include <string>
    #include <future>
    struct X
    {
    	void foo(int,std::string const&);
    	std::string bar(std::string const&);
    };
    X x;
    auto f1=std::async(&X::foo,&x,42,"hello");
    auto f2=std::async(&X::bar,x,"goodbye");
    struct Y
    {
    	double operator()(double);
    };
    Y y;
    auto f3=std::async(Y(),3.141);
    auto f4=std::async(std::ref(y),2.718);
    X baz(X&);
    std::async(baz,std::ref(x));
    class move_only
    {
    public:
    	move_only();
    	move_only(move_only&&)
    	move_only(move_only const&) = delete;
    	move_only& operator=(move_only&&);
    	move_only& operator=(move_only const&) = delete;
    	void operator()();
    };
    auto f5=std::async(move_only());

auto f1=std::async(&X::foo,&x,42,“hello”);-》Calls p->foo(42,“hello”) where p is &x  
auto f2=std::async(&X::bar,x,“goodbye”);-》Calls tmpx.bar(“goodbye”) where tmpx is a copy of x  
auto f3=std::async(Y(),3.141);-》Calls tmpy(3.141) where tmpy is move-constructed from Y()  
auto f4=std::async(std::ref(y),2.718);-》Calls y(2.718)  
std::async(baz,std::ref(x));-》Calls baz(x)  
auto f5=std::async(move\_only());-》Calls tmp() where tmp is constructed from std::move(move\_only())  
在默认情况下，期望是否进行等待取决于std::async是否启动一个线程或是有任务正在进行同步。在大多数情况下，但是也可以在函数调用之前，向std::async传递一个额外参数。这个参数的类型是std::launch，还可以是std::launch::defered，用来表明函数调用被延迟到wait()或get()函数调用时才执行，std::launch::async表明函数必须在其所在的独立线程上执行，std::launch::deferred | std::launch::async表明实现可以选择这两种方式的一种。最后一个选项是默认的。当函数调用被延迟，它可能不会再运行了。

auto f6=std::async(std::launch::async,Y(),1.2); //Run in new thread
    auto f7=std::async(std::launch::deferred,baz,std::ref(x)); //Run in wait() or get()
    //Implementation chooses
    auto f8=std::async(std::launch::deferred|std::launch::async,baz,std::ref(x));
    auto f9=std::async(baz,std::ref(x));
    f7.wait(); //Invoke deferred function

#### 任务与期望 ####

std::packaged\_task<>对一个函数或可调用对象，绑定一个期望。当std::packaged\_task<>对象被调用，它就会调用相关函数或可调用对象，将期望状态置为就绪，返回值也会被存储为相关数据。这可以用在构建线程池的结构单元，或用于其他任务的管理，比如在任务所在线程上运行任务，或将它们顺序运行在一个特殊后台线程上。当一个粒度较大的操作可以被分解为独立的子任务时，其中每个子任务就可以包含在一个std::packaged\_task<> 实例中，之后这个实例将传递到任务调度器或线程池中。对任务的细节进行抽象，调度器仅处理std::package\_task<>实例，而非处理单独的函数。  
std::packaged\_task<>的模板参数是一个函数签名，比如void()就是一个没有参数也没有返回值的函数，或int(std::string&, double\*)就是有一个非const引用的std::string和一个指向double类型的指针，并且返回类型是int。当你构造出一个std::packaged\_task<>实例时，你必须传入一个函数或可调用对象，这个函数或可调用的对象需要能接收指定的参数和返回可转换为指定返回类型的值。类型可以不完全匹配；你可以用一个int类型的参数和返回一个float类型的函数，来构建 std::packaged\_task<double(double)> 的实例，因为在这里，类型可以隐式转换。  
指定函数签名的返回类型可以用来标识，从get\_future()返回的std::future<>的类型，函数签名的参数列表，用来指定“打包任务”的函数调用操作符。

std::packaged\_task<>的偏特化

template<>
    class packaged_task<std::string(std::vector<char>*,int)>
    {
    public:
    	template<typename Callable>
    	explicit packaged_task(Callable&& f);
    	std::future<std::string> get_future();
    	void operator()(std::vector<char>*,int);
    };

这里的std::package\_task对象是一个可调用对象，并且可以包含在一个std::function对象中，传递到std::thread对象中，可以作为线程函数。当std::packaged\_task作为一个函数调用时，可为函数调用操作符提供所需的参数，并且返回值作为异步结果存储在std::future，可以通过get\_future()获取。

使用std::promises  
std::promise<T>提供设定值的方式（类型为T），这个类型会从后面看到的关联的std::future<T>对象读出。一对std::promise/std::future会为这种方式提供一个可行的机制；在期望上可以阻塞等待线程，同时，提供数据的线程可以使用组合中的“承诺promises”来对相关值进行设置，以及将期望的状态置为就绪。可以通过get\_future()成员函数来获取与一个给定的std::promise相关的std::future对象，就像是与std::packaged\_task相关。当promise的值已经设置完毕（使用set\_value()成员函数），对应期望的状态变为就绪，并且可用于检索已存储的值。当你在设置值之前销毁std::promise，将会存储一个异常。

为“期望”存储“异常”

double square_root(double x)
    {
    	if(x<0)
    	{
    		throw std::out_of_range(“x<0”);
    	}
    	return sqrt(x);
    }

异步调用：

std::future<double> f=std::async(square_root,-1);
    double y=f.get();

函数作为std::async的一部分时，当在调用时抛出一个异常，那么这个异常就会存储到“期望”的结果数据中，之后“期望”的状态被置为就绪，之后调用get()会抛出这个存储的异常。（注意：标准没有指定重新抛出这个异常是原始的异常对象，还是一个拷贝；不同编译器和库将会在这个方面做出不同的选择）。当你将函数打包入std::packaged\_task任务包中后，这个任务被调用时，同样的事情也会发生；当打包函数抛出一个异常，这个异常将被存储在“期望”的结果中，准备在调用get()再次抛出。当你希望存入的是一个异常而非一个数值时，需要调用set\_exception()成员函数，而非set\_value()。

extern std::promise<double> some_promise;
    try
    {
    	some_promise.set_value(calculate_value());
    }
    catch(...)
    {
    	some_promise.set_exception(std::current_exception());
    }

这里使用了std::current\_exception()来检索抛出的异常；可用std::copy\_exception()作为一个替代方案，std::copy\_exception()会直接存储一个新异常而不抛出。

some_promise.set_exception(std::copy_exception(std::logic_error("foo ")));

### 限定等待时间 ###

waiting with a time limit

### 使用同步操作简化代码 ###

using synchronization of operations to simplify code