caffe代码阅读3：data_reader、internalthread以及blocking_queue的实现细节-2016.3.15
柯翔
（1）data_reader.cpp
首先介绍一下boost::weak_ptr;
弱引用是为了解决shared_ptr在循环引用下的内存释放问题而产生的。
弱引用当引用的对象活着的时候不一定存在。仅仅是当它存在的时候的一个引用。弱引用并不修改该对象的引用计数，这意味这弱引用它并不对对象的内存进行管理，在功能上类似于普通指针，然而一个比较大的区别是，
弱引用能检测到所管理的对象是否已经被释放，从而避免访问非法内存。
由于弱引用不更改引用计数，类似普通指针，只要把循环引用的一方使用弱引用，即可解除循环引用。
1）DataReader类中的变量：
shared_ptr<Body> body_;
static map<const string, boost::weak_ptr<DataReader::Body> > bodies_;
const shared_ptr<QueuePair> queue_pair_;2）此外还有构造函数：
explicit DataReader(const LayerParameter& param);
内联函数：
inline BlockingQueue<Datum*>& free() const {
return queue_pair_->free_;
}
inline BlockingQueue<Datum*>& full() const {
return queue_pair_->full_;
}3）除此之外：
内部还定义了一个Body类，该类是继承于InternalThread
内部还定义了一个QueuePair类，该类有free和full函数，该类用于在body和readers之间进行数据分享
（2）此外该类还涉及到另一个类BlockingQueue，该类位于/util/block_queue.hpp里
1）BlockingQueue类有成员函数
void push(const T& t);
bool try_pop(T* t);
T pop(const string& log_on_wait = "");
bool try_peek(T* t);
T peek();2）此外该类内部还有一个sync的类（该类内部有同步机制和互斥机制）
该类的定义如下：
template<typename T>
class BlockingQueue<T>::sync {
public:
mutable boost::mutex mutex_;
boost::condition_variable condition_;
};
该类内部包含一个mutex_互斥量
还有一个条件变量condition_
3）局部的变量有：
std::queue<T> queue_;
shared_ptr<sync> sync_;
BlockingQueue的push函数的实现如下：
void BlockingQueue<T>::push(const T& t) {
boost::mutex::scoped_lock lock(sync_->mutex_); //关于锁后面会详细讲
queue_.push(t);
lock.unlock();
sync_->condition_.notify_one();
}
首先尝试锁住，然后将数据push到队列（queue_ 是std::queue<T> 类型的），然后unlock，条件变量通知。
BlockingQueue的try_pop函数的实现如下：
template<typename T>
bool BlockingQueue<T>::try_pop(T* t) {
boost::mutex::scoped_lock lock(sync_->mutex_); //
if (queue_.empty()) {
return false;
}
*t = queue_.front();
queue_.pop();
return true;
}这里插播一段关于互斥锁的知识：
上述的代码中：
typedef unique_lock<mutex> scoped_lock;
scoped_lock是unique_lock<mutex>类型，因此通过查看boost的文档知道：
std::unique_lock<std::mutex> is the tool of choice when your locking needs are more complex than a simple lock at the beginning followed unconditionally by an unlock at the end.
也就是说当你的锁需求比简单的情况：一般的应用都是以lock开始，然后最后再unlock这样的情况，但是更复杂的时候你就需要scoped_lock。
参考文档：
为了解释这种锁的必要性，考虑下面的例子：
class A
{
mutable std::mutex mut_;
std::vector<double> data_;
public:
// ...
A& operator=(const A& rhs)
{
if (this != &rhs)
{
std::unique_lock<std::mutex> lhs_lock(mut_);
std::unique_lock<std::mutex> rhs_lock(rhs.mut_); // 死锁
// assign data ...
data_ = rhs.data_;
}
return *this;
}
// ...
};
如果线程1：
A a1();
另一个线程2复制：
A a2=a1;
而原先的线程1此时再赋值：
a1=a2;
这个时候就死锁了。。。碰到这个问题真是无解。。。
不过幸好我们还有解决方法，可以将上述代码写成：
class A
{
mutable std::mutex mut_;
std::vector<double> data_;
public:
// ...
A& operator=(const A& rhs)
{
if (this != &rhs)
{
std::unique_lock<std::mutex> lhs_lock( mut_, std::defer_lock); // 其定义为：struct defer_lock_t {};一个空的标记类而已 通常作为参数传入给 unique_lock 或 lock_guard 的构造函数
std::unique_lock<std::mutex> rhs_lock(rhs.mut_, std::defer_lock);
std::lock(lhs_lock, rhs_lock);
// assign data ...
data_ = rhs.data_;
}
return *this;
}
// ...
};
通过std::lock同时锁住两个，这样就能防止死锁了。
那么为什么新的代码能够避免这个问题：
a）首先lhs_lock和rhs_lock构建的时候是没有锁住的，因为unique_locks并没有引用他们（用了这个参数std::defer_lock ）
b）std::lock(lhs_lock, rhs_lock);同时所住着两个mutex，而不会死锁，这是它的功能
c）这儿不能用lock_guard是因为lock并不拥有所引用的mutex的模式，如果尝试编译safe_guard的话那么就无法编译
总结：也就是说遇到这种循环引用的，要先构建两个不锁的mutex，然后同时上锁（将两个资源上锁）。错误的代码是先锁住其中一个，然后再锁另一个。。。
这里再插播关于条件变量的知识：
条件变量是提供了一种机制，该机制能够等待另一个线程发来的通知，如果另一个线程满足某个条件的话。通常使用条件变量是这样的，一个线程锁住mutex，然后wait，当该线程醒来的时候会检查条件变量的值是否true，如果是则放行，否则继续睡。。。
为了介绍条件变量，给出下面的例子：
boost::condition_variable cond;
boost::mutex mut;bool data_ready;
void process_data();
void wait_for_data_to_process(){
boost::unique_lock<boost::mutex> lock(mut);
while(!data_ready)// lock保护变量data_ready
{
cond.wait(lock);
}
process_data();
}
上述代码的含义是：先定义一个lock，注意，此时是使用的unique_lock，并且mutex是关联上lock，也就是说此时是互斥的，假设处理数据的线程是多个的，然后用条件变量的wait，将线程陷入睡眠
此时另一个线程在准备数据
void retrieve_data();
void prepare_data();
void prepare_data_for_processing(){
retrieve_data();
prepare_data();
{
boost::lock_guard<boost::mutex> lock(mut);
data_ready=true;// lock保护变量data_ready
}
cond.notify_one();
}
当多个准备数据线程坑次坑次把数据搞定后，发送通知，那么原来的线程就醒来开始干活。
接下来继续BlockingQueue的实现代码：
BlockingQueue的pop函数的实现如下：
template<typename T>
T BlockingQueue<T>::pop(const string& log_on_wait) {
boost::mutex::scoped_lock lock(sync_->mutex_); // 锁住
while (queue_.empty()) {
if (!log_on_wait.empty()) {
LOG_EVERY_N(INFO, 1000)<< log_on_wait;
}
sync_->condition_.wait(lock); // 如果队列一直为空则一直在等待
}
T t = queue_.front(); // 否则取出
queue_.pop();
return t;
}
BlockingQueue的try_peek函数的实现如下：
该函数是判断队列首部是不是有数据
template<typename T>
bool BlockingQueue<T>::try_peek(T* t) {
boost::mutex::scoped_lock lock(sync_->mutex_);
if (queue_.empty()) {
return false;
}
*t = queue_.front();
return true;
}
BlockingQueue的peek 函数的实现如下：
该函数取出队列首部的数据，同样也是使用的条件变量来实现同步
template<typename T>
T BlockingQueue<T>::peek() {
boost::mutex::scoped_lock lock(sync_->mutex_);
while (queue_.empty()) {
sync_->condition_.wait(lock);
}
return queue_.front();
}
BlockingQueue的size 函数的实现如下：
template<typename T>
size_t BlockingQueue<T>::size() const {
boost::mutex::scoped_lock lock(sync_->mutex_);
return queue_.size();
}
最后定义了几个类型的BlockingQueue类
template class BlockingQueue<Batch<float>*>;
template class BlockingQueue<Batch<double>*>;
template class BlockingQueue<Datum*>;
template class BlockingQueue<shared_ptr<DataReader::QueuePair> >;
template class BlockingQueue<P2PSync<float>*>;
template class BlockingQueue<P2PSync<double>*>;讲完了BlockingQueue类接下来讲DataReader内部的QueuePair类的实现：
首先甩出定义：
class QueuePair {
public:
explicit QueuePair(int size);
~QueuePair();
BlockingQueue<Datum*> free_;
BlockingQueue<Datum*> full_;
DISABLE_COPY_AND_ASSIGN(QueuePair);
};
从定义里面可以看出定义了两个阻塞队列free_和full_，刚才分析了阻塞队列之后，这次回头看就不懵逼了。
接着看看具体实现：
构造函数做了些啥呢？
就是根据给定的size初始化的若干个Datum（本文最后会给出该数据结构的定义）的实例到free里面。
DataReader::QueuePair::QueuePair(int size) {
// Initialize the free queue with requested number of datums
for (int i = 0; i < size; ++i) {
free_.push(new Datum());
}
}
析构函数做了些啥呢？
就是将full_和free_这两个队列里面的Datum对象全部delete。
DataReader::QueuePair::~QueuePair() {
Datum* datum;
while (free_.try_pop(&datum)) {
delete datum;
}
while (full_.try_pop(&datum)) {
delete datum;
}
}接下来看看Body类的实现，该类是继承自InternalThread 这个类的
class Body : public InternalThread {
public:
explicit Body(const LayerParameter& param);
virtual ~Body();
protected:
void InternalThreadEntry();
void read_one(db::Cursor* cursor, QueuePair* qp);
const LayerParameter param_;
BlockingQueue<shared_ptr<QueuePair> > new_queue_pairs_;
friend class DataReader;
DISABLE_COPY_AND_ASSIGN(Body);
};
Body里面重写了InternalThread内部的InternalThreadEntry函数，此外还添加了read_one函数
Body内部有DataReader的友元，以及BlockingQueue<shared_ptr<QueuePair> > new_queue_pairs_;
为了弄清楚究竟干啥，有必要了解InternalThread这个类究竟干了哪些工作？
InternalThread类实际上就是boost库的thread的封装
首先看看该类的定义是啥：
class InternalThread {
public:
// 构造函数和析构函数
InternalThread() : thread_() {}
virtual ~InternalThread();
/**
* Caffe's thread local state will be initialized using the current
* thread values, e.g. device id, solver index etc. The random seed
* is initialized using caffe_rng_rand.
* caffe的线程局部状态将会使用当前线程值来进行初始化，当前的线程的值有设备id，solver的编号、随机数种子等
*/
void StartInternalThread();
/** Will not return until the internal thread has exited. */
// 是否知道线程退出才返回
void StopInternalThread();
// 线程是否已经起来了
bool is_started() const;
protected:
/* Implement this method in your subclass
with the code you want your thread to run. */
// 定义了一个虚函数，要求继承该类的必须要实现之
virtual void InternalThreadEntry() {}
/* Should be tested when running loops to exit when requested. */
// 在当请求退出的时候应该调用该函数
bool must_stop();
private:
void entry(int device, Caffe::Brew mode, int rand_seed, int solver_count,
bool root_solver);
// 内部的成员变量
shared_ptr<boost::thread> thread_;
};
} // namespace caffe
好了，看完类的定义代码的注释之后。我们来看看具体的实现
namespace caffe {
// 析构函数，调用停止内部线程函数
InternalThread::~InternalThread() {
StopInternalThread();
}
// 测试线程是否起来
bool InternalThread::is_started() const {
return thread_ && thread_->joinable(); // 首先thread_指针不能为空，然后该线程是可等待的（joinable）
}
bool InternalThread::must_stop() {
// if interruption has been requested for the current thread, false otherwise. 见boost的doc
return thread_ && thread_->interruption_requested();
}
// 初始化工作，然后
void InternalThread::StartInternalThread() {
CHECK(!is_started()) << "Threads should persist and not be restarted.";
int device = 0;
#ifndef CPU_ONLY
CUDA_CHECK(cudaGetDevice(&device));
#endif
Caffe::Brew mode = Caffe::mode();
int rand_seed = caffe_rng_rand();
int solver_count = Caffe::solver_count();
bool root_solver = Caffe::root_solver();
try {// 然后重新实例化一个thread对象给thread_指针，该线程的执行的是entry函数
thread_.reset(new boost::thread(&InternalThread::entry, this, device, mode,
rand_seed, solver_count, root_solver));
} catch (std::exception& e) {
LOG(FATAL) << "Thread exception: " << e.what();
}
}
// 线程所要执行的函数
void InternalThread::entry(int device, Caffe::Brew mode, int rand_seed,
int solver_count, bool root_solver) {
#ifndef CPU_ONLY
CUDA_CHECK(cudaSetDevice(device));
#endif
Caffe::set_mode(mode);
Caffe::set_random_seed(rand_seed);
Caffe::set_solver_count(solver_count);
Caffe::set_root_solver(root_solver);
InternalThreadEntry();
}
// 停止线程
void InternalThread::StopInternalThread() {
if (is_started()) {// 如果线程已经开始
thread_->interrupt();// 那么打断
try {
thread_->join();// 等待线程结束
} catch (boost::thread_interrupted&) {//如果被打断，啥也不干，因为是自己要打断的^_^
} catch (std::exception& e) {// 如果发生其他错误则记录到日志
LOG(FATAL) << "Thread exception: " << e.what();
}
}
}
} // namespace caffe
总结一下：无非就是获取线程的状态、启动线程、以及定义的线程入口函数InternalThread::entry ，这个入口函数很有意思，里面调用了虚函数InternalThreadEntry，并且在调用之前，帮用户做好了初始化的工作（随机数种子，CUDA、工作模式及GPU还是CPU、solver的类型）。
好了插播了这么多，咱们回头继续看Body类的情况，
class Body : public InternalThread {
public:
explicit Body(const LayerParameter& param);
virtual ~Body();
protected:
void InternalThreadEntry();
void read_one(db::Cursor* cursor, QueuePair* qp);
const LayerParameter param_;
BlockingQueue<shared_ptr<QueuePair> > new_queue_pairs_;
friend class DataReader;
DISABLE_COPY_AND_ASSIGN(Body);
};
Body类里面果然重写了InternalThread的虚函数InternalThreadEntry。
我们来看看Body的情况
//Body类的构造函数，实际上是给定网络的参数，然后开始启动内部线程
DataReader::Body::Body(const LayerParameter& param)
: param_(param),
new_queue_pairs_() {
StartInternalThread();// 调用InternalThread内部的函数来初始化运行环境以及新建线程去执行虚函数InternalThreadEntry的内容
}
// 析构，停止线程
DataReader::Body::~Body() {
StopInternalThread();
}
// 自己实现的需要执行的函数
// 首先打开数据库，然后设置游标，然后设置QueuePair指针容器
void DataReader::Body::InternalThreadEntry() {
// 获取所给定的数据源的类型来得到DB的指针
shared_ptr<db::DB> db(db::GetDB(param_.data_param().backend()));
// 从网络参数中给定的DB的位置打开DB
db->Open(param_.data_param().source(), db::READ);
// 新建游标指针
shared_ptr<db::Cursor> cursor(db->NewCursor());
// 新建QueuePair指针容器，QueuePair里面包含了free_和full_这两个阻塞队列
vector<shared_ptr<QueuePair> > qps;
try {
// 根据网络参数的阶段来设置solver_count
int solver_count = param_.phase() == TRAIN ? Caffe::solver_count() : 1;
// To ensure deterministic runs, only start running once all solvers
// are ready. But solvers need to peek on one item during initialization,
// so read one item, then wait for the next solver.
for (int i = 0; i < solver_count; ++i) {
shared_ptr<QueuePair> qp(new_queue_pairs_.pop());
read_one(cursor.get(), qp.get());// 读取一个数据
qps.push_back(qp);压入
}
// Main loop
while (!must_stop()) {
for (int i = 0; i < solver_count; ++i) {
read_one(cursor.get(), qps[i].get());
}
// Check no additional readers have been created. This can happen if
// more than one net is trained at a time per process, whether single
// or multi solver. It might also happen if two data layers have same
// name and same source.
CHECK_EQ(new_queue_pairs_.size(), 0);
}
} catch (boost::thread_interrupted&) {
// Interrupted exception is expected on shutdown
}
}// 从数据库中获取一个数据
void DataReader::Body::read_one(db::Cursor* cursor, QueuePair* qp) {
// 从QueuePair中的free_队列pop出一个
Datum* datum = qp->free_.pop();
// TODO deserialize in-place instead of copy?
// 然后解析cursor中的值
datum->ParseFromString(cursor->value());
// 然后压入QueuePair中的full_队列
qp->full_.push(datum);
// go to the next iter
// 游标指向下一个
cursor->Next();
if (!cursor->valid()) {
DLOG(INFO) << "Restarting data prefetching from start.";
cursor->SeekToFirst();// 如果游标指向的位置已经无效了则指向第一个位置
}
}OK接下来我们收拾DataReader类剩下的部分，这里我就偷个懒把DataReader类的所有代码的注释都贴上去。
#include <boost/thread.hpp>
#include <map>
#include <string>
#include <vector>
#include "caffe/common.hpp"
#include "caffe/data_reader.hpp"
#include "caffe/layers/data_layer.hpp"
#include "caffe/proto/caffe.pb.h"
namespace caffe {
// 用于解决share_ptr在循环引用的时候的内存释放
using boost::weak_ptr;
map<const string, weak_ptr<DataReader::Body> > DataReader::bodies_;
static boost::mutex bodies_mutex_;
// 构造函数，传入的是网络的参数、
// 初始化queue_pair_（里面包含两个阻塞队列free_和full_）
DataReader::DataReader(const LayerParameter& param)
: queue_pair_(new QueuePair( //
param.data_param().prefetch() * param.data_param().batch_size())) {
// Get or create a body
// 首先创建或者获取一个body实例
boost::mutex::scoped_lock lock(bodies_mutex_);
string key = source_key(param);// 从网络参数中获取key
weak_ptr<Body>& weak = bodies_[key];// bodies_是存放的string到Body的映射
body_ = weak.lock();
if (!body_) {// 如果bodies是空的
body_.reset(new Body(param));// 则新建Body实例到body_
bodies_[key] = weak_ptr<Body>(body_);// 然后存放到bodies_中去
}
body_->new_queue_pairs_.push(queue_pair_); // 并将queue_pair放入body_中的new_queue_pairs_中去
}
// 析构函数
DataReader::~DataReader() {
string key = source_key(body_->param_);
body_.reset();
boost::mutex::scoped_lock lock(bodies_mutex_);// 上锁
if (bodies_[key].expired()) {
bodies_.erase(key);// map里面的erase
}
}
//
DataReader::QueuePair::QueuePair(int size) {
// Initialize the free queue with requested number of datums
// 一开始全部压入free
for (int i = 0; i < size; ++i) {
free_.push(new Datum());
}
}
// 删除free_和full_内的datum
DataReader::QueuePair::~QueuePair() {
Datum* datum;
while (free_.try_pop(&datum)) {
delete datum;
}
while (full_.try_pop(&datum)) {
delete datum;
}
}
//Body类的构造函数，实际上是给定网络的参数，然后开始启动内部线程
DataReader::Body::Body(const LayerParameter& param)
: param_(param),
new_queue_pairs_() {
StartInternalThread();// 调用InternalThread内部的函数来初始化运行环境以及新建线程去执行虚函数InternalThreadEntry的内容
}
// 析构，停止线程
DataReader::Body::~Body() {
StopInternalThread();
}
// 自己实现的需要执行的函数
// 首先打开数据库，然后设置游标，然后设置QueuePair指针容器
void DataReader::Body::InternalThreadEntry() {
// 获取所给定的数据源的类型来得到DB的指针
shared_ptr<db::DB> db(db::GetDB(param_.data_param().backend()));
// 从网络参数中给定的DB的位置打开DB
db->Open(param_.data_param().source(), db::READ);
// 新建游标指针
shared_ptr<db::Cursor> cursor(db->NewCursor());
// 新建QueuePair指针容器，QueuePair里面包含了free_和full_这两个阻塞队列
vector<shared_ptr<QueuePair> > qps;
try {
// 根据网络参数的阶段来设置solver_count
int solver_count = param_.phase() == TRAIN ? Caffe::solver_count() : 1;
// To ensure deterministic runs, only start running once all solvers
// are ready. But solvers need to peek on one item during initialization,
// so read one item, then wait for the next solver.
for (int i = 0; i < solver_count; ++i) {
shared_ptr<QueuePair> qp(new_queue_pairs_.pop());
read_one(cursor.get(), qp.get());// 读取一个数据
qps.push_back(qp);压入
}
// Main loop
while (!must_stop()) {
for (int i = 0; i < solver_count; ++i) {
read_one(cursor.get(), qps[i].get());
}
// Check no additional readers have been created. This can happen if
// more than one net is trained at a time per process, whether single
// or multi solver. It might also happen if two data layers have same
// name and same source.
CHECK_EQ(new_queue_pairs_.size(), 0);
}
} catch (boost::thread_interrupted&) {
// Interrupted exception is expected on shutdown
}
}
// 从数据库中获取一个数据
void DataReader::Body::read_one(db::Cursor* cursor, QueuePair* qp) {
// 从QueuePair中的free_队列pop出一个
Datum* datum = qp->free_.pop();
// TODO deserialize in-place instead of copy?
// 然后解析cursor中的值
datum->ParseFromString(cursor->value());
// 然后压入QueuePair中的full_队列
qp->full_.push(datum);
// go to the next iter
// 游标指向下一个
cursor->Next();
if (!cursor->valid()) {
DLOG(INFO) << "Restarting data prefetching from start.";
cursor->SeekToFirst();// 如果游标指向的位置已经无效了则指向第一个位置
}
}
} // namespace caffe
总结：实际上该数据层就是调用了封装层的DB来读取数据，此外还简单封装了boost的线程库，然后自己封装了个阻塞队列。
最后还有Datum究竟是哈
可以看caffe.proto文件中的定义
message Datum {
optional int32 channels = 1;
optional int32 height = 2;
optional int32 width = 3;
// the actual image data, in bytes
optional bytes data = 4;
optional int32 label = 5;
// Optionally, the datum could also hold float data.
repeated float float_data = 6;
// If true data contains an encoded image that need to be decoded
optional bool encoded = 7 [default = false];
}
参考：
[1]我猜你有可能需要boost的知识
关于unique_lock
file:///C:/Program%20Files/boost_1_60_0/doc/html/thread/synchronization.html#thread.synchronization.mutex_types.mutex
关于同步机制的(Handling mutexes in C++)
[2]如果你安装了boost的文档，你可以在找到关于线程的知识
file:///C:/Program%20Files/boost_1_60_0/doc/html/thread/thread_management.html#thread.thread_management.this_thread.interruption_requested
[3]关于弱指针的知识
http://baike.baidu.com/link?url=-mb6Lc2iMwP0kzcAyszaJ1gugtcnlSLHeq2UT5SGdVXVgsg_ppDcin4PLTVrfAlsrm4t5focfsS9d9-Z-ZOWBq
类似资料：