Merge pull request #1000 from kloudkl/boost-thread

Replace pthread with boost::thread

Makefile

@@ -161,7 +161,7 @@ endif
 LIBRARIES += \
 	glog gflags pthread protobuf leveldb snappy \
 	lmdb \
-	boost_system \
+	boost_system boost_thread \
 	hdf5_hl hdf5 \
 	opencv_core opencv_highgui opencv_imgproc
 PYTHON_LIBRARIES := boost_python python2.7

include/caffe/internal_thread.hpp

@@ -1,42 +1,36 @@
 #ifndef CAFFE_INTERNAL_THREAD_HPP_
 #define CAFFE_INTERNAL_THREAD_HPP_
 
-#include <pthread.h>
+#include <boost/thread.hpp>
+
+#include "caffe/common.hpp"
 
 namespace caffe {
 
 /**
- * Virutal class encapsulate pthread for use in base class
- * The child class will acquire the ability to run a single pthread,
+ * Virutal class encapsulate boost::thread for use in base class
+ * The child class will acquire the ability to run a single thread,
  * by reimplementing the virutal function InternalThreadEntry.
  */
 class InternalThread {
  public:
-  InternalThread() {}
-  virtual ~InternalThread() {}
+  InternalThread() : thread_(NULL) {}
+  virtual ~InternalThread();
 
   /** Returns true if the thread was successfully started. **/
-  bool StartInternalThread() {
-    return (pthread_create(&_thread, NULL, InternalThreadEntryFunc, this) == 0);
-  }
+  bool StartInternalThread();
 
   /** Will not return until the internal thread has exited. */
-  bool WaitForInternalThreadToExit() {
-    return pthread_join(_thread, NULL);
-  }
+  bool WaitForInternalThreadToExit();
+
+  bool is_started() const { return thread_ != NULL && thread_->joinable(); }
 
  protected:
   /* Implement this method in your subclass
       with the code you want your thread to run. */
-  virtual void InternalThreadEntry() = 0;
-
- private:
-  static void * InternalThreadEntryFunc(void * This) {
-    reinterpret_cast<InternalThread *>(This)->InternalThreadEntry();
-    return NULL;
-  }
+  virtual void InternalThreadEntry() {}
 
-  pthread_t _thread;
+  boost::thread* thread_;
 };
 
 }  // namespace caffe

scripts/travis/travis_install.sh

@@ -6,7 +6,7 @@ sudo apt-get install \
     wget git curl \
     python-dev python-numpy \
     libleveldb-dev libsnappy-dev libopencv-dev \
-    libboost-dev libboost-system-dev libboost-python-dev \
+    libboost-dev libboost-system-dev libboost-python-dev libboost-thread-dev \
     libprotobuf-dev protobuf-compiler \
     libatlas-dev libatlas-base-dev \
     libhdf5-serial-dev \

src/caffe/CMakeLists.txt

@@ -57,7 +57,7 @@ find_package(LMDB REQUIRED)
 include_directories(${LMDB_INCLUDE_DIR})
 
 #    Boost
-find_package(Boost 1.46 COMPONENTS system REQUIRED)
+find_package(Boost 1.46 COMPONENTS system thread REQUIRED)
 include_directories( ${Boost_INCLUDE_DIR} )
 link_directories( ${Boost_LIBRARY_DIRS} )
 
@@ -99,14 +99,15 @@ if(NOT CPU_ONLY)
 endif()
 
 target_link_libraries(caffe proto
-        ${GLOG_LIBRARIES}
+        ${BLAS_LIBRARIES}
+        ${Boost_LIBRARIES}
+        ${CMAKE_THREAD_LIBS_INIT}
         ${GFLAGS_LIBRARIES}
+        ${GLOG_LIBRARIES}
         ${HDF5_LIBRARIES}
-        ${OpenCV_LIBS}
         ${LEVELDB_LIBS}
         ${LMDB_LIBRARIES}
-        ${BLAS_LIBRARIES}
-        ${CMAKE_THREAD_LIBS_INIT}
+        ${OpenCV_LIBS}
 )
 
 #set output directory

src/caffe/internal_thread.cpp

@@ -0,0 +1,33 @@
+#include "caffe/internal_thread.hpp"
+
+namespace caffe {
+
+InternalThread::~InternalThread() {
+  WaitForInternalThreadToExit();
+  if (thread_ != NULL) {
+    delete thread_;
+  }
+}
+
+bool InternalThread::StartInternalThread() {
+  try {
+    thread_ = new boost::thread(&InternalThread::InternalThreadEntry, this);
+  } catch (...) {
+    return false;
+  }
+  return true;
+}
+
+/** Will not return until the internal thread has exited. */
+bool InternalThread::WaitForInternalThreadToExit() {
+  if (is_started()) {
+    try {
+      thread_->join();
+    } catch (...) {
+      return false;
+    }
+  }
+  return true;
+}
+
+}  // namespace caffe

src/caffe/layers/data_layer.cpp

@@ -244,12 +244,12 @@ void DataLayer<Dtype>::CreatePrefetchThread() {
 
   data_transformer_.InitRand();
 
-  CHECK(StartInternalThread()) << "Pthread execution failed";
+  CHECK(StartInternalThread()) << "Thread execution failed";
 }
 
 template <typename Dtype>
 void DataLayer<Dtype>::JoinPrefetchThread() {
-  CHECK(!WaitForInternalThreadToExit()) << "Pthread joining failed";
+  CHECK(WaitForInternalThreadToExit()) << "Thread joining failed";
 }
 
 template <typename Dtype>

src/caffe/layers/image_data_layer.cpp

@@ -159,7 +159,7 @@ void ImageDataLayer<Dtype>::CreatePrefetchThread() {
   data_transformer_.InitRand();
 
   // Create the thread.
-  CHECK(StartInternalThread()) << "Pthread execution failed";
+  CHECK(StartInternalThread()) << "Thread execution failed";
 }
 
 template <typename Dtype>
@@ -172,7 +172,7 @@ void ImageDataLayer<Dtype>::ShuffleImages() {
 
 template <typename Dtype>
 void ImageDataLayer<Dtype>::JoinPrefetchThread() {
-  CHECK(!WaitForInternalThreadToExit()) << "Pthread joining failed";
+  CHECK(WaitForInternalThreadToExit()) << "Thread joining failed";
 }
 
 template <typename Dtype>

src/caffe/layers/window_data_layer.cpp

@@ -408,12 +408,12 @@ void WindowDataLayer<Dtype>::CreatePrefetchThread() {
     prefetch_rng_.reset();
   }
   // Create the thread.
-  CHECK(StartInternalThread()) << "Pthread execution failed.";
+  CHECK(StartInternalThread()) << "Thread execution failed.";
 }
 
 template <typename Dtype>
 void WindowDataLayer<Dtype>::JoinPrefetchThread() {
-  CHECK(!WaitForInternalThreadToExit()) << "Pthread joining failed.";
+  CHECK(WaitForInternalThreadToExit()) << "Thread joining failed.";
 }
 
 template <typename Dtype>

src/caffe/test/test_internal_thread.cpp

@@ -0,0 +1,23 @@
+#include "glog/logging.h"
+#include "gtest/gtest.h"
+
+#include "caffe/internal_thread.hpp"
+
+#include "caffe/test/test_caffe_main.hpp"
+
+namespace caffe {
+
+
+class InternalThreadTest : public ::testing::Test {};
+
+TEST_F(InternalThreadTest, TestStartAndExit) {
+  InternalThread thread;
+  EXPECT_FALSE(thread.is_started());
+  EXPECT_TRUE(thread.StartInternalThread());
+  EXPECT_TRUE(thread.is_started());
+  EXPECT_TRUE(thread.WaitForInternalThreadToExit());
+  EXPECT_FALSE(thread.is_started());
+}
+
+}  // namespace caffe
+