【C++】多线程与条件变量【三】

1 条件变量是什么？

condition_variable 类是同步原语，能用于阻塞一个线程，或同时阻塞多个线程，直至另一线程修改共享变量（条件）并通知 condition_variable 。

有意修改变量的线程必须

1. 获得 std::mutex （常通过 std::lock_guard ）
2. 在保有锁时进行修改
3. 在 std::condition_variable 上执行 notify_one 或 notify_all （不需要为通知保有锁）

即使共享变量是原子的，也必须在互斥下修改它，以正确地发布修改到等待的线程。

任何有意在 std::condition_variable 上等待的线程必须

1. 在与用于保护共享变量者相同的互斥上获得 std::unique_lock<std::mutex>
2. 执行下列之一：

std::condition_variable 只可与 std::unique_lock<std::mutex> 一同使用；此限制在一些平台上允许最大效率。 std::condition_variable_any 提供可与任何基本可锁定 (BasicLockable) 对象，例如 std::shared_lock 一同使用的条件变量。

condition_variable 容许 wait 、 wait_for 、 wait_until 、 notify_one 及 notify_all 成员函数的同时调用。

类 std::condition_variable 是标准布局类型 (StandardLayoutType) 。它非可复制构造 (CopyConstructible) 、可移动构造 (MoveConstructible) 、可复制赋值 (CopyAssignable) 或可移动赋值 (MoveAssignable) 。

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-uCh6xXA1-1609602482375)(C:\Users\guoqi\AppData\Roaming\Typora\typora-user-images\1609594667734.png)]

实例1：

#include <iostream>
#include <string>
#include <thread>
#include <mutex>
#include <condition_variable>std::mutex m;
std::condition_variable cv;
std::string data;
bool ready = false;
bool processed = false;void worker_thread()
{// 等待直至 main() 发送数据std::unique_lock<std::mutex> lk(m);cv.wait(lk, [] {return ready; });// 等待后，我们占有锁。std::cout << "Worker thread is processing data\n";data += " after processing";// 发送数据回 main()processed = true;std::cout << "Worker thread signals data processing completed\n";// 通知前完成手动解锁，以避免等待线程才被唤醒就阻塞（细节见 notify_one ）lk.unlock();cv.notify_one();
}int main()
{std::thread worker(worker_thread);data = "Example data";// 发送数据到 worker 线程{std::lock_guard<std::mutex> lk(m);ready = true;std::cout << "main() signals data ready for processing\n";}cv.notify_one();// 等候 worker{std::unique_lock<std::mutex> lk(m);cv.wait(lk, [] {return processed; });}std::cout << "Back in main(), data = " << data << '\n';worker.join();getchar();
}

2 条件变量本质？

条件变量是操作系统实现的。关键在于理解为啥要有它，而且需注意一点，条件变量自身并不包含条件。因为它通常和 if (或者while) 一起用，所以叫条件变量。

并发有两大需求，一是互斥，二是等待。互斥是因为线程间存在共享数据，等待则是因为线程间存在依赖。

条件变量，是为了解决等待需求。考虑实现生产者消费者队列，生产者和消费者各是一个线程。一个明显的依赖是，消费者线程依赖生产者线程 push 元素进队列。

典型的如CS架构， 基于请求响应的模式，即客户端向发送服务器请求，然后服务器把请求结果返回给客户端。

而服务器可以用生产者消费者模型来处理客户端的请求：

• 服务器有一些线程(或者进程)，把客户端的请求转换成统一格式压入消息队列，这些线程称为生产者。
• 另有一些线程(或者进程)，不断从消息队列取出消息来处理，这些线程称为消费者。

线程同步的原理和实现使用互斥锁解决数据竞争访问问题，算是线程同步的加锁原语，用于排他性的访问共享数据。我们在使用mutex时，一般都会期望加锁不要阻塞，总是能立刻拿到锁，然后尽快访问数据，用完之后尽快解锁，这样才能不影响并发性和性能。

**条件变量是线程的另外一种有效同步机制。**这些同步对象为线程提供了交互的场所（一个线程给另外的一个或者多个线程发送消息），我们指定在条件变量这个地方发生，一个线程用于修改这个变量使其满足其它线程继续往下执行的条件，其它线程则等待接收条件已经发生改变的信号。当条件变量同互斥锁一起使用时，条件变量允许线程以一种无竞争的方式等待任意条件的发生。

3 引入条件变量的原因？

前一章【C++】多线程与互斥锁【二】介绍了多线程并发访问共享数据时遇到的数据竞争问题，通过互斥锁保护共享数据，保证多线程对共享数据的访问同步有序。但如果一个线程需要等待一个互斥锁的释放，该线程通常需要轮询该互斥锁是否已被释放，我们也很难找到适当的轮训周期，如果轮询周期太短则太浪费CPU资源，如果轮询周期太长则可能互斥锁已被释放而该线程还在睡眠导致发生延误。

针对典型的如CS架构，如果没有条件变量，你会怎么实现消费者呢？让消费者线程一直轮询队列（需要加 mutex)。如果是队列里有值，就去消费；如果为空，要么是继续查( spin 策略)，要么 sleep 一下，让系统过一会再唤醒你，你再次查。可以想到，无论哪种策略，都不通用，要么费 cpu，要么线程过分 sleep，影响该线程的性能。有条件变量后，就可以使用事件模式了。上面的消费者线程，发现队列为空，就告诉操作系统，我要 wait，一会肯定有其他线程发信号来唤醒我的。这个『其他线程』，实际上就是生产者线程。生产者线程 push 队列之后，则调用 signal，告诉操作系统，之前有个线程在 wait，你现在可以唤醒它了。**上述两种等待方式，前者是轮询(poll)，后者是事件(event)。**一般来说，事件方式比较通用，性能不会太差(但存在切换上下文的开销)。轮询方式的性能，就非常依赖并发 pattern，也特别消耗 cpu。

实例2：

如下给出一个实例：一个线程往队列中放入数据，一个线程从队列中提取数据，取数据前需要判断一下队列中确实有数据，由于这个队列是线程间共享的，所以，需要使用互斥锁进行保护，一个线程在往队列添加数据的时候，另一个线程不能取，反之亦然。程序实现代码如下：

//cond_var1.cpp用互斥锁实现一个生产者消费者模型#include <iostream>
#include <deque>
#include <thread>
#include <mutex>std::deque<int> q;						//双端队列标准容器全局变量
std::mutex mu;							//互斥锁全局变量
//生产者，往队列放入数据
void function_1() {int count = 10;while (count > 0) {std::unique_lock<std::mutex> locker(mu);q.push_front(count);			//数据入队锁保护locker.unlock();std::this_thread::sleep_for(std::chrono::seconds(1));		//延时1秒count--;}
}
//消费者，从队列提取数据
void function_2() {int data = 0;while ( data != 1) {std::unique_lock<std::mutex> locker(mu);if (!q.empty()) {			//判断队列是否为空data = q.back();q.pop_back();			//数据出队锁保护locker.unlock();std::cout << "t2 got a value from t1: " << data << std::endl;} else {locker.unlock();}}
}int main() {std::thread t1(function_1);std::thread t2(function_2);t1.join();t2.join();getchar();return 0;
}

在生产过程中，因每放入一个数据有1秒延时，所以这个生产的过程是很慢的；在消费过程中，存在着一个while循环，只有在接收到表示结束的数据的时候，才会停止，每次循环内部，都是先加锁，判断队列不空，然后就取出一个数，最后解锁。所以说，在1s内，做了很多无用功！这样的话，CPU占用率会很高，可能达到100%（单核）

实例3：

由于消费者在while循环内因等待数据做了过多的无用功导致CPU占有率过高，我们可以考虑在消费者发现队列为空时，让消费者小睡一会儿，即增加一个小延时（比如500ms），相当于增大了轮询间隔周期，应该能降低CPU的占用率。按该方案修改后的消费者代码如下：

//cond_var1.cpp用互斥锁实现一个生产者消费者模型
#include <iostream>
#include <deque>
#include <thread>
#include <mutex>std::deque<int> q;						//双端队列标准容器全局变量
std::mutex mu;							//互斥锁全局变量//生产者，往队列放入数据
void function_1() {int count = 10;while (count > 0) {std::unique_lock<std::mutex> locker(mu);q.push_front(count);			//数据入队锁保护locker.unlock();std::this_thread::sleep_for(std::chrono::seconds(1));//延时1秒std::cout << "t1 sent a value from t1: " << count << std::endl;count--;}
}
//消费者，从队列提取数据
void function_2() {int data = 0;while (data != 1) {std::unique_lock<std::mutex> locker(mu);if (!q.empty()) {			//判断队列是否为空data = q.back();q.pop_back();			//数据出队锁保护locker.unlock();std::cout << "t2 got a value from t1: " << data << std::endl;}else {locker.unlock();}}
}//消费者，从队列提取数据
void function_22() {int data = 0;while (data != 1) {std::unique_lock<std::mutex> locker(mu);if (!q.empty()) {			//判断队列是否为空data = q.back();q.pop_back();			//数据出队锁保护locker.unlock();std::cout << "t2 got a value from t1: " << data << std::endl;}else {locker.unlock();std::this_thread::sleep_for(std::chrono::milliseconds(500));//延时500毫秒}}
}int main() {std::thread t1(function_1);std::thread t2(function_22);t1.join();t2.join();getchar();return 0;
}

前面也说了，困难之处在于如何确定这个延长时间（即轮询间隔周期），如果间隔太短会过多占用CPU资源，如果间隔太长会因无法及时响应造成延误。

这就引入了条件变量来解决该问题：条件变量使用“通知—唤醒”模型，****生产者生产出一个数据后通知消费者使用，消费者在未接到通知前处于休眠状态节约CPU资源；当消费者收到通知后，赶紧从休眠状态被唤醒来处理数据，使用了事件驱动模型，在保证不误事儿的情况下尽可能减少无用功降低对资源的消耗。

实例4：

由于有很多线程对消息队列进行操作，所以我们需要用锁来保证队列操作的正确。我们当然可以使用一个互斥体，在生产者线程压入消息，或消费者线程弹出消息时使用它，这样就保证了线程同步。但是这里存在一个效率问题，消费者线程只是循环获得锁，然后判断消息队列是否有消息。大多数时候队列可能是没有消息的，这样就比较浪费运算了。

最好的情况是这样的，消费者线程判断队列没有消息后，进入休眠状态，直到有别的线程告诉它有消息了才醒过来，此时消费者继续取消息来处理。条件变量就能满足这样的要求，pthread的条件变量API是这样的：

// 初始化一个条件变量
int pthread_cond_init(pthread_cond_t *restrict cond,const pthread_condattr_t *restrict attr);
// 释放一个条件变量
int pthread_cond_destroy(pthread_cond_t *cond);
// 等待条件满足
int pthread_cond_wait(pthread_cond_t *restrict cond,pthread_mutex_t *restrict mutex);
// 唤醒至少一个等待条件的线程
int pthread_cond_signal(pthread_cond_t *cond);
// 唤醒所有等待条件的线程
int pthread_cond_broadcast(pthread_cond_t *cond);

• 消费者调用pthread_cond_wait等待条件满足，这里的条件满足就是队列中有消息。
• 生产者压入消息后，调用pthread_cond_signal或pthread_cond_broadcast通知消费者。
• 条件变量只是用于通知，但它不是锁，所以生产者或消费者仍然需要用互斥体来保护消息队列，这就是为什么条件变量需要和互斥体一起使用的原因。

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <pthread.h>// 消息结构
struct msg {struct msg *next;int data;       // 消息数据
};struct msg *queue;  // 消息队列
pthread_cond_t qcond = PTHREAD_COND_INITIALIZER;    // 简化初始化条件变量和互斥体
pthread_mutex_t qlock = PTHREAD_MUTEX_INITIALIZER;// 随机数范围[mi, ma]
int randint(int mi, int ma) {double r = (double)rand() * (1.0 / ((double)RAND_MAX + 1.0));r *= (double)(ma - mi) + 1.0;return (int)r + mi;
}// 打印消息
void print_msg(struct msg *m) {printf(">>>>msg: %d\n", m->data);
}// 压入消息
void push_msg(struct msg *m) {pthread_mutex_lock(&qlock);m->next = queue;queue = m;pthread_mutex_unlock(&qlock);// 通知条件满足pthread_cond_signal(&qcond);
}// 生产者线程：
void* product(void *data) {while (1) {usleep(randint(1000*100, 1000*200));struct msg *m = malloc(sizeof(*m));memset(m, 0, sizeof(*m));m->data = randint(0, 1000);push_msg(m);}
}// 弹出消息
struct msg* pop_msg() {struct msg *m;pthread_mutex_lock(&qlock);// 等待条件满足while (queue == NULL) pthread_cond_wait(&qcond, &qlock);m = queue;queue = m->next;pthread_mutex_unlock(&qlock);return m;
}// 消费者线程
void* consum(void *data) {whlie (1) {struct msg *m = pop_msg();print_msg(m);free(m);}
}int main() {
#define PRO_NUM 3
#define CON_NUM 3pthread_t tid_p[PRO_NUM];pthread_t tid_c[CON_NUM];int i;for (i = 0; i < PRO_NUM; ++i) {pthread_create(&tid_p[i], NULL, product, NULL);}for (i = 0; i < CON_NUM; ++i) {pthread_create(&tid_c[i], NULL, consum, NULL);}for (i = 0; i < PRO_NUM; ++i) {pthread_join(tid_p[i], NULL);}for (i = 0; i < CON_NUM; ++i) {pthread_join(tid_c[i], NULL);}return 0;
}

• 程序创建了几个生产者线程和消费者线程，一个条件变量和一个互斥体。

• 生产者不断通过push_msg向queue压入消息，注意这里使用了互斥体(qlock)，因为有多个线程在生产消息和消费消息，所以必须使用qlock保护消息队列。当消息压入完成后，调用pthread_cond_signal通知消费者有消息了。

• 消费者不断通过pop_msg从queue取出消息,它也是通过pthread_mutex_lock先加锁，成功获得锁后，有一个while循环： c while (queue == NULL) pthread_cond_wait(&qcond, &qlock); 这里有两个地方要解释一下：

• • 为什么pthread_cond_wait需要传入qlock？因为前面已经获得了锁，所以在线程进入休眠之前，pthread_cond_wait要先解锁。如果pthread_cond_wait里面不先解锁，该线程进入休眠状态，此时其他的消费者或生产者调用pthread_mutex_lock也进入休眠状态，那么他们永远也等不到解锁的时刻，这时就出现死锁的情况了。接着描述：解锁后线程进入休眠，某个生产者线程成功获得锁，向队列压入消息，然后调用pthread_cond_signal；此时等待的消费者醒过来，它马上又调用pthread_mutex_lock尝试获得锁，获得锁后才可以从消息队列取出消息来处理。这就是pthread_cond_wait做的事情：先解锁，然后休眠，得到通知后醒来再加锁。
  • 为什么要用while循环判断queue是否为空？这是因为消费者有多个，pthread_cond_signal可能会唤醒多个消费者，假如A先获得了锁，从队列取出了消息，然后解锁；B接着获得了锁，但队列已经空了，所以需要用while循环判断，队列不为空才往下处理。

4 如何使用条件变量？

4.1 std::condition_variable

Condition variable作用：一个可以阻塞线程的对象，直至被告知继续。

A condition variable is an object able to block the calling thread until notified to resume.

It uses a unique_lock (over a mutex) to lock the thread when one of its wait functions is called. The thread remains blocked until woken up by another thread that calls a notification function on the same condition_variable object.

当它的一个等待函数被调用时，它使用unique_lock(通过互斥锁)来锁定线程。该线程保持阻塞状态，直到被另一个线程唤醒，该线程调用同一个condition_variable对象上的通知函数。
condition_variable类型的对象总是使用unique_lock来等待:对于任何类型的可锁定类型，请参阅condition_variable_any

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-8GtVvFgq-1609602482383)(C:\Users\guoqi\AppData\Roaming\Typora\typora-user-images\1609595732375.png)]

实例5：

// condition_variable example
#include <iostream>           // std::cout
#include <thread>             // std::thread
#include <mutex>              // std::mutex, std::unique_lock
#include <condition_variable> // std::condition_variablestd::mutex mtx;
std::condition_variable cv;
bool ready = false;void print_id(int id) {std::unique_lock<std::mutex> lck(mtx);while (!ready) cv.wait(lck);// ...std::cout << "thread " << id << '\n';
}void go() {std::unique_lock<std::mutex> lck(mtx);ready = true;cv.notify_all();
}int main()
{std::thread threads[10];// spawn (生产) 10 threads:for (int i = 0; i<10; ++i)threads[i] = std::thread(print_id, i);std::cout << "10 threads ready to race...\n";go();                       // go!for (auto& th : threads) th.join();getchar();return 0;
}

4.2 std::condition_variable_any

Condition variable (any lock)

Same as condition_variable, except that its wait functions can take any lockable type as argument (condition_variable objects can only take unique_lock). Other than that, they are identical.

与condition_variable相同，不同的是它的等待函数可以接受任何可锁定类型作为参数(condition_variable对象只能接受unique_lock)。除此之外，它们是相同的。

实例6：

C++标准库在< condition_variable >中提供了条件变量，借由它，一个线程可以唤醒一个或多个其他等待中的线程。原则上，条件变量的运作如下：

• 你必须同时包含< mutex >和< condition_variable >，并声明一个mutex和一个condition_variable变量；
• 那个通知“条件已满足”的线程（或多个线程之一）必须调用notify_one()或notify_all()，以便条件满足时唤醒处于等待中的一个条件变量；
• 那个等待"条件被满足"的线程必须调用wait()，可以让线程在条件未被满足时陷入休眠状态，当接收到通知时被唤醒去处理相应的任务；

将上面的实例2，实例3程序使用条件变量解决轮询间隔难题的示例代码如下：

//cond_var2.cpp用条件变量解决轮询间隔难题#include <iostream>
#include <deque>
#include <thread>
#include <mutex>
#include <condition_variable>std::deque<int> q;						//双端队列标准容器全局变量
std::mutex mu;							//互斥锁全局变量
std::condition_variable cond;           //全局条件变量//生产者，往队列放入数据
void function_13() {int count = 10;while (count > 0) {std::unique_lock<std::mutex> locker(mu);q.push_front(count);			//数据入队锁保护locker.unlock();cond.notify_one();              // 向一个等待线程发出“条件已满足”的通知std::this_thread::sleep_for(std::chrono::seconds(1));//延时1秒std::cout << "t1 sent a value from t1: " << count << std::endl;count--;}
}
//消费者，从队列提取数据
void function_23() {int data = 0;while (data != 1) {std::unique_lock<std::mutex> locker(mu);while (q.empty())        //判断队列是否为空cond.wait(locker); // 解锁互斥量并陷入休眠以等待通知被唤醒，被唤醒后加锁以保护共享数据data = q.back();q.pop_back();			//数据出队锁保护locker.unlock();std::cout << "t2 got a value from t1: " << data << std::endl;}
}int main() {std::thread t1(function_13);std::thread t2(function_23);t1.join();t2.join();getchar();return 0;
}

使用条件变量对CPU的占用率也很低，而且免去了轮询间隔该设多长的难题！

上面的代码有三个注意事项：

1. 在function_23中，在判断队列是否为空的时候，使用的是while(q.empty())，而不是if(q.empty())，这是因为wait()从阻塞到返回，不一定就是由于notify_one()函数造成的，还有可能由于系统的不确定原因唤醒（可能和条件变量的实现机制有关），这个的时机和频率都是不确定的，被称作伪唤醒。如果在错误的时候被唤醒了，执行后面的语句就会错误，所以需要再次判断队列是否为空，如果还是为空，就继续wait()阻塞；
2. 在管理互斥锁的时候，使用的是std::unique_lock而不是std::lock_guard，而且事实上也不能使用std::lock_guard。这需要先解释下wait()函数所做的事情，可以看到，在wait()函数之前，使用互斥锁保护了，如果wait的时候什么都没做，岂不是一直持有互斥锁？那生产者也会一直卡住，不能够将数据放入队列中了。所以，wait()函数会先调用互斥锁的unlock()函数，然后再将自己睡眠，在被唤醒后，又会继续持有锁，保护后面的队列操作。lock_guard没有lock和unlock接口，而unique_lock提供了，这就是必须使用unique_lock的原因；
3. 使用细粒度锁，尽量减小锁的范围，在notify_one()的时候，不需要处于互斥锁的保护范围内，所以在唤醒条件变量之前可以将锁unlock()。

实例7：

还可以将cond.wait(locker)换一种写法，wait()的第二个参数可以传入一个函数表示检查条件，这里使用lambda函数最为简单，如果这个函数返回的是true，wait()函数不会阻塞会直接返回，如果这个函数返回的是false，wait()函数就会阻塞着等待唤醒，如果被伪唤醒，会继续判断函数返回值。代码示例如下：

#include <iostream>
#include <deque>
#include <thread>
#include <mutex>
#include <condition_variable>std::deque<int> q;						//双端队列标准容器全局变量
std::mutex mu;							//互斥锁全局变量
std::condition_variable cond;           //全局条件变量//生产者，往队列放入数据
void function_14() {int count = 10;while (count > 0) {std::unique_lock<std::mutex> locker(mu);q.push_front(count);			//数据入队锁保护locker.unlock();cond.notify_one();              // 向一个等待线程发出“条件已满足”的通知std::this_thread::sleep_for(std::chrono::seconds(1));//延时1秒std::cout << "t1 sent a value from t1: " << count << std::endl;count--;}
}
//消费者，从队列提取数据
void function_23() {int data = 0;while (data != 1) {std::unique_lock<std::mutex> locker(mu);while (q.empty())        //判断队列是否为空cond.wait(locker); // 解锁互斥量并陷入休眠以等待通知被唤醒，被唤醒后加锁以保护共享数据data = q.back();q.pop_back();			//数据出队锁保护locker.unlock();std::cout << "t2 got a value from t1: " << data << std::endl;}
}
//消费者，从队列提取数据
void function_24() {int data = 0;while (data != 1) {std::unique_lock<std::mutex> locker(mu);//如果条件变量被唤醒，检查队列非空条件是否为真，为真则直接返回，为假则继续等待cond.wait(locker, []() { return !q.empty(); });data = q.back();q.pop_back();			//数据出队锁保护locker.unlock();std::cout << "t2 got a value from t1: " << data << std::endl;}
}int main() {std::thread t1(function_14);std::thread t2(function_24);t1.join();t2.join();getchar();return 0;
}

4.3 std::condition_variable::wait

Wait until notified 等到通知

当前线程(应该已经锁定了lck的互斥锁)的执行会被阻塞，直到收到通知。

在阻塞线程的那一刻，函数自动调用lock .unlock()，允许其他锁定的线程继续。

一旦得到通知(由其他线程显式地发出)，该函数将解除阻塞并调用lck.lock()，使lck处于与函数被调用时相同的状态。然后函数返回(注意最后的互斥锁在返回之前可能会再次阻塞线程)。

通常，函数会被另一个线程中的notify_one或notify_all成员的调用唤醒。但某些实现可能会在不调用任何这些函数的情况下产生虚假的唤醒调用。因此，使用该功能的用户应确保满足其恢复的条件。

如果指定了pred(2)，函数只有在pred返回false时才会阻塞，并且只有当它变为true时，通知才能解除线程阻塞(这对于检查虚假的唤醒调用特别有用)。这个版本(2)的行为就像实现了:

实例8：

while (!pred()) wait(lck);

// condition_variable::wait (with predicate)
#include <iostream>           // std::cout
#include <thread>             // std::thread, std::this_thread::yield
#include <mutex>              // std::mutex, std::unique_lock
#include <condition_variable> // std::condition_variablestd::mutex mtx;
std::condition_variable cv;int cargo = 0;
bool shipment_available() {return cargo!=0;}void consume (int n) {for (int i=0; i<n; ++i) {std::unique_lock<std::mutex> lck(mtx);cv.wait(lck,shipment_available);// consume:std::cout << cargo << '\n';cargo=0;}
}int main ()
{std::thread consumer_thread (consume,10);// produce 10 items when needed:for (int i=0; i<10; ++i) {while (shipment_available()) std::this_thread::yield();std::unique_lock<std::mutex> lck(mtx);cargo = i+1;cv.notify_one();}consumer_thread.join();getchar();	return 0;
}

实例9：

ailable() {return cargo!=0;}void consume (int n) {for (int i=0; i<n; ++i) {std::unique_lock<std::mutex> lck(mtx);cv.wait(lck,shipment_available);// consume:std::cout << cargo << '\n';cargo=0;}
}int main ()
{std::thread consumer_thread (consume,10);// produce 10 items when needed:for (int i=0; i<10; ++i) {while (shipment_available()) std::this_thread::yield();std::unique_lock<std::mutex> lck(mtx);cargo = i+1;cv.notify_one();}consumer_thread.join();getchar();	return 0;
}