From f0c3b2368478af977e9c2ebe931b7fa92276ac55 Mon Sep 17 00:00:00 2001 From: Mike Vitus Date: Wed, 28 Feb 2018 13:08:48 -0800 Subject: [PATCH] Increases the performance of the C++11 threading. Previously, the thread ID was acquired and released on every iteration of the for loop. The C++11 concurrent queue implementation is much slower than TBB's version and consequently this was a huge bottleneck. This introduces another ParallelFor API which takes the thread ID as a parameter in the evaluation function. This allows us to acquire and release the thread ID for each block of work which drastically improves the performance. This change brings us on par with OpenMP and TBB. See below for a timing comparison. Note: in this example this CLs C++11 version is faster to compute the residuals because TBB still must acquire the thread ID on every iteration, which has some overhead. Tested by building and running tests for no threading, OpenMP, TBB, and C++11 threads. Also ran bazel tests. ./bin/bundle_adjuster --input=problem-744-543562-pre.txt --num_threads=8 C++11 @Head Time (in seconds): Residual only evaluation 7.819692 (5) Jacobian & residual evaluation 11.606063 (6) Linear solver 47.860195 (5) Minimizer 70.877072 Total 90.806338 --------------------------------------------------- C++11 (This CL) Time (in seconds): Residual only evaluation 1.217500 (5) Jacobian & residual evaluation 5.796112 (6) Linear solver 44.080873 (5) Minimizer 54.635524 Total 77.640072 --------------------------------------------------- OpenMP Time (in seconds): Residual only evaluation 0.797023 (5) Jacobian & residual evaluation 5.633916 (6) Linear solver 43.280020 (5) Minimizer 53.199058 Total 76.250861 --------------------------------------------------- TBB Time (in seconds): Residual only evaluation 1.911095 (5) Jacobian & residual evaluation 5.557807 (6) Linear solver 44.074680 (5) Minimizer 55.002688 Total 78.052687 --------------------------------------------------- No Threads Time (in seconds): Residual only evaluation 2.939212 (5) Jacobian & residual evaluation 18.519874 (6) Linear solver 74.017837 (5) Minimizer 98.980080 Total 122.216391 Change-Id: I3af959b0771bbdfe8cad8c13896191d6ac903181 --- internal/ceres/covariance_impl.cc | 26 ++++++++- internal/ceres/parallel_for.h | 10 ++++ internal/ceres/parallel_for_cxx.cc | 77 ++++++++++++++++++++------ internal/ceres/parallel_for_tbb.cc | 18 ++++++ internal/ceres/parallel_for_test.cc | 73 ++++++++++++++++++++++++ internal/ceres/program_evaluator.h | 11 +++- internal/ceres/schur_eliminator_impl.h | 13 +++-- internal/ceres/thread_pool_test.cc | 3 + internal/ceres/thread_token_provider.h | 13 +---- 9 files changed, 207 insertions(+), 37 deletions(-) diff --git a/internal/ceres/covariance_impl.cc b/internal/ceres/covariance_impl.cc index 8b34bc14e..d77e5e121 100644 --- a/internal/ceres/covariance_impl.cc +++ b/internal/ceres/covariance_impl.cc @@ -342,7 +342,9 @@ bool CovarianceImpl::GetCovarianceMatrixInTangentOrAmbientSpace( bool success = true; +#if !(defined(CERES_USE_TBB) || defined(CERES_USE_CXX11_THREADS)) ThreadTokenProvider thread_token_provider(num_threads); +#endif // !(defined(CERES_USE_TBB) || defined(CERES_USE_CXX11_THREADS)) // Technically the following code is a double nested loop where // i = 1:n, j = i:n. @@ -361,7 +363,7 @@ bool CovarianceImpl::GetCovarianceMatrixInTangentOrAmbientSpace( 0, num_parameters * num_parameters, num_threads, - [&](int k) { + [&](int thread_id, int k) { #endif // !(defined(CERES_USE_TBB) || defined(CERES_USE_CXX11_THREADS)) int i = k / num_parameters; int j = k % num_parameters; @@ -377,8 +379,10 @@ bool CovarianceImpl::GetCovarianceMatrixInTangentOrAmbientSpace( int covariance_col_idx = cum_parameter_size[j]; int size_i = parameter_sizes[i]; int size_j = parameter_sizes[j]; +#if !(defined(CERES_USE_TBB) || defined(CERES_USE_CXX11_THREADS)) const ScopedThreadToken scoped_thread_token(&thread_token_provider); const int thread_id = scoped_thread_token.token(); +#endif // !(defined(CERES_USE_TBB) || defined(CERES_USE_CXX11_THREADS)) double* covariance_block = workspace.get() + thread_id * max_covariance_block_size * max_covariance_block_size; @@ -716,7 +720,9 @@ bool CovarianceImpl::ComputeCovarianceValuesUsingSuiteSparseQR() { const int num_threads = options_.num_threads; scoped_array workspace(new double[num_threads * num_cols]); +#if !(defined(CERES_USE_TBB) || defined(CERES_USE_CXX11_THREADS)) ThreadTokenProvider thread_token_provider(num_threads); +#endif // !(defined(CERES_USE_TBB) || defined(CERES_USE_CXX11_THREADS)) #ifdef CERES_USE_OPENMP #pragma omp parallel for num_threads(num_threads) schedule(dynamic) @@ -726,14 +732,20 @@ bool CovarianceImpl::ComputeCovarianceValuesUsingSuiteSparseQR() { for (int r = 0; r < num_cols; ++r) { #else problem_->context()->EnsureMinimumThreads(num_threads); - ParallelFor(problem_->context(), 0, num_cols, num_threads, [&](int r) { + ParallelFor(problem_->context(), + 0, + num_cols, + num_threads, + [&](int thread_id, int r) { #endif // !(defined(CERES_USE_TBB) || defined(CERES_USE_CXX11_THREADS)) const int row_begin = rows[r]; const int row_end = rows[r + 1]; if (row_end != row_begin) { +#if !(defined(CERES_USE_TBB) || defined(CERES_USE_CXX11_THREADS)) const ScopedThreadToken scoped_thread_token(&thread_token_provider); const int thread_id = scoped_thread_token.token(); +#endif // !(defined(CERES_USE_TBB) || defined(CERES_USE_CXX11_THREADS)) double* solution = workspace.get() + thread_id * num_cols; SolveRTRWithSparseRHS( @@ -917,7 +929,9 @@ bool CovarianceImpl::ComputeCovarianceValuesUsingEigenSparseQR() { const int num_threads = options_.num_threads; scoped_array workspace(new double[num_threads * num_cols]); +#if !(defined(CERES_USE_TBB) || defined(CERES_USE_CXX11_THREADS)) ThreadTokenProvider thread_token_provider(num_threads); +#endif // !(defined(CERES_USE_TBB) || defined(CERES_USE_CXX11_THREADS)) #ifdef CERES_USE_OPENMP #pragma omp parallel for num_threads(num_threads) schedule(dynamic) @@ -927,14 +941,20 @@ bool CovarianceImpl::ComputeCovarianceValuesUsingEigenSparseQR() { for (int r = 0; r < num_cols; ++r) { #else problem_->context()->EnsureMinimumThreads(num_threads); - ParallelFor(problem_->context(), 0, num_cols, num_threads, [&](int r) { + ParallelFor(problem_->context(), + 0, + num_cols, + num_threads, + [&](int thread_id, int r) { #endif // !(defined(CERES_USE_TBB) || defined(CERES_USE_CXX11_THREADS)) const int row_begin = rows[r]; const int row_end = rows[r + 1]; if (row_end != row_begin) { +#if !(defined(CERES_USE_TBB) || defined(CERES_USE_CXX11_THREADS)) const ScopedThreadToken scoped_thread_token(&thread_token_provider); const int thread_id = scoped_thread_token.token(); +#endif // !(defined(CERES_USE_TBB) || defined(CERES_USE_CXX11_THREADS)) double* solution = workspace.get() + thread_id * num_cols; SolveRTRWithSparseRHS( diff --git a/internal/ceres/parallel_for.h b/internal/ceres/parallel_for.h index 603c609bf..e54a1b363 100644 --- a/internal/ceres/parallel_for.h +++ b/internal/ceres/parallel_for.h @@ -47,6 +47,16 @@ void ParallelFor(ContextImpl* context, int num_threads, const std::function& function); +// Execute the function for every element in the range [start, end) with at most +// num_threads. It will execute all the work on the calling thread if +// num_threads is 1. Each invocation of function() will be passed a thread_id +// in [0, num_threads) that is guaranteed to be distinct from the value passed +// to any concurrent execution of function(). +void ParallelFor(ContextImpl* context, + int start, + int end, + int num_threads, + const std::function& function); } // namespace internal } // namespace ceres diff --git a/internal/ceres/parallel_for_cxx.cc b/internal/ceres/parallel_for_cxx.cc index f03ddb278..bfc8973c0 100644 --- a/internal/ceres/parallel_for_cxx.cc +++ b/internal/ceres/parallel_for_cxx.cc @@ -41,7 +41,8 @@ #include #include "ceres/concurrent_queue.h" -#include "ceres/thread_pool.h" +#include "ceres/scoped_thread_token.h" +#include "ceres/thread_token_provider.h" #include "glog/logging.h" namespace ceres { @@ -90,6 +91,7 @@ struct SharedState { end(end), num_work_items(num_work_items), i(0), + thread_token_provider(num_work_items), block_until_finished(num_work_items) {} // The start and end index of the for loop. @@ -105,27 +107,17 @@ struct SharedState { int i; std::mutex mutex_i; - // Used to signal when all the work has been completed. + // Provides a unique thread ID among all active threads working on the same + // group of tasks. Thread-safe. + ThreadTokenProvider thread_token_provider; + + // Used to signal when all the work has been completed. Thread safe. BlockUntilFinished block_until_finished; }; } // namespace -// This implementation uses a fixed size max worker pool with a shared task -// queue. The problem of executing the function for the interval of [start, end) -// is broken up into at most num_threads blocks and added to the thread pool. To -// avoid deadlocks, the calling thread is allowed to steal work from the worker -// pool. This is implemented via a shared state between the tasks. In order for -// the calling thread or thread pool to get a block of work, it will query the -// shared state for the next block of work to be done. If there is nothing left, -// it will return. We will exit the ParallelFor call when all of the work has -// been done, not when all of the tasks have been popped off the task queue. -// -// A performance analysis has shown this implementation is about ~20% slower -// than OpenMP or TBB. This native implementation is a fix for platforms that do -// not have access to OpenMP or TBB. The gain in enabling multi-threaded Ceres -// is much more significant so we decided to not chase the performance of these -// two libraries. +// See ParallelFor (below) for more details. void ParallelFor(ContextImpl* context, int start, int end, @@ -145,6 +137,51 @@ void ParallelFor(ContextImpl* context, return; } + ParallelFor(context, start, end, num_threads, + [&function](int /*thread_id*/, int i) { function(i); }); +} + +// This implementation uses a fixed size max worker pool with a shared task +// queue. The problem of executing the function for the interval of [start, end) +// is broken up into at most num_threads blocks and added to the thread pool. To +// avoid deadlocks, the calling thread is allowed to steal work from the worker +// pool. This is implemented via a shared state between the tasks. In order for +// the calling thread or thread pool to get a block of work, it will query the +// shared state for the next block of work to be done. If there is nothing left, +// it will return. We will exit the ParallelFor call when all of the work has +// been done, not when all of the tasks have been popped off the task queue. +// +// A unique thread ID among all active tasks will be acquired once for each +// block of work. This avoids the significant performance penalty for acquiring +// it on every iteration of the for loop. The thread ID is guaranteed to be in +// [0, num_threads). +// +// A performance analysis has shown this implementation is onpar with OpenMP and +// TBB. +void ParallelFor(ContextImpl* context, + int start, + int end, + int num_threads, + const std::function& function) { + CHECK_GT(num_threads, 0); + CHECK(context != NULL); + if (end <= start) { + return; + } + + // Fast path for when it is single threaded. + if (num_threads == 1) { + // Even though we only have one thread, use the thread token provider to + // guarantee the exact same behavior when running with multiple threads. + ThreadTokenProvider thread_token_provider(num_threads); + const ScopedThreadToken scoped_thread_token(&thread_token_provider); + const int thread_id = scoped_thread_token.token(); + for (int i = start; i < end; ++i) { + function(thread_id, i); + } + return; + } + // We use a shared_ptr because the main thread can finish all the work before // the tasks have been popped off the queue. So the shared state needs to // exist for the duration of all the tasks. @@ -167,11 +204,15 @@ void ParallelFor(ContextImpl* context, ++shared_state->i; } + const ScopedThreadToken scoped_thread_token( + &shared_state->thread_token_provider); + const int thread_id = scoped_thread_token.token(); + // Perform each task. for (int j = shared_state->start + i; j < shared_state->end; j += shared_state->num_work_items) { - function(j); + function(thread_id, j); } shared_state->block_until_finished.Finished(); return true; diff --git a/internal/ceres/parallel_for_tbb.cc b/internal/ceres/parallel_for_tbb.cc index 82fbf100f..e235b5062 100644 --- a/internal/ceres/parallel_for_tbb.cc +++ b/internal/ceres/parallel_for_tbb.cc @@ -38,6 +38,8 @@ #include #include +#include "ceres/scoped_thread_token.h" +#include "ceres/thread_token_provider.h" #include "glog/logging.h" namespace ceres { @@ -68,6 +70,22 @@ void ParallelFor(ContextImpl* context, }); } +void ParallelFor(ContextImpl* context, + int start, + int end, + int num_threads, + const std::function& function) { + CHECK(context != NULL); + + ThreadTokenProvider thread_token_provider(num_threads); + ParallelFor(context, start, end, num_threads, [&](int i) { + const ScopedThreadToken scoped_thread_token(&thread_token_provider); + const int thread_id = scoped_thread_token.token(); + function(thread_id, i); + }); +} + + } // namespace internal } // namespace ceres diff --git a/internal/ceres/parallel_for_test.cc b/internal/ceres/parallel_for_test.cc index a3cf2dcd7..2de48657d 100644 --- a/internal/ceres/parallel_for_test.cc +++ b/internal/ceres/parallel_for_test.cc @@ -36,9 +36,13 @@ #include "ceres/parallel_for.h" #include +#include +#include +#include #include #include "ceres/context_impl.h" +#include "glog/logging.h" #include "gmock/gmock.h" #include "gtest/gtest.h" @@ -46,6 +50,7 @@ namespace ceres { namespace internal { using testing::ElementsAreArray; +using testing::UnorderedElementsAreArray; // Tests the parallel for loop computes the correct result for various number of // threads. @@ -67,6 +72,26 @@ TEST(ParallelFor, NumThreads) { } } +// Tests the parallel for loop with the thread ID interface computes the correct +// result for various number of threads. +TEST(ParallelForWithThreadId, NumThreads) { + ContextImpl context; + context.EnsureMinimumThreads(/*num_threads=*/2); + + const int size = 16; + std::vector expected_results(size, 0); + for (int i = 0; i < size; ++i) { + expected_results[i] = std::sqrt(i); + } + + for (int num_threads = 1; num_threads <= 8; ++num_threads) { + std::vector values(size, 0); + ParallelFor(&context, 0, size, num_threads, + [&values](int thread_id, int i) { values[i] = std::sqrt(i); }); + EXPECT_THAT(values, ElementsAreArray(expected_results)); + } +} + // Tests nested for loops do not result in a deadlock. TEST(ParallelFor, NestedParallelForDeadlock) { ContextImpl context; @@ -85,6 +110,54 @@ TEST(ParallelFor, NestedParallelForDeadlock) { } } +// Tests nested for loops do not result in a deadlock for the parallel for with +// thread ID interface. +TEST(ParallelForWithThreadId, NestedParallelForDeadlock) { + ContextImpl context; + context.EnsureMinimumThreads(/*num_threads=*/2); + + // Increment each element in the 2D matrix. + std::vector> x(3, {1, 2, 3}); + ParallelFor(&context, 0, 3, 2, [&x, &context](int thread_id, int i) { + std::vector& y = x.at(i); + ParallelFor(&context, 0, 3, 2, [&y](int thread_id, int j) { ++y.at(j); }); + }); + + const std::vector results = {2, 3, 4}; + for (const std::vector& value : x) { + EXPECT_THAT(value, ElementsAreArray(results)); + } +} + +TEST(ParallelForWithThreadId, UniqueThreadIds) { + // Ensure the hardware supports more than 1 thread to ensure the test will + // pass. + const int num_hardware_threads = std::thread::hardware_concurrency(); + if (num_hardware_threads <= 1) { + LOG(ERROR) + << "Test not supported, the hardware does not support threading."; + return; + } + + ContextImpl context; + context.EnsureMinimumThreads(/*num_threads=*/2); + // Increment each element in the 2D matrix. + std::vector x(2, -1); + std::mutex mutex; + std::condition_variable condition; + int count = 0; + ParallelFor(&context, 0, 2, 2, + [&x, &mutex, &condition, &count](int thread_id, int i) { + std::unique_lock lock(mutex); + x[i] = thread_id; + ++count; + condition.notify_all(); + condition.wait(lock, [&]() { return count == 2; }); + }); + + EXPECT_THAT(x, UnorderedElementsAreArray({0,1})); +} + } // namespace internal } // namespace ceres diff --git a/internal/ceres/program_evaluator.h b/internal/ceres/program_evaluator.h index ca6090298..5986d81ac 100644 --- a/internal/ceres/program_evaluator.h +++ b/internal/ceres/program_evaluator.h @@ -175,7 +175,9 @@ class ProgramEvaluator : public Evaluator { const int num_residual_blocks = program_->NumResidualBlocks(); +#if !(defined(CERES_USE_TBB) || defined(CERES_USE_CXX11_THREADS)) ThreadTokenProvider thread_token_provider(options_.num_threads); +#endif // !(defined(CERES_USE_TBB) || defined(CERES_USE_CXX11_THREADS)) #ifdef CERES_USE_OPENMP // This bool is used to disable the loop if an error is encountered @@ -196,8 +198,11 @@ class ProgramEvaluator : public Evaluator { #if defined(CERES_USE_TBB) || defined(CERES_USE_CXX11_THREADS) std::atomic_bool abort(false); - ParallelFor(options_.context, 0, num_residual_blocks, options_.num_threads, - [&](int i) { + ParallelFor(options_.context, + 0, + num_residual_blocks, + options_.num_threads, + [&](int thread_id, int i) { #endif // defined(CERES_USE_TBB) || defined(CERES_USE_CXX11_THREADS) if (abort) { @@ -208,8 +213,10 @@ class ProgramEvaluator : public Evaluator { #endif // defined(CERES_USE_TBB) || defined(CERES_USE_CXX11_THREADS) } +#if !(defined(CERES_USE_TBB) || defined(CERES_USE_CXX11_THREADS)) const ScopedThreadToken scoped_thread_token(&thread_token_provider); const int thread_id = scoped_thread_token.token(); +#endif // !(defined(CERES_USE_TBB) || defined(CERES_USE_CXX11_THREADS)) EvaluatePreparer* preparer = &evaluate_preparers_[thread_id]; EvaluateScratch* scratch = &evaluate_scratch_[thread_id]; diff --git a/internal/ceres/schur_eliminator_impl.h b/internal/ceres/schur_eliminator_impl.h index ca8f30e9f..8baa03936 100644 --- a/internal/ceres/schur_eliminator_impl.h +++ b/internal/ceres/schur_eliminator_impl.h @@ -222,7 +222,9 @@ Eliminate(const BlockSparseMatrix* A, #endif // defined(CERES_USE_TBB) || defined(CERES_USE_CXX11_THREADS) } +#if !(defined(CERES_USE_TBB) || defined(CERES_USE_CXX11_THREADS)) ThreadTokenProvider thread_token_provider(num_threads_); +#endif // !(defined(CERES_USE_TBB) || defined(CERES_USE_CXX11_THREADS)) #ifdef CERES_USE_OPENMP // Eliminate y blocks one chunk at a time. For each chunk, compute @@ -243,12 +245,15 @@ Eliminate(const BlockSparseMatrix* A, #if !(defined(CERES_USE_TBB) || defined(CERES_USE_CXX11_THREADS)) for (int i = 0; i < chunks_.size(); ++i) { -#else - ParallelFor(context_, 0, int(chunks_.size()), num_threads_, [&](int i) { -#endif // !(defined(CERES_USE_TBB) || defined(CERES_USE_CXX11_THREADS)) - const ScopedThreadToken scoped_thread_token(&thread_token_provider); const int thread_id = scoped_thread_token.token(); +#else + ParallelFor(context_, + 0, + int(chunks_.size()), + num_threads_, + [&](int thread_id, int i) { +#endif // !(defined(CERES_USE_TBB) || defined(CERES_USE_CXX11_THREADS)) double* buffer = buffer_.get() + thread_id * buffer_size_; const Chunk& chunk = chunks_[i]; diff --git a/internal/ceres/thread_pool_test.cc b/internal/ceres/thread_pool_test.cc index 1aa81f20b..5485fe417 100644 --- a/internal/ceres/thread_pool_test.cc +++ b/internal/ceres/thread_pool_test.cc @@ -36,10 +36,13 @@ #include "ceres/thread_pool.h" #include +#include +#include #include #include "gmock/gmock.h" #include "gtest/gtest.h" +#include "glog/logging.h" namespace ceres { namespace internal { diff --git a/internal/ceres/thread_token_provider.h b/internal/ceres/thread_token_provider.h index 841d31273..e4c9cd838 100644 --- a/internal/ceres/thread_token_provider.h +++ b/internal/ceres/thread_token_provider.h @@ -95,16 +95,9 @@ class ThreadTokenProvider { #ifdef CERES_USE_CXX11_THREADS // This queue initially holds a sequence from 0..num_threads-1. Every // Acquire() call the first number is removed from here. When the token is not - // needed anymore it shall be given back with corresponding Release() call. - // - // The thread number is acquired on every for loop iteration. The - // ConcurrentQueue uses a mutex to enable thread safety, however, this can - // lead to a large amount of contention between the threads which can cause a - // loss in performance. This is noticable for problems with inexpensive - // residual computations. - // - // TODO(vitus): We should either implement a more performant queue (such as - // lock free), or get the thread ID from the shared state. + // needed anymore it shall be given back with corresponding Release() + // call. This concurrent queue is more expensive than TBB's version, so you + // should not acquire the thread ID on every for loop iteration. ConcurrentQueue pool_; #endif