本文转自:https://www.cnblogs.com/neopenx/p/5200519.html
用户线程与内核线程
广义上线程分为用户线程和内核线程。
前者已经绝迹,它一般只存在于早期不支持多线程的系统中。
它用模拟的方式实现一个模拟的多线程系统,不支持异步。
即,一个线程被阻塞了,其它线程也会被阻塞。
当今的操作系统几乎都默认提供了内核线程API,底层由操作系统实现。
内核线程的好处在于,它们之间支持异步,是"真"多线程。
操作系统的流氓软件
不过,内核线程也给线程的使用带来了操作系统捆绑性。
不同操作系统平台,其内核线程的实现与提供的API不同,给跨平台带来麻烦。
比如在Windows上,MFC就是封装了Windows内核线程。
在Linux上,广泛使用的pThread就是POSIX系列系统流传下来的内核线程。
第三方跨平台内核线程库
有幸的是,历史上有许多跨平台的项目库。
我最早知道是Qt,GTK,这俩个比较特殊,因为它们是Application Framework。
是在90年代左右,C++为了对抗Java等后期开发之秀,而专门写成的跨平台C++库。
主要以GUI为作战武器,对抗Java。
Boost库同样提供了优秀了内核线程库,还是跨平台的。
所以Caffe移植到Windows,是不需要改动线程系统的。
何以线程用?
生产者与消费者
生产者与消费者是一个经典的资源分配问题。
它的核心要点主要体现在两方面:
①阻塞
②临界
首先看①:
为什么需要阻塞?因为生产者比较快,消费者比较慢。
一次消费过程,包括整个正向传播和反向传播,这需要不少的时间。
而一次生产过程,就是对一个Batch数据的预缓冲,这不需要很多时间。
生产者总不能一直生产下去,然后爆掉缓冲区吧?
所以,生产者在检测到缓冲区满了之后,就要进入阻塞状态。
那么问题来了,如果我们不用多线程,将阻塞代码放在主进程中执行,会怎么样?
读取,阻塞,前向传播失败,反向传播不可能,死锁。
这是为什么I/O代码需要多线程处理的根本原因。
再来看②:
为什么需要临界?因为生产者和消费者之间必然是异步的,换言之,存在于不同线程。
证明如下:
生产者有一定概率被阻塞,假设生产者和消费者在同一个线程实体中,那么必然触发死锁。
又因为系统执行必然不应该有死锁,所以假设不成立。
结论:生产者和消费者之间必然是异步的。
异步临界资源问题,则需要加锁(Mutex)来维持资源的一致性,具体在第陆章中详解。
破除因果律
多线程程序设计的核心原则就是:将非因果连续的代码,并行化。
也就是说,只要代码前后不是上下文相关的,都能够并行执行。
那么Caffe的I/O模型中,有哪几处不是上下文相关的?
答案有二:
①Datum和Blob(Batch)不是上下文相关的。
Blob包含着正向传播的shape信息,这些信息只有初始化网络在初始化时才能确定。
而Datum则只是与输入样本有关。
所以,Datum的读取工作可以在网络未初始化之前就开始,这就是DataReader采用线程设计的内涵。
同时,这种不相关性,也为生产者和消费者对于临界资源访问的设计埋下伏笔。
②GPU之间不是上下文相关的。
Caffe的多GPU方案,是让不同GPU覆盖不同段的数据,最后不在网络结构上做融合。
这点和AlexNet略有不同(AlexNet两个GPU的网络结构最后交叉了)
这样的多GPU方案,使得每个GPU至少存在一个DataReader,覆盖不一样的数据段。
在网络结构上,通过共享root网络即可,如图所示:
上图是一个经典的多GPU流水线编程方案。3个GPU拥有各自的DataReader,但是共享所有Layer。
GPU0由主进程控制,GPU1由线程1控制,GPU2由线程2控制。
Caffe在主机端,也就是CPU主进程和次线程,每个Layer的前向传播被一个mutex锁住,而反向传播却没有。
这样,尽管主进程、线程1、线程2是并行调用Layer.Foward(),但不能同时访问同一Layer,此时Layer为互斥临界资源。
这种行为会构造出一个人工的流水线,比如:
GPU0在Conv1时,GPU1、GPU2会被锁住。
GPU0在Conv3时,Conv1和Conv2是空闲的,会被其它GPU占用。
反向传播之所以不锁,是因为前向传播和反向传播是符合因果律的,前向传播成流水线,
反向传播自然也是流水线,非常优美的设计。
影分身之术
俗话说,一个好汉三个帮。
本篇所述的多线程,均指的是CPU多线程。
实际上,由于GPGPU的异构计算引入,CPU线程基本都在做一些后勤工作。
主要是数据的读取、与GPU显存的数据交换。
CPU主进程负责全程调度GPU执行计算代码,在这点上,CPU利用率并不高。
而从线程对数据的预缓冲任务也不是很艰难。
所以,相对于计算密集型CPU多线程设计而言,Caffe的多线程任务相对轻松。
我们很难将CPU的利用率榨到100%,在这点上,为深度学习Online应用系统埋下伏笔。
试想一下,在后台构建一个基于Socket的深度学习应用服务器,同时CPU并发线程可达几千,
我觉得只有这样,才能真正榨干CPU计算力。至于Caffe的训练,其实对CPU的要求不是很高。
代码实战
Boost线程的创建
使用boost::thread, 需要#include "boost/thread/thread.hpp"
与Qt、MFC等Application Framework提供的线程库不同,
boost::thread的封装性比较强,一般不建议继承和改写boost::thread类(没见过有人这么用)
为了线程能够执行自定义代码,需要在其构造时,传入执行函数的函数指针。
函数指针分为两类:
①普通函数指针
②类成员函数指针
boost::function结合bind函数提供了一个函数指针的载体。(style1)
也可以直接将函数指针的构造方式传入thread。(style2)
建议配合boost::shared_ptr使用。(style3)
若是普通函数指针,可用:
// style 1 void helloworld(int a,string b); boost::function<void()> f=bind(helloworld,1,"helloworld"); boost::thread(f); // style 2 boost::thread(helloworld,1,"helloworld") // style 3(Caffe style) boost::shared_ptr<boost::thread> thread; thread.reset(new boost::thread(helloworld,1,"helloworld"));
当然,为了系统的开发,我们显然需要一个封装类,如将boost::thread封装为DragonThread类。
即,将boost::shared_ptr<boost::thread> thread作为类成员。
基于类的函数指针绑定需要传入类this指针,写法做如下更改:
class DragonThread{ void helloworld(int a,string b); }; // style 1 boost::function<void()> f=bind(&DragonThread::helloworld,this,1,"helloworld"); boost::thread(f); // style 2 boost::thread(&DragonThread::helloworld,this,1,"helloworld"); // style 3(Caffe style) boost::shared_ptr<boost::thread> thread; thread.reset(new boost::thread(&DragonThread::helloworld,this,1,"helloworld"));
Boost线程的生与死
boost::thread一旦被构造后,就会立刻以异步的方式执行传入的函数。
在debug线程的过程中,注意boost::thread将晚于主进程的代码的执行。
如果不在main函数中循环等待,很有可能boost::thread还没有执行,main函数已经退出了。
想要立刻终结一个boost线程是不可能的,一些强大的Application Framework的线程库
通常会提供thread.terminate(),来立刻终结线程的执行(比如Qt),但是boost没有提供。
因为这种方式是相当不安全的,在Java设计模式中,更鼓励开发者让线程函数自动检测终结条件而退出。
这种检测函数在Caffe里是must_stop()函数,它使用了boost::thread提供的中断点检测功能。
bool DragonThread::must_stop(){ return boost::this_thread::interruption_requested(); }
注意,中断请求的检测,只能在异步线程执行函数中执行,主进程从外部调用是无效的。
主进程可以从外部触发interrupt操作,通知正在异步执行的线程,该方法封装为stopThread函数:
void DragonThread::stopThread(){ if (is_start()){ thread->interrupt(); } try{thread->join();} catch (boost::thread_interrupted&) {} catch (std::exception& e){ LOG(FATAL) << "Thread exception: " << e.what(); } }
有时候,interrupt的线程可能处于阻塞睡眠状态,我们需要从外部立即唤醒它,让其检测中断请求。
所以在interrupt操作后,需要立即后接join操作。最后,还可以选择性地补上异常检测。
数据结构
建立dragon_thread.hpp。
class DragonThread { public: DragonThread() {} virtual ~DragonThread(); void initializeThread(int device, Dragon::Mode mode, int rand_seed, int solver_count, bool root_solver); void startThread(); void stopThread(); //the interface implements for specific working task virtual void interfaceKernel() {} bool is_start(); bool must_stop(); boost::shared_ptr<thread> thread; };
在第叁章,我们提到了全局管理器是线程独立的,因此每一个dragon线程,
需要从主管理器转移一些参数,包括(GPU设备、计算模式、随机种子、root_solver&solver_count)
成员函数包括:
boost::thread的传入函数initializeThread,这个函数里最后又嵌套了interfaceKernel。
前者负责转移参数,后者默认是一个空函数,你也可以写成纯虚函数。
由于boost::thread没有继承的用法,所以Caffe二度封装以后,提供了这种用法。
所有继承DragonThread的类,只要重载interfaceKernel()这个虚函数就行了。
startThread应该最先被执行,它包括获取主进程管理器参数,以及构造thread。
由于thread构造结束,就会立刻执行,所以startThread不负其名,就是启动了线程。
stopThread的功能如上所述。
实现
建立dragon_thread.cpp。
首先是thread的传入函数initializeThread:
void DragonThread::initializeThread(int device, Dragon::Mode mode, int rand_seed, int solver_count, bool root_solver){ #ifndef CPU_ONLY CUDA_CHECK(cudaSetDevice(device)); #endif Dragon::set_random_seed(rand_seed); Dragon::set_mode(mode); Dragon::set_solver_count(solver_count); Dragon::set_root_solver(root_solver); interfaceKernel(); //do nothing }
然后是外部调用的startThread函数:
void DragonThread::startThread(){ CHECK(!is_start()); int device = 0; #ifndef CPU_ONLY CUDA_CHECK(cudaGetDevice(&device)); #endif Dragon::Mode mode = Dragon::get_mode(); unsigned int seed = Dragon::get_random_value(); int solver_count = Dragon::get_solver_count(); bool root_solver = Dragon::get_root_solver(); try{ thread.reset(new boost::thread(&DragonThread::initializeThread, this, device, mode, seed, solver_count, root_solver)); } catch (std::exception& e){ LOG(FATAL) << "Thread exception: " << e.what(); } }
由于该函数是在主进程中执行,Dragon::get()与initializeThread里的Dragon::set()
操作的其实不是同一个全局管理器,所以需要这样麻烦的转移参数过程。
最后是线程控制与析构:
void DragonThread::stopThread(){ if (is_start()){ thread->interrupt(); } try{thread->join();} catch (boost::thread_interrupted&) {} catch (std::exception& e){ LOG(FATAL) << "Thread exception: " << e.what(); } } bool DragonThread::is_start(){ return thread&&thread->joinable(); } bool DragonThread::must_stop(){ return boost::this_thread::interruption_requested(); } DragonThread::~DragonThread(){ stopThread(); }
完整代码
dragon_thread.hpp:
https://github.com/neopenx/Dragon/blob/master/Dragon/include/dragon_thread.hpp
dragon_thread.cpp:
https://github.com/neopenx/Dragon/blob/master/Dragon/src/dragon_thread.cpp
本站文章如无特殊说明,均为本站原创,如若转载,请注明出处:从零开始山寨Caffe·肆:线程系统 - Python技术站