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.

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<void()> Task;
   typedef RunQueue<Task, Tag, 1024> 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<void()> 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;
           }
         }

onnxruntime/core/platform/env.h

@@ -79,9 +79,6 @@ struct ThreadOptions {
 /// multiple threads without any external synchronization.
 class Env {
  public:
-  struct Task {
-    std::function<void()> 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<void()> 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.

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<void()> 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

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<void()> f) {
-    return Task{std::move(f)};
-  }
-  void ExecuteTask(const Task& t) {
-    t.f();
-  }
 
   void SleepForMicroseconds(int64_t micros) const override {
     Sleep(static_cast<DWORD>(micros) / 1000);