worker 主循环

fn run(&self, mut core: Box<Core>) {
    // Reset `lifo_enabled` here in case the core was previously stolen from
    // a task that had the LIFO slot disabled.
    self.reset_lifo_enabled(&mut core);

    // ...

    while !core.is_shutdown {
        // ...

        // Increment the tick
        core.tick();

        // Run maintenance, if needed
        core = self.maintenance(core);

        // First, check work available to the current worker.
        if let Some(task) = core.next_task(&self.worker) {
            core = match self.run_task(task, core) {
                ControlFlow::Continue(core) => core,
                ControlFlow::Break(()) => return,
            };
            continue;
        }

        // ...

        // There is no more **local** work to process, try to steal work
        // from other workers.
        if let Some(task) = core.steal_work(&self.worker) {
            // Found work, switch back to processing
            core.stats.start_processing_scheduled_tasks();
            core = match self.run_task(task, core) {
                ControlFlow::Continue(core) => core,
                ControlFlow::Break(()) => return,
            };
        } else {
            // Wait for work
            core = if !self.defer.is_empty() {
                self.park_yield(core)
            } else {
                self.park(core)
            };
            core.stats.start_processing_scheduled_tasks();
        }
    }

    // ...

    core.pre_shutdown(&self.worker);
    // Signal shutdown
    self.worker.handle.shutdown_core(core);
}

fn next_local_task(&mut self) -> Option<Notified> {
    self.lifo_slot.take().or_else(|| self.run_queue.pop())
}

fn schedule_local(&self, core: &mut Core, task: Notified, is_yield: bool) {
    core.stats.inc_local_schedule_count();

    // Spawning from the worker thread. If scheduling a "yield" then the
    // task must always be pushed to the back of the queue, enabling other
    // tasks to be executed. If **not** a yield, then there is more
    // flexibility and the task may go to the front of the queue.
    let should_notify = if is_yield || !core.lifo_enabled {
        core.run_queue
            .push_back_or_overflow(task, self, &mut core.stats);
        true
    } else {
        // Push to the LIFO slot
        let prev = core.lifo_slot.take();
        let ret = prev.is_some();

        if let Some(prev) = prev {
            core.run_queue
                .push_back_or_overflow(prev, self, &mut core.stats);
        }

        core.lifo_slot = Some(task);

        ret
    };

    // Only notify if not currently parked. If `park` is `None`, then the
    // scheduling is from a resource driver. As notifications often come in
    // batches, the notification is delayed until the park is complete.
    if should_notify && core.park.is_some() {
        self.notify_parked_local();
    }
}

let task = self.worker.handle.shared.owned.assert_owner(task);

// Make sure the worker is not in the **searching** state. This enables
// another idle worker to try to steal work.
let notified_parked_worker = core.transition_from_searching(&self.worker);

// ...

// Make the core available to the runtime context
*self.core.borrow_mut() = Some(core);

// Run the task
coop::budget(|| {
    // ...
    task.run();
    // ...

    let mut lifo_polls = 0;

    // As long as there is budget remaining and a task exists in the
    // `lifo_slot`, then keep running.
    loop {
        // Check if we still have the core. If not, the core was stolen
        // by another worker.
        let mut core = match self.core.borrow_mut().take() {
            Some(core) => core,
            None => {
                // ...
                return ControlFlow::Break(());
            }
        };

        // Check for a task in the LIFO slot
        let task = match core.lifo_slot.take() {
            Some(task) => task,
            None => {
                self.reset_lifo_enabled(&mut core);
                core.stats.end_poll();
                return ControlFlow::Continue(core);
            }
        };

        if !coop::has_budget_remaining() {
            core.stats.end_poll();

            // Not enough budget left to run the LIFO task, push it to
            // the back of the queue and return.
            core.run_queue.push_back_or_overflow(
                task,
                &*self.worker.handle,
                &mut core.stats,
            );
            // ...
            return ControlFlow::Continue(core);
        }

        // Track that we are about to run a task from the LIFO slot.
        lifo_polls += 1;
        // ...

        // Disable the LIFO slot if we reach our limit
        //
        // In ping-ping style workloads where task A notifies task B,
        // which notifies task A again, continuously prioritizing the
        // LIFO slot can cause starvation as these two tasks will
        // repeatedly schedule the other. To mitigate this, we limit the
        // number of times the LIFO slot is prioritized.
        if lifo_polls >= MAX_LIFO_POLLS_PER_TICK {
            core.lifo_enabled = false;
            // ...
        }

        let task = self.worker.handle.shared.owned.assert_owner(task);

        // ...

        *self.core.borrow_mut() = Some(core);

        // ...
        task.run();
        // ...
    }
})

use std::time::Duration;

fn main() {
  for disable in [false, true] {
    let mut b = tokio::runtime::Builder::new_multi_thread();
    b.worker_threads(1).enable_all();
    #[cfg(tokio_unstable)]
    if disable {
      b.disable_lifo_slot();
    }
    #[cfg(not(tokio_unstable))]
    if disable {
      continue;
    }
    let rt = b.build().unwrap();
    println!("disable_lifo_slot = {disable}");
    rt.block_on(async {
      tokio::spawn(async {
        for name in ["A", "B", "C"] {
          tokio::spawn(async move { println!("  {name}") });
        }
        println!("  parent done");
      });
      tokio::time::sleep(Duration::from_millis(100)).await;
    });
  }
}

disable_lifo_slot = false
  parent done
  C
  A
  B
disable_lifo_slot = true
  parent done
  A
  B
  C

fn maintenance(&self, mut core: Box<Core>) -> Box<Core> {
    if core.tick % self.worker.handle.shared.config.event_interval == 0 {
        super::counters::inc_num_maintenance();

        core.stats.end_processing_scheduled_tasks();

        // Call `park` with a 0 timeout. This enables the I/O driver, timer, ...
        // to run without actually putting the thread to sleep.
        core = self.park_yield(core);

        // Run regularly scheduled maintenance
        core.maintenance(&self.worker);

        core.stats.start_processing_scheduled_tasks();
    }

    core
}

fn park_internal(&self, mut core: Box<Core>, duration: Option<Duration>) -> Box<Core> {
    // ...

    // Take the parker out of core
    let mut park = core.park.take().expect("park missing");
    // Store `core` in context
    *self.core.borrow_mut() = Some(core);

    // ...

    // Park thread
    let had_driver = if let Some(timeout) = duration {
        park.park_timeout(&self.worker.handle.driver, timeout)
    } else {
        park.park(&self.worker.handle.driver)
    };

    self.defer.wake();

    // ...

    // Remove `core` from context
    core = self.core.borrow_mut().take().expect("core missing");

    // Place `park` back in `core`
    core.park = Some(park);
    core.had_driver = had_driver;

    if core.should_notify_others() {
        self.worker.handle.notify_parked_local();
    }
    core
}

fn chain(i: u32, max: u32) {
  println!("  c{i}");
  if i < max {
    tokio::spawn(async move { chain(i + 1, max) });
  }
}

fn main() {
  let rt = tokio::runtime::Builder::new_multi_thread()
    .worker_threads(1)
    .enable_all()
    .build()
    .unwrap();
  rt.block_on(async {
    tokio::spawn(async {
      tokio::spawn(async { println!("  B") });
      tokio::spawn(async { chain(1, 8) });
      println!("  parent done");
    })
    .await
    .unwrap();
    tokio::time::sleep(std::time::Duration::from_millis(100)).await;
  });
}

  parent done
  c1
  c2
  c3
  B
  c4
  c5
  c6
  c7
  c8