saymrwulf/onnxruntime
1
0
Fork 0
mirror of https://github.com/saymrwulf/onnxruntime.git synced 2026-07-30 20:18:08 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Remove non-trivially-destructible thread-local from thread pool state, blocking ARM64 builds (#4336)

Browse source 
- Move thread hint vectors from thread-local struct

- Add static_assert that the per-thread state in the thread pool is trivially-destructible

- Rename "thread_data" to "worker_data" (only allocated for workers in the pool, not threads calling into the pool)
This commit is contained in:
Tim Harris 2020-06-25 19:04:31 +01:00 committed by GitHub
parent a3b466cdf1
commit 3fc68cb150
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
1 changed files with 52 additions and 39 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -9,6 +9,8 @@
/* Modifications Copyright (c) Microsoft. */
#include <type_traits>
#pragma once
#include "onnxruntime_config.h"
// build/external/eigen/unsupported/Eigen/CXX11/src/Tensor/TensorEvaluator.h:162:71:
@ -403,7 +405,7 @@ class ThreadPoolTempl : public onnxruntime::concurrency::ExtendedThreadPoolInter
: env_(env),
num_threads_(num_threads),
allow_spinning_(allow_spinning),
thread_data_(num_threads),
worker_data_(num_threads),
all_coprimes_(num_threads),
blocked_(0),
done_(false),
@ -427,9 +429,9 @@ class ThreadPoolTempl : public onnxruntime::concurrency::ExtendedThreadPoolInter
num_hint_words_ = static_cast<int>((num_threads_ + bits_per_hint_word_ - 1) / bits_per_hint_word_);
good_worker_hints_ = onnxruntime::make_unique<std::atomic<uint64_t>[]>(num_hint_words_);
thread_data_.resize(num_threads_);
worker_data_.resize(num_threads_);
for (int i = 0; i < num_threads_; i++) {
thread_data_[i].thread.reset(env_.CreateThread(name, i, WorkerLoop, this, thread_options));
worker_data_[i].thread.reset(env_.CreateThread(name, i, WorkerLoop, this, thread_options));
}
}
@ -444,13 +446,13 @@ class ThreadPoolTempl : public onnxruntime::concurrency::ExtendedThreadPoolInter
} else {
// Since we were cancelled, there might be entries in the queues.
// Empty them to prevent their destructor from asserting.
for (size_t i = 0; i < thread_data_.size(); i++) {
thread_data_[i].queue.Flush();
for (size_t i = 0; i < worker_data_.size(); i++) {
worker_data_[i].queue.Flush();
}
}
// Join threads explicitly (by destroying) to avoid destruction order within
// this class.
for (size_t i = 0; i < thread_data_.size(); ++i) thread_data_[i].thread.reset();
for (size_t i = 0; i < worker_data_.size(); ++i) worker_data_[i].thread.reset();
}
void Schedule(std::function<void()> fn) override {
@ -458,13 +460,13 @@ class ThreadPoolTempl : public onnxruntime::concurrency::ExtendedThreadPoolInter
PerThread* pt = GetPerThread();
if (pt->pool == this) {
// Worker thread of this pool, push onto the thread's queue.
Queue& q = thread_data_[pt->thread_id].queue;
Queue& q = worker_data_[pt->thread_id].queue;
t = q.PushFront(std::move(t));
} else {
// A free-standing thread (or worker of another pool), push onto a random
// queue.
int q_idx = Rand(&pt->rand) % num_threads_;
ThreadData &td = thread_data_[q_idx];
WorkerData &td = worker_data_[q_idx];
Queue& q = td.queue;
t = q.PushBack(std::move(t));
if (t.f) {
@ -539,9 +541,9 @@ void GetGoodWorkerHints(int n, std::vector<unsigned>& good_hints, std::vector<un
}
void RunInParallel(std::function<void()> fn, unsigned n) override {
PerThread* pt = GetPerThread();
PerThread* my_pt = GetPerThread();
assert(n>=1);
if (n == 1 || pt->in_parallel) {
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
@ -550,14 +552,13 @@ void RunInParallel(std::function<void()> fn, unsigned n) override {
std::vector<std::pair<int, unsigned>> pending_items;
Barrier b(n);
pt->in_parallel = true;
if (!pt->tag.Get()) {
pt->tag = Tag::GetNext();
my_pt->in_parallel = true;
if (!my_pt->tag.Get()) {
my_pt->tag = Tag::GetNext();
}
// Push up to n-1 copies of the work item into the queues
std::vector<unsigned>& good_hints = pt->good_hints;
std::vector<unsigned>& alt_hints = pt->alt_hints;
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]() {
@ -572,13 +573,13 @@ 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(&pt->rand) % num_threads_;
q_idx = Rand(&my_pt->rand) % num_threads_;
}
}
ThreadData& td = thread_data_[q_idx];
WorkerData& td = worker_data_[q_idx];
Queue& q = td.queue;
unsigned w_idx;
t = q.PushBackWithTag(std::move(t), pt->tag, 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
@ -600,8 +601,8 @@ void RunInParallel(std::function<void()> fn, unsigned n) override {
// revoke from the work queues
int notifications_needed = 1;
for (auto& item : pending_items) {
Queue& q = thread_data_[item.first].queue;
if (q.RevokeWithTag(pt->tag, item.second)) {
Queue& q = worker_data_[item.first].queue;
if (q.RevokeWithTag(my_pt->tag, item.second)) {
notifications_needed++;
}
}
@ -609,20 +610,20 @@ void RunInParallel(std::function<void()> fn, unsigned n) override {
// Synchronize with any work items that are still running
b.Wait();
pt->in_parallel = false;
my_pt->in_parallel = false;
}
}
void Cancel() override {
cancelled_ = true;
// If done_ is true, which means this object is being destructing.
// Therefore thread_data_[i].thread could be NULL.
// Therefore worker_data_[i].thread could be NULL.
if (!done_) {
done_ = true;
// Let each thread know it's been cancelled.
for (size_t i = 0; i < thread_data_.size(); i++) {
assert(thread_data_[i].thread != nullptr);
thread_data_[i].thread->OnCancel();
for (size_t i = 0; i < worker_data_.size(); i++) {
assert(worker_data_[i].thread != nullptr);
worker_data_[i].thread->OnCancel();
}
}
@ -672,7 +673,19 @@ int CurrentThreadId() const EIGEN_FINAL {
}
typedef typename Environment::EnvThread Thread;
struct ThreadData;
struct WorkerData;
// PerThread objects are allocated in thread-local storage and allocated
// on the thread's first call to GetPerThread. The object should
// remain trivially-destructable, with other state placed in the
// WorkerData objects that are allocated and cleaned-up explicitly.
//
// PerThread objects are allocated for all threads that submit work to
// the thread pool, in addition to threads within the pool.
//
// In contrast, the WorkerData objects are allocated only for the
// threads in the pool, and their lifetime is managed along with the
// pool.
struct PerThread {
constexpr PerThread() : pool(nullptr) {
@ -682,12 +695,12 @@ int CurrentThreadId() const EIGEN_FINAL {
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)
std::vector<unsigned> good_hints; // Vector used to pass hints of workers to use
std::vector<unsigned> alt_hints; // Vector used to pass hints of workers to use if not sufficient good hints
};
struct ThreadData {
constexpr ThreadData() : thread(), queue() {
static_assert(std::is_trivially_destructible<PerThread>::value, "Per-thread state should be trivially destructible");
struct WorkerData {
constexpr WorkerData() : thread(), queue() {
}
std::unique_ptr<Thread> thread;
Queue queue;
@ -781,7 +794,7 @@ int CurrentThreadId() const EIGEN_FINAL {
Environment& env_;
const int num_threads_;
const bool allow_spinning_;
Eigen::MaxSizeVector<ThreadData> thread_data_;
Eigen::MaxSizeVector<WorkerData> worker_data_;
Eigen::MaxSizeVector<Eigen::MaxSizeVector<unsigned>> all_coprimes_;
std::atomic<unsigned> blocked_; // Count of blocked workers, used as a termination condition
std::atomic<bool> done_;
@ -804,7 +817,7 @@ int CurrentThreadId() const EIGEN_FINAL {
// items in the work queues.
void WakeAllWorkersForExit() {
for (auto &td: thread_data_) {
for (auto &td: worker_data_) {
td.EnsureAwake();
}
}
@ -812,14 +825,14 @@ int CurrentThreadId() const EIGEN_FINAL {
// Main worker thread loop.
void WorkerLoop(int thread_id) {
PerThread* pt = GetPerThread();
ThreadData& td = thread_data_[thread_id];
WorkerData& td = worker_data_[thread_id];
Queue& q = td.queue;
bool should_exit = false;
pt->pool = this;
pt->rand = GlobalThreadIdHash();
pt->thread_id = thread_id;
assert(td.GetStatus() == ThreadData::ThreadStatus::Spinning);
assert(td.GetStatus() == WorkerData::ThreadStatus::Spinning);
SetGoodWorkerHint(thread_id, true /* Is good */);
const int log2_spin = 20;
@ -857,7 +870,7 @@ int CurrentThreadId() const EIGEN_FINAL {
if (victim != -1) {
should_block = false;
if (!cancelled_) {
t = thread_data_[victim].queue.PopBack();
t = worker_data_[victim].queue.PopBack();
}
}
// Number of blocked threads is used as termination condition.
@ -932,8 +945,8 @@ int CurrentThreadId() const EIGEN_FINAL {
for (unsigned i = 0; i < size; i++) {
assert(victim < size);
if (round == 1 ||
thread_data_[victim].GetStatus() == ThreadData::ThreadStatus::Active) {
Task t = thread_data_[victim].queue.PopBack();
worker_data_[victim].GetStatus() == WorkerData::ThreadStatus::Active) {
Task t = worker_data_[victim].queue.PopBack();
if (t.f) {
return t;
}
@ -957,12 +970,12 @@ int CurrentThreadId() const EIGEN_FINAL {
int NonEmptyQueueIndex() {
PerThread* pt = GetPerThread();
const unsigned size = static_cast<unsigned>(thread_data_.size());
const unsigned size = static_cast<unsigned>(worker_data_.size());
unsigned r = Rand(&pt->rand);
unsigned inc = all_coprimes_[size - 1][r % all_coprimes_[size - 1].size()];
unsigned victim = r % size;
for (unsigned i = 0; i < size; i++) {
if (!thread_data_[victim].queue.Empty()) {
if (!worker_data_[victim].queue.Empty()) {
return victim;
}
victim += inc;