From 5e44d25c5a57fe81a2df700d42d99e821eadd576 Mon Sep 17 00:00:00 2001 From: Tim Harris Date: Tue, 10 Nov 2020 12:24:57 +0000 Subject: [PATCH] Support multi-loop parallel sections, use multi-loop sections in GRU (#5602) This PR updates the ThreadPool API to support multi-loop parallel sections. As with the OpenMP "parallel" construct, this allows per-loop work to be amortized over a series of loops. For ORT, it also promotes locality between successive loops in the sense that iteration X of one loop will tend to run on the same worker thread as iteration X of preceding loops. The change was developed while optimizing the implementation of a model that performed better with OpenMP. Profiling indicated that OpenMP was providing lower loop entry/exit costs and that, via OpenMP's static scheduling, it was leading to a lower L2 miss rate in the series of parallel loops used in GRU. The main changes are: - Addition of ThreadPool::ParallelSection and underlying support in the modified Eigen thread pool. - In EigenNonBlockingThreadPool.h, refactoring the RunInParallel method to support two variants: one that takes an existing parallel section object created by the caller, and another (used by default) that creates its own parallel section. - Simplify ThreadPool::LoopCounter (used by worker threads to claim loop iterations), basing it an ID supplied by the underlying Eigen thread pool for affinity in a series of loops. - Fix a possible perf issue where a loop with iterations scheduled in batches would have more threads than batches available. - Use of parallel sections in the GRU operator. - Additional test cases in threadpool_test.h. - Additional comments at the top of threadpool.h and EigenNonBlockingThreadPool.h. --- docs/NotesOnThreading.md | 5 + .../platform/EigenNonBlockingThreadPool.h | 503 +++++++++++++++--- .../onnxruntime/core/platform/threadpool.h | 156 +++++- onnxruntime/core/common/threadpool.cc | 158 ++++-- .../core/providers/cpu/rnn/deep_cpu_gru.cc | 338 ++++++------ onnxruntime/test/platform/threadpool_test.cc | 167 ++++-- 6 files changed, 984 insertions(+), 343 deletions(-) diff --git a/docs/NotesOnThreading.md b/docs/NotesOnThreading.md index c0e8bf4aee..25f504c02c 100644 --- a/docs/NotesOnThreading.md +++ b/docs/NotesOnThreading.md @@ -17,6 +17,11 @@ Examples of these abstractions are: ([threadpool.h](https://github.com/microsoft These static methods abstract over the different implementation choices. They can run over the ORT thread pool, or run over OpenMP, or run sequentially. +In addition, ThreadPool::ParallelSection allows a series of loops to +be grouped together in a single parallel section. This allows an +operator to amortize loop entry/exit costs in cases where it is +impractical to refactor code into a single large loop. + **Please do not write #ifdef pragma omp in operator code**. For intra op parallelism ORT users can use either OpenMP or ORT threadpool. The choice of using OpenMP is indicated by building ORT with ```--use_openmp``` switch. For inter op parallelism, however, we always use the ORT threadpool. diff --git a/include/onnxruntime/core/platform/EigenNonBlockingThreadPool.h b/include/onnxruntime/core/platform/EigenNonBlockingThreadPool.h index 13acaedbcf..5940200d7c 100644 --- a/include/onnxruntime/core/platform/EigenNonBlockingThreadPool.h +++ b/include/onnxruntime/core/platform/EigenNonBlockingThreadPool.h @@ -38,24 +38,194 @@ #include "core/platform/ort_mutex.h" #include "core/platform/Barrier.h" -namespace onnxruntime { +// ORT thread pool overview +// ------------------------ +// +// The ORT thread pool implementation is split into two layers. This +// file provides the low-level component. See the accompanying +// comments in threadpool.h for the high-level component. +// +// The code here is derived from the Eigen non-blocking thread pool, +// although many parts have been updated over time. The main +// abstractions used here are: +// +// - The thread pool maintains a set of OS threads running +// ThreadPoolTempl::WorkerLoop. +// +// Each thread has its own RunQueue object, holding a queue of tasks +// that have been pushed to the thread for execution. The main work +// loop is to pop a task from the head of the queue, and to execute +// it to completion. If the worker's run queue is empty then it +// will spin waiting for work, then attempt to steal tasks from +// other threads' queues, and then block in the OS if it cannot find +// work. +// +// This spin-then-block behavior is configured via a flag provided +// when creating the thread pool, and by the constant spin_count. +// +// - Although all tasks are simple void()->void functions, +// conceptually there are three different kinds: +// +// - One-shot tasks submitted externally via the Schedule() method. +// These tasks are used to support asynchronous work. These are +// used in the parallel executor, but otherwise are not widely +// used outside of test harnesses (see threadpool_test.cc for some +// examples). +// +// - Tasks for running a parallel loop. +// +// The tasks themselves are defined in threadpool.cc, and are +// submitted to the run queues via RunInParallel->SummonWorkers. +// Each task will loop internally, picking off iterations from the +// user's code via atoic-fetch-and-add, until the loop is +// complete. +// +// This two-layer approach lets us separate out the +// super-lightweight per-iteration-batch work from the more +// costsly per-loop work of managing Task objects. +// +// - Tasks for running a parallel section. This is an extension of +// the approach taken for parallel loops. However, the Tasks are +// defined in this file, and can pick up iterations from a series +// of different parallel loops. The tasks are defined in +// RunInParallelSection->SummonWorkers. +// +// The additional layer of parallel sections is a further way to +// amortize costs: the work done creating the tasks can be +// performed once, and then exploited over a series of loops. +// +// There are a few aspects of the modified Eigen thread pool to +// highlight: +// +// - The run queues follow the usual approach of having push/pop +// operations on the front/back, and optimizing the PopFront case +// for single-threaded use by the thread owning the run queue. +// +// However, we support an additional Revoke operation to replace an +// item in the middle of a queue with a tombstone. This operation +// is used at the end of parallel loops and parallel sections to +// remove any tasks that were created but not yet executed. Once +// revoked, a thread can rely on the fact that the task will no +// longer execute. Revocation helps manage captured state in +// parallel loops: the alternatives would be (i) waiting for all +// tasks that captured state to reach the head of their queues and +// execute, or (ii) use heap-allocated state in tasks, and use a +// technique such as reference counting to de-allocate it. +// +// To support revoation, each thread has a unique "Tag" to identify +// the items that it adds to the work queues. A thread can revoke +// an item only if it has the thread's own tag. +// +// - The worker threads maintain a best-effort bitmap in +// good_worker_hints_ of which threads to push work to. A thread +// controls its status via SetGoodWorkerHint. A thread is a "good" +// worker when it is actively spinning for work, meaning both that +// it is not blocked in the OS, and that it is not busy with work +// already. +// +// This heuristic aims to avoid waking additional sleeping threads +// where possible, and in a series of parallel loops or parallel +// sections to push the work to the same set of threads each time. +namespace onnxruntime { namespace concurrency { -// Extended Eigen thread pool interface, avoiding the need to modify the ThreadPoolInterface.h -// header from the external Eigen repository. +class ThreadPoolParallelSection; +class ThreadPoolLoop; + +// Extended Eigen thread pool interface, avoiding the need to modify +// the ThreadPoolInterface.h header from the external Eigen +// repository. class ExtendedThreadPoolInterface : public Eigen::ThreadPoolInterface { public: - // Run fn with up to n degree-of-parallelism enlisting the thread pool for - // help. The degree-of-parallelism includes the caller, and so if n==1 - // then the function will run directly in the caller. The fork-join - // synchronization is handled in the thread pool, and so any state captured - // by fn() is safe from concurrent access once RunInParallel returns. - virtual void RunInParallel(std::function fn, unsigned n) = 0; + // Start/end a parallel section, within which calls to + // RunInParallelSection may be made. Parallel sections are + // non-nesting. + virtual std::unique_ptr AllocateParallelSection() = 0; + virtual void StartParallelSection(ThreadPoolParallelSection &ps) = 0; + virtual void EndParallelSection(ThreadPoolParallelSection &ps) = 0; + + // Run fn with up to n degree-of-parallelism enlisting the thread + // pool for help. The degree-of-parallelism includes the caller, + // and so if n==1 then the function will run directly in the caller. + // + // The fork-join synchronization is handled in the thread pool, and + // so any state captured by fn() is safe from concurrent access once + // RunInParallelSection returns. + // + // The parameter idx provides a loop-local thread ID in the range + // [0,k) where k<=n. + virtual void RunInParallelSection(ThreadPoolParallelSection &ps, + std::function fn, + unsigned n) = 0; + + // Special case alternative to RunInParallelSection for use without + // an existing parallel section. Ideally we would use a single + // iplemenation and a stack-allocated ThreadPoolParallelSection. + // + // However, on the BM_ThreadPoolParallelFor microbenchmark I saw + // ~20% overhead on the resulting single-loop parallel sections. + // There are some additional costs (~5%) for additional invocations + // through lambda functions on loop entry. Most significantly, on + // loop exit, we incurred ~15% cost by no longer being able to + // overlap clean-up of unused Task objects in EndParallelSection + // with waiting for loop iterations to complete. + // + // [ Note that this 20% overhead is more than paid for when we have + // two loops execute in series in a parallel section. ] + virtual void RunInParallel(std::function fn, + unsigned n) = 0; }; -} // namespace concurrency + +class ThreadPoolParallelSection { + public: + // State accessed only by the main thread + // -------------------------------------- + + // Tasks successfully submitted to the work queues. This sets the + // maximum degree of parallelism that the section will support. + std::vector> tasks; + + // State shared between the main thread and worker threads + // ------------------------------------------------------- + + // Flag to signal termination of the parallel section + std::atomic active{false}; + + std::atomic worker_idx{0}; + + // Count of the number of tasks that completed normally. Other + // tasks may be running currently, or may be present in work queues, + // or may have been removed from the queues by + // RunQueue::RevokeWithTag. + std::atomic tasks_finished{0}; + + // If non-null, the current loop that tasks should be executing. We + // need to be careful on access to the contents of current_loop + // because it can be stack allocated on the thread entering the + // loop: + // + // - Readers increment workers_in_loop and then read current_loop + // + // - Writers wishing to deallocate *current_loop must first clear + // current_loop and then wait for workers_in_loop==0 + std::atomic current_loop{nullptr}; + std::atomic workers_in_loop{0}; +}; + +class ThreadPoolLoop { + public: + ThreadPoolLoop(std::function f, unsigned t) : fn(std::move(f)), threads_needed(t) { + } + + const std::function fn; + const unsigned threads_needed; + + private: + ORT_DISALLOW_COPY_ASSIGNMENT_AND_MOVE(ThreadPoolLoop); +}; template class RunQueue { @@ -212,6 +382,11 @@ class RunQueue { std::unique_lock lock(mutex_); Elem& e = array_[w_idx]; ElemState s = e.state.load(std::memory_order_relaxed); + + // We have acquired a lock on the queue, synchronizing with + // operations aside from the PopFront fast-path. Synchronize with + // that by attempting the same kReady->kBusy transition via CAS. + if (s == ElemState::kReady && e.state.compare_exchange_strong(s, ElemState::kBusy, std::memory_order_acquire)) { if (e.tag == tag) { @@ -349,6 +524,8 @@ template class ThreadPoolTempl : public onnxruntime::concurrency::ExtendedThreadPoolInterface { private: + struct PerThread; + static unsigned WorkerLoop(int id, Eigen::ThreadPoolInterface* param) { // unsafe downcast ThreadPoolTempl* this_ptr = (ThreadPoolTempl*)param; @@ -366,12 +543,13 @@ class ThreadPoolTempl : public onnxruntime::concurrency::ExtendedThreadPoolInter Tag(uint32_t v) : v_(v) { } - // Allocate a new tag to use to identify work items from a given thread - // in RunInParallel. Ideally, threads will have unique tags, but re-use - // is not incorrect if the counter wraps (for intsance, if a long-running - // workload is calling into ORT from a fresh thread for each request). - // We must not re-use the default tag 0 which is used to identify work - // items added via Schedule as opposed to requests for help in RunInParallel. + // Allocate a new tag to use to identify work items from a given + // thread in a parallel section. Ideally, threads will have + // unique tags, but re-use is not incorrect if the counter wraps + // (for intsance, if a long-running workload is calling into ORT + // from a fresh thread for each request). We must not re-use the + // default tag 0 which is used to identify work items added via + // Schedule as opposed to requests for help in parallel sections. static Tag GetNext() { Tag t = Tag(next_tag++); @@ -392,10 +570,6 @@ class ThreadPoolTempl : public onnxruntime::concurrency::ExtendedThreadPoolInter uint32_t v_ = 0; }; - static Tag GetNextTag() { - return Tag(next_tag++); - } - typedef RunQueue Queue; #ifdef _WIN32 using CHAR_TYPE = wchar_t; @@ -511,9 +685,9 @@ void SetGoodWorkerHint(int idx, bool is_good) { // good_hints, letting the caller avoid distributing more than one work item to // any individual thread. -void GetGoodWorkerHints(int n, std::vector& good_hints, std::vector& alt_hints) { +void GetGoodWorkerHints(unsigned n, std::vector& good_hints, std::vector& alt_hints) { PerThread* pt = GetPerThread(); - int need_alt = n; + unsigned need_alt = n; good_hints.clear(); alt_hints.clear(); @@ -522,14 +696,14 @@ void GetGoodWorkerHints(int n, std::vector& good_hints, std::vectorrand) % num_hint_words_; - for (int i = 0; n && (i < num_hint_words_); i++) { + for (unsigned i = 0u; n && (i < num_hint_words_); i++) { int u64_idx = (base + i) % num_hint_words_; std::atomic* u64 = &good_worker_hints_[u64_idx]; uint64_t saw = u64->load(); uint64_t want = saw; // Pick up to n bits that are set in the current word - for (int j = 0; n && (j < bits_per_hint_word_); j++) { + for (unsigned j = 0u; n && (j < bits_per_hint_word_); j++) { uint64_t bit = 1ull << j; int thread = u64_idx * bits_per_hint_word_ + j; if (saw & bit) { @@ -550,31 +724,146 @@ void GetGoodWorkerHints(int n, std::vector& good_hints, std::vector fn, unsigned n) override { - PerThread* my_pt = GetPerThread(); - assert(n>=1); - if (n == 1 || my_pt->in_parallel) { - fn(); - } else { - // We build a list of pairs for each of the queues that accepts a work - // item. This lets us remove any work items that do not get executed by the threads - // that we push them to. - std::vector> pending_items; - Barrier b(n, allow_spinning_); +//...................................................................... +// +// Parallel sections +// ----------------- +// +// Allocate a new ThreadPoolParallelSection, owned by the returned +// unique_ptr. The explicit deleter avoids the Eigen-specific +// definition of ThreadPoolParallelSection needing to be avilable in +// threadpool.h where the user-facing parallel section API is defined. - my_pt->in_parallel = true; - if (!my_pt->tag.Get()) { - my_pt->tag = Tag::GetNext(); +std::unique_ptr AllocateParallelSection() override { + return std::unique_ptr + (new ThreadPoolParallelSection, + [](ThreadPoolParallelSection *tps) { + delete tps; + }); +} + +// Start a parallel section, using a caller-provided +// ThreadPoolParallelSection for maintaining the per-section state. +// Starting a parallel section is just book-keeping; threads are +// "summoned" to help with the parallel section once it enters +// parallel loops. The threads are then retained until the end of the +// section, being re-used over subsequent loops. + +void StartParallelSectionInternal(PerThread &pt, + ThreadPoolParallelSection &ps) { + assert((!pt.leading_par_section) && "Nested parallelism not supported"); + assert((!ps.active) && "Starting parallel section, but active already"); + pt.leading_par_section = true; + if (!pt.tag.Get()) { + pt.tag = Tag::GetNext(); + } + ps.active = true; +} + +void StartParallelSection(ThreadPoolParallelSection &ps) override { + PerThread* pt = GetPerThread(); + StartParallelSectionInternal(*pt, ps); +} + +// End a parallel section, waiting for all worker threads to exit from +// section. Hence, on return, the ThreadPoolParallelSection object +// can be dealloacted. + +void EndParallelSectionInternal(PerThread &pt, + ThreadPoolParallelSection &ps) { + assert((pt.leading_par_section) && "Ending parallel section, but none started"); + assert((ps.active) && "Ending parallel section, but not active"); + pt.leading_par_section = false; + + // Notify workers to exit from the section + ps.active = false; + + // Attempt to revoke any tasks that were sent to workers but not + // started. + unsigned tasks_started = static_cast(ps.tasks.size()); + unsigned tasks_revoked = 0; + while (!ps.tasks.empty()) { + const auto& item = ps.tasks.back(); + Queue& q = worker_data_[item.first].queue; + if (q.RevokeWithTag(pt.tag, item.second)) { + tasks_revoked++; } + ps.tasks.pop_back(); + } - // Push up to n-1 copies of the work item into the queues + // Wait for workers to exit ParLoopWorker + auto tasks_to_wait_for = tasks_started - tasks_revoked; + while (ps.tasks_finished < tasks_to_wait_for) { + onnxruntime::concurrency::SpinPause(); + } + + // Clear status to allow the ThreadPoolParallelSection to be + // re-used. + ps.tasks_finished = 0; +} + +void EndParallelSection(ThreadPoolParallelSection &ps) override { + PerThread* pt = GetPerThread(); + EndParallelSectionInternal(*pt, ps); +} + +//...................................................................... +// +// Parallel loops +// -------------- +// +// Ensure that the ThreadPoolParallelSection has sufficient workers to +// execute a loop with degree of parallelism n. We track the number +// of workers already avaiable to the parallel section, prior to +// submitting tasks to the work queues to make up the total. +// +// Each worker will call in to worker_fn(idx) with a per-worker thread +// ID. Note there are different levels of indirection here: +// +// - In a single-loop parallel section, worker_fn will directly +// execute the threadpool.cc code that implements the parallel loop. +// +// - In a multi-loop parallel section, worker_fn is an intermediate +// function that is long-lived (i.e., that lasts until the end of +// the parallel section, as opposed to just a single loop's +// duration). + +void SummonWorkers(PerThread &pt, + ThreadPoolParallelSection &ps, + unsigned n, + const std::function &worker_fn) { + // Wrap the user's worker function with one that allocates a unique + // worker index for the loop, and synchronizes (as the last step) + // with the exit path in EndParallelSection. In principle we could + // allocate worker IDs during the loop below and capture them by + // value. However, the costs of creating distinct lambda for each + // iteration appeared more costly than the cost of synchronization + // on a shared counter. + auto call_worker_fn = [&ps, worker_fn]() { + unsigned my_idx = ++ps.worker_idx; + worker_fn(my_idx); + // After the assignment to ps.tasks_finished, the stack-allocated + // ThreadPoolParallelSection object may be destroyed. + ps.tasks_finished++; + }; + + // Identify whether we need to create additional workers. + // Throughout the threadpool implementation, degrees of parallelism + // ("n" here) refer to the total parallelism including the main + // thread. Hence we consider the number of existing tasks + 1. + unsigned current_dop = static_cast(ps.tasks.size()) + 1; + if (n > current_dop) { + unsigned extra_needed = n - current_dop; + + // Obtain hints for which worker threads to push the tasks to. + // This uses a best-effort assessment of which threads are + // spinning. std::vector good_hints, alt_hints; - GetGoodWorkerHints(n - 1, good_hints, alt_hints); - for (unsigned i = 0; i < n - 1; i++) { - Task t = env_.CreateTask([&b, &fn]() { - fn(); - b.Notify(1); - }); + GetGoodWorkerHints(extra_needed, good_hints, alt_hints); + + // Create the additional tasks, and push them to workers. + for (auto i = 0u; i < extra_needed; i++) { + Task t = env_.CreateTask(call_worker_fn); int q_idx; if (i < good_hints.size()) { q_idx = good_hints[i]; @@ -583,47 +872,100 @@ void RunInParallel(std::function fn, unsigned n) override { if (alt_i < alt_hints.size()) { q_idx = alt_hints[alt_i]; } else { - q_idx = Rand(&my_pt->rand) % num_threads_; + q_idx = Rand(&pt.rand) % num_threads_; } } + + // If the worker's queue accepts the task, then record it in + // the vector of tasks that we will need to synchronize with on + // exiting the parallel section. If the queue rejects the task + // (perhaps because it is full) then we take no further action: + // in a parallel loop we will always be running work on the + // main thread, providing progress. WorkerData& td = worker_data_[q_idx]; Queue& q = td.queue; unsigned w_idx; - t = q.PushBackWithTag(std::move(t), my_pt->tag, w_idx); - if (t.f) { - // The queue rejected the work. Account for the missing capacity for work - // on the synchronization barrier. The semantics for RunInParallel are that - // the function is called with up to n-way parallelism, and so the - // work itself will be performed in the current thread's call to fn() - // after finishing adding work to the pool. - b.Notify(1); - } else { - // The queue accepted the work, ensure that the thread is servicing the queue - pending_items.push_back({q_idx, w_idx}); + t = q.PushBackWithTag(std::move(t), pt.tag, w_idx); + if (!t.f) { + ps.tasks.push_back({q_idx, w_idx}); td.EnsureAwake(); } } + } +} - // Run the final copy ourselves, for the total of n degree-of-parallelism - fn(); +// Run a single parallel loop in an existing parallel section. This +// maps directly onto SummonWorkers to create sufficient worker +// threads for the desired degree of parallelism, followed by +// dispatching the loop to those workers. - // Notify the barrier for the work we completed, plus any work that we successfully - // revoke from the work queues - int notifications_needed = 1; - for (auto& item : pending_items) { - Queue& q = worker_data_[item.first].queue; - if (q.RevokeWithTag(my_pt->tag, item.second)) { - notifications_needed++; +void RunInParallelSection(ThreadPoolParallelSection &ps, + std::function fn, + unsigned n) override { + PerThread* pt = GetPerThread(); + assert(pt->leading_par_section && "RunInParallel, but not in parallel section"); + assert((n > 1) && "Trivial parallel section; should be avoided by caller"); + + // Publish the work to any existing workers in the parallel + // section, and ensure it is visible to any new threads created + // below. + assert((!ps.current_loop) && "RunInParallelSection, but loop already active"); + ThreadPoolLoop loop{std::move(fn), n}; + ps.current_loop = &loop; + + // Increase the worker count if needed. Each worker will pick up + // loops to execute from the current parallel section. + const auto worker_fn = [&ps](unsigned my_idx) { + while (ps.active) { + if (!ps.current_loop) { + onnxruntime::concurrency::SpinPause(); + } else { + ps.workers_in_loop++; + ThreadPoolLoop *work_item = ps.current_loop; + if (work_item && my_idx < work_item->threads_needed) { + work_item->fn(my_idx); + } + ps.workers_in_loop--; } } - b.Notify(notifications_needed); + }; + SummonWorkers(*pt, ps, n, worker_fn); - // Synchronize with any work items that are still running - b.Wait(); - my_pt->in_parallel = false; + // Run work in the main thread + loop.fn(0); + + // Wait for workers to exit the loop + ps.current_loop = 0; + while (ps.workers_in_loop) { + onnxruntime::concurrency::SpinPause(); } } +// Run a single parallel loop _without_ a parallel section. This is a +// special case of RunInParallelSection, avoiding code paths for +// handing off multiple loops to the pool of workers. + +void RunInParallel(std::function fn, unsigned n) override { + PerThread *pt = GetPerThread(); + ThreadPoolParallelSection ps; + StartParallelSectionInternal(*pt, ps); + + // Summon workers to run the function (n is the desired maximum + // degree of parallelism, including the main thread). Unlike the + // multi-loop RunInParallelSection, this single-loop worker can run + // fn directly without needing to receive it via ps.current_loop. + SummonWorkers(*pt, ps, n, fn); + + // Run work in the main thread + fn(0); + + // Wait for workers to exit the parallel section and hence to have + // completed the loop (i.e., ps.tasks_finished matches the number of + // tasks that have been created less the number successfully + // revoked). + EndParallelSectionInternal(*pt, ps); +} + void Cancel() override { cancelled_ = true; // If done_ is true, which means this object is being destructing. @@ -700,14 +1042,15 @@ int CurrentThreadId() const EIGEN_FINAL { struct PerThread { constexpr PerThread() : pool(nullptr) { } - ThreadPoolTempl* pool; // Parent pool, or null for normal threads. - uint64_t rand{0}; // Random generator state. - int thread_id{-1}; // Worker thread index in pool. - Tag tag{}; // Work item tag used to identify this thread. - bool in_parallel{false}; // Inside a parallel section (hence tag not unique if we re-use) + ThreadPoolTempl* pool; // Parent pool, or null for normal threads. + uint64_t rand{0}; // Random generator state. + int thread_id{-1}; // Worker thread index in pool. + Tag tag{}; // Work item tag used to identify this thread. + bool leading_par_section{false}; // Leading a parallel section (used only for asserts) }; - static_assert(std::is_trivially_destructible::value, "Per-thread state should be trivially destructible"); + static_assert(std::is_trivially_destructible::value, + "Per-thread state should be trivially destructible"); struct WorkerData { constexpr WorkerData() : thread(), queue() { @@ -817,8 +1160,8 @@ int CurrentThreadId() const EIGEN_FINAL { // reduce contention by having different threads start work searching for hints // at different locations in the bitmap. - static const int bits_per_hint_word_ = 4; - int num_hint_words_; + static const unsigned bits_per_hint_word_ = 4; + unsigned num_hint_words_; std::unique_ptr[]> good_worker_hints_; // Wake any blocked workers so that they can cleanly exit WorkerLoop(). For an @@ -1004,7 +1347,7 @@ int CurrentThreadId() const EIGEN_FINAL { return std::hash()(std::this_thread::get_id()); } - EIGEN_STRONG_INLINE PerThread* GetPerThread() { + static EIGEN_STRONG_INLINE PerThread* GetPerThread() { static thread_local PerThread per_thread_; PerThread* pt = &per_thread_; return pt; @@ -1019,4 +1362,6 @@ int CurrentThreadId() const EIGEN_FINAL { } }; + } // namespace concurrency + } // namespace onnxruntime diff --git a/include/onnxruntime/core/platform/threadpool.h b/include/onnxruntime/core/platform/threadpool.h index d44e941a91..126dd133fa 100644 --- a/include/onnxruntime/core/platform/threadpool.h +++ b/include/onnxruntime/core/platform/threadpool.h @@ -26,8 +26,94 @@ limitations under the License. #include #include -// This file use PIMPL to avoid having eigen headers here +// ORT thread pool overview +// ------------------------ +// +// The ORT thread pool implementation is split into two layers. This +// file provides the high-level component. See the accompanying +// comments in EigenNonBlockingThreadPool.h for the low-level +// component. +// +// threadpool.h defines the user-facing functions for use in +// operators. The main abstraction are parallel loops +// (ThreadPool::TryParallelFor*), although we also support scheduling +// of asynchronous tasks (ThreadPool::Schedule), and the construction +// of multi-loop parallel sections (ThreadPool::ParallelSection). +// +// This high level API is accessed via static methods on the +// ThreadPool class. These methods map the operations onto one of +// three low-level implementations: (#1) direct execution of the +// operations if there is no thread pool configured, (#2) execution of +// the operations using the modified Eigen threadpool, (#3) execution +// of the operations using OpenMP. Option #1 enables execution in +// simple settings without needing threads. Option #2 is the +// preferred approach for use in settings with parallelism. +// +// The high-level part of the thread pool is responsible for: +// +// - Exposing the desired degree of parallelism to user code, and to +// libraries such as MLAS. This lets the libraries tailor the +// extent to which they parallelize work. +// +// - Handling trivial cases (such as directly running parallel loops +// with only a single iteration, or with no iterations at all). +// +// - Deciding how to divide work efficiently between the threads +// available. +// +// The ThreadPool::TryParallelFor methods do this based on cost +// estimates supplied by the caller, and are designed to support +// loops with small amounts of work per iteration. The loop body is +// supplied as a function taking a [start,end) range of iterations +// to execute (avoiding the need for per-iteration std::function +// calls, or a reliance upon inlining to avoid those calls). +// +// ThreadPool::TrySimpleParallelFor uses a simpler single-iteration +// API based on the assumption that the caller has divided work to +// an appropriate granularity. +// +// - When used with the Eigen-based thread pool, the implementation of +// all of the loops maps down onto +// ThreadPool::ParallelForFixedBlockSizeScheduling. This method +// takes the degree of parallelism (d_of_p) and work distribution +// block size (from the cost-based heuristics), and creates a set of +// tasks in the underlying thread pool (via +// ThreadPool::RunInParallel). +// +// These tasks then run a loop which picks off batches of iterations +// from the user's code. The distribution of these batches is +// handled dynmamically via LoopCounter::ClaimIterations. This +// dynamic balancing behavior helps make performance robust to any +// variability in the execution time across iterations, and to +// situations such as multiple loops running concurrently on the +// same thread pool. +// +// - When running a series of loops inside a parallel section, the +// LoopCounter also helps obtain affinity between these loops (i.e., +// iteration X of one loop will tend to run on the same thread that +// ran iteration X of prior loops). This locality helps improve hit +// rates in per-core caches across the series of short loops used in +// operators like GRU. +// +// There are some known areas for exploration here: +// +// - The cost-based heuristics were developed prior to recent changes +// to the thread pool. The heuristics seem to work well, but we +// should revisit the tuning periodically. +// +// - Can we unify the APIs for the different kinds of parallel loop? +// +// In particular, we may be able to replace the current use of +// TryBatchParallelFor with appropriate costs for each call site, +// and then use TryParallelFor. This would allow for more dynamic +// re-balancing of work between threads than the current +// ThreadPool::PartitionWork function provides. +// +// - Given the extensive modifications to original Eigen code, should +// we separate that out as a new class and remove the dependence on +// other Eigen components. +// This file use PIMPL to avoid having eigen headers here namespace Eigen { class Allocator; class ThreadPoolInterface; @@ -41,13 +127,15 @@ struct TensorOpCost { double compute_cycles; }; -template -class ThreadPoolTempl; namespace concurrency { +template +class ThreadPoolTempl; + class ExtendedThreadPoolInterface; class LoopCounter; +class ThreadPoolParallelSection; class ThreadPool { public: @@ -65,7 +153,6 @@ class ThreadPool { // operations like I/O the hint should be set to false. // // REQUIRES: degree_of_parallelism > 0 - // The allocator parameter is only used for creating a Eigen::ThreadPoolDevice to be used with Eigen Tensor classes. ThreadPool(Env* env, const ThreadOptions& thread_options, const NAME_CHAR_TYPE* name, @@ -76,6 +163,65 @@ class ThreadPool { // set of threads. ~ThreadPool(); + // Start and end a multi-loop parallel section. Parallel loops can + // be executed directly (without using this API), but entering a + // parallel section allows the runtime system to amortize loop + // entry/exit costs over multiple loops, and allows it to promote + // affinity between corresponding iterations of different loops. + // + // Multi-loop sections would typically be used in cases where a + // series of loops executes without much code in between them, and + // where it is impractical to refactor code into a single loop. For + // instance: + // + // { + // onnxruntime::concurrency::ThreadPoool::ParallelSection ps(tp); + // for (int x = 0; x < seq_len; x++) { + // TrySimpleParallelFor(tp, 16, [&]() { ... }); + // } + // } + // + // The parallel section is entered via the constructor of + // ThreadPool::ParallelSection, and exited via the destructor. + // Currently, thread-local state is used to track whether or not the + // current thread is inside a parallel section. In contrast to + // handling parallel section objects explicitly in user code, this + // approach allows code such as MLAS to operate with/without the use + // of parallel sections. + // + // Parallel sections are only implemented with the Eigen threadpool. + // They have no effect when using OpenMP. + // + // Parallel sections may not be nested, and may not be used inside + // parallel loops. + + class ParallelSection { + public: + explicit ParallelSection(ThreadPool *tp); + ~ParallelSection(); + + private: + friend class ThreadPool; + + // Owning reference for the underlying ThreadPoolParallelSection + // which implements the thread management. We use an explicit + // deleter here so that the definition of + // ThreadPoolParallelSection does not need to be available at this + // point to avoid a dependence on the Eigen headers. + std::unique_ptr + ps_{nullptr, [](ThreadPoolParallelSection*){}}; +#ifndef _OPENMP + ThreadPool *tp_; +#endif + ORT_DISALLOW_COPY_ASSIGNMENT_AND_MOVE(ParallelSection); + + // Non-owning reference to the current thread's paralel section + // (or nullptr outside parallel sections). + static thread_local ParallelSection *current_parallel_section; + static_assert(std::is_trivially_destructible::value, + "Per-thread state should be trivially destructible"); + }; + // Schedules fn() for execution in the pool of threads. The function may run // synchronously if it cannot be enqueued. This will occur if the thread pool's // degree-of-parallelism is 1, but it may also occur for implementation-dependent @@ -250,7 +396,7 @@ class ThreadPool { // then the function will run directly in the caller. The fork-join // synchronization is handled in the thread pool, and so any state captured // by fn() is safe from concurrent access once RunWithHelp returns. - void RunInParallel(std::function fn, int n); + void RunInParallel(std::function fn, unsigned n); // Divides the work represented by the range [0, total) into k shards. // Calls fn(i*block_size, (i+1)*block_size) from the ith shard (0 <= i < k). diff --git a/onnxruntime/core/common/threadpool.cc b/onnxruntime/core/common/threadpool.cc index f0026de9b7..4c2a8a8c7f 100644 --- a/onnxruntime/core/common/threadpool.cc +++ b/onnxruntime/core/common/threadpool.cc @@ -39,64 +39,57 @@ namespace concurrency { #endif static constexpr int CACHE_LINE_BYTES = 64; -static constexpr int NUM_SHARDS = 8; +static constexpr unsigned MAX_SHARDS = 8; struct alignas(CACHE_LINE_BYTES) LoopCounterShard { - ::std::atomic _next; - uint64_t _end; + ::std::atomic _next{0}; + uint64_t _end{0}; }; +static_assert(sizeof(LoopCounterShard) == CACHE_LINE_BYTES, "Expected loop counter shards to match cache-line size"); + class alignas(CACHE_LINE_BYTES) LoopCounter { public: - LoopCounter(const ThreadPool& tp, - uint64_t num_iterations, - uint64_t block_size = 1) : _tp(tp), - _block_size(block_size) { - assert(sizeof(LoopCounterShard) == 64); - assert(block_size != 0); + LoopCounter(uint64_t num_iterations, + uint64_t block_size = 1) : _block_size(block_size), + _num_shards(GetNumShards(num_iterations, block_size)) { + // Divide the iteration space between the shards. If the iteration + // space does not divide evenly into shards of multiples of + // block_size then the final shard is left uneven. - // Divide the iteration space into NUM_SHARDS pieces. If the iteration space does not - // divide evenly into shards of multiples of block_size then the final shard is left uneven. - double iterations_per_shard = static_cast(num_iterations) / NUM_SHARDS; - uint64_t split = 0; - for (uint64_t shard = 0; shard < NUM_SHARDS; shard++) { - _shards[shard]._next = split; - split = (static_cast((shard + 1) * iterations_per_shard) / block_size) * block_size; - _shards[shard]._end = split; + auto num_blocks = num_iterations / block_size; + auto blocks_per_shard = num_blocks / _num_shards; + auto iterations_per_shard = blocks_per_shard * block_size; + + for (uint64_t shard = 0; shard < _num_shards; shard++) { + // Initialize with a relaxed store; synchronization with worker + // threads is provided via the thread pool + _shards[shard]._next.store(shard * iterations_per_shard, + ::std::memory_order_relaxed); + _shards[shard]._end = (shard+1) * iterations_per_shard; } - // Ensure that the final shard finishes precisely at the end of the iteration space - _shards[NUM_SHARDS - 1]._end = num_iterations; + _shards[_num_shards - 1]._end = num_iterations; } - int GetHomeShard() const { - // Allocate each thread to a home shard, from which it starts claiming iterations. The allocation - // does not need to be unique, but we aim for a good distribution, particularly in the case where - // most/all of the thread pool's threads are active in the loop. Threads outside the pool may - // also be claiming work, with CurrentThreadId -1. - int d_of_p = ThreadPool::DegreeOfParallelism(&_tp); - int my_thread_idx = (_tp.CurrentThreadId() + 1) % d_of_p; - assert(my_thread_idx >= 0 && my_thread_idx < d_of_p); + // Allocate each thread to a home shard, from which it starts + // claiming iterations. + // + // We use the worker ID provided by the thread pool as the basis of + // this allocation. Doing so promotes locality between successive + // loops: the worker that runs a given iteration in one loop will + // tend to run the same iterations in the next loop. This helps + // operators with a series of short loops, such as GRU. - int home_shard; - if (d_of_p >= NUM_SHARDS) { - // More threads than shards => allocate them home shards round-robin, aiming to sprace the load across - // the shards - home_shard = my_thread_idx % NUM_SHARDS; - } else { - // Fewer threads than shards => spread the threads evenly across the shards, so each will work - // on a run of successive shards before contention - home_shard = (my_thread_idx * NUM_SHARDS) / d_of_p; - } - assert(home_shard >= 0 && home_shard < NUM_SHARDS); - return home_shard; - } + unsigned GetHomeShard(unsigned idx) const { + return idx % _num_shards; + } // Attempt to claim iterations from the sharded counter. The function either // returns true, along with a block of exactly block_size iterations, or it returns false // if all of the iterations have been claimed. - bool ClaimIterations(int my_home_shard, - int& my_shard, + bool ClaimIterations(unsigned my_home_shard, + unsigned& my_shard, uint64_t& my_start, uint64_t& my_end) { do { @@ -111,15 +104,32 @@ public: } // Work in the current shard is exhausted, move to the next shard, until // we are back at the home shard. - my_shard = (my_shard + 1) % NUM_SHARDS; + my_shard = (my_shard + 1) % _num_shards; } while (my_shard != my_home_shard); return false; } private: - alignas(CACHE_LINE_BYTES) LoopCounterShard _shards[NUM_SHARDS]; - const ThreadPool& _tp; + // Derive the number of shards to use for a given loop. We require + // at least one block of work per shard, and subject to that + // constraint we use [1,MAX_SHARDS) shards. + static unsigned GetNumShards(uint64_t num_iterations, + uint64_t block_size) { + unsigned num_shards; + auto num_blocks = num_iterations / block_size; + if (num_blocks == 0) { + num_shards = 1; + } else if (num_blocks < MAX_SHARDS) { + num_shards = static_cast(num_blocks); + } else { + num_shards = MAX_SHARDS; + } + return num_shards; + } + + alignas(CACHE_LINE_BYTES) LoopCounterShard _shards[MAX_SHARDS]; const uint64_t _block_size; + const unsigned _num_shards; }; #ifdef _MSC_VER @@ -135,7 +145,7 @@ ThreadPool::ThreadPool(Env* env, // In the current implementation, a thread pool with degree_of_parallelism==1 uses // the caller as one of the threads for executing work. Hence we only create // additional thread(s) for degree_of_parallelism>=2. - ORT_ENFORCE(degree_of_parallelism >= 1); + assert(degree_of_parallelism >= 1); if (degree_of_parallelism >= 2) { int threads_to_create = degree_of_parallelism - 1; extended_eigen_threadpool_ = @@ -167,13 +177,14 @@ void ThreadPool::ParallelForFixedBlockSizeScheduling(const std::ptrdiff_t total, // Split the work across threads in the pool. Each work item will run a loop claiming iterations, // hence we need at most one for each thread, even if the numberof blocks of iterations is larger. auto d_of_p = DegreeOfParallelism(this); - int num_work_items = static_cast(std::min(static_cast(d_of_p), total)); + auto num_blocks = total / block_size; + int num_work_items = static_cast(std::min(static_cast(d_of_p), num_blocks)); assert(num_work_items > 0); - LoopCounter lc(*this, total, block_size); - std::function run_work = [&]() { - int my_home_shard = lc.GetHomeShard(); - int my_shard = my_home_shard; + LoopCounter lc(total, block_size); + std::function run_work = [&](unsigned idx) { + unsigned my_home_shard = lc.GetHomeShard(idx); + unsigned my_shard = my_home_shard; uint64_t my_iter_start, my_iter_end; while (lc.ClaimIterations(my_home_shard, my_shard, my_iter_start, my_iter_end)) { fn(static_cast(my_iter_start), @@ -196,7 +207,6 @@ void ThreadPool::SimpleParallelFor(std::ptrdiff_t total, const std::function fn) { - ORT_ENFORCE(fn != nullptr); if (underlying_threadpool_) { underlying_threadpool_->Schedule(std::move(fn)); } else { @@ -204,12 +214,48 @@ void ThreadPool::Schedule(std::function fn) { } } -void ThreadPool::RunInParallel(std::function fn, int n) { - ORT_ENFORCE(fn != nullptr); +thread_local ThreadPool::ParallelSection *ThreadPool::ParallelSection::current_parallel_section{nullptr}; + +ThreadPool::ParallelSection::ParallelSection(ThreadPool *tp) { +#ifdef _OPENMP + // Nothing + ORT_UNUSED_PARAMETER(tp); +#else + ORT_ENFORCE(!current_parallel_section, "Nested parallelism not supported"); + ORT_ENFORCE(!ps_.get()); + tp_ = tp; + if (tp && tp->underlying_threadpool_) { + ps_ = tp->underlying_threadpool_->AllocateParallelSection(); + tp_->underlying_threadpool_->StartParallelSection(*ps_.get()); + current_parallel_section = this; + } +#endif +} + +ThreadPool::ParallelSection::~ParallelSection() { +#ifdef _OPENMP + // Nothing +#else + if (current_parallel_section) { + tp_->underlying_threadpool_->EndParallelSection(*ps_.get()); + ps_.reset(); + current_parallel_section = nullptr; + } +#endif +} + +void ThreadPool::RunInParallel(std::function fn, unsigned n) { if (underlying_threadpool_) { - underlying_threadpool_->RunInParallel(std::move(fn), n); + if (ThreadPool::ParallelSection::current_parallel_section) { + underlying_threadpool_->RunInParallelSection(*(ThreadPool::ParallelSection::current_parallel_section->ps_.get()), + std::move(fn), + n); + } else { + underlying_threadpool_->RunInParallel(std::move(fn), + n); + } } else { - fn(); + fn(0); } } diff --git a/onnxruntime/core/providers/cpu/rnn/deep_cpu_gru.cc b/onnxruntime/core/providers/cpu/rnn/deep_cpu_gru.cc index ca395808c3..59e788a4cb 100644 --- a/onnxruntime/core/providers/cpu/rnn/deep_cpu_gru.cc +++ b/onnxruntime/core/providers/cpu/rnn/deep_cpu_gru.cc @@ -580,198 +580,206 @@ void UniDirectionalGru::Compute(const gsl::span& inputs_arg, } } - // for each item in sequence run all calculations - for (int step = 0; step < max_sequence_length; step++) { + { + // Enter a parallel section encompassing the kernels invoked + // below. This lets the runtime system amortize loop entry/exit + // costs over a series of short kernels, and promotes cache + // affinity between iterations of successive loops. + onnxruntime::concurrency::ThreadPool::ParallelSection ps(ttp_); + + // for each item in sequence run all calculations + for (int step = 0; step < max_sequence_length; step++) { #if defined(DUMP_MATRIXES) - const std::string seqno_str = " [seqno=" + std::to_string(step) + "]"; + const std::string seqno_str = " [seqno=" + std::to_string(step) + "]"; #endif - DumpMatrix("Ht-1" + seqno_str, &*prev_Ht, batch_size_, hidden_size_); + DumpMatrix("Ht-1" + seqno_str, &*prev_Ht, batch_size_, hidden_size_); - out_added_offset = (step * batch_size_) * hidden_size_x3; + out_added_offset = (step * batch_size_) * hidden_size_x3; - // calculate Ht-1*R[zr], and add to the weighted inputs that are in outputZRH_ - // Ht-1 * R[zr] + Xt*(W[zr]^T) - ComputeGemm(batch_size_, hidden_size_x2, hidden_size_, alpha, - prev_Ht, prev_Ht_end, - hidden_size_, - recurrent_weightsZR.cbegin(), recurrent_weightsZR.cend(), - hidden_size_, 1.f, // beta == 1 so we add existing values in outputZRH_ - outputZRH_.begin() + out_added_offset, outputZRH_.end(), - hidden_size_x3, ttp_); - - DumpMatrix("Ht-1 * R[zr] + Xt*(W[zr]^T)" + seqno_str, - outputZRH_.data() + out_added_offset, batch_size_, hidden_size_x2, 0, hidden_size_x3); - - if (linear_before_reset_) { - // copy Rbh to linear output - if (use_bias_) { - gsl::copy(batched_bias_Rh_.subspan(batched_bias_Rh_local - batched_bias_Rh_.begin(), - batched_bias_Rh_local_end - batched_bias_Rh_local), - linear_output_); - } - - // compute Ht-1 * (Rh^T) + Rbh - ComputeGemm(batch_size_, hidden_size_, hidden_size_, alpha, - prev_Ht, prev_Ht_end, // Ht-1 + // calculate Ht-1*R[zr], and add to the weighted inputs that are in outputZRH_ + // Ht-1 * R[zr] + Xt*(W[zr]^T) + ComputeGemm(batch_size_, hidden_size_x2, hidden_size_, alpha, + prev_Ht, prev_Ht_end, hidden_size_, - recurrent_weightsH.cbegin(), recurrent_weightsH.cend(), // Rh^T - hidden_size_, - use_bias_ ? 1.f : 0.f, // don't add values in linear_output_ if no bias input - linear_output_.begin(), - linear_output_.end(), // pre: Rbh if use_bias_, post:output - hidden_size_, ttp_); + recurrent_weightsZR.cbegin(), recurrent_weightsZR.cend(), + hidden_size_, 1.f, // beta == 1 so we add existing values in outputZRH_ + outputZRH_.begin() + out_added_offset, outputZRH_.end(), + hidden_size_x3, ttp_); - DumpMatrix("Ht-1 * (Rh^T) + Rbh " + seqno_str, linear_output_.data(), batch_size_, hidden_size_); - } - - // 1st Set Of Activations - for (int r = 0; r < batch_size_; r++) { - const T* p_bias_r = use_bias_ ? SafeRawConstPointer(batched_bias_WRr_local + r * hidden_size_, - batched_bias_WRr_local_end, hidden_size_) - : nullptr; - - // initialize p_rt with input to calculate rt. outputZRH_ has Xt*(Wr^T) + Ht-1*(Rr^T). - T* p_rt = SafeRawPointer(outputZRH_, out_added_offset + r * hidden_size_x3 + hidden_size_, hidden_size_); - - // add the bias and clip. post: p_rt == Xt*(Wr^T) + Ht-1*(Rr^T) + Wbr + Rbr - clip_with_bias_ptr_(clip_, p_bias_r, p_rt, hidden_size_); + DumpMatrix("Ht-1 * R[zr] + Xt*(W[zr]^T)" + seqno_str, + outputZRH_.data() + out_added_offset, batch_size_, hidden_size_x2, 0, hidden_size_x3); if (linear_before_reset_) { - // p_linear_output = Ht-1 * (Rh^T) + Rbh - T* p_linear_output = SafeRawPointer(linear_output_, r * hidden_size_, hidden_size_); - T* p_cur_h = SafeRawPointer(cur_h_local + r * hidden_size_, cur_h_local_end, hidden_size_); + // copy Rbh to linear output + if (use_bias_) { + gsl::copy(batched_bias_Rh_.subspan(batched_bias_Rh_local - batched_bias_Rh_.begin(), + batched_bias_Rh_local_end - batched_bias_Rh_local), + linear_output_); + } - // calculate rt in-place [p_rt = f(p_rt)] - // calculate rt (.) (Ht-1 * (Rh^T) + Rbh) using p_linear_output. write to p_cur_h - reset_gate_(p_linear_output, p_rt, p_cur_h, hidden_size_, zr_alpha_, zr_beta_); + // compute Ht-1 * (Rh^T) + Rbh + ComputeGemm(batch_size_, hidden_size_, hidden_size_, alpha, + prev_Ht, prev_Ht_end, // Ht-1 + hidden_size_, + recurrent_weightsH.cbegin(), recurrent_weightsH.cend(), // Rh^T + hidden_size_, + use_bias_ ? 1.f : 0.f, // don't add values in linear_output_ if no bias input + linear_output_.begin(), + linear_output_.end(), // pre: Rbh if use_bias_, post:output + hidden_size_, ttp_); - } else { - const T* p_prev_Ht = SafeRawConstPointer(prev_Ht + r * hidden_size_, prev_Ht_end, hidden_size_); - T* p_cur_h = SafeRawPointer(cur_h_local + r * hidden_size_, cur_h_local_end, hidden_size_); - - // calculate rt in-place [p_rt = f(p_rt)] - // calculate rt (.) Ht-1 using p_prev_Ht, and write to p_cur_h - reset_gate_(p_prev_Ht, p_rt, p_cur_h, hidden_size_, zr_alpha_, zr_beta_); + DumpMatrix("Ht-1 * (Rh^T) + Rbh " + seqno_str, linear_output_.data(), batch_size_, hidden_size_); } - } - -#if defined(DUMP_MATRIXES) - std::string label = linear_before_reset_ ? "rt (.) (Ht-1 * (Rh^T) + Rbh)" : "rt (.) Ht-1"; -#endif - DumpMatrix(label + seqno_str, &*cur_h_local, batch_size_, hidden_size_); - - if (linear_before_reset_) { - // input contains rt (.) (Ht-1*(Rh^T) + Rbh) - auto input = cur_h_local; - // out_H currently contains Xt*(W[zrh]^T). - auto out_H = outputZRH_.begin() + out_added_offset; + // 1st Set Of Activations for (int r = 0; r < batch_size_; r++) { - // skip over the inputs with Z and R weights - out_H += hidden_size_x2; - for (int h = 0; h < hidden_size_; ++h) { - *out_H += *input; - ++out_H; - ++input; - } - } - } else { -#if defined(DUMP_MATRIXES) - label += " * Rh^T"; -#endif + const T* p_bias_r = use_bias_ ? SafeRawConstPointer(batched_bias_WRr_local + r * hidden_size_, + batched_bias_WRr_local_end, hidden_size_) + : nullptr; - // out_H currently contains Xt*(Wh^T). - auto out_H = outputZRH_.begin() + out_added_offset + hidden_size_x2; + // initialize p_rt with input to calculate rt. outputZRH_ has Xt*(Wr^T) + Ht-1*(Rr^T). + T* p_rt = SafeRawPointer(outputZRH_, out_added_offset + r * hidden_size_x3 + hidden_size_, hidden_size_); - // Calculate Xt*(Wh^T) + rt (.) Ht-1 * Rh - ComputeGemm(batch_size_, hidden_size_, hidden_size_, alpha, - cur_h_local, cur_h_local_end, // rt (.) Ht-1 - hidden_size_, - recurrent_weightsH.cbegin(), recurrent_weightsH.cend(), // Rh^T - hidden_size_, 1.f, // beta == 1 to add Xt*(Wh^T) from out_H - out_H, outputZRH_.end(), - hidden_size_x3, ttp_); - } + // add the bias and clip. post: p_rt == Xt*(Wr^T) + Ht-1*(Rr^T) + Wbr + Rbr + clip_with_bias_ptr_(clip_, p_bias_r, p_rt, hidden_size_); - DumpMatrix("Xt*(Wh^T) + (" + label + ")" + seqno_str, outputZRH_.data() + out_added_offset, - batch_size_, hidden_size_, hidden_size_x2, hidden_size_x3); - - //2nd Set of Activations - span_T_iter output; - span_T_iter output_end; - if (output_sequence) { - output = outputs.begin() + step * output_step_length; - output_end = outputs.end(); - - } else { - output = final_hidden_state.begin(); - output_end = final_hidden_state.end(); - } - - for (int r = 0; r < batch_size_; r++) { - if (step >= min_sequence_length && step >= sequence_lengths[r]) { - // if we need output for every step, - // or we need to set prev_Ht for an empty sequence to avoid warnings about using uninitialized values - if (output_sequence || (step == 0 && sequence_lengths[r] == 0)) { - auto fill_output = output + r * hidden_size_; - std::fill_n(&*fill_output, hidden_size_, T{}); - } - - continue; - } - - const T* p_bias_z = use_bias_ ? SafeRawConstPointer(batched_bias_WRz_local, - batched_bias_WRz_local_end, hidden_size_) - : nullptr; - - // initialize p_zt with Xt*(Wz^T) + Ht-1*(Rz^T), which is most of the input to calculate zt: - T* p_zt = SafeRawPointer(outputZRH_, out_added_offset + r * hidden_size_x3, hidden_size_); - - // using p_zt, add bias and clip in-place - clip_with_bias_ptr_(clip_, p_bias_z, p_zt, hidden_size_); - - // calculate zt in-place. p_zt = f(p_zt) - update_gate_(p_zt, hidden_size_, zr_alpha_, zr_beta_); - - DumpMatrix("zt[" + std::to_string(r) + "]" + seqno_str, p_zt, 1, hidden_size_); - - const T* p_bias_h = nullptr; - if (use_bias_) { if (linear_before_reset_) { - // Wbh - p_bias_h = SafeRawConstPointer(batched_bias_Wh_local + r * hidden_size_, - batched_bias_Wh_local_end, hidden_size_); + // p_linear_output = Ht-1 * (Rh^T) + Rbh + T* p_linear_output = SafeRawPointer(linear_output_, r * hidden_size_, hidden_size_); + T* p_cur_h = SafeRawPointer(cur_h_local + r * hidden_size_, cur_h_local_end, hidden_size_); + + // calculate rt in-place [p_rt = f(p_rt)] + // calculate rt (.) (Ht-1 * (Rh^T) + Rbh) using p_linear_output. write to p_cur_h + reset_gate_(p_linear_output, p_rt, p_cur_h, hidden_size_, zr_alpha_, zr_beta_); } else { - // Wbh + Wrh - p_bias_h = SafeRawConstPointer(batched_bias_WRh_local + r * hidden_size_, - batched_bias_WRh_local_end, hidden_size_); + const T* p_prev_Ht = SafeRawConstPointer(prev_Ht + r * hidden_size_, prev_Ht_end, hidden_size_); + T* p_cur_h = SafeRawPointer(cur_h_local + r * hidden_size_, cur_h_local_end, hidden_size_); + + // calculate rt in-place [p_rt = f(p_rt)] + // calculate rt (.) Ht-1 using p_prev_Ht, and write to p_cur_h + reset_gate_(p_prev_Ht, p_rt, p_cur_h, hidden_size_, zr_alpha_, zr_beta_); } } - // setup p_ht with input to calculate ht - // p_ht = Xt*(Wh^T) + (rt (.) Ht-1 * Rh^T) # linear_before_reset_ == false - // = Xt*(Wh^T) + (rt (.) (Ht-1*(Rh^T) + Rbh)) # linear_before_reset_ == true - T* p_ht = SafeRawPointer(outputZRH_, out_added_offset + r * hidden_size_x3 + hidden_size_x2, hidden_size_); +#if defined(DUMP_MATRIXES) + std::string label = linear_before_reset_ ? "rt (.) (Ht-1 * (Rh^T) + Rbh)" : "rt (.) Ht-1"; +#endif + DumpMatrix(label + seqno_str, &*cur_h_local, batch_size_, hidden_size_); - // add Wbh [and Wrh] and clip - clip_with_bias_ptr_(clip_, p_bias_h, p_ht, hidden_size_); // post: p_ht == input to g() for calculating ht + if (linear_before_reset_) { + // input contains rt (.) (Ht-1*(Rh^T) + Rbh) + auto input = cur_h_local; + // out_H currently contains Xt*(W[zrh]^T). + auto out_H = outputZRH_.begin() + out_added_offset; - DumpMatrix("ht input [" + std::to_string(r) + "]" + seqno_str, p_ht, 1, hidden_size_); + for (int r = 0; r < batch_size_; r++) { + // skip over the inputs with Z and R weights + out_H += hidden_size_x2; + for (int h = 0; h < hidden_size_; ++h) { + *out_H += *input; + ++out_H; + ++input; + } + } + } else { +#if defined(DUMP_MATRIXES) + label += " * Rh^T"; +#endif - const T* p_prev_Ht = SafeRawConstPointer(prev_Ht + r * hidden_size_, prev_Ht_end, hidden_size_); - T* p_Ht = SafeRawPointer(output + r * hidden_size_, output_end, hidden_size_); + // out_H currently contains Xt*(Wh^T). + auto out_H = outputZRH_.begin() + out_added_offset + hidden_size_x2; - // calculate ht = g(p_ht) and write in-place to p_ht - // calculate Ht = (1 - zt) (.) ht + zt (.) Ht-1 and write to p_Ht - output_gate_(p_ht, p_zt, p_prev_Ht, p_Ht, hidden_size_, h_alpha_, h_beta_); // calculate ht and Ht + // Calculate Xt*(Wh^T) + rt (.) Ht-1 * Rh + ComputeGemm(batch_size_, hidden_size_, hidden_size_, alpha, + cur_h_local, cur_h_local_end, // rt (.) Ht-1 + hidden_size_, + recurrent_weightsH.cbegin(), recurrent_weightsH.cend(), // Rh^T + hidden_size_, 1.f, // beta == 1 to add Xt*(Wh^T) from out_H + out_H, outputZRH_.end(), + hidden_size_x3, ttp_); + } + + DumpMatrix("Xt*(Wh^T) + (" + label + ")" + seqno_str, outputZRH_.data() + out_added_offset, + batch_size_, hidden_size_, hidden_size_x2, hidden_size_x3); + + //2nd Set of Activations + span_T_iter output; + span_T_iter output_end; + if (output_sequence) { + output = outputs.begin() + step * output_step_length; + output_end = outputs.end(); + + } else { + output = final_hidden_state.begin(); + output_end = final_hidden_state.end(); + } + + for (int r = 0; r < batch_size_; r++) { + if (step >= min_sequence_length && step >= sequence_lengths[r]) { + // if we need output for every step, + // or we need to set prev_Ht for an empty sequence to avoid warnings about using uninitialized values + if (output_sequence || (step == 0 && sequence_lengths[r] == 0)) { + auto fill_output = output + r * hidden_size_; + std::fill_n(&*fill_output, hidden_size_, T{}); + } + + continue; + } + + const T* p_bias_z = use_bias_ ? SafeRawConstPointer(batched_bias_WRz_local, + batched_bias_WRz_local_end, hidden_size_) + : nullptr; + + // initialize p_zt with Xt*(Wz^T) + Ht-1*(Rz^T), which is most of the input to calculate zt: + T* p_zt = SafeRawPointer(outputZRH_, out_added_offset + r * hidden_size_x3, hidden_size_); + + // using p_zt, add bias and clip in-place + clip_with_bias_ptr_(clip_, p_bias_z, p_zt, hidden_size_); + + // calculate zt in-place. p_zt = f(p_zt) + update_gate_(p_zt, hidden_size_, zr_alpha_, zr_beta_); + + DumpMatrix("zt[" + std::to_string(r) + "]" + seqno_str, p_zt, 1, hidden_size_); + + const T* p_bias_h = nullptr; + if (use_bias_) { + if (linear_before_reset_) { + // Wbh + p_bias_h = SafeRawConstPointer(batched_bias_Wh_local + r * hidden_size_, + batched_bias_Wh_local_end, hidden_size_); + + } else { + // Wbh + Wrh + p_bias_h = SafeRawConstPointer(batched_bias_WRh_local + r * hidden_size_, + batched_bias_WRh_local_end, hidden_size_); + } + } + + // setup p_ht with input to calculate ht + // p_ht = Xt*(Wh^T) + (rt (.) Ht-1 * Rh^T) # linear_before_reset_ == false + // = Xt*(Wh^T) + (rt (.) (Ht-1*(Rh^T) + Rbh)) # linear_before_reset_ == true + T* p_ht = SafeRawPointer(outputZRH_, out_added_offset + r * hidden_size_x3 + hidden_size_x2, hidden_size_); + + // add Wbh [and Wrh] and clip + clip_with_bias_ptr_(clip_, p_bias_h, p_ht, hidden_size_); // post: p_ht == input to g() for calculating ht + + DumpMatrix("ht input [" + std::to_string(r) + "]" + seqno_str, p_ht, 1, hidden_size_); + + const T* p_prev_Ht = SafeRawConstPointer(prev_Ht + r * hidden_size_, prev_Ht_end, hidden_size_); + T* p_Ht = SafeRawPointer(output + r * hidden_size_, output_end, hidden_size_); + + // calculate ht = g(p_ht) and write in-place to p_ht + // calculate Ht = (1 - zt) (.) ht + zt (.) Ht-1 and write to p_Ht + output_gate_(p_ht, p_zt, p_prev_Ht, p_Ht, hidden_size_, h_alpha_, h_beta_); // calculate ht and Ht + } + + DumpMatrix("output" + seqno_str, &*output, batch_size_, hidden_size_); + + prev_Ht = output; + prev_Ht_end = output_end; } - - DumpMatrix("output" + seqno_str, &*output, batch_size_, hidden_size_); - - prev_Ht = output; - prev_Ht_end = output_end; - } + } // End parallel section // copy last output to final_hidden_state for (int i = 0; i < batch_size_; i++) { diff --git a/onnxruntime/test/platform/threadpool_test.cc b/onnxruntime/test/platform/threadpool_test.cc index f2e4930081..ba6e8d96dd 100644 --- a/onnxruntime/test/platform/threadpool_test.cc +++ b/onnxruntime/test/platform/threadpool_test.cc @@ -39,14 +39,23 @@ void IncrementElement(TestData& test_data, ptrdiff_t i) { test_data.data[i]++; } -void ValidateTestData(TestData& test_data) { - ASSERT_TRUE(std::count_if(test_data.data.cbegin(), test_data.data.cend(), [](int i) { return i != 1; }) == 0); +void ValidateTestData(TestData& test_data, int expected=1) { + ASSERT_TRUE(std::count_if(test_data.data.cbegin(), test_data.data.cend(), [&](int i) { return i != expected; }) == 0); } +// Run a test with a new thread pool created with num_threads threads +// in total (including the main thread). If num_threads is 0 then we +// test the function with a null pointer, reflecting scenarios where we +// run with just the main thread. Note that the thread pool API uses +// static methods and should operate across all of these cases. void CreateThreadPoolAndTest(const std::string&, int num_threads, const std::function& test_body) { - auto tp = onnxruntime::make_unique(&onnxruntime::Env::Default(), onnxruntime::ThreadOptions(), nullptr, - num_threads, true); - test_body(tp.get()); + if (num_threads > 0) { + auto tp = onnxruntime::make_unique(&onnxruntime::Env::Default(), onnxruntime::ThreadOptions(), nullptr, + num_threads, true); + test_body(tp.get()); + } else { + test_body(nullptr); + } } void TestParallelFor(const std::string& name, int num_threads, int num_tasks) { @@ -67,7 +76,7 @@ void TestBatchParallelFor(const std::string& name, int num_threads, int num_task ValidateTestData(*test_data); } -void TestMultipleParallelFor(const std::string& name, int num_threads, int num_concurrent, int num_tasks) { +void TestConcurrentParallelFor(const std::string& name, int num_threads, int num_concurrent, int num_tasks) { // Test running multiple concurrent loops over the same thread pool. This aims to provoke a // more diverse mix of interleavings than with a single loop running at a time. for (int rep = 0; rep < 5; rep++) { @@ -161,9 +170,35 @@ void TestPoolCreation(const std::string&, int iter) { ASSERT_EQ(ctr, iter * per_iter); } +void TestMultiLoopSections(const std::string& name, int num_threads, int num_loops) { + for (int rep = 0; rep < 5; rep++) { + const int num_tasks = 1024; + auto test_data = CreateTestData(num_tasks); + CreateThreadPoolAndTest(name, num_threads, [&](ThreadPool* tp) { + ThreadPool::ParallelSection ps(tp); + for (int l = 0; l < num_loops; l++) { + ThreadPool::TrySimpleParallelFor(tp, + num_tasks, + [&](std::ptrdiff_t i) { + IncrementElement(*test_data, i); + }); + } + }); + ValidateTestData(*test_data, num_loops); + } +} + } // namespace namespace onnxruntime { +TEST(ThreadPoolTest, TestParallelFor_0_Thread_NoTask) { + TestParallelFor("TestParallelFor_0_Thread_NoTask", 0, 0); +} + +TEST(ThreadPoolTest, TestParallelFor_0_Thread_50_Task) { + TestParallelFor("TestParallelFor_0_Thread_50_Task", 0, 50); +} + TEST(ThreadPoolTest, TestParallelFor_2_Thread_NoTask) { TestParallelFor("TestParallelFor_2_Thread_NoTask", 2, 0); } @@ -176,6 +211,10 @@ TEST(ThreadPoolTest, TestParallelFor_1_Thread_50_Task) { TestParallelFor("TestParallelFor_1_Thread_50_Task", 1, 50); } +TEST(ThreadPoolTest, TestBatchParallelFor_0_Thread_50_Task_10_Batch) { + TestBatchParallelFor("TestBatchParallelFor_0_Thread_50_Task_10_Batch", 0, 50, 10); +} + TEST(ThreadPoolTest, TestBatchParallelFor_2_Thread_50_Task_10_Batch) { TestBatchParallelFor("TestBatchParallelFor_2_Thread_50_Task_10_Batch", 2, 50, 10); } @@ -196,68 +235,72 @@ TEST(ThreadPoolTest, TestBatchParallelFor_2_Thread_81_Task_20_Batch) { TestBatchParallelFor("TestBatchParallelFor_2_Thread_81_Task_20_Batch", 2, 81, 20); } -TEST(ThreadPoolTest, TestMultipleParallelFor_1Thread_1Conc_0Tasks) { - TestMultipleParallelFor("TestMultipleParallelFor_1Thread_1Conc_0Tasks", 1, 1, 0); +TEST(ThreadPoolTest, TestConcurrentParallelFor_0Thread_1Conc_0Tasks) { + TestConcurrentParallelFor("TestConcurrentParallelFor_0Thread_1Conc_0Tasks", 0, 1, 0); } -TEST(ThreadPoolTest, TestMultipleParallelFor_1Thread_1Conc_1Tasks) { - TestMultipleParallelFor("TestMultipleParallelFor_1Thread_1Conc_1Tasks", 1, 1, 1); +TEST(ThreadPoolTest, TestConcurrentParallelFor_1Thread_1Conc_0Tasks) { + TestConcurrentParallelFor("TestConcurrentParallelFor_1Thread_1Conc_0Tasks", 1, 1, 0); } -TEST(ThreadPoolTest, TestMultipleParallelFor_1Thread_1Conc_8Tasks) { - TestMultipleParallelFor("TestMultipleParallelFor_1Thread_1Conc_8Tasks", 1, 1, 8); +TEST(ThreadPoolTest, TestConcurrentParallelFor_1Thread_1Conc_1Tasks) { + TestConcurrentParallelFor("TestConcurrentParallelFor_1Thread_1Conc_1Tasks", 1, 1, 1); } -TEST(ThreadPoolTest, TestMultipleParallelFor_1Thread_1Conc_1MTasks) { - TestMultipleParallelFor("TestMultipleParallelFor_1Thread_1Conc_1MTasks", 1, 1, 1000000); +TEST(ThreadPoolTest, TestConcurrentParallelFor_1Thread_1Conc_8Tasks) { + TestConcurrentParallelFor("TestConcurrentParallelFor_1Thread_1Conc_8Tasks", 1, 1, 8); } -TEST(ThreadPoolTest, TestMultipleParallelFor_1Thread_4Conc_0Tasks) { - TestMultipleParallelFor("TestMultipleParallelFor_1Thread_4Conc_0Tasks", 1, 4, 0); +TEST(ThreadPoolTest, TestConcurrentParallelFor_1Thread_1Conc_1MTasks) { + TestConcurrentParallelFor("TestConcurrentParallelFor_1Thread_1Conc_1MTasks", 1, 1, 1000000); } -TEST(ThreadPoolTest, TestMultipleParallelFor_1Thread_4Conc_1Tasks) { - TestMultipleParallelFor("TestMultipleParallelFor_1Thread_4Conc_1Tasks", 1, 4, 1); +TEST(ThreadPoolTest, TestConcurrentParallelFor_1Thread_4Conc_0Tasks) { + TestConcurrentParallelFor("TestConcurrentParallelFor_1Thread_4Conc_0Tasks", 1, 4, 0); } -TEST(ThreadPoolTest, TestMultipleParallelFor_1Thread_4Conc_8Tasks) { - TestMultipleParallelFor("TestMultipleParallelFor_1Thread_4Conc_8Tasks", 1, 4, 8); +TEST(ThreadPoolTest, TestConcurrentParallelFor_1Thread_4Conc_1Tasks) { + TestConcurrentParallelFor("TestConcurrentParallelFor_1Thread_4Conc_1Tasks", 1, 4, 1); } -TEST(ThreadPoolTest, TestMultipleParallelFor_1Thread_4Conc_1MTasks) { - TestMultipleParallelFor("TestMultipleParallelFor_1Thread_4Conc_1MTasks", 1, 4, 1000000); +TEST(ThreadPoolTest, TestConcurrentParallelFor_1Thread_4Conc_8Tasks) { + TestConcurrentParallelFor("TestConcurrentParallelFor_1Thread_4Conc_8Tasks", 1, 4, 8); } -TEST(ThreadPoolTest, TestMultipleParallelFor_4Thread_1Conc_0Tasks) { - TestMultipleParallelFor("TestMultipleParallelFor_4Thread_4Conc_0Tasks", 4, 1, 0); +TEST(ThreadPoolTest, TestConcurrentParallelFor_1Thread_4Conc_1MTasks) { + TestConcurrentParallelFor("TestConcurrentParallelFor_1Thread_4Conc_1MTasks", 1, 4, 1000000); } -TEST(ThreadPoolTest, TestMultipleParallelFor_4Thread_1Conc_1Tasks) { - TestMultipleParallelFor("TestMultipleParallelFor_4Thread_4Conc_1Tasks", 4, 1, 1); +TEST(ThreadPoolTest, TestConcurrentParallelFor_4Thread_1Conc_0Tasks) { + TestConcurrentParallelFor("TestConcurrentParallelFor_4Thread_4Conc_0Tasks", 4, 1, 0); } -TEST(ThreadPoolTest, TestMultipleParallelFor_4Thread_1Conc_8Tasks) { - TestMultipleParallelFor("TestMultipleParallelFor_4Thread_4Conc_8Tasks", 4, 1, 8); +TEST(ThreadPoolTest, TestConcurrentParallelFor_4Thread_1Conc_1Tasks) { + TestConcurrentParallelFor("TestConcurrentParallelFor_4Thread_4Conc_1Tasks", 4, 1, 1); } -TEST(ThreadPoolTest, TestMultipleParallelFor_4Thread_1Conc_1MTasks) { - TestMultipleParallelFor("TestMultipleParallelFor_4Thread_4Conc_1MTasks", 4, 1, 1000000); +TEST(ThreadPoolTest, TestConcurrentParallelFor_4Thread_1Conc_8Tasks) { + TestConcurrentParallelFor("TestConcurrentParallelFor_4Thread_4Conc_8Tasks", 4, 1, 8); } -TEST(ThreadPoolTest, TestMultipleParallelFor_4Thread_4Conc_0Tasks) { - TestMultipleParallelFor("TestMultipleParallelFor_4Thread_4Conc_0Tasks", 4, 4, 0); +TEST(ThreadPoolTest, TestConcurrentParallelFor_4Thread_1Conc_1MTasks) { + TestConcurrentParallelFor("TestConcurrentParallelFor_4Thread_4Conc_1MTasks", 4, 1, 1000000); } -TEST(ThreadPoolTest, TestMultipleParallelFor_4Thread_4Conc_1Tasks) { - TestMultipleParallelFor("TestMultipleParallelFor_4Thread_4Conc_1Tasks", 4, 4, 1); +TEST(ThreadPoolTest, TestConcurrentParallelFor_4Thread_4Conc_0Tasks) { + TestConcurrentParallelFor("TestConcurrentParallelFor_4Thread_4Conc_0Tasks", 4, 4, 0); } -TEST(ThreadPoolTest, TestMultipleParallelFor_4Thread_4Conc_8Tasks) { - TestMultipleParallelFor("TestMultipleParallelFor_4Thread_4Conc_8Tasks", 4, 4, 8); +TEST(ThreadPoolTest, TestConcurrentParallelFor_4Thread_4Conc_1Tasks) { + TestConcurrentParallelFor("TestConcurrentParallelFor_4Thread_4Conc_1Tasks", 4, 4, 1); } -TEST(ThreadPoolTest, TestMultipleParallelFor_4Thread_4Conc_1MTasks) { - TestMultipleParallelFor("TestMultipleParallelFor_4Thread_4Conc_1MTasks", 4, 4, 1000000); +TEST(ThreadPoolTest, TestConcurrentParallelFor_4Thread_4Conc_8Tasks) { + TestConcurrentParallelFor("TestConcurrentParallelFor_4Thread_4Conc_8Tasks", 4, 4, 8); +} + +TEST(ThreadPoolTest, TestConcurrentParallelFor_4Thread_4Conc_1MTasks) { + TestConcurrentParallelFor("TestConcurrentParallelFor_4Thread_4Conc_1MTasks", 4, 4, 1000000); } TEST(ThreadPoolTest, TestBurstScheduling_0Tasks) { @@ -290,6 +333,54 @@ TEST(ThreadPoolTest, TestPoolCreation_100Iter) { TestPoolCreation("TestPoolCreation_100Iter", 100); } +TEST(ThreadPoolTest, TestMultiLoopSections_0Thread_0Loop) { + TestMultiLoopSections("TestMultiLoopSections_0Thread_0Loop", 0, 0); +} + +TEST(ThreadPoolTest, TestMultiLoopSections_0Thread_1Loop) { + TestMultiLoopSections("TestMultiLoopSections_0Thread_1Loop", 0, 1); +} + +TEST(ThreadPoolTest, TestMultiLoopSections_0Thread_100Loop) { + TestMultiLoopSections("TestMultiLoopSections_0Thread_100Loop", 0, 100); +} + +TEST(ThreadPoolTest, TestMultiLoopSections_1Thread_0Loop) { + TestMultiLoopSections("TestMultiLoopSections_1Thread_0Loop", 1, 0); +} + +TEST(ThreadPoolTest, TestMultiLoopSections_1Thread_1Loop) { + TestMultiLoopSections("TestMultiLoopSections_1Thread_1Loop", 1, 1); +} + +TEST(ThreadPoolTest, TestMultiLoopSections_2Thread_0Loop) { + TestMultiLoopSections("TestMultiLoopSections_2Thread_0Loop", 2, 0); +} + +TEST(ThreadPoolTest, TestMultiLoopSections_2Thread_1Loop) { + TestMultiLoopSections("TestMultiLoopSections_2Thread_1Loop", 2, 1); +} + +TEST(ThreadPoolTest, TestMultiLoopSections_2Thread_2Loop) { + TestMultiLoopSections("TestMultiLoopSections_2Thread_2Loop", 2, 2); +} + +TEST(ThreadPoolTest, TestMultiLoopSections_2Thread_100Loop) { + TestMultiLoopSections("TestMultiLoopSections_2Thread_100Loop", 2, 100); +} + +TEST(ThreadPoolTest, TestMultiLoopSections_4Thread_1Loop) { + TestMultiLoopSections("TestMultiLoopSections_4Thread_1Loop", 4, 1); +} + +TEST(ThreadPoolTest, TestMultiLoopSections_4Thread_10Loop) { + TestMultiLoopSections("TestMultiLoopSections_4Thread_10Loop", 4, 10); +} + +TEST(ThreadPoolTest, TestMultiLoopSections_4Thread_100Loop) { + TestMultiLoopSections("TestMultiLoopSections_4Thread_100Loop", 4, 100); +} + #ifdef _WIN32 TEST(ThreadPoolTest, TestStackSize) { ThreadOptions to;