From e6a1444facfc72276987f4138784686507d1c61c Mon Sep 17 00:00:00 2001 From: Carl Lerche Date: Mon, 12 Jun 2023 16:14:33 -0700 Subject: [PATCH] fix more bugz --- tokio/src/runtime/driver.rs | 6 ++ .../runtime/scheduler/multi_thread/idle.rs | 14 +++-- .../runtime/scheduler/multi_thread/worker.rs | 60 +++++++++++++++++-- 3 files changed, 69 insertions(+), 11 deletions(-) diff --git a/tokio/src/runtime/driver.rs b/tokio/src/runtime/driver.rs index 572fdefb0..7b065f048 100644 --- a/tokio/src/runtime/driver.rs +++ b/tokio/src/runtime/driver.rs @@ -14,6 +14,12 @@ pub(crate) struct Driver { inner: TimeDriver, } +impl Drop for Driver { + fn drop(&mut self) { + println!(" + DROP DRIVER"); + } +} + #[derive(Debug)] pub(crate) struct Handle { /// IO driver handle diff --git a/tokio/src/runtime/scheduler/multi_thread/idle.rs b/tokio/src/runtime/scheduler/multi_thread/idle.rs index 433f33bf2..08951fd8d 100644 --- a/tokio/src/runtime/scheduler/multi_thread/idle.rs +++ b/tokio/src/runtime/scheduler/multi_thread/idle.rs @@ -126,8 +126,7 @@ impl Idle { shared.condvars[worker].notify_one(); return; } else { - // synced.idle.sleepers.push(worker); - panic!("[tokio] unexpected condition"); + synced.idle.sleepers.push(worker); } } @@ -150,7 +149,7 @@ impl Idle { workers: &mut Vec, num: usize, ) { - let mut did_notify = false; + debug_assert!(workers.is_empty()); for _ in 0..num { if let Some(worker) = synced.idle.sleepers.pop() { @@ -160,18 +159,17 @@ impl Idle { synced.assigned_cores[worker] = Some(core); workers.push(worker); - did_notify = true; continue; } else { - panic!("[tokio] unexpected condition"); + synced.idle.sleepers.push(worker); } } break; } - if did_notify { + if !workers.is_empty() { let num_idle = synced.idle.available_cores.len(); self.num_idle.store(num_idle, Release); } else { @@ -184,6 +182,7 @@ impl Idle { } pub(super) fn shutdown(&self, synced: &mut worker::Synced, shared: &Shared) { + println!(" + start shutdown"); // First, set the shutdown flag on each core for core in &mut synced.idle.available_cores { core.is_shutdown = true; @@ -199,6 +198,8 @@ impl Idle { synced.assigned_cores[worker] = Some(core); shared.condvars[worker].notify_one(); + println!(" + notify worker shutdown w/ core"); + self.num_idle .store(synced.idle.available_cores.len(), Release); } @@ -206,6 +207,7 @@ impl Idle { // Wake up any other workers while let Some(index) = synced.idle.sleepers.pop() { shared.condvars[index].notify_one(); + println!(" + notify worker shutdown NO core"); } } diff --git a/tokio/src/runtime/scheduler/multi_thread/worker.rs b/tokio/src/runtime/scheduler/multi_thread/worker.rs index c988a0c9a..efdc53085 100644 --- a/tokio/src/runtime/scheduler/multi_thread/worker.rs +++ b/tokio/src/runtime/scheduler/multi_thread/worker.rs @@ -107,6 +107,10 @@ pub(super) struct Core { /// Used to schedule bookkeeping tasks every so often. tick: u32, + /// Counter used to track when to poll from the local queue vs. the + /// injection queue + num_seq_local_queue_polls: u32, + /// When a task is scheduled from a worker, it is stored in this slot. The /// worker will check this slot for a task **before** checking the run /// queue. This effectively results in the **last** scheduled task to be run @@ -141,6 +145,12 @@ pub(super) struct Core { rand: FastRand, } +impl Drop for Core { + fn drop(&mut self) { + println!(" DROPPED CORE"); + } +} + /// State shared across all workers pub(crate) struct Shared { /// Per-core remote state. @@ -268,6 +278,7 @@ pub(super) fn create( cores.push(Box::new(Core { index: i, tick: 0, + num_seq_local_queue_polls: 0, lifo_slot: None, lifo_enabled: !config.disable_lifo_slot, run_queue, @@ -560,6 +571,7 @@ impl Worker { .idle .try_acquire_available_core(&mut synced.idle) { + println!(" + acquired_core; {}", self.index); self.reset_acquired_core(cx, synced, &mut core); Some(core) } else { @@ -569,6 +581,7 @@ impl Worker { // Block the current thread, waiting for an available core fn wait_for_core(&self, cx: &Context, mut synced: MutexGuard<'_, Synced>) -> NextTaskResult { + println!(" + wait_for_core; {}", self.index); cx.shared() .idle .transition_worker_to_parked(&mut synced, self.index); @@ -582,6 +595,7 @@ impl Worker { // If shutting down, abort if cx.shared().inject.is_closed(&synced.inject) { self.shutdown_clear_defer(cx); + println!(" + wait_for_core; shutdown; {}", self.index); return Err(()); } @@ -591,19 +605,22 @@ impl Worker { self.reset_acquired_core(cx, &mut synced, &mut core); if core.is_shutdown { + println!(" + wait_for_core; CORE(shutdown) {}", self.index); // Currently shutting down, don't do any more work return Ok((None, core)); } // The core was notified to search for work, don't try to take tasks from the injection queue if core.is_searching { + println!(" + wait_for_core; SEARCHING"); return Ok((None, core)); } - // TODO: don't hardcode 128 let n = core.run_queue.max_capacity() / 2; let maybe_task = self.next_remote_task_batch(cx, &mut synced, &mut core, n); + println!(" + wait_for_core; task={:?}", maybe_task.is_some()); + Ok((maybe_task, core)) } @@ -666,7 +683,12 @@ impl Worker { } fn next_notified_task(&self, cx: &Context, core: &mut Core) -> Option { - if core.tick % core.global_queue_interval == 0 { + core.num_seq_local_queue_polls += 1; + + if core.num_seq_local_queue_polls % core.global_queue_interval == 0 { + core.num_seq_local_queue_polls = 0; + + println!(" + next_notified_task; REMOTE FIRST"); // Update the global queue interval, if needed self.tune_global_queue_interval(cx, core); @@ -873,7 +895,7 @@ impl Worker { fn schedule_deferred_with_core<'a>( &mut self, cx: &'a Context, - core: Box, + mut core: Box, synced: impl FnOnce() -> MutexGuard<'a, Synced>, ) -> NextTaskResult { let mut defer = cx.defer.borrow_mut(); @@ -909,6 +931,16 @@ impl Worker { cx.shared().condvars[worker].notify_one() } + if !defer.is_empty() { + // Push the rest of the tasks on the local queue + for task in defer.drain(..) { + core.run_queue + .push_back_or_overflow(task, cx.shared(), &mut core.stats); + } + + cx.shared().notify_parked_local(); + } + Ok((task, core)) } @@ -962,18 +994,27 @@ impl Worker { } fn park_yield(&mut self, cx: &Context, mut core: Box) -> NextTaskResult { + println!(" + park_yield; tick={}", core.tick); + let mut maybe_task = None; + // Call `park` with a 0 timeout. This enables the I/O driver, timer, ... // to run without actually putting the thread to sleep. if let Some(mut driver) = cx.shared().driver.take() { driver.park_timeout(&cx.handle.driver, Duration::from_millis(0)); + cx.shared().driver.set(driver); + // If there are more I/O events, schedule them. - core = n!(self.schedule_deferred_with_core(cx, core, || cx.shared().synced.lock())); + let res = self.schedule_deferred_with_core(cx, core, || cx.shared().synced.lock())?; + + maybe_task = res.0; + core = res.1; } self.flush_metrics(cx, &mut core); self.update_global_flags(cx, &mut cx.shared().synced.lock(), &mut core); - Ok((None, core)) + + Ok((maybe_task, core)) } fn park(&mut self, cx: &Context, mut core: Box) -> NextTaskResult { @@ -1050,8 +1091,11 @@ impl Worker { // Drop the lock before parking on the driver drop(synced); + println!(" + driver::park"); + // Wait for driver events driver.park(&self.handle.driver); + println!(" + driver::park / done"); synced = cx.shared().synced.lock(); @@ -1123,16 +1167,22 @@ impl Worker { } fn shutdown_finalize(&self, cx: &Context, mut synced: MutexGuard<'_, Synced>) { + println!(" + shutdown core"); + // Wait for all cores if synced.shutdown_cores.len() != cx.shared().remotes.len() { return; } + println!(" + shutdown_finalize; all cores"); + // Wait for driver if cx.shared().driver.is_none() { return; } + println!(" + shutdown_finalize; have driver"); + debug_assert!(cx.shared().owned.is_empty()); for mut core in synced.shutdown_cores.drain(..) {