std::async Tutorial & Example

本文将介绍在C++11中如何执行异步任务。

what is std::async()

std::async()是一个函数模版，可以接收一个回调作为参数，并异步执行该函数。

1
2

template <class Fn, class... Args>
future<typename result_of<Fn(Args...)>::type> async (launch policy, Fn&& fn, Args&&... args);

返回值是一个future对象，该对象存着异步执行的函数的返回值。

第一个参数launch policy是一个启动策略，它控制着异步执行的行为，我们可以创建三种异步启动的策略：

• std::launch::async：传递进来的函数在独立的线程中执行；
• std::launch::deferred：只有某个线程对future对象调用get()的时候才会调用函数；
• std::launch::async | std::launch::deferred：默认行为；

我们可以传递以下的回调到std::async中：

• 函数指针；
• 函数对象；
• lambda函数；

假设这样的一个场景，异步调用函数：

1
2
3
4
5
6
7

std::future<std::string> resultFromDB = std::async(std::launch::async, fetchDataFromDB, "Data");
 
// Do Some Stuff 
 
//Fetch Data from DB
// Will block till data is available in future<std::string> object.
std::string dbData = resultFromDB.get();

std::async()的行为如下：

1. 自动创建一个线程或者从线程池中获取一个，还会有创建一个promise对象；
2. 然后将promise对象传递进线程，并返回future对象；
3. 当传递进的参数，即函数退出时，它的返回值会被promise对象给设置进去；

std::future, std::promise and Returning values from Thread

std::future对象常与异步一起使用。本文将主要关注使用std::future与std::promise。

一般情况下，我们希望一个线程能够返回一个结果。假设这样的一个场景，我们的应用创建一个线程去压缩文件夹，然后我们希望得到返回的zip包名和大小。

1. 老的方法：在多个线程之间通过指针共享数据

往新的线程传递进指针，该线程将会设置其中的数据。然后在主线程中继续使用条件变量进行等待，当新线程设置数据并发出条件变量信号时，主线程将被唤醒并从该指针获取数据。

这种方法使用了一个条件变量、一个互斥锁和一个指针。但假如我们希望这个线程在不同的时间点返回三个不同的值，问题就变得复杂了。一个简单的方法是使用std::future。

2. c++11的方法：使用std::future和std::promise

std::future是一个类模版，对象存储的是future value——一个在内部存储中将在未来分配的值，提供了get()访问，但在这个未来值不可用时，调用get()将会阻塞。

std::promise也是一个类模版，它的对象承诺将来会设置值，每个对象都有一个关联的std::future对象。一个std::promise对象与关联的std::future对象共享数据。

使用方法

1. 首先在线程1创建std::promise对象，该对象将传到线程2以便设置值

1

std::promise<int> promiseObj;

2. 在将promiseObj从线程1传递给线程2之前，可以先获得一个关联的future值

1

std::future<int> futureObj = promiseObj.get_future();

3. 将promiseObj传递给线程2
4. 线程1尝试去获取线程2设置的值

1

int val = futureObj.get();

5. 在线程2设置值之前，线程1会阻塞住

1

promiseObj.set_value(45);

注意

如果在设置值之前销毁了std::promise对象，则关联的std::future对象调用get将会抛出异常。

如果希望线程在不同的时间点返回多个值，那么只需要在线程中传递多个std::promise对象。

Condition Variables Explained

Condition Variables

条件变量是一种用于在若干个线程之间发送信号的事件，一个或者多个线程可以等待其他线程发出信号。条件变量在C++11的头文件是：

1

#include <condition_variable>

条件变量往往与锁一起工作。

How things actually work with condition variable,

• 线程1调用条件变量的wait，改变量在内部获取互斥锁并检查是否满足所需条件；
• 如果不满足。则释放锁并等待条件变量发出信号，此时现场阻塞。条件变量的wait函数以原子的方式提供这两个操作；
• 如果条件满足，线程2会向条件变量发出信号；
• 一旦条件变量发出信号，正在等待它的线程1恢复，获取互斥锁，并检查是否实际满足与条件变量相关的条件，或者它是否是上级调用。如果有多个线程在等待，那么notify_one()将只解除阻塞一个线程；
• 如果它是一个上级调用，那么它再次调用wait()函数；

Main member functions for std::condition_variable are

Wait()

该函数使得当前线程阻塞，直到条件变量发出信号或者发生虚假唤醒。

其以原子方式释放附属的的互斥锁，阻塞当前线程，并将其添加到等待当前条件变量对象的线程列表中。当某个线程在相同的条件变量对象上调用notify_one()或notify_all()时，该线程将被解除阻塞。

回调作为参数传递给该函数，回调将被调用以检查它是否属于虚假调用或真的满足条件。

当线程解锁时，wait()函数重新获取互斥锁并检查是否满足实际条件。如果不满足条件，则再次以原子方式释放附加的互斥锁，阻塞当前线程，并将其添加到等待当前条件变量对象的线程列表中。

notify_one()

如果存在线程正在等待相同的条件变量对象，则notify_one将解除其中一个等待线程的阻塞。

notify_all()

如果存在线程正在等待相同的条件变量对象，则notify_all将解除阻塞所有等待的线程。

Joint 3D Face Reconstruction and Dense Alignment with Position Map Regression Network

本文提出了一种直观的方法，可以同时完成3D人脸结构的重建和人脸对齐。作者设计了一种称为UV位置图的2D表示方法，它记录了UV空间中完整面部的3D形状。该方法并不依赖任何的先验脸部模型，而是直接重建完整的人脸几何与语义。

Introduction

在早期，2D基准点的人脸检测通常作为其它一些人脸研究任务的基础条件。随着深度学习的发展，一些任务开始用CNN去估计3D可变模型（3DMM）的系数或3D模型变形函数，以此来从2D面部图像恢复相应的3D信息。然而，由于面部模型或模板定义的3D空间的限制，这些方法的性能受到限制。包括透视投影或3D Thin Plate Spline（TPS）转换在内的所需操作也增加了整个过程的复杂性。

最近有一些基于端到端的方法绕开了模型的限制，实现了很好的性能，但却丢失了点的语义。

本文提出了一种称为位置映射回归网络（PRN）的端到端方法，以预测人脸密集对齐并重建3D人脸形状。具体来说，作者设计了一个UV位置图，它是一个2D图像，记录了完整的面部点云的3D坐标，同时保持每个UV位置的语义。然后，文中训练一个带有加权损失的简单编码器 - 解码器网络，该网络更多地关注重要区域，以从单个2D面部图像回归UV位置图。

Proposed Method

3D Face Representation

本文的目标是从单个2D图像中回归3D人脸几何体以及起对齐信息，因此需要一个可以通过深度网络来进行预测的适当表示。为了解决一些以前的工作留下的弊端，本文提出了UV位置图作为具有人脸对齐信息的3D人脸结构呈现，UV位置图记录了UV空间中所有点的3D位置。

与传统UV坐标不同，本文使用UV空间来存储与2D图像相对应的3D人脸模型点的空间位置，其中\(Pos(u_i, v_i)=(x_i, y_i, z_i)\)，其中\((u_i, v_i)\)表示的是人脸第i个点的UV坐标，因此\((u_i, v_i)\)与\((x_i, y_i)\)表示相同的人脸位置。如下图：

因此，该位置图以其语义含义记录了3D人脸的密集点集，通过使用CNN直接从无约束的3D图像中回归位置图，就能够同时获得3D人脸结构以及密集对齐的结果。

由于要直接从2D图像回归3D完整结构，因此端到端训练需要无约束的2D面部图像及其相应的3D形状。300W-LP是一个大型数据集，包含超过60K的无约束图像和3DMM参数，适合形成训练集。

Network Architecture and Loss Function

论文的网络结构通过将输入的RGB图像转为位置图图像，采用了编码器-解码器的结构来学校变换函数。网络的编码器以一个卷积层开始，后面紧跟10个残差块，这样就可以将图像(256x256x3)转变为特征图(8x8x512)。解码器部分使用了17个转置卷积层，kernel大小为4，并使用ReLU层进行激活。架构如下：

为了学习网络参数，论文使用了MSE作为损失函数，但这里做了一些改动。因为MSE平等对待所有的点，而图像中人脸的中心区域比其它区域具备更多辨别特征，因此作者使用了权重掩码来构造损失函数。如下图所示，权重掩码记录了图中每个点的权重。按照设计目的，论文将点分为四个类别，分别是68个关键点、眼睛鼻子嘴巴、其它脸部区域和脖子区域，而脖子区域因为容易被毛发或者衣服遮盖，所以分配了0权重。

因此损失函数的公式为： \[ Loss = \sum || Pos(u,v) - \overline{Pos}(w,v)|| \cdot W(u,v) \]

Training Details

论文使用了300W-LP的数据集来构造训练集，这是一个包含了不同角度人脸的图像和预测的3DMM系数，从中可以生成3D点云。具体实现上，首先根据ground-truth边框来裁剪图像，然后缩放到256x256大小，然后利用带注释的3DMM参数生成相应的3D位置，并将它们渲染到UV空间中以获得地面实况位置图。

另外，论文还推荐通过在2D图像平面中随机旋转和平移目标面来扰乱训练集。论文使用了Adam优化器，学习率从0.00001开始，并在每5个epochs后衰减一半，batch size设为16。

Need of Event Handling

本文将集中讨论多线程环境下的事件处理机制。有时线程需要等待某事件发生，例如条件变量成为真或由另一个线程完成的任务。

举个例子，应用处理以下的任务：

1. 与服务器握手；
2. 从文件读取数据；
3. 处理读出来的数据；

在上面的任务中，任务1是独立的，而任务3则依赖于任务2。因此我们可以将该应用拆分为2个线程去处理：

创建一个默认值为false的布尔全局变量。在线程2中将其值设置为true，并且线程1在循环中检查其值，一旦变为真，线程1将继续处理数据。但由于它是两个线程共享的全局变量，因此需要与互斥锁同步。

不过这种方式有以下的缺点：

线程需要持续争取锁，而这是为了检查全局变量，这种方式会消耗CPU使得线程1变慢。

比较好的方式是使用信号量来解决这个问题。

Using mutex to fix Race Conditions

本文主要讨论如何使用锁，以此保护多线程共享的数据，避免race conditions

std::mutex

c++11的线程库中，锁的使用在头文件<mutex>里。mutex有两个重要的方法：

• lock()
• unlock

接着上一篇文章，我们可以对wallet的变量进行锁保护，如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

#include<iostream>
#include<thread>
#include<vector>
#include<mutex>
 
class Wallet
{
	int mMoney;
	std::mutex mutex;
public:
	Wallet() :mMoney(0){}
    int getMoney()   { 	return mMoney; }
    void addMoney(int money)
    {
      mutex.lock();
        for(int i = 0; i < money; ++i)
        {
          	mMoney++;
        }
      mutex.unlock();
    }
};

在这种情况下，可能遇到的一个问题是，一个线程获取到锁之后，可能在释放锁之前就退出了。这种情况可能发生在上锁后出现异常退出。为了避免这种情况，我们应该使用std::lock_guard。

std::lock_guard

std::lock_guard是一个类模版，它实现了互斥锁的RAII。通过将锁包装在其对象中，并把互斥锁添加到构造函数中。当调用析构函数时，它就会释放互斥锁。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

	
class Wallet
{
	int mMoney;
	std::mutex mutex;
public:
    Wallet() :mMoney(0){}
    int getMoney()   { 	return mMoney; }
    void addMoney(int money)
    {
      std::lock_guard<std::mutex> lockGuard(mutex);
      // In constructor it locks the mutex

        for(int i = 0; i < money; ++i)
        {
            // If some exception occurs at this
            // poin then destructor of lockGuard
            // will be called due to stack unwinding.
            //
            mMoney++;
        }
      // Once function exits, then destructor
      // of lockGuard Object will be called.
      // In destructor it unlocks the mutex.
    }
 };

Sharing Data & Race Conditions

多线程编程的一个经典问题就是Race Condition。

What is a Race Condition?

race condition是多线程编程的一个经典bug，它指的是当多个线程并行执行某些操作时，它们访问公共的内存区域。此时有若干个线程回去修改这篇内存区域的的数据。

A Practical example of Race Condition

假设有这样的一个钱包类，我们提供一个成员函数addMoney()：

1
2
3
4
5
6
7
8
9
10
11
12
13
14

class Wallet
{
    int mMoney;
public:
    Wallet() :mMoney(0){}
    int getMoney() { return mMoney; }
    void addMoney(int money)
    {
       for(int i = 0; i < money; ++i)
       {
          mMoney++;
       }
    }
};

接着我们创建5个线程，共享这个钱包类的对象，并行地添加1000到哪步的money变量，预期结果钱包类的结果应该是5000。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

int testMultithreadedWallet()
{
   Wallet walletObject;
   std::vector<std::thread> threads;
   for(int i = 0; i < 5; ++i){
        threads.push_back(std::thread(&Wallet::addMoney, &walletObject, 1000));
   }
 
   for(int i = 0; i < threads.size() ; i++)
   {
       threads.at(i).join();
   }
   return walletObject.getMoney();
}
 
int main()
{
 
  int val = 0;
  for(int k = 0; k < 1000; k++)
  {
     if((val = testMultithreadedWallet()) != 5000)
     {
       std::cout << "Error at count = "<<k<<" Money in Wallet = "<<val << std::endl;
     }
  }
  return 0;
}

但由于并发时，mMoney++并不是原子操作，而是分为三个机器指令：

• 从存储加载mMoney到寄存器；
• 递增寄存器的值；
• 把寄存器的值更新回存储；

以下是出现的几种非预期结果：

1
2
3
4
5
6

Error at count = 14 Money in Wallet = 4000
Error at count = 44 Money in Wallet = 4112
Error at count = 52 Money in Wallet = 4387
Error at count = 65 Money in Wallet = 4904
Error at count = 81 Money in Wallet = 4907
Error at count = 98 Money in Wallet = 4666

Passing Arguments to Threads

要想传递参数给线程相关的回调对象或者函数，可以通过在std::thread的构造器中传递，默认情况下，所有的参数都会被拷贝进线程的内部存储。

Passing simple arguments to a std::thread in C++11

1
2
3
4
5
6
7
8
9

void threadCallback(int x, std::string str){ ... }
int main()  
{
    int x = 10;
    std::string str = "Sample String";
    std::thread threadObj(threadCallback, x, str);
    threadObj.join();
    return 0;
}

How not to pass arguments to threads in C++11

不要将变量的地址从本地栈传递给线程的回调函数，因为线程1中的局部变量可能已经不在作用域，但线程2访问了它的非法地址。

1
2
3
4
5
6
7

void newThreadCallback(int * p) { ... }
void startNewThread()
{
    int i = 10;
    std::thread t(newThreadCallback,&i);
    t.detach();
}

同样要注意的是，不要把指向位于堆上内存的指针传递进线程，因为有可能在新线程访问之前，原线程已经删除了该内存。

1
2
3
4
5
6
7
8
9
10

void newThreadCallback(int * p) { ... }
void startNewThread()
{
    int * p = new int();
    *p = 10;
    std::thread t(newThreadCallback,p);
    t.detach();
  	delete p;
    p = NULL;
}

How to pass references to std::thread in C++11

如果是传递引用，参数会被拷贝进线程栈：

1
2
3
4
5
6
7

void threadCallback(int const & x) {} //change x
int main()
{
    std::thread threadObj(threadCallback, x);
    threadObj.join();
    return 0;
}

这种情况下，线程内部对于x的改动，外部域是看不到的。因为线程函数threadCallback中的x引用了在新线程堆栈中复制的临时值。

如果想要外部也是可视的，可以使用std::ref。

1
2
3
4
5
6
7

void threadCallback(int const & x) {} //hange x
int main()
{
    std::thread threadObj(threadCallback, std::ref(x));
    threadObj.join();
    return 0;
}

Assigning pointer to member function of a class as thread function

将指向类成员函数的指针传递给回调函数，并将指针作为第二个参数传递给object：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

#include <iostream>
#include <thread>
class DummyClass {
public:
    DummyClass()
    {}
    DummyClass(const DummyClass & obj)
    {}
    void sampleMemberFunction(int x)
    {
        std::cout<<"Inside sampleMemberFunction "<<x<<std::endl;
    }
};
int main() {
 
    DummyClass dummyObj;
    int x = 10;
    std::thread threadObj(&DummyClass::sampleMemberFunction,&dummyObj, x);
    threadObj.join();
    return 0;
}

C++11 Multithreading – Part 2: Joining and Detaching Threads

Joining Threads with std::thread::join()

一个线程启动后，另一个线程可以等待该线程完成。这个特别的需求可以用join来实现：

1
2

std::thread th(funcPtr);
th.join();

Detaching Threads using std::thread::detach()

被分离的线程也叫守护(daemon)/后台(background)线程，这个需求可以调用std::function来完成：

1
2

std::thread th(funcPtr);
th.detach();

Be careful with calling detach() and join() on Thread Handles

Case 1: Never call join() or detach() on std::thread object with no associated executing thread std::thread threadObj( (WorkerThread()) ); threadObj.join(); threadObj.join(); // It will cause Program to Terminate

比如在一个没有相关线程的线程上执行join或者detach都会导致程序终止：

1
2
3
4
5
6
7
8

std::thread threadObj( (WorkerThread()) );
threadObj.join();
threadObj.join(); // It will cause Program to Terminate

	
std::thread threadObj( (WorkerThread()) );
threadObj.detach();
threadObj.detach(); // It will cause Program to Terminate

为了避免这个问题，每次调用join()或者detach()之前都应该检查线程状态：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

std::thread threadObj( (WorkerThread()) );
if(threadObj.joinable())
{
    std::cout<<"Detaching Thread "<<std::endl;
    threadObj.detach();
}
if(threadObj.joinable())    
{
    std::cout<<"Detaching Thread "<<std::endl;
    threadObj.detach();
}

std::thread threadObj2( (WorkerThread()) );
if(threadObj2.joinable())
{
    std::cout<<"Joining Thread "<<std::endl;
    threadObj2.join();
}
if(threadObj2.joinable())    
{
    std::cout<<"Joining Thread "<<std::endl;
    threadObj2.join();
}

Case 2 : Never forget to call either join or detach on a std::thread object with associated executing thread

如果一个线程还有相关联的在执行的线程，但却没有对此执行join()或者detach()，那么在destructor中会终止该程序。因为destructor会检查线程是否joinable。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

#include <iostream>
#include <thread>
#include <algorithm>
class WorkerThread
{
public:
    void operator()()     
    {
        std::cout<<"Worker Thread "<<std::endl;
    }
};
int main()  
{
    std::thread threadObj( (WorkerThread()) );
    // Program will terminate as we have't called either join or detach with the std::thread object.
    // Hence std::thread's object destructor will terminate the program
    return 0;
}

C++11 Multithreading – Part 1 : Three Different ways to Create Threads

Thread Creation in C++11

每个C++应用中都存在一个默认的主线程——main()函数。在C++11中，我们可以通过创建std::thread的对象类创建额外的线程。头文件为：

1

#include <thread>

What std::thread accepts in constructor ?

我们可以为std::thread对象附上一个callback，在线程开始时执行，这些callbacks可以是：

1. Function Pointer
2. Function Objects
3. Lambda functions

调用方式如下：

1

std::thread theObj(<CALLBACK>)

新的线程会在对象创建出来之后开始运行，并且会并行地执行传递进来的回调。此外，任何线程都可以通过调用该线程对象上的join()函数来等待另一个线程退出。

让我们来看看三种不同的回调机制：

1
2
3
4
5
6

void thread_function(){...}

int main()
{
  std::thread threadObj(thread_function);
}

1
2
3
4
5
6
7
8
9
10

class DisplayThread
{
public:
    void operator()(){...}
};

int main()
{
  std::thread threadObj(DisplayThread());
}

1

std::thread threadObj([]{...});

Differentiating between threads

每个std::thread对象都有一个关联的ID。

以下成员函数给出了关联的线程对象ID：

1

std::thread::get_id()

要得到当前线程的ID，可以通过：

1

std::this_thread::get_id()