diff --git a/core/sched.cc b/core/sched.cc index 07801a291..495ff18de 100644 --- a/core/sched.cc +++ b/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++; + } } }