Prefetching HDF5DataLayer (currently broken)

include/caffe/data_layers.hpp

@@ -144,16 +144,18 @@ class DummyDataLayer : public Layer<Dtype> {
  * TODO(dox): thorough documentation for Forward and proto params.
  */
 template <typename Dtype>
-class HDF5DataLayer : public Layer<Dtype> {
+class HDF5DataLayer : public BasePrefetchingDataLayer<Dtype> {
  public:
   explicit HDF5DataLayer(const LayerParameter& param)
-      : Layer<Dtype>(param) {}
+      : BasePrefetchingDataLayer<Dtype>(param) {}
   virtual ~HDF5DataLayer();
   virtual void LayerSetUp(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) {}
+  virtual void Reset();
+  virtual void InternalThreadEntry();
 
   virtual inline LayerParameter_LayerType type() const {
     return LayerParameter_LayerType_HDF5_DATA;
@@ -166,16 +168,12 @@ class HDF5DataLayer : public Layer<Dtype> {
       const vector<Blob<Dtype>*>& top);
   virtual void Forward_gpu(const vector<Blob<Dtype>*>& bottom,
       const vector<Blob<Dtype>*>& top);
-  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) {}
-  virtual void LoadHDF5FileData(const char* filename);
+  virtual void FillHDF5FileData();
 
   std::vector<std::string> hdf_filenames_;
   unsigned int num_files_;
   unsigned int current_file_;
-  hsize_t current_row_;
+  int current_row_;
   std::vector<shared_ptr<Blob<Dtype> > > hdf_blobs_;
 };
 

include/caffe/util/io.hpp

@@ -165,15 +165,34 @@ inline cv::Mat DecodeDatumToCVMat(const Datum& datum) {
 void CVMatToDatum(const cv::Mat& cv_img, Datum* datum);
 #endif
 
+/**
+ * @brief Shapes a Blob to read "num" rows of HDF5 data.  If num == -1, take
+ *        the num of the HDF5 dataset.
+ *
+ * @param file_id      the HDF5 file handle
+ * @param dataset_name the name of the HDF5 dataset to read
+ * @param num          the number of rows to read: either num >= 0,
+ *                     or num == -1 for the number of rows in the HDF5 dataset
+ * @param blob         the Blob to shape
+ *
+ * The HDF5 dataset must have 1-4 dimensions. blob will be shaped like the
+ * the HDF5 dataset, except that the HDF5 dataset's first dimension is ignored
+ * and replaced by num, and if the dataset has \@$ D < 4 \@$ dimensions, the
+ * remaining \@$ 4 - D \@$ dimensions are replaced with 1's -- so an
+ * \@$ N \times D \times H \@$ HDF5 dataset will result in a
+ * \@$ \mathrm{num} \times D \times H \times 1 \@$ Blob.
+ */
 template <typename Dtype>
-void hdf5_load_nd_dataset_helper(
-  hid_t file_id, const char* dataset_name_, int min_dim, int max_dim,
-  Blob<Dtype>* blob);
+void HDF5PrepareBlob(hid_t file_id, const char* dataset_name, int num,
+                     Blob<Dtype>* blob);
 
+/**
+ * @brief Reads rows [offset, offset + data->num() - 1] into Blob* data, which
+ *        must have been pre-shaped using HDF5PrepareBlob (or otherwise).
+ */
 template <typename Dtype>
-void hdf5_load_nd_dataset(
-  hid_t file_id, const char* dataset_name_, int min_dim, int max_dim,
-  Blob<Dtype>* blob);
+int HDF5ReadRowsToBlob(hid_t file_id, const char* dataset_name,
+                       int h5_offset, int blob_offset, Blob<Dtype>* blob);
 
 template <typename Dtype>
 void hdf5_save_nd_dataset(

scripts/cpp_lint.py

@@ -1609,6 +1609,7 @@ def CheckCaffeDataLayerSetUp(filename, clean_lines, linenum, error):
   ix = line.find('DataLayer<Dtype>::LayerSetUp')
   if ix >= 0 and (
        line.find('void DataLayer<Dtype>::LayerSetUp') != -1 or
+       line.find('void HDF5DataLayer<Dtype>::LayerSetUp') == -1 and
        line.find('void ImageDataLayer<Dtype>::LayerSetUp') != -1 or
        line.find('void MemoryDataLayer<Dtype>::LayerSetUp') != -1 or
        line.find('void WindowDataLayer<Dtype>::LayerSetUp') != -1):
@@ -1621,6 +1622,7 @@ def CheckCaffeDataLayerSetUp(filename, clean_lines, linenum, error):
   if ix >= 0 and (
        line.find('void Base') == -1 and
        line.find('void DataLayer<Dtype>::DataLayerSetUp') == -1 and
+       line.find('void HDF5DataLayer<Dtype>::DataLayerSetUp') == -1 and
        line.find('void ImageDataLayer<Dtype>::DataLayerSetUp') == -1 and
        line.find('void MemoryDataLayer<Dtype>::DataLayerSetUp') == -1 and
        line.find('void WindowDataLayer<Dtype>::DataLayerSetUp') == -1):

src/caffe/layers/hdf5_data_layer.cpp

@@ -25,34 +25,49 @@ HDF5DataLayer<Dtype>::~HDF5DataLayer<Dtype>() { }
 
 // Load data and label from HDF5 filename into the class property blobs.
 template <typename Dtype>
-void HDF5DataLayer<Dtype>::LoadHDF5FileData(const char* filename) {
-  DLOG(INFO) << "Loading HDF5 file: " << filename;
-  hid_t file_id = H5Fopen(filename, H5F_ACC_RDONLY, H5P_DEFAULT);
-  if (file_id < 0) {
-    LOG(FATAL) << "Failed opening HDF5 file: " << filename;
-  }
-
-  int top_size = this->layer_param_.top_size();
-  hdf_blobs_.resize(top_size);
-
-  const int MIN_DATA_DIM = 1;
-  const int MAX_DATA_DIM = 4;
-
-  for (int i = 0; i < top_size; ++i) {
-    hdf_blobs_[i] = shared_ptr<Blob<Dtype> >(new Blob<Dtype>());
-    hdf5_load_nd_dataset(file_id, this->layer_param_.top(i).c_str(),
-        MIN_DATA_DIM, MAX_DATA_DIM, hdf_blobs_[i].get());
-  }
-
-  herr_t status = H5Fclose(file_id);
-  CHECK_GE(status, 0) << "Failed to close HDF5 file: " << filename;
-
-  // MinTopBlobs==1 guarantees at least one top blob
-  int num = hdf_blobs_[0]->num();
-  for (int i = 1; i < top_size; ++i) {
-    CHECK_EQ(hdf_blobs_[i]->num(), num);
+void HDF5DataLayer<Dtype>::FillHDF5FileData() {
+  int num_rows_filled = 0;
+  while (true) {
+    CHECK_LT(current_file_, hdf_filenames_.size());
+    const char* filename = hdf_filenames_[current_file_].c_str();
+    DLOG(INFO) << "Loading HDF5 file: " << filename;
+    hid_t file_id = H5Fopen(filename, H5F_ACC_RDONLY, H5P_DEFAULT);
+    if (file_id < 0) {
+      LOG(FATAL) << "Failed opening HDF5 file: " << filename;
+    }
+    int rows_read = -1;
+    for (int i = 0; i < hdf_blobs_.size(); ++i) {
+      const int current_rows_read = HDF5ReadRowsToBlob(
+          file_id, this->layer_param_.top(i).c_str(),
+          current_row_, num_rows_filled, hdf_blobs_[i].get());
+      if (rows_read == -1) {
+        CHECK_GE(current_rows_read, 0);
+        rows_read = current_rows_read;
+      }
+      CHECK_EQ(rows_read, current_rows_read);
+    }
+    num_rows_filled += rows_read;
+    CHECK_LE(num_rows_filled, hdf_blobs_[0]->num());
+    herr_t status = H5Fclose(file_id);
+    CHECK_GE(status, 0) << "Failed to close HDF5 file: " << filename;
+    DLOG(INFO) << "Successully loaded " << rows_read << " rows from: "
+               << filename;
+    // If we didn't fill up the blob, should move onto the next file.
+    // If we did fill the blob, we may or may not be at the end.
+    if (num_rows_filled < hdf_blobs_[0]->num()) {
+      if (num_files_ > 1) {
+        ++current_file_;
+        if (current_file_ == num_files_) {
+          current_file_ = 0;
+          DLOG(INFO) << "Looping around to first file.";
+        }
+      }
+      current_row_ = 0;
+    } else {
+      current_row_ += rows_read;
+      break;
+    }
   }
-  DLOG(INFO) << "Successully loaded " << hdf_blobs_[0]->num() << " rows";
 }
 
 template <typename Dtype>
@@ -71,47 +86,56 @@ void HDF5DataLayer<Dtype>::LayerSetUp(const vector<Blob<Dtype>*>& bottom,
   } else {
     LOG(FATAL) << "Failed to open source file: " << source;
   }
+  CHECK_GT(hdf_filenames_.size(), 0)
+      << "Source file must contain at least 1 filename: " << source;
   source_file.close();
   num_files_ = hdf_filenames_.size();
   current_file_ = 0;
   LOG(INFO) << "Number of HDF5 files: " << num_files_;
 
-  // Load the first HDF5 file and initialize the line counter.
-  LoadHDF5FileData(hdf_filenames_[current_file_].c_str());
-  current_row_ = 0;
-
   // Reshape blobs.
   const int batch_size = this->layer_param_.hdf5_data_param().batch_size();
   const int top_size = this->layer_param_.top_size();
+  hdf_blobs_.resize(top_size);
+  hid_t file_id = H5Fopen(hdf_filenames_[0].c_str(), H5F_ACC_RDONLY,
+                          H5P_DEFAULT);
   for (int i = 0; i < top_size; ++i) {
-    top[i]->Reshape(batch_size, hdf_blobs_[i]->channels(),
-                    hdf_blobs_[i]->height(), hdf_blobs_[i]->width());
+    hdf_blobs_[i].reset(new Blob<Dtype>(1, 1, 1, 1));
+    HDF5PrepareBlob(file_id, this->layer_param_.top(i).c_str(), batch_size,
+                    hdf_blobs_[i].get());
+    hdf_blobs_[i]->mutable_cpu_data();
+    top[i]->ReshapeLike(*hdf_blobs_[i]);
   }
+  herr_t status = H5Fclose(file_id);
+  CHECK_GE(status, 0) << "Failed to close HDF5 file: " << hdf_filenames_[0];
+
+  Reset();
+
+  DLOG(INFO) << "Initializing prefetch";
+  this->CreatePrefetchThread();
+  DLOG(INFO) << "Prefetch initialized.";
+}
+
+template <typename Dtype>
+void HDF5DataLayer<Dtype>::Reset() {
+  current_file_ = 0;
+  current_row_ = 0;
+}
+
+template <typename Dtype>
+void HDF5DataLayer<Dtype>::InternalThreadEntry() {
+  FillHDF5FileData();
 }
 
 template <typename Dtype>
 void HDF5DataLayer<Dtype>::Forward_cpu(const vector<Blob<Dtype>*>& bottom,
       const vector<Blob<Dtype>*>& top) {
-  const int batch_size = this->layer_param_.hdf5_data_param().batch_size();
-  for (int i = 0; i < batch_size; ++i, ++current_row_) {
-    if (current_row_ == hdf_blobs_[0]->num()) {
-      if (num_files_ > 1) {
-        ++current_file_;
-        if (current_file_ == num_files_) {
-          current_file_ = 0;
-          DLOG(INFO) << "Looping around to first file.";
-        }
-        LoadHDF5FileData(hdf_filenames_[current_file_].c_str());
-      }
-      current_row_ = 0;
-    }
-    for (int j = 0; j < this->layer_param_.top_size(); ++j) {
-      int data_dim = top[j]->count() / top[j]->num();
-      caffe_copy(data_dim,
-          &hdf_blobs_[j]->cpu_data()[current_row_ * data_dim],
-          &top[j]->mutable_cpu_data()[i * data_dim]);
-    }
+  this->JoinPrefetchThread();
+  for (int i = 0; i < top.size(); ++i) {
+    const int count = top[i]->count();
+    caffe_copy(count, hdf_blobs_[i]->cpu_data(), top[i]->mutable_cpu_data());
   }
+  this->CreatePrefetchThread();
 }
 
 #ifdef CPU_ONLY

src/caffe/layers/hdf5_data_layer.cu

@@ -19,28 +19,14 @@ namespace caffe {
 template <typename Dtype>
 void HDF5DataLayer<Dtype>::Forward_gpu(const vector<Blob<Dtype>*>& bottom,
       const vector<Blob<Dtype>*>& top) {
-  const int batch_size = this->layer_param_.hdf5_data_param().batch_size();
-  for (int i = 0; i < batch_size; ++i, ++current_row_) {
-    if (current_row_ == hdf_blobs_[0]->num()) {
-      if (num_files_ > 1) {
-        current_file_ += 1;
-        if (current_file_ == num_files_) {
-          current_file_ = 0;
-          DLOG(INFO) << "Looping around to first file.";
-        }
-        LoadHDF5FileData(hdf_filenames_[current_file_].c_str());
-      }
-      current_row_ = 0;
-    }
-    for (int j = 0; j < this->layer_param_.top_size(); ++j) {
-      int data_dim = top[j]->count() / top[j]->num();
-      caffe_copy(data_dim,
-          &hdf_blobs_[j]->cpu_data()[current_row_ * data_dim],
-          &top[j]->mutable_gpu_data()[i * data_dim]);
-    }
+  this->JoinPrefetchThread();
+  for (int i = 0; i < top.size(); ++i) {
+    const int count = top[i]->count();
+    caffe_copy(count, hdf_blobs_[i]->gpu_data(), top[i]->mutable_gpu_data());
   }
+  this->CreatePrefetchThread();
 }
 
-INSTANTIATE_LAYER_GPU_FUNCS(HDF5DataLayer);
+INSTANTIATE_LAYER_GPU_FORWARD(HDF5DataLayer);
 
 }  // namespace caffe

src/caffe/test/test_hdf5_output_layer.cpp

@@ -76,10 +76,10 @@ TYPED_TEST(HDF5OutputLayerTest, TestForward) {
                           H5P_DEFAULT);
   ASSERT_GE(file_id, 0)<< "Failed to open HDF5 file" <<
       this->input_file_name_;
-  hdf5_load_nd_dataset(file_id, HDF5_DATA_DATASET_NAME, 0, 4,
-                       this->blob_data_);
-  hdf5_load_nd_dataset(file_id, HDF5_DATA_LABEL_NAME, 0, 4,
-                       this->blob_label_);
+  HDF5PrepareBlob(file_id, HDF5_DATA_DATASET_NAME, -1, this->blob_data_);
+  HDF5ReadRowsToBlob(file_id, HDF5_DATA_DATASET_NAME, 0, 0, this->blob_data_);
+  HDF5PrepareBlob(file_id, HDF5_DATA_LABEL_NAME, -1, this->blob_label_);
+  HDF5ReadRowsToBlob(file_id, HDF5_DATA_LABEL_NAME, 0, 0, this->blob_label_);
   herr_t status = H5Fclose(file_id);
   EXPECT_GE(status, 0)<< "Failed to close HDF5 file " <<
       this->input_file_name_;
@@ -103,13 +103,13 @@ TYPED_TEST(HDF5OutputLayerTest, TestForward) {
           this->input_file_name_;
 
   Blob<Dtype>* blob_data = new Blob<Dtype>();
-  hdf5_load_nd_dataset(file_id, HDF5_DATA_DATASET_NAME, 0, 4,
-                       blob_data);
+  HDF5PrepareBlob(file_id, HDF5_DATA_DATASET_NAME, -1, blob_data);
+  HDF5ReadRowsToBlob(file_id, HDF5_DATA_DATASET_NAME, 0, 0, blob_data);
   this->CheckBlobEqual(*(this->blob_data_), *blob_data);
 
   Blob<Dtype>* blob_label = new Blob<Dtype>();
-  hdf5_load_nd_dataset(file_id, HDF5_DATA_LABEL_NAME, 0, 4,
-                       blob_label);
+  HDF5PrepareBlob(file_id, HDF5_DATA_LABEL_NAME, -1, blob_label);
+  HDF5ReadRowsToBlob(file_id, HDF5_DATA_LABEL_NAME, 0, 0, blob_label);
   this->CheckBlobEqual(*(this->blob_label_), *blob_label);
 
   status = H5Fclose(file_id);

src/caffe/test/test_hdf5data_layer.cpp

@@ -30,8 +30,8 @@ class HDF5DataLayerTest : public MultiDeviceTest<TypeParam> {
 
     // Check out generate_sample_data.py in the same directory.
     filename = new string(
-    CMAKE_SOURCE_DIR "caffe/test/test_data/sample_data_list.txt" CMAKE_EXT);
-    LOG(INFO)<< "Using sample HDF5 data file " << filename;
+        CMAKE_SOURCE_DIR "caffe/test/test_data/sample_data_list.txt" CMAKE_EXT);
+    LOG(INFO) << "Using sample HDF5 data file " << *filename;
   }
 
   virtual ~HDF5DataLayerTest() {