diff --git a/tokio/src/runtime/scheduler/multi_thread/worker.rs b/tokio/src/runtime/scheduler/multi_thread/worker.rs index efdc53085..10de632d1 100644 --- a/tokio/src/runtime/scheduler/multi_thread/worker.rs +++ b/tokio/src/runtime/scheduler/multi_thread/worker.rs @@ -559,7 +559,9 @@ impl Worker { // Signal shutdown self.shutdown_core(cx, core); - debug_assert!(cx.defer.borrow().is_empty()); + // It is possible that tasks wake others during drop, so we need to + // clear the defer list. + self.shutdown_clear_defer(cx); Err(()) } @@ -688,7 +690,6 @@ impl Worker { 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); @@ -966,6 +967,7 @@ impl Worker { fn maybe_maintenance(&mut self, cx: &Context, mut core: Box) -> NextTaskResult { if core.tick % cx.shared().config.event_interval == 0 { + println!(" + MAINT {}", core.tick); super::counters::inc_num_maintenance(); core.stats.end_processing_scheduled_tasks(); @@ -994,23 +996,21 @@ 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() { + println!(" + DRIVER POLL"); driver.park_timeout(&cx.handle.driver, Duration::from_millis(0)); cx.shared().driver.set(driver); - - // If there are more I/O events, schedule them. - let res = self.schedule_deferred_with_core(cx, core, || cx.shared().synced.lock())?; - - maybe_task = res.0; - core = res.1; } + // If there are more I/O events, schedule them. + let res = self.schedule_deferred_with_core(cx, core, || cx.shared().synced.lock())?; + + let 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); @@ -1278,16 +1278,21 @@ impl Shared { } fn schedule_local(&self, core: &mut Core, task: Notified) { - // Push to the LIFO slot - let prev = core.lifo_slot.take(); + if core.lifo_enabled { + // Push to the LIFO slot + let prev = core.lifo_slot.take(); - if let Some(prev) = prev { + if let Some(prev) = prev { + core.run_queue + .push_back_or_overflow(prev, self, &mut core.stats); + } + + core.lifo_slot = Some(task); + } else { core.run_queue - .push_back_or_overflow(prev, self, &mut core.stats); + .push_back_or_overflow(task, self, &mut core.stats); } - core.lifo_slot = Some(task); - self.notify_parked_local(); }