"Sticky" allocation of worker threads (#7551)

[ PR previously merged as https://github.com//pull/7372, then reverted pending investigation of lost-wake-up issue seen with ParallelExecutor. Issue was a missing test for new work pushed to thread concurrent with a worker blocking. Change from 7372 is the addition of: https://github.com/microsoft/onnxruntime/blob/tiharr/dev-sticky-4/include/onnxruntime/core/platform/EigenNonBlockingThreadPool.h#L1473-L1492 ]

Description: This change updates the heuristics used when a thread selects which worker threads to push work to on entering a parallel loop. Previously, worker threads would maintain a best-effort bitmap of "good worker hints" indicating the threads that were likely to be spinning waiting for work. This change uses a simpler heuristic where a thread records which workers ran its previous loop, and then re-submits its next loop to those same workers. The aim is to retain affinity between a thread and a set of workers, and to avoid maintaining the "good worker hints" bitmaps.

Motivation and Context: Profiling suggested that maintaining the "good worker hints" was taking unexpected time, particularly on NUMA systems. In addition, when running many concurrent workloads, the hints did not provide a way to help retain locality of workers and hence data in caches. Testing to confirm no regressions on microbenchmark (./build/Linux/Release/onnxruntime_benchmark --benchmark_filter=BM_ThreadPoolParallelFor) and on Linux mobilenet_v1_1.0_224.onnx, comparing p50 and p99 with vs without this change:

1 concurrent:
p50 0.0172s vs 0.0181s
p99 0.0204s vs 0.0216s

2 concurrent:
p50 0.0172s vs 0.0181s
p99 0.0213s vs 0.0221s

onnxruntime/test/platform/threadpool_test.cc

@@ -168,6 +168,7 @@ void TestPoolCreation(const std::string&, int iter) {
   ASSERT_EQ(ctr, iter * per_iter);
 }
 
+// Test multi-loop parallel sections, with a series of fixed-size loops
 void TestMultiLoopSections(const std::string& name, int num_threads, int num_loops) {
   for (int rep = 0; rep < 5; rep++) {
     const int num_tasks = 1024;
@@ -186,6 +187,35 @@ void TestMultiLoopSections(const std::string& name, int num_threads, int num_loo
   }
 }
 
+// Test multi-loop parallel sections, with alternating larger and
+// smaller loops.  This helps test that we can dispatch work to
+// differing numbers of threads over time.
+void TestStagedMultiLoopSections(const std::string& name, int num_threads, int num_loops) {
+  for (int rep = 0; rep < 5; rep++) {
+    auto test_data1 = CreateTestData(num_threads/2);
+    auto test_data2 = CreateTestData(num_threads);
+    CreateThreadPoolAndTest(name, num_threads, [&](ThreadPool* tp) {
+	ThreadPool::ParallelSection ps(tp);
+	for (int l = 0; l < num_loops; l++) {
+          // Loop needing few threads
+          ThreadPool::TrySimpleParallelFor(tp,
+                                           num_threads / 2,
+                                           [&](std::ptrdiff_t i) {
+                                             IncrementElement(*test_data1, i);
+                                           });
+          // Loop needing more threads, forcing growth of set of threads in use
+          ThreadPool::TrySimpleParallelFor(tp,
+                                           num_threads,
+                                           [&](std::ptrdiff_t i) {
+                                             IncrementElement(*test_data2, i);
+                                           });
+	}
+      });
+    ValidateTestData(*test_data1, num_loops);
+    ValidateTestData(*test_data2, num_loops);
+  }
+}
+
 }  // namespace
 
 namespace onnxruntime {
@@ -351,6 +381,10 @@ TEST(ThreadPoolTest, TestMultiLoopSections_1Thread_1Loop) {
   TestMultiLoopSections("TestMultiLoopSections_1Thread_1Loop", 1, 1);
 }
 
+TEST(ThreadPoolTest, TestMultiLoopSections_1Thread_2Loop) {
+  TestMultiLoopSections("TestMultiLoopSections_1Thread_2Loop", 1, 2);
+}
+
 TEST(ThreadPoolTest, TestMultiLoopSections_2Thread_0Loop) {
   TestMultiLoopSections("TestMultiLoopSections_2Thread_0Loop", 2, 0);
 }
@@ -379,6 +413,17 @@ TEST(ThreadPoolTest, TestMultiLoopSections_4Thread_100Loop) {
   TestMultiLoopSections("TestMultiLoopSections_4Thread_100Loop", 4, 100);
 }
 
+TEST(ThreadPoolTest, TestStagedMultiLoopSections_4Thread_1Loop) {
+  TestStagedMultiLoopSections("TestStagedMultiLoopSections_4Thread_1Loop", 4, 1);
+}
+
+TEST(ThreadPoolTest, TestStagedMultiLoopSections_4Thread_10Loop) {
+  TestStagedMultiLoopSections("TestStagedMultiLoopSections_4Thread_10Loop", 4, 10);
+}
+
+TEST(ThreadPoolTest, TestStagedMultiLoopSections_4Thread_100Loop) {
+  TestStagedMultiLoopSections("TestStagedMultiLoopSections_4Thread_100Loop", 4, 100);
+}
 #ifdef _WIN32
 #pragma warning(push)
 #pragma warning(disable : 6387)