saymrwulf/onnxruntime
1
0
Fork 0
mirror of https://github.com/saymrwulf/onnxruntime.git synced 2026-07-09 17:28:58 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Support multi-loop parallel sections, use multi-loop sections in GRU (#5602)

Browse source 
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.
This commit is contained in:
Tim Harris 2020-11-10 12:24:57 +00:00 committed by GitHub
parent 919c270f3c
commit 5e44d25c5a
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
6 changed files with 984 additions and 343 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

5
docs/NotesOnThreading.md
View file

@ -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.

503
include/onnxruntime/core/platform/EigenNonBlockingThreadPool.h
View file

@ -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<void()> 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<ThreadPoolParallelSection, void(*)(ThreadPoolParallelSection*)> 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<void(unsigned idx)> 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<void(unsigned idx)> 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<std::pair<int,unsigned>> tasks;
// State shared between the main thread and worker threads
// -------------------------------------------------------
// Flag to signal termination of the parallel section
std::atomic<bool> active{false};
std::atomic<unsigned> 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<unsigned> 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<ThreadPoolLoop *> current_loop{nullptr};
std::atomic<unsigned> workers_in_loop{0};
};
class ThreadPoolLoop {
public:
ThreadPoolLoop(std::function<void(unsigned)> f, unsigned t) : fn(std::move(f)), threads_needed(t) {
}
const std::function<void(unsigned)> fn;
const unsigned threads_needed;
private:
ORT_DISALLOW_COPY_ASSIGNMENT_AND_MOVE(ThreadPoolLoop);
};
template <typename Work, typename Tag, unsigned kSize>
class RunQueue {
@ -212,6 +382,11 @@ class RunQueue {
std::unique_lock<OrtMutex> 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 <typename Environment>
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<Task, Tag, 1024> 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<unsigned>& good_hints, std::vector<unsigned>& alt_hints) {
void GetGoodWorkerHints(unsigned n, std::vector<unsigned>& good_hints, std::vector<unsigned>& 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<unsigned>& good_hints, std::vector<un
// have multiple threads scheduling work concurrently.
unsigned base = Rand(&pt->rand) % 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<uint64_t>* 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<unsigned>& good_hints, std::vector<un
}
}
void RunInParallel(std::function<void()> 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 <thread,idx> 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<std::pair<int, unsigned>> 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<ThreadPoolParallelSection, void(*)(ThreadPoolParallelSection*)> AllocateParallelSection() override {
return std::unique_ptr<ThreadPoolParallelSection, void(*)(ThreadPoolParallelSection*)>
(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<unsigned>(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<void(unsigned)> &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<unsigned>(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<unsigned> 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<void()> 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<void(unsigned idx)> 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<void(unsigned idx)> 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<PerThread>::value, "Per-thread state should be trivially destructible");
static_assert(std::is_trivially_destructible<PerThread>::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<std::atomic<uint64_t>[]> 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::thread::id>()(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

156
include/onnxruntime/core/platform/threadpool.h
View file

@ -26,8 +26,94 @@ limitations under the License.
#include <functional>
#include <memory>
// 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 <typename Environment>
class ThreadPoolTempl;
namespace concurrency {
template <typename Environment>
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<ThreadPoolParallelSection, void(*)(ThreadPoolParallelSection*)>
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<decltype(current_parallel_section)>::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<void()> fn, int n);
void RunInParallel(std::function<void(unsigned idx)> 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).

158
onnxruntime/core/common/threadpool.cc
View file

@ -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<uint64_t> _next;
uint64_t _end;
::std::atomic<uint64_t> _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<double>(num_iterations) / NUM_SHARDS;
uint64_t split = 0;
for (uint64_t shard = 0; shard < NUM_SHARDS; shard++) {
_shards[shard]._next = split;
split = (static_cast<uint64_t>((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<unsigned>(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<int>(std::min(static_cast<std::ptrdiff_t>(d_of_p), total));
auto num_blocks = total / block_size;
int num_work_items = static_cast<int>(std::min(static_cast<std::ptrdiff_t>(d_of_p), num_blocks));
assert(num_work_items > 0);
LoopCounter lc(*this, total, block_size);
std::function<void()> run_work = [&]() {
int my_home_shard = lc.GetHomeShard();
int my_shard = my_home_shard;
LoopCounter lc(total, block_size);
std::function<void(unsigned)> 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<std::ptrdiff_t>(my_iter_start),
@ -196,7 +207,6 @@ void ThreadPool::SimpleParallelFor(std::ptrdiff_t total, const std::function<voi
}
void ThreadPool::Schedule(std::function<void()> fn) {
ORT_ENFORCE(fn != nullptr);
if (underlying_threadpool_) {
underlying_threadpool_->Schedule(std::move(fn));
} else {
@ -204,12 +214,48 @@ void ThreadPool::Schedule(std::function<void()> fn) {
}
}
void ThreadPool::RunInParallel(std::function<void()> 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<void(unsigned idx)> 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);
}
}

338
onnxruntime/core/providers/cpu/rnn/deep_cpu_gru.cc
View file

@ -580,198 +580,206 @@ void UniDirectionalGru<T>::Compute(const gsl::span<const T>& 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<T>(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<T>(linear_output_, r * hidden_size_, hidden_size_);
T* p_cur_h = SafeRawPointer<T>(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<T>(prev_Ht + r * hidden_size_, prev_Ht_end, hidden_size_);
T* p_cur_h = SafeRawPointer<T>(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<T>(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<T>(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<T>(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<T>(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<T>(linear_output_, r * hidden_size_, hidden_size_);
T* p_cur_h = SafeRawPointer<T>(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<T>(batched_bias_WRh_local + r * hidden_size_,
batched_bias_WRh_local_end, hidden_size_);
const T* p_prev_Ht = SafeRawConstPointer<T>(prev_Ht + r * hidden_size_, prev_Ht_end, hidden_size_);
T* p_cur_h = SafeRawPointer<T>(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<T>(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<T>(prev_Ht + r * hidden_size_, prev_Ht_end, hidden_size_);
T* p_Ht = SafeRawPointer<T>(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<T>(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<T>(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<T>(batched_bias_Wh_local + r * hidden_size_,
batched_bias_Wh_local_end, hidden_size_);
} else {
// Wbh + Wrh
p_bias_h = SafeRawConstPointer<T>(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<T>(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<T>(prev_Ht + r * hidden_size_, prev_Ht_end, hidden_size_);
T* p_Ht = SafeRawPointer<T>(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++) {

167
onnxruntime/test/platform/threadpool_test.cc
View file

@ -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<void(ThreadPool*)>& test_body) {
auto tp = onnxruntime::make_unique<ThreadPool>(&onnxruntime::Env::Default(), onnxruntime::ThreadOptions(), nullptr,
num_threads, true);
test_body(tp.get());
if (num_threads > 0) {
auto tp = onnxruntime::make_unique<ThreadPool>(&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;