C++11 并发与多线程（一）

std::thread 类

1.1， 什么叫并发 concurrency?

一遍走路一边说话；你打球我游泳

单核计算机上的并发是个假象，其实只是任务切换（task switching）需要上下文切换

多处理器或一个处理器上有多个核上的并发才是自然的并发，叫硬件并发



并发种类：

1，多进程并发

这些进程间可通过正常的进程通信渠道（信号，套接字，文件，管道等）

缺点：1；通信建立较复杂或者慢；2；操作系统需要花时间启动进程和管理进程资源等

优点：1；更容易写安全的并发代码比如Erlang语言 2；可以运行在不同的机器上



2，多线程并发

线程像轻量级的进程；每个线程互相独立，一个进程的所有线程分享同一个地址空间，多数数据可以共享，比如全局变量，指针和引用可以在线程间传递；

优点：共享地址空间和没有数据保护使得使用多线程程序比多进程程序代价小

The shared address space and lack of protection of data between threads makes the overhead associated with using multi- ple threads much smaller than that from using multiple pro- cesses



1.2 为什么使用并发

1，分离关注点

考虑一个桌面DVD应用

one thread can handle the user interface and another the DVD playback；

该应用线程的个数不受当前核数影响，而是根据设计而非纯粹提高吞吐量

2，性能

1，任务并行： 把任务划分为几部分并行执行 ；任务之间很可能有依赖

2，数据并行： 每个线程完成不同数据部分的相同操作

易并行算法 embarrassingly parallel、naturally parallel、 conveniently concurrent；扩展性好

nice scalability, cpu 核越多越好，视频处理，不同部分的图像可以并行处理

1.3 什么时候不使用并发

一般使用并发的代码更难理解，尤其是有额外复杂性的多线程代码将会导致更多的bug. 除非能带来性能提升或分离关注点，否则不要使用多线程

每个线程都会消耗stack 1M左右 比如32位系统， 总共4G，最多支持4096个线程，没有多余空间作为静态数据和堆数据了，64位系统也会只有有限的其它资源；通过线程池可以限制线程的数量（比如服务端程序限制连接数目）

1.4 C98

Posix C pthread lib 不支持跨平台

Boost and ACE 支持跨平台

C++11 支持 新的线程意识的内存模型；管理线程；保护共享数据；线程同步；低级原子操作

许多命名和结构从boost过来

支持原子操作让程序员写出更高效的跨平台代码

1.5 Hello world

#include <iostream>
#include <thread>

void hello()
{
std::cout<<"Hello Concurrent World\n";
}

int main()
{
std::thread t(hello);
t.join();
}

g++ -o test.o ch1.1.cpp -pthread -std=c++11

2. 管理线程

一个程序至少有一个线程即运行main()的线程，也可以运行其它线程（以其它函数作为入口点）；如果使用的是std::thread 对象，可以等它完成。

函数对象作为开始

class background_task
{
public:
void operator()() const
{
do_something();
do_something_else();
}
};
background_task f;
std::thread my_thread(f);

std::thread my_thread(background_task()); //wrong

std::thread my_thread((background_task())); //right

std::thread my_thread{background_task()}; //right

lamda

std::thread my_thread([]{

do_something();

do_something_else();

});

线程访问局部变量（指针或引用）的同时函数返回；应避免

#include <thread>
#include <iostream>

void do_something(int& i)
{
++i;
if(i%100==0)
std::cout<<"cur:"<<i<<std::endl;
}

struct func
{
int& i;

func(int& i_):i(i_){}

void operator()()
{
for(unsigned j=0;j<1000000;++j)
{
do_something(i);   //潜在的访问局部变量的引用
}
}
};

void oops()
{
int some_local_state=0;
func my_func(some_local_state);
std::thread my_thread(my_func);
my_thread.detach();   //没有等待线程结束
}

int main()
{
oops();
}

解决办法： 1，拷贝数据 2 join 保证函数返回前线程执行完毕 线程对象执行 join() eg: my_thread.join()

join 是简单粗暴的，要等线程完成或不完成，若要更好的控制需要条件变量与未来，join会清除线程的存储，所以线程只能join一次，joinable() 一直返回false

在销毁一个线程对象之前需要确保调用了detach() or join() //生命周期问题

异常发生时的处理

需要确保join在异常发生时被调用

void f()
{
int some_local_state=0;
func my_func(some_local_state);
std::thread t(my_func);
try
{
do_something_in_current_thread();
}
catch(...)
{
t.join();   //do_something_in_current_thread 异常发生后的清理工作
throw;
}
t.join();
}

一个更简单，精确的方法是使用RAII 资源获取即初始化时刻

#include <thread>

class thread_guard
{
std::thread& t;
public:
explicit thread_guard(std::thread& t_):
t(t_)
{}
~thread_guard()
{
if(t.joinable())    //一个线程只能被join一次
{
t.join();      //放在析构函数里自动做
}
}
thread_guard(thread_guard const&)=delete;    //不需要拷贝构造
thread_guard& operator=(thread_guard const&)=delete; //不需要赋值
};

void do_something(int& i)
{
++i;
}

struct func
{
int& i;

func(int& i_):i(i_){}

void operator()()
{
for(unsigned j=0;j<1000000;++j)
{
do_something(i);
}
}
};

void do_something_in_current_thread()
{}

void f()
{
int some_local_state;
func my_func(some_local_state);
std::thread t(my_func);
thread_guard g(t);    //封装thread 对象

do_something_in_current_thread();
}

int main()
{
f();
}

后台运行线程：

又叫守护线程，比如检测文件系统，一般生命周期和应用程序差不多

std::thread t(do_background_work);
t.detach();
assert(!t.joinable());

如何给线程函数传递参数？

void f(int i,std::string const& s);

std::thread t(f,3,std::string(buffer));

void update_data_for_widget(widget_id w,widget_data& data);

std::thread t(update_data_for_widget,w,std::ref(data)); //引用类型

//成员函数

class X

{

public:

void do_lengthy_work();

};

X my_x;

std::thread t(&X::do_lengthy_work,&my_x);

有些参数不是拷贝而是移动

void process_big_object(std::unique_ptr);

std::unique_ptr p(new big_object);

p->prepare_data(42);

std::thread t(process_big_object,std::move(p));

产生很多线程等待他们完成

#include <vector>
#include <thread>
#include <algorithm>
#include <functional>

void do_work(unsigned id)
{}

void f()
{
std::vector<std::thread> threads;
for(unsigned i=0;i<20;++i)
{
threads.push_back(std::thread(do_work,i));
}
std::for_each(threads.begin(),threads.end(),
std::mem_fn(&std::thread::join));
}

int main()
{
f();
}

//转移线程的控制权

#include <thread>
#include <utility>

class scoped_thread
{
std::thread t;
public:
explicit scoped_thread(std::thread t_):
t(std::move(t_))
{
if(!t.joinable())
throw std::logic_error("No thread");
}
~scoped_thread()
{
t.join();
}
scoped_thread(scoped_thread const&)=delete;
scoped_thread& operator=(scoped_thread const&)=delete;
};

void do_something(int& i)
{
++i;
}

struct func
{
int& i;

func(int& i_):i(i_){}

void operator()()
{
for(unsigned j=0;j<1000000;++j)
{
do_something(i);
}
}
};

void do_something_in_current_thread()
{}

void f()
{
int some_local_state;
scoped_thread t(std::thread(func(some_local_state)));

do_something_in_current_thread();
}

int main()
{
f();
}

运行时选择线程数目

#include <thread>
#include <numeric>
#include <algorithm>
#include <functional>
#include <vector>
#include <iostream>

template<typename Iterator,typename T>
struct accumulate_block
{
void operator()(Iterator first,Iterator last,T& result)
{
result=std::accumulate(first,last,result);
}
};

template<typename Iterator,typename T>
T parallel_accumulate(Iterator first,Iterator last,T init)
{
unsigned long const length=std::distance(first,last);

if(!length)
return init;

unsigned long const min_per_thread=25;
unsigned long const max_threads=
(length+min_per_thread-1)/min_per_thread;

unsigned long const hardware_threads=
std::thread::hardware_concurrency();

unsigned long const num_threads=
std::min(hardware_threads!=0?hardware_threads:2,max_threads);

unsigned long const block_size=length/num_threads;

std::vector<T> results(num_threads);
std::vector<std::thread>  threads(num_threads-1);

Iterator block_start=first;
for(unsigned long i=0;i<(num_threads-1);++i)
{
Iterator block_end=block_start;
std::advance(block_end,block_size);
threads[i]=std::thread(
accumulate_block<Iterator,T>(),
block_start,block_end,std::ref(results[i]));
block_start=block_end;
}
accumulate_block<Iterator,T>()(block_start,last,results[num_threads-1]);

std::for_each(threads.begin(),threads.end(),
std::mem_fn(&std::thread::join));

return std::accumulate(results.begin(),results.end(),init);
}

int main()
{
std::vector<int> vi;
for(int i=0;i<10;++i)
{
vi.push_back(10);
}
int sum=parallel_accumulate(vi.begin(),vi.end(),5);
std::cout<<"sum="<<sum<<std::endl;
}

如何区分线程？

线程标识符id可以通过thread::get_id()获得，若thread obejct没有和任何线程关联则返回一个NULL的std::thread::id表示没有任何线程。当前线程若想获得自己的id可以调用std::this_thread::get_id()。

thread::id对象可以被任意复制和比较。这里的比较语义是：若相等表示是同一个线程或者都没有线程，不等表示不同的线程。

bool operator== (thread::id lhs, thread::id rhs) noexcept;

bool operator!= (thread::id lhs, thread::id rhs) noexcept;

bool operator< (thread::id lhs, thread::id rhs) noexcept;

bool operator>= (thread::id lhs, thread::id rhs) noexcept;

bool operator> (thread::id lhs, thread::id rhs) noexcept;

bool operator>= (thread::id lhs, thread::id rhs) noexcept

thread::id可以用于关联容器的key，可以用于排序，用于比较等用途。比如std::hash

主线程在启动子线程之前记录下自己的master_thread，然后每个子线程启动时都去比较这个ID，若不是则执行do_common_work()，主线程则执行do_master_thread_work()，这样就可以将主线程和子线程的工作统一到一个函数中，但是主线程和子线程的工作又不一样。

// thread::get_id / this_thread::get_id
#include <iostream>       // std::cout
#include <thread>         // std::thread, std::thread::id, std::this_thread::get_id
#include <chrono>         // std::chrono::seconds

std::thread::id main_thread_id = std::this_thread::get_id();

void is_main_thread() {
if ( main_thread_id == std::this_thread::get_id() )
std::cout << "This is the main thread.\n";
else
std::cout << "This is not the main thread.\n";
}

int main()
{
is_main_thread();
std::thread th (is_main_thread);
th.join();
}

Output:

This is the main thread.

This is not the main thread.

thread::id可以作为关联容器的key，关联容器中可以根据key来存放线程的私有数据。

输出线程标识符std::cout<

std::thread::joinable 使用方法

bool joinable() const noexcept;

Check if joinable

Returns whether the thread object is joinable.

线程对象代表一个执行线程的时候是joinable的

线程对象何时不能 joinable ？

1，如何线程对象是缺省构造的

2，如果线程对象已经被移走（构造了另一个或者赋值给另一个线程对象）

3，已经调用了join 或 detach 成员函数

// example for thread::joinable
#include <iostream>       // std::cout
#include <thread>         // std::thread

void mythread()
{
// do stuff...
}

int main()
{
std::thread foo;
std::thread bar(mythread);

std::cout << "Joinable after construction:\n" << std::boolalpha;
std::cout << "foo: " << foo.joinable() << '\n';
std::cout << "bar: " << bar.joinable() << '\n';

if (foo.joinable()) foo.join();
if (bar.joinable()) bar.join();

std::cout << "Joinable after joining:\n" << std::boolalpha;
std::cout << "foo: " << foo.joinable() << '\n';
std::cout << "bar: " << bar.joinable() << '\n';

return 0;
}

Joinable after construction:

foo: false

bar: true

Joinable after joining:

foo: false

bar: false

std::thread::detach 使用方法

void detach();

Detach thread

Detaches the thread represented by the object from the calling thread, allowing them to execute independently from each other.

Both threads continue without blocking nor synchronizing in any way. Note that when either one ends execution, its resources are released.

After a call to this function, the thread object becomes non-joinable and can be destroyed safely.

#include <iostream>       // std::cout
#include <thread>         // std::thread, std::this_thread::sleep_for
#include <chrono>         // std::chrono::seconds

void pause_thread(int n)
{
std::this_thread::sleep_for (std::chrono::seconds(n));
std::cout << "pause of " << n << " seconds ended\n";
}

int main()
{
std::cout << "Spawning and detaching 3 threads...\n";
std::thread (pause_thread,1).detach();
std::thread (pause_thread,2).detach();
std::thread (pause_thread,3).detach();
std::cout << "Done spawning threads.\n";

std::cout << "(the main thread will now pause for 5 seconds)\n";
// give the detached threads time to finish (but not guaranteed!):
pause_thread(5);
return 0;
}

Spawning and detaching 3 threads…

Done spawning threads.

(the main thread will now pause for 5 seconds)

pause of 1 seconds ended

pause of 2 seconds ended

pause of 3 seconds ended

pause of 5 seconds ended