sched: disperse threads faster in the load balancer

core/sched.cc

@@ -797,20 +797,38 @@ void cpu::load_balance()
     notifier::fire();
     timer tmr(*thread::current());
     while (true) {
-        tmr.set(osv::clock::uptime::now() + 100_ms);
+        // This balancer is the only mechanism that spreads threads across CPUs:
+        // thread::start() places a new thread on its creator's CPU and
+        // thread::wake() re-wakes a blocked thread on the CPU it last ran on,
+        // so neither disperses work by itself.  Running once every 100ms and
+        // migrating a single thread per wakeup is far too slow for a workload
+        // that fans out many threads from one parent (e.g. a server that forks
+        // or spawns a worker per connection): the workers pile onto the few
+        // CPUs they were started/woken on while the rest of the machine sits
+        // idle, and a 1-thread-per-100ms drip cannot catch up under a high
+        // request rate.  Run more often, and on each pass keep migrating the
+        // most-migratable queued thread to the least-loaded CPU until this CPU
+        // is no longer meaningfully more loaded -- draining the whole imbalance
+        // in one pass instead of one thread of it.
+        tmr.set(osv::clock::uptime::now() + 10_ms);
         thread::wait_until([&] { return tmr.expired(); });
         if (runqueue.empty()) {
             continue;
         }
+        // Bound the work per pass so a transient spike cannot become an
+        // unbounded migration storm.
+        unsigned migrated = 0;
+        const unsigned max_migrations_per_pass = cpus.size();
+        while (migrated < max_migrations_per_pass) {
         auto min = *std::min_element(cpus.begin(), cpus.end(),
                 [](cpu* c1, cpu* c2) { return c1->load() < c2->load(); });
         if (min == this) {
-            continue;
+            break;
         }
         // This CPU is temporarily running one extra thread (this thread),
         // so don't migrate a thread away if the difference is only 1.
         if (min->load() >= (load() - 1)) {
-            continue;
+            break;
         }
 #if CONF_lazy_stack_invariant
         assert(!thread::current()->is_app());
@@ -819,7 +837,7 @@ void cpu::load_balance()
             auto i = std::find_if(runqueue.rbegin(), runqueue.rend(),
                     [](thread& t) { return t._migration_lock_counter == 0; });
             if (i == runqueue.rend()) {
-                continue;
+                break;
             }
             auto& mig = *i;
             trace_sched_migrate(&mig, min->id);
@@ -841,6 +859,8 @@ void cpu::load_balance()
             // FIXME: warrant an interruption
             min->send_wakeup_ipi();
         }
+        migrated++;
+        }
     }
 }