From 48b14b52b8b476a839a307d307e0a39ee4ab63de Mon Sep 17 00:00:00 2001 From: Tim Harris Date: Tue, 10 Nov 2020 20:22:07 +0000 Subject: [PATCH] Remove Env::Task wrapper around std::function (#5753) This is a small perf / clean-up change. It removes the Env::Task abstraction which wraps a single std::function field, and adds at least one virtual method call overhead when creating a Task and when executing it. The POSIX and Windows implementations are now identical. --- .../platform/EigenNonBlockingThreadPool.h | 34 ++++++++----------- onnxruntime/core/platform/env.h | 8 ----- onnxruntime/core/platform/posix/env.cc | 6 ---- onnxruntime/core/platform/windows/env.cc | 6 ---- 4 files changed, 15 insertions(+), 39 deletions(-) diff --git a/include/onnxruntime/core/platform/EigenNonBlockingThreadPool.h b/include/onnxruntime/core/platform/EigenNonBlockingThreadPool.h index 5940200d7c..ca247f08b4 100644 --- a/include/onnxruntime/core/platform/EigenNonBlockingThreadPool.h +++ b/include/onnxruntime/core/platform/EigenNonBlockingThreadPool.h @@ -534,8 +534,6 @@ class ThreadPoolTempl : public onnxruntime::concurrency::ExtendedThreadPoolInter } public: - typedef typename Environment::Task Task; - struct Tag { constexpr Tag() : v_(0) { } @@ -570,6 +568,7 @@ class ThreadPoolTempl : public onnxruntime::concurrency::ExtendedThreadPoolInter uint32_t v_ = 0; }; + typedef std::function Task; typedef RunQueue Queue; #ifdef _WIN32 using CHAR_TYPE = wchar_t; @@ -638,29 +637,26 @@ class ThreadPoolTempl : public onnxruntime::concurrency::ExtendedThreadPoolInter // reject work if the queue of pending work is full. void Schedule(std::function fn) override { - Task t = env_.CreateTask(std::move(fn)); PerThread* pt = GetPerThread(); if (pt->pool == this) { // Worker thread of this pool, push onto the thread's queue. Queue& q = worker_data_[pt->thread_id].queue; - t = q.PushFront(std::move(t)); + fn = q.PushFront(std::move(fn)); } else { // A free-standing thread (or worker of another pool), push onto a random // queue. int q_idx = Rand(&pt->rand) % num_threads_; WorkerData &td = worker_data_[q_idx]; Queue& q = td.queue; - t = q.PushBack(std::move(t)); - if (!t.f) { + fn = q.PushBack(std::move(fn)); + if (!fn) { // The queue accepted the work; ensure that the thread will pick it up td.EnsureAwake(); } } // Run the work directly if the queue rejected the work - if (t.f) { - env_.ExecuteTask(t); - } + if (fn) fn(); } // The thread pool maintains a set of hints for which threads will be good to distribute @@ -863,7 +859,7 @@ void SummonWorkers(PerThread &pt, // Create the additional tasks, and push them to workers. for (auto i = 0u; i < extra_needed; i++) { - Task t = env_.CreateTask(call_worker_fn); + Task t; int q_idx; if (i < good_hints.size()) { q_idx = good_hints[i]; @@ -885,8 +881,8 @@ void SummonWorkers(PerThread &pt, WorkerData& td = worker_data_[q_idx]; Queue& q = td.queue; unsigned w_idx; - t = q.PushBackWithTag(std::move(t), pt.tag, w_idx); - if (!t.f) { + t = q.PushBackWithTag(call_worker_fn, pt.tag, w_idx); + if (!t) { ps.tasks.push_back({q_idx, w_idx}); td.EnsureAwake(); } @@ -1197,26 +1193,26 @@ int CurrentThreadId() const EIGEN_FINAL { while (!cancelled_ && !should_exit) { Task t = q.PopFront(); - if (!t.f) { + if (!t) { // Spin waiting for work. We indicate, via SetGOodWorkerHint that we are // spinning. This will bias other threads toward pushing work to our queue. // In addition, priodically make a best-effort attempt to steal from other // threads which are not themselves spinning. SetGoodWorkerHint(thread_id, true); - for (int i = 0; i < spin_count && !t.f && !cancelled_ && !done_; i++) { + for (int i = 0; i < spin_count && !t && !cancelled_ && !done_; i++) { t = ((i+1)%steal_count == 0) ? TrySteal() : q.PopFront(); onnxruntime::concurrency::SpinPause(); } SetGoodWorkerHint(thread_id, false); - if (!t.f) { + if (!t) { // No work passed to us while spinning; make a further full attempt to // steal work from other threads prior to blocking. if (num_threads_ != 1) { t = Steal(true /* true => check all queues */); } - if (!t.f) { + if (!t) { td.SetBlocked( // Pre-block test [&]() -> bool { @@ -1264,9 +1260,9 @@ int CurrentThreadId() const EIGEN_FINAL { } } } - if (t.f) { + if (t) { td.SetActive(); - env_.ExecuteTask(t); + t(); td.SetSpinning(); } } @@ -1304,7 +1300,7 @@ int CurrentThreadId() const EIGEN_FINAL { if (round == 1 || worker_data_[victim].GetStatus() == WorkerData::ThreadStatus::Active) { Task t = worker_data_[victim].queue.PopBack(); - if (t.f) { + if (t) { return t; } } diff --git a/onnxruntime/core/platform/env.h b/onnxruntime/core/platform/env.h index 73068e01dd..268f729acf 100644 --- a/onnxruntime/core/platform/env.h +++ b/onnxruntime/core/platform/env.h @@ -79,9 +79,6 @@ struct ThreadOptions { /// multiple threads without any external synchronization. class Env { public: - struct Task { - std::function f; - }; using EnvThread = onnxruntime::EnvThread; virtual ~Env() = default; // clang-format off @@ -99,11 +96,6 @@ class Env { virtual EnvThread* CreateThread(_In_opt_z_ const ORTCHAR_T* name_prefix, int index, _In_ unsigned (*start_address)(int id, Eigen::ThreadPoolInterface* param), Eigen::ThreadPoolInterface* threadpool, const ThreadOptions& thread_options) = 0; - virtual Task CreateTask(std::function f) = 0; - /** - * Execute the task 't' in current thread - */ - virtual void ExecuteTask(const Task& t) = 0; /// \brief Returns a default environment suitable for the current operating /// system. diff --git a/onnxruntime/core/platform/posix/env.cc b/onnxruntime/core/platform/posix/env.cc index 2631f5f2c6..399a0e3b8c 100644 --- a/onnxruntime/core/platform/posix/env.cc +++ b/onnxruntime/core/platform/posix/env.cc @@ -186,12 +186,6 @@ class PosixEnv : public Env { Eigen::ThreadPoolInterface* param, const ThreadOptions& thread_options) override { return new PosixThread(name_prefix, index, start_address, param, thread_options); } - Task CreateTask(std::function f) override { - return Task{std::move(f)}; - } - void ExecuteTask(const Task& t) override { - t.f(); - } int GetNumCpuCores() const override { // TODO if you need the number of physical cores you'll need to parse diff --git a/onnxruntime/core/platform/windows/env.cc b/onnxruntime/core/platform/windows/env.cc index fa45c61b95..4b2f701bb1 100644 --- a/onnxruntime/core/platform/windows/env.cc +++ b/onnxruntime/core/platform/windows/env.cc @@ -118,12 +118,6 @@ class WindowsEnv : public Env { Eigen::ThreadPoolInterface* param, const ThreadOptions& thread_options) { return new WindowsThread(name_prefix, index, start_address, param, thread_options); } - Task CreateTask(std::function f) { - return Task{std::move(f)}; - } - void ExecuteTask(const Task& t) { - t.f(); - } void SleepForMicroseconds(int64_t micros) const override { Sleep(static_cast(micros) / 1000);