multi_thread 总览

struct Core {
    /// Used to schedule bookkeeping tasks every so often.
    tick: 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
    /// next (LIFO). This is an optimization for improving locality which
    /// benefits message passing patterns and helps to reduce latency.
    lifo_slot: Option<Notified>,

    /// When `true`, locally scheduled tasks go to the LIFO slot. When `false`,
    /// they go to the back of the `run_queue`.
    lifo_enabled: bool,

    /// The worker-local run queue.
    run_queue: queue::Local<Arc<Handle>>,

    // ...

    /// True if the worker is currently searching for more work. Searching
    /// involves attempting to steal from other workers.
    is_searching: bool,

    /// True if the scheduler is being shutdown
    is_shutdown: bool,

    // ...

    /// Parker
    ///
    /// Stored in an `Option` as the parker is added / removed to make the
    /// borrow checker happy.
    park: Option<Parker>,

    /// Per-worker runtime stats
    stats: Stats,

    /// How often to check the global queue
    global_queue_interval: u32,

    /// Fast random number generator.
    rand: FastRand,
}

pub(crate) struct Shared {
    /// Per-worker remote state. All other workers have access to this and is
    /// how they communicate between each other.
    remotes: Box<[Remote]>,

    /// Global task queue used for:
    ///  1. Submit work to the scheduler while **not** currently on a worker thread.
    ///  2. Submit work to the scheduler when a worker run queue is saturated
    pub(super) inject: CachePadded<InjectQueue<Arc<Handle>>>,

    /// Coordinates idle workers
    idle: CachePadded<Idle>,

    /// Collection of all active tasks spawned onto this executor.
    pub(crate) owned: OwnedTasks<Arc<Handle>>,

    /// Data synchronized by the scheduler mutex
    pub(super) synced: Mutex<Synced>,

    // ...
}

/// Used to communicate with a worker from other threads.
struct Remote {
    /// Steals tasks from this worker.
    pub(super) steal: queue::Steal<Arc<Handle>>,

    /// Unparks the associated worker thread
    unpark: Unparker,
}

pub(crate) fn local<T: 'static>() -> (Steal<T>, Local<T>) {

// Acquire a core. If this fails, then another thread is running this
// worker and there is nothing further to do.
let core = match worker.core.take() {
    Some(core) => core,
    None => return,
};

// 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 => {
        // In this case, we cannot call `reset_lifo_enabled()`
        // because the core was stolen. The stealer will handle
        // that at the top of `Context::run`
        return ControlFlow::Break(());
    }
};

for _ in 0..size {
    let (steal, run_queue) = queue::local();

    let park = park.clone();
    let unpark = park.unpark();
    let metrics = WorkerMetrics::from_config(&config);
    let stats = Stats::new(&metrics);

    cores.push(Box::new(Core {
        tick: 0,
        lifo_slot: None,
        lifo_enabled: !config.disable_lifo_slot,
        run_queue,
        // ...
        park: Some(park),
        global_queue_interval: stats.tuned_global_queue_interval(&config),
        stats,
        rand: FastRand::from_seed(config.seed_generator.next_seed()),
    }));

    remotes.push(Remote { steal, unpark });
    // ...
}

pub(crate) fn launch(mut self) {
    for worker in self.0.drain(..) {
        runtime::spawn_blocking(move || run(worker));
    }
}

crate::runtime::context::enter_runtime(&handle, true, |_| {
    // Set the worker context.
    let cx = scheduler::Context::MultiThread(Context {
        worker,
        core: RefCell::new(None),
        defer: Defer::new(),
    });

    context::set_scheduler(&cx, || {
        let cx = cx.expect_multi_thread();

        cx.run(core);

        // Check if there are any deferred tasks to notify. This can happen when
        // the worker core is lost due to `block_in_place()` being called from
        // within the task.
        cx.defer.wake();
    });
});

use std::thread;
use std::time::Duration;

fn main() {
  for (sleep_ms, rounds) in [(0u64, 200), (50, 20)] {
    let rt = tokio::runtime::Builder::new_multi_thread()
      .worker_threads(1)
      .enable_all()
      .build()
      .unwrap();
    let (mut same, mut moved) = (0, 0);
    for _ in 0..rounds {
      let (before, after) = rt.block_on(async move {
        tokio::spawn(async move {
          let before = thread::current().id();
          tokio::task::block_in_place(|| thread::sleep(Duration::from_millis(sleep_ms)));
          tokio::task::yield_now().await;
          (before, thread::current().id())
        })
        .await
        .unwrap()
      });
      if before == after {
        same += 1;
      } else {
        moved += 1;
      }
    }
    println!("sleep {sleep_ms}ms: same thread {same}, other thread {moved}");
  }
}

sleep 0ms: same thread 200, other thread 0
sleep 50ms: same thread 0, other thread 20