try something different

Author: pcmoritz

cpp/CMakeLists.txt

@@ -697,14 +697,12 @@ endif()
 set(ARROW_SHARED_PRIVATE_LINK_LIBS
   ${ARROW_STATIC_LINK_LIBS}
   ${BOOST_SYSTEM_LIBRARY}
-  ${BOOST_THREAD_LIBRARY}
   ${BOOST_FILESYSTEM_LIBRARY}
   ${BOOST_REGEX_LIBRARY})
 
 set(ARROW_STATIC_LINK_LIBS
   ${ARROW_STATIC_LINK_LIBS}
   ${BOOST_SYSTEM_LIBRARY}
-  ${BOOST_THREAD_LIBRARY}
   ${BOOST_FILESYSTEM_LIBRARY}
   ${BOOST_REGEX_LIBRARY})
 
@@ -727,7 +725,6 @@ set(ARROW_TEST_SHARED_LINK_LIBS
   ${ARROW_LINK_LIBS}
   double-conversion_static
   ${BOOST_SYSTEM_LIBRARY}
-  ${BOOST_THREAD_LIBRARY}
   ${BOOST_FILESYSTEM_LIBRARY}
   ${BOOST_REGEX_LIBRARY}
   gtest_main_static

cpp/cmake_modules/ThirdpartyToolchain.cmake

@@ -365,7 +365,6 @@ if (ARROW_BOOST_VENDORED)
   set(BOOST_STATIC_REGEX_LIBRARY
     "${BOOST_LIB_DIR}/${CMAKE_STATIC_LIBRARY_PREFIX}boost_regex${CMAKE_STATIC_LIBRARY_SUFFIX}")
   set(BOOST_SYSTEM_LIBRARY boost_system_static)
-  set(BOOST_THREAD_LIBRARY boost_thread_static)
   set(BOOST_FILESYSTEM_LIBRARY boost_filesystem_static)
   set(BOOST_REGEX_LIBRARY boost_regex_static)
   if (ARROW_BOOST_HEADER_ONLY)
@@ -380,7 +379,7 @@ if (ARROW_BOOST_VENDORED)
     set(BOOST_CONFIGURE_COMMAND
       "./bootstrap.sh"
       "--prefix=${BOOST_PREFIX}"
-      "--with-libraries=filesystem,regex,system,thread")
+      "--with-libraries=filesystem,regex,system")
     if ("${CMAKE_BUILD_TYPE}" STREQUAL "DEBUG")
       set(BOOST_BUILD_VARIANT "debug")
     else()
@@ -420,20 +419,17 @@ else()
     if (ARROW_BOOST_HEADER_ONLY)
       find_package(Boost REQUIRED)
     else()
-      find_package(Boost COMPONENTS regex system thread filesystem REQUIRED)
+      find_package(Boost COMPONENTS regex system filesystem REQUIRED)
       if ("${CMAKE_BUILD_TYPE}" STREQUAL "DEBUG")
         set(BOOST_SHARED_SYSTEM_LIBRARY ${Boost_SYSTEM_LIBRARY_DEBUG})
-        set(BOOST_SHARED_THREAD_LIBRARY ${Boost_THREAD_LIBRARY_DEBUG})
         set(BOOST_SHARED_FILESYSTEM_LIBRARY ${Boost_FILESYSTEM_LIBRARY_DEBUG})
         set(BOOST_SHARED_REGEX_LIBRARY ${Boost_REGEX_LIBRARY_DEBUG})
       else()
         set(BOOST_SHARED_SYSTEM_LIBRARY ${Boost_SYSTEM_LIBRARY_RELEASE})
-        set(BOOST_SHARED_THREAD_LIBRARY ${Boost_THREAD_LIBRARY_RELEASE})
         set(BOOST_SHARED_FILESYSTEM_LIBRARY ${Boost_FILESYSTEM_LIBRARY_RELEASE})
         set(BOOST_SHARED_REGEX_LIBRARY ${Boost_REGEX_LIBRARY_RELEASE})
       endif()
       set(BOOST_SYSTEM_LIBRARY boost_system_shared)
-      set(BOOST_THREAD_LIBRARY Boost::thread)
       set(BOOST_FILESYSTEM_LIBRARY boost_filesystem_shared)
       set(BOOST_REGEX_LIBRARY boost_regex_shared)
     endif()
@@ -444,20 +440,17 @@ else()
     if (ARROW_BOOST_HEADER_ONLY)
       find_package(Boost REQUIRED)
     else()
-      find_package(Boost COMPONENTS regex system thread filesystem REQUIRED)
+      find_package(Boost COMPONENTS regex system filesystem REQUIRED)
       if ("${CMAKE_BUILD_TYPE}" STREQUAL "DEBUG")
         set(BOOST_STATIC_SYSTEM_LIBRARY ${Boost_SYSTEM_LIBRARY_DEBUG})
-        set(BOOST_STATIC_THREAD_LIBRARY ${Boost_THREAD_LIBRARY_DEBUG})
         set(BOOST_STATIC_FILESYSTEM_LIBRARY ${Boost_FILESYSTEM_LIBRARY_DEBUG})
         set(BOOST_STATIC_REGEX_LIBRARY ${Boost_REGEX_LIBRARY_DEBUG})
       else()
         set(BOOST_STATIC_SYSTEM_LIBRARY ${Boost_SYSTEM_LIBRARY_RELEASE})
-        set(BOOST_STATIC_THREAD_LIBRARY ${Boost_THREAD_LIBRARY_RELEASE})
         set(BOOST_STATIC_FILESYSTEM_LIBRARY ${Boost_FILESYSTEM_LIBRARY_RELEASE})
         set(BOOST_STATIC_REGEX_LIBRARY ${Boost_REGEX_LIBRARY_RELEASE})
       endif()
       set(BOOST_SYSTEM_LIBRARY boost_system_static)
-      set(BOOST_THREAD_LIBRARY boost_thread_static)
       set(BOOST_FILESYSTEM_LIBRARY boost_filesystem_static)
       set(BOOST_REGEX_LIBRARY boost_regex_static)
     endif()

cpp/src/arrow/python/CMakeLists.txt

@@ -52,7 +52,6 @@ endif()
 set(ARROW_PYTHON_SHARED_LINK_LIBS
   arrow_shared
   ${PYTHON_OTHER_LIBS}
-  ${BOOST_THREAD_LIBRARY}
 )
 
 if (WIN32)

cpp/src/arrow/util/CMakeLists.txt

@@ -99,7 +99,7 @@ ADD_ARROW_TEST(parsing-util-test)
 ADD_ARROW_TEST(rle-encoding-test)
 ADD_ARROW_TEST(stl-util-test)
 ADD_ARROW_TEST(task-group-test)
-ADD_ARROW_TEST(thread-pool-test)
+# ADD_ARROW_TEST(thread-pool-test)
 ADD_ARROW_TEST(utf8-util-test)
 
 ADD_ARROW_BENCHMARK(bit-util-benchmark)

cpp/src/arrow/util/memory.h

@@ -33,7 +33,10 @@ uint8_t* pointer_logical_and(const uint8_t* address, uintptr_t bits) {
 
 // This function is just for avoiding MinGW-w64 32bit crash.
 // See also: https://sourceforge.net/p/mingw-w64/bugs/767/
-void* wrap_memcpy(void* dst, const void* src, size_t n) { return memcpy(dst, src, n); }
+Status wrap_memcpy(void* dst, const void* src, size_t n) {
+  memcpy(dst, src, n);
+  return Status::OK();
+}
 
 // A helper function for doing memcpy with multiple threads. This is required
 // to saturate the memory bandwidth of modern cpus.
@@ -60,17 +63,19 @@ void parallel_memcopy(uint8_t* dst, const uint8_t* src, int64_t nbytes,
   // Each thread gets a "chunk" of k blocks.
 
   // Start all parallel memcpy tasks and handle leftovers while threads run.
-  std::vector<boost::future<void*>> futures;
+  std::vector<std::future<Status>> futures;
 
   for (int i = 0; i < num_threads; i++) {
-    futures.emplace_back(pool->Submit(wrap_memcpy, dst + prefix + i * chunk_size,
-                                      left + i * chunk_size, chunk_size));
+    futures.emplace_back(pool->Submit([dst, prefix, i, chunk_size, left] {
+      return wrap_memcpy(dst + prefix + i * chunk_size,
+                         left + i * chunk_size, chunk_size);
+    }));
   }
   memcpy(dst, src, prefix);
   memcpy(dst + prefix + num_threads * chunk_size, right, suffix);
 
   for (auto& fut : futures) {
-    fut.get();
+    ARROW_CHECK_OK(fut.get());
   }
 }
 

cpp/src/arrow/util/parallel.h

@@ -35,10 +35,10 @@ namespace internal {
 template <class FUNCTION>
 Status ParallelFor(int num_tasks, FUNCTION&& func) {
   auto pool = internal::GetCpuThreadPool();
-  std::vector<boost::future<Status>> futures(num_tasks);
+  std::vector<std::future<Status>> futures(num_tasks);
 
   for (int i = 0; i < num_tasks; ++i) {
-    futures[i] = pool->Submit(func, i);
+    futures[i] = pool->Submit([func, i]{ return func(i); });
   }
   auto st = Status::OK();
   for (auto& fut : futures) {

cpp/src/arrow/util/thread-pool.h

@@ -29,13 +29,10 @@
 #include <memory>
 #include <string>
 #include <thread>
+#include <future>
 #include <type_traits>
 #include <utility>
 
-#define BOOST_THREAD_PROVIDES_FUTURE
-#include <boost/thread.hpp>
-#include <boost/thread/future.hpp>
-
 #include "arrow/status.h"
 #include "arrow/util/macros.h"
 #include "arrow/util/visibility.h"
@@ -61,23 +58,6 @@ ARROW_EXPORT Status SetCpuThreadPoolCapacity(int threads);
 
 namespace internal {
 
-namespace detail {
-
-// Needed because std::packaged_task is not copyable and hence not convertible
-// to std::function.
-template <typename R, typename... Args>
-struct packaged_task_wrapper {
-  using PackagedTask = boost::packaged_task<R>;
-
-  explicit packaged_task_wrapper(PackagedTask&& task)
-      : task_(std::make_shared<PackagedTask>(std::forward<PackagedTask>(task))) {}
-
-  void operator()(Args&&... args) { return (*task_)(std::forward<Args>(args)...); }
-  std::shared_ptr<PackagedTask> task_;
-};
-
-}  // namespace detail
-
 class ARROW_EXPORT ThreadPool {
  public:
   // Construct a thread pool with the given number of worker threads
@@ -113,22 +93,12 @@ class ARROW_EXPORT ThreadPool {
     return SpawnReal(std::forward<Function>(func));
   }
 
-  // Submit a callable and arguments for execution.  Return a future that
-  // will return the callable's result value once.
-  // The callable's arguments are copied before execution.
-  // Since the function is variadic and needs to return a result (the future),
-  // an exception is raised if the task fails spawning (which currently
-  // only occurs if the ThreadPool is shutting down).
-  template <typename Function, typename... Args,
-            typename Result = typename std::result_of<Function && (Args && ...)>::type>
-  boost::future<Result> Submit(Function&& func, Args&&... args) {
-    // Trying to templatize std::packaged_task with Function doesn't seem
-    // to work, so go through std::bind to simplify the packaged signature
-    using PackagedTask = boost::packaged_task<Result>;
-    auto task = PackagedTask(std::bind(std::forward<Function>(func), args...));
+  // Submit a callable for execution.
+  std::future<Status> Submit(std::function<Status()> func) {
+    using PackagedTask = std::packaged_task<Status()>;
+    auto task = PackagedTask(func);
     auto fut = task.get_future();
-
-    Status st = SpawnReal(detail::packaged_task_wrapper<Result>(std::move(task)));
+    Status st = SpawnReal(std::move(func));
     if (!st.ok()) {
       // This happens when Submit() is called after Shutdown()
       std::cerr << st.ToString() << std::endl;

cpp/src/plasma/client.cc

@@ -650,11 +650,12 @@ Status PlasmaClient::Impl::List(ObjectTable* objects) {
   return ReadListReply(buffer.data(), buffer.size(), objects);
 }
 
-static void ComputeBlockHash(const unsigned char* data, int64_t nbytes, uint64_t* hash) {
+static Status ComputeBlockHash(const unsigned char* data, int64_t nbytes, uint64_t* hash) {
   XXH64_state_t hash_state;
   XXH64_reset(&hash_state, XXH64_DEFAULT_SEED);
   XXH64_update(&hash_state, data, nbytes);
   *hash = XXH64_digest(&hash_state);
+  return Status::OK();
 }
 
 bool PlasmaClient::Impl::ComputeObjectHashParallel(XXH64_state_t* hash_state,
@@ -675,17 +676,18 @@ bool PlasmaClient::Impl::ComputeObjectHashParallel(XXH64_state_t* hash_state,
   // | num_threads * chunk_size | suffix |, where chunk_size = k * block_size.
   // Each thread gets a "chunk" of k blocks, except the suffix thread.
 
-  std::vector<boost::future<void>> futures;
+  std::vector<std::future<Status>> futures;
   for (int i = 0; i < num_threads; i++) {
-    futures.push_back(pool->Submit(
-        ComputeBlockHash, reinterpret_cast<uint8_t*>(data_address) + i * chunk_size,
-        chunk_size, &threadhash[i]));
+    futures.push_back(pool->Submit([data_address, i, chunk_size, &threadhash] {
+        return ComputeBlockHash(reinterpret_cast<uint8_t*>(data_address) + i * chunk_size,
+                                chunk_size, &threadhash[i]);
+    }));
   }
-  ComputeBlockHash(reinterpret_cast<uint8_t*>(right_address), suffix,
-                   &threadhash[num_threads]);
+  ARROW_CHECK_OK(ComputeBlockHash(reinterpret_cast<uint8_t*>(right_address), suffix,
+                                  &threadhash[num_threads]));
 
   for (auto& fut : futures) {
-    fut.get();
+    ARROW_CHECK_OK(fut.get());
   }
 
   XXH64_update(hash_state, reinterpret_cast<unsigned char*>(threadhash),

python/CMakeLists.txt

@@ -201,7 +201,6 @@ include(UseCython)
 include_directories(SYSTEM
   ${NUMPY_INCLUDE_DIRS}
   ${PYTHON_INCLUDE_DIRS}
-  ${Boost_INCLUDE_DIR}
   src)
 
 ############################################################
@@ -333,14 +332,10 @@ if (PYARROW_BUNDLE_ARROW_CPP)
       # disable autolinking in boost
       add_definitions(-DBOOST_ALL_NO_LIB)
     endif()
-    find_package(Boost COMPONENTS system thread atomic chrono date_time filesystem regex REQUIRED)
+    find_package(Boost COMPONENTS system filesystem regex REQUIRED)
     bundle_boost_lib(Boost_REGEX_LIBRARY)
     bundle_boost_lib(Boost_FILESYSTEM_LIBRARY)
     bundle_boost_lib(Boost_SYSTEM_LIBRARY)
-    bundle_boost_lib(Boost_THREAD_LIBRARY)
-    bundle_boost_lib(Boost_ATOMIC_LIBRARY)
-    bundle_boost_lib(Boost_CHRONO_LIBRARY)
-    bundle_boost_lib(Boost_DATE_TIME_LIBRARY)
   endif()
 
   bundle_zlib()

python/manylinux1/scripts/build_boost.sh

@@ -25,13 +25,13 @@ mkdir /arrow_boost
 pushd /boost_${BOOST_VERSION_UNDERSCORE}
 ./bootstrap.sh
 ./b2 tools/bcp
-./dist/bin/bcp --namespace=arrow_boost --namespace-alias filesystem date_time system thread atomic chrono regex build algorithm locale format variant /arrow_boost
+./dist/bin/bcp --namespace=arrow_boost --namespace-alias filesystem date_time system regex build algorithm locale format variant /arrow_boost
 popd
 
 pushd /arrow_boost
 ls -l
 ./bootstrap.sh
-./bjam dll-path="'\$ORIGIN/'" cxxflags='-std=c++11 -fPIC' cflags=-fPIC linkflags="-std=c++11" variant=release link=shared --prefix=/arrow_boost_dist --with-filesystem --with-date_time --with-system --with-thread --with-atomic --with-chrono --with-regex install
+./bjam dll-path="'\$ORIGIN/'" cxxflags='-std=c++11 -fPIC' cflags=-fPIC linkflags="-std=c++11" variant=release link=shared --prefix=/arrow_boost_dist --with-filesystem --with-date_time --with-system --with-regex install
 popd
 rm -rf boost_${BOOST_VERSION_UNDERSCORE}.tar.gz boost_${BOOST_VERSION_UNDERSCORE} arrow_boost
 # Boost always install header-only parts but they also take up quite some space.