【撸码caffe 五】数据层搭建

caffe.cpp中的train函数内声明了一个类型为Solver类的智能指针solver：

// Train / Finetune a model.
int train() {
……
  shared_ptr<caffe::Solver<float> >
      solver(caffe::SolverRegistry<float>::CreateSolver(solver_param));
……  
}

之后调用Solver类的构造函数，在构造函数内执行了 Init（param）函数：

template <typename Dtype>
Solver<Dtype>::Solver(const SolverParameter& param, const Solver* root_solver)
    : net_(), callbacks_(), root_solver_(root_solver),
      requested_early_exit_(false) {
  Init(param);
}

param是一个SolverParameter类对象，SolverParameter类继承自google的protobuf类，在类内定义了网络模型的参数和对网络的各种操作。

在Init函数里，又分别执行了一个InitTrainNet和InitTestNet函数，功能分别是构建训练网络和测试网络：

template <typename Dtype>
void Solver<Dtype>::Init(const SolverParameter& param) {
  ……
  InitTrainNet();
  if (Caffe::root_solver()) {
    InitTestNets();
    LOG(INFO) << "Solver scaffolding done.";
  }
  ……
}

InitTrainNet函数里执行了一些检查工作，接着判断是否是root_solver，之后在net_.reset函数的入参里，以net_param为参数实例化了一个Net类对象：

template <typename Dtype>
void Solver<Dtype>::InitTrainNet() {
  ……
  if (Caffe::root_solver()) {
    net_.reset(new Net<Dtype>(net_param));
  } else {
    net_.reset(new Net<Dtype>(net_param, root_solver_->net_.get()));
  }
}

在Net的构造函数里，执行了Net类的Init函数，这个Init函数完成了网络模型各个层的构建工作：

template <typename Dtype>
Net<Dtype>::Net(const NetParameter& param, const Net* root_net)
    : root_net_(root_net) {
  Init(param);
}

param.layer_size()函数获取到传入的param模型的网络层数，通过for循环，逐个构建网络的每个层，在Lenet的训练网络中，一共有9层：

template <typename Dtype>
void Net<Dtype>::Init(const NetParameter& in_param) {
……
for (int layer_id = 0; layer_id < param.layer_size(); ++layer_id) {
……
  layers_[layer_id]->SetUp(bottom_vecs_[layer_id], top_vecs_[layer_id]);
……
}
}

SetUp是在layer.hpp中定义的，用于构建网络层，修改输出数据维度，以及设置损失权重：

void SetUp(const vector<Blob<Dtype>*>& bottom,
      const vector<Blob<Dtype>*>& top) {
    InitMutex();
    CheckBlobCounts(bottom, top);
	//配置网络模型的每一层
    LayerSetUp(bottom, top);
	//修改输出数据的维度
    Reshape(bottom, top);
	//设置损失权重
    SetLossWeights(top);
  }

数据层是网络模型的最底层，用于把数据封装成blob送入到网络中执行训练，也是SetUp里LaverSetUp第一个配置的网络层，lenet_train_test.prototxt中定义的训练网络的数据层：

layer {
  name: "mnist"
  type: "Data"
  top: "data"
  top: "label"
  include {
    phase: TRAIN
  }
  transform_param {
    scale: 0.00390625
  }
  data_param {
    source: "D:/Software/Caffe/caffe-master/examples/mnist/mnist_train_lmdb"
    batch_size: 64
    backend: LMDB
  }
}

具体的数据层构建是在base_data_layer.cpp和data_layer.cpp中完成的。

base_data_layer.hpp：

#ifndef CAFFE_DATA_LAYERS_HPP_
#define CAFFE_DATA_LAYERS_HPP_

#include <vector>

#include "caffe/blob.hpp"
#include "caffe/data_transformer.hpp"
#include "caffe/internal_thread.hpp"
#include "caffe/layer.hpp"
#include "caffe/proto/caffe.pb.h"
#include "caffe/util/blocking_queue.hpp"

namespace caffe {

/**
 * @brief Provides base for data layers that feed blobs to the Net.
 *
 * TODO(dox): thorough documentation for Forward and proto params.
 */
template <typename Dtype>
//BaseDataLayer 继承自Layer类
class BaseDataLayer : public Layer<Dtype> {
 public:
	 //LayerParameter类型的参数param是传入的网络模型
  explicit BaseDataLayer(const LayerParameter& param);
  // LayerSetUp: implements common data layer setup functionality, and calls
  // DataLayerSetUp to do special data layer setup for individual layer types.
  // This method may not be overridden except by the BasePrefetchingDataLayer.
  virtual void LayerSetUp(const vector<Blob<Dtype>*>& bottom,
      const vector<Blob<Dtype>*>& top);
  //数据层可以并行solvers共享
  // Data layers should be shared by multiple solvers in parallel
  virtual inline bool ShareInParallel() const { return true; }
  //数据层设置
  virtual void DataLayerSetUp(const vector<Blob<Dtype>*>& bottom,
      const vector<Blob<Dtype>*>& top) {}
  //数据层没有更底层，所有不涉及维度变换
  // Data layers have no bottoms, so reshaping is trivial.
  virtual void Reshape(const vector<Blob<Dtype>*>& bottom,
      const vector<Blob<Dtype>*>& top) {}

  //cpu与gpu上的后向传播
  virtual void Backward_cpu(const vector<Blob<Dtype>*>& top,
      const vector<bool>& propagate_down, const vector<Blob<Dtype>*>& bottom) {}
  virtual void Backward_gpu(const vector<Blob<Dtype>*>& top,
      const vector<bool>& propagate_down, const vector<Blob<Dtype>*>& bottom) {}

 protected:
  TransformationParameter transform_param_;
  shared_ptr<DataTransformer<Dtype> > data_transformer_;
  bool output_labels_;     //label标签
};

//Batch类包含数据和标签数据
template <typename Dtype>
class Batch {
 public:
  Blob<Dtype> data_, label_;
};

template <typename Dtype>
class BasePrefetchingDataLayer :
    public BaseDataLayer<Dtype>, public InternalThread {
 public:
  explicit BasePrefetchingDataLayer(const LayerParameter& param);
  // LayerSetUp: implements common data layer setup functionality, and calls
  // DataLayerSetUp to do special data layer setup for individual layer types.
  // This method may not be overridden.
  void LayerSetUp(const vector<Blob<Dtype>*>& bottom,
      const vector<Blob<Dtype>*>& top);

  //数据层的前向传播
  virtual void Forward_cpu(const vector<Blob<Dtype>*>& bottom,
      const vector<Blob<Dtype>*>& top);
  virtual void Forward_gpu(const vector<Blob<Dtype>*>& bottom,
      const vector<Blob<Dtype>*>& top);

  //GPU预先读取的batches组
  // Prefetches batches (asynchronously if to GPU memory)
  static const int PREFETCH_COUNT = 3;

 protected:
  virtual void InternalThreadEntry();
  //加载batch
  virtual void load_batch(Batch<Dtype>* batch) = 0;

  //batch数值，包含PREFETCH_COUNT个batch数据组
  Batch<Dtype> prefetch_[PREFETCH_COUNT];
  BlockingQueue<Batch<Dtype>*> prefetch_free_;
  BlockingQueue<Batch<Dtype>*> prefetch_full_;

  Blob<Dtype> transformed_data_;
};

}  // namespace caffe

#endif  // CAFFE_DATA_LAYERS_HPP_

base_data_layer.cpp：

#include <boost/thread.hpp>
#include <vector>

#include "caffe/blob.hpp"
#include "caffe/data_transformer.hpp"
#include "caffe/internal_thread.hpp"
#include "caffe/layer.hpp"
#include "caffe/layers/base_data_layer.hpp"
#include "caffe/proto/caffe.pb.h"
#include "caffe/util/blocking_queue.hpp"

namespace caffe {

template <typename Dtype>
//BaseDataLayer 类继承自Layer类
BaseDataLayer<Dtype>::BaseDataLayer(const LayerParameter& param)
    : Layer<Dtype>(param),
      transform_param_(param.transform_param()) {
}

//数据层设置
template <typename Dtype>
void BaseDataLayer<Dtype>::LayerSetUp(const vector<Blob<Dtype>*>& bottom,
      const vector<Blob<Dtype>*>& top) {
  if (top.size() == 1) {   //判断数据是否带label标签
    output_labels_ = false;
  } else {
    output_labels_ = true;
  }
  //数据预处理
  data_transformer_.reset(
      new DataTransformer<Dtype>(transform_param_, this->phase_));
  //生成随机数种子
  data_transformer_->InitRand();
  // The subclasses should setup the size of bottom and top
  DataLayerSetUp(bottom, top);  //数据层设置
}

template <typename Dtype>
BasePrefetchingDataLayer<Dtype>::BasePrefetchingDataLayer(
    const LayerParameter& param)
    : BaseDataLayer<Dtype>(param),
      prefetch_free_(), prefetch_full_() {
  for (int i = 0; i < PREFETCH_COUNT; ++i) {
    prefetch_free_.push(&prefetch_[i]);
  }
}

template <typename Dtype>
void BasePrefetchingDataLayer<Dtype>::LayerSetUp(
    const vector<Blob<Dtype>*>& bottom, const vector<Blob<Dtype>*>& top) {
  BaseDataLayer<Dtype>::LayerSetUp(bottom, top);
  // Before starting the prefetch thread, we make cpu_data and gpu_data
  // calls so that the prefetch thread does not accidentally make simultaneous
  // cudaMalloc calls when the main thread is running. In some GPUs this
  // seems to cause failures if we do not so.
  for (int i = 0; i < PREFETCH_COUNT; ++i) {
    prefetch_[i].data_.mutable_cpu_data();
    if (this->output_labels_) {
      prefetch_[i].label_.mutable_cpu_data();
    }
  }
#ifndef CPU_ONLY
  if (Caffe::mode() == Caffe::GPU) {
    for (int i = 0; i < PREFETCH_COUNT; ++i) {
		prefetch_[i].data_.mutable_gpu_data();   //依次给队列中每个batch的数据blob分配cpu内存
      if (this->output_labels_) {
        prefetch_[i].label_.mutable_gpu_data(); //依次给队列中每个batch的标签blob分配cpu内存
      }
    }
  }
#endif
  DLOG(INFO) << "Initializing prefetch";  //初始化预取数据
  this->data_transformer_->InitRand();   //随机数种子，每次随机取
  StartInternalThread();   //启动读取数据线程
  DLOG(INFO) << "Prefetch initialized.";  //预取数据初始化完成
}

template <typename Dtype>
void BasePrefetchingDataLayer<Dtype>::InternalThreadEntry() {
#ifndef CPU_ONLY
  cudaStream_t stream;
  if (Caffe::mode() == Caffe::GPU) {
    CUDA_CHECK(cudaStreamCreateWithFlags(&stream, cudaStreamNonBlocking));
  }
#endif

  try {
    while (!must_stop()) {
      Batch<Dtype>* batch = prefetch_free_.pop();
      load_batch(batch);
#ifndef CPU_ONLY
      if (Caffe::mode() == Caffe::GPU) {
        batch->data_.data().get()->async_gpu_push(stream);
        CUDA_CHECK(cudaStreamSynchronize(stream));
      }
#endif
      prefetch_full_.push(batch);
    }
  } catch (boost::thread_interrupted&) {
    // Interrupted exception is expected on shutdown
  }
#ifndef CPU_ONLY
  if (Caffe::mode() == Caffe::GPU) {
    CUDA_CHECK(cudaStreamDestroy(stream));
  }
#endif
}

// 将预处理过的batch，送到top
// 数据层的forward函数不进行计算，不使用bottom，只是准备数据，填充到top
template <typename Dtype>
void BasePrefetchingDataLayer<Dtype>::Forward_cpu(
    const vector<Blob<Dtype>*>& bottom, const vector<Blob<Dtype>*>& top) {
  Batch<Dtype>* batch = prefetch_full_.pop("Data layer prefetch queue empty");
  // Reshape to loaded data.
  //调整数据维度，一次读取一个batch大小的数据
  top[0]->ReshapeLike(batch->data_);
  // Copy the data
  caffe_copy(batch->data_.count(), batch->data_.cpu_data(),
             top[0]->mutable_cpu_data());  //拷贝数据到输出中
  DLOG(INFO) << "Prefetch copied";
  if (this->output_labels_) {
    // Reshape to loaded labels.
    top[1]->ReshapeLike(batch->label_);
    // Copy the labels.
    caffe_copy(batch->label_.count(), batch->label_.cpu_data(),
        top[1]->mutable_cpu_data());   //拷贝标签到输出中
  }

  prefetch_free_.push(batch);
}

#ifdef CPU_ONLY
STUB_GPU_FORWARD(BasePrefetchingDataLayer, Forward);
#endif

INSTANTIATE_CLASS(BaseDataLayer);
INSTANTIATE_CLASS(BasePrefetchingDataLayer);

}  // namespace caffe

data_layer.cpp：

template <typename Dtype>  
DataLayer<Dtype>::DataLayer(const LayerParameter& param)  
  : BasePrefetchingDataLayer<Dtype>(param),  
    reader_(param) {  
}  
  
  
template <typename Dtype>  
DataLayer<Dtype>::~DataLayer() {  
  this->StopInternalThread();  
}  
  
  
//主要工作是：Reshape top blob 和 prefetch得到的batch的data_ blob、label_ blob  
template <typename Dtype>  
void DataLayer<Dtype>::DataLayerSetUp(const vector<Blob<Dtype>*>& bottom,  
      const vector<Blob<Dtype>*>& top) {  
  const int batch_size = this->layer_param_.data_param().batch_size();  
  // Read a data point, and use it to initialize the top blob.  
  Datum& datum = *(reader_.full().peek());  
  
  
  // Use data_transformer to infer the expected blob shape from datum.  
  vector<int> top_shape = this->data_transformer_->InferBlobShape(datum);  
  this->transformed_data_.Reshape(top_shape);//transformed_data_只是存储一张图片的数据，所以'0'维度依旧保持默认值'1'  
  // Reshape top[0] and prefetch_data according to the batch_size.  
  top_shape[0] = batch_size;//InferBlobShape(datum)返回的top_shape[0]为1  
  top[0]->Reshape(top_shape);  
  for (int i = 0; i < this->PREFETCH_COUNT; ++i) {  
    this->prefetch_[i].data_.Reshape(top_shape);  
  }  
  LOG(INFO) << "output data size: " << top[0]->num() << ","  
      << top[0]->channels() << "," << top[0]->height() << ","  
      << top[0]->width();  
  // label  
  if (this->output_labels_) {  
    vector<int> label_shape(1, batch_size);  
    top[1]->Reshape(label_shape);  
    for (int i = 0; i < this->PREFETCH_COUNT; ++i) {  
      this->prefetch_[i].label_.Reshape(label_shape);  
    }  
  }  
}  
  
  
// This function is called on prefetch thread  
// 经过load_batch后，batch所指的数据显然发生了变化——> 虽然是以&(this->transformed_data_作为实参传递给Transform但是该地址与batch的data_ blob中每张图片的地址是相吻合的。  
// load_batch(Batch<Dtype>* batch)方法Reshape了其中的data_ Blob,并且更新了数据成员transformed_data_。  
// 因为Batch<Dtype>* batch仅仅是个指针，对其Reshape已经为这个Blob分配了所需要的内存，做到这一点已经足够了，毕竟prefetch_free_成员里存储的也只是指针。  
template<typename Dtype>  
void DataLayer<Dtype>::load_batch(Batch<Dtype>* batch) {  
  CPUTimer batch_timer;  
  batch_timer.Start();  
  double read_time = 0;  
  double trans_time = 0;  
  CPUTimer timer;  
  //返回count_。count_表示Blob存储的元素个数（shape_所有元素乘积）. 如果是默认构造函数构造Blob，count_ capacity_为0。  
  //但是，经过Datalayer::DataLayerSetup函数的调用后，btach中data_/label_ blob都已经Reshape了,所以cout_,capacity_就不再为0了。  
  CHECK(batch->data_.count());  
  CHECK(this->transformed_data_.count());  
  
  
  // Reshape according to the first datum of each batch  
  // on single input batches allows for inputs of varying dimension.  
  const int batch_size = this->layer_param_.data_param().batch_size();  
  Datum& datum = *(reader_.full().peek());  
  // Use data_transformer to infer the expected blob shape from datum.  
  vector<int> top_shape = this->data_transformer_->InferBlobShape(datum);//从reader_中获取一个datum来猜测top_shape。  
  this->transformed_data_.Reshape(top_shape);  
  // Reshape batch according to the batch_size.  
  top_shape[0] = batch_size;  
  batch->data_.Reshape(top_shape);//reshape data_ blob的大小  
  
  
  Dtype* top_data = batch->data_.mutable_cpu_data();  
  Dtype* top_label = NULL;  // suppress warnings about uninitialized variables  
  
  
  if (this->output_labels_) {  
    top_label = batch->label_.mutable_cpu_data();  
  }  
  for (int item_id = 0; item_id < batch_size; ++item_id) {  
    timer.Start();  
    // get a datum  
    Datum& datum = *(reader_.full().pop("Waiting for data"));//从reader_获取一张图片的Datum.  
    read_time += timer.MicroSeconds();  
    timer.Start();  
    // Apply data transformations (mirror, scale, crop...)  
    int offset = batch->data_.offset(item_id);//获取一张图片的offset,然后transform  
    //设置this->transformed_data_这个Blob的data_成员所指向的SyncedMemory类型对象的CPU内存指针cpu_ptr_设置为"top_data + offset"。  
    this->transformed_data_.set_cpu_data(top_data + offset);//简言之，将cpu_ptr定位到batch的data_ blob的"top_data + offset"位置处，使其指向当前即将要处理的一张图片，其实真实的过程是拷贝datum中的数据（或经过处理）至this->transformed_data_所指处。通过for循环，处理每张图片，从而更新transformed_data_。  
    this->data_transformer_->Transform(datum, &(this->transformed_data_));//调用后，this->transformed_data_所指向的内存会发生变化，即经过变换后的数据。如此更新数据成员transformed_data_，该成员是BasePrefetchingDataLayer类及其子类的数据成员  
    // Copy label.  
    if (this->output_labels_) {  
      top_label[item_id] = datum.label();  
    }  
    trans_time += timer.MicroSeconds();  
  
  
    reader_.free().push(const_cast<Datum*>(&datum));  
  }  
  timer.Stop();  
  batch_timer.Stop();  
  DLOG(INFO) << "Prefetch batch: " << batch_timer.MilliSeconds() << " ms.";  
  DLOG(INFO) << "     Read time: " << read_time / 1000 << " ms.";  
  DLOG(INFO) << "Transform time: " << trans_time / 1000 << " ms.";  
}