全局注入队列

pub(crate) struct Inject<T: 'static> {
    shared: Shared<T>,
    synced: Mutex<Synced>,
}

pub(crate) struct Synced {
    /// True if the queue is closed.
    pub(super) is_closed: bool,

    /// Linked-list head.
    pub(super) head: Option<task::RawTask>,

    /// Linked-list tail.
    pub(super) tail: Option<task::RawTask>,
}

pub(crate) unsafe fn push(&self, synced: &mut Synced, task: task::Notified<T>) {
    if synced.is_closed {
        return;
    }

    // safety: only mutated with the lock held
    let len = unsafe { self.len.unsync_load() };
    let task = task.into_raw();

    // The next pointer should already be null
    debug_assert!(unsafe { task.get_queue_next().is_none() });

    if let Some(tail) = synced.tail {
        // safety: Holding the Notified for a task guarantees exclusive
        // access to the `queue_next` field.
        unsafe { tail.set_queue_next(Some(task)) };
    } else {
        synced.head = Some(task);
    }

    synced.tail = Some(task);
    self.len.store(len + 1, Release);
}

let maybe_task = self.next_local_task();

if maybe_task.is_some() {
    return maybe_task;
}

if worker.inject().is_empty() {
    return None;
}

let cap = usize::min(
    // Other threads can only **remove** tasks from the current
    // worker's `run_queue`. So, we can be confident that by the
    // time we call `run_queue.push_back` below, there will be *at
    // least* `cap` available slots in the queue.
    // ...
    self.run_queue.remaining_slots(),
    // We want to make sure that all of the tasks we take end up in
    // the first half of the local queue. This ensures that the
    // tasks do not get pushed to the inject queue again if overflow
    // occurs, as overflow only affects tasks in the second half of
    // the local queue.
    // ...
    self.run_queue.max_capacity() / 2,
);

// The worker is currently idle, pull a batch of work from the
// injection queue. We don't want to pull *all* the work so other
// workers can also get some.
let n = usize::min(
    worker.inject().len() / worker.handle.shared.remotes.len() + 1,
    cap,
);

// Take at least one task since the first task is returned directly
// and not pushed onto the local queue.
let n = usize::max(1, n);

worker.inject().pop_n(n, |mut tasks| {
    // Pop the first task to return immediately
    let ret = tasks.next();

    // Push the rest of the on the run queue
    self.run_queue.push_back(tasks);

    ret
})

if self.tick % self.global_queue_interval == 0 {
    // Update the global queue interval, if needed
    self.tune_global_queue_interval(worker);

    worker
        .handle
        .next_remote_task()
        .or_else(|| self.next_local_task())
}

/// How to weigh each individual poll time, value is plucked from thin air.
const TASK_POLL_TIME_EWMA_ALPHA: f64 = 0.1;

/// Ideally, we wouldn't go above this, value is plucked from thin air.
const TARGET_GLOBAL_QUEUE_INTERVAL: f64 = Duration::from_micros(200).as_nanos() as f64;

/// Max value for the global queue interval. This is 2x the previous default
const MAX_TASKS_POLLED_PER_GLOBAL_QUEUE_INTERVAL: u32 = 127;

/// This is the previous default
const TARGET_TASKS_POLLED_PER_GLOBAL_QUEUE_INTERVAL: u32 = 61;

pub(crate) fn tuned_global_queue_interval(&self, config: &Config) -> u32 {
    // If an interval is explicitly set, don't tune.
    if let Some(configured) = config.global_queue_interval {
        return configured;
    }

    // As of Rust 1.45, casts from f64 -> u32 are saturating, which is fine here.
    let tasks_per_interval = (TARGET_GLOBAL_QUEUE_INTERVAL / self.task_poll_time_ewma) as u32;

    // If we are using self-tuning, we don't want to return less than 2 as that would result in the
    // global queue always getting checked first.
    tasks_per_interval.clamp(2, MAX_TASKS_POLLED_PER_GLOBAL_QUEUE_INTERVAL)
}

// Calculate the mean poll duration for a single task in the batch
let mean_poll_duration = elapsed / num_polls;

// Compute the alpha weighted by the number of tasks polled this batch.
let weighted_alpha = 1.0 - (1.0 - TASK_POLL_TIME_EWMA_ALPHA).powf(num_polls);

// Now compute the new weighted average task poll time.
self.task_poll_time_ewma = weighted_alpha * mean_poll_duration
    + (1.0 - weighted_alpha) * self.task_poll_time_ewma;

use std::sync::mpsc;

fn main() {
  for n in [300, 100] {
    let rt = tokio::runtime::Builder::new_multi_thread()
      .worker_threads(1)
      .build()
      .unwrap();
    // Occupy the only worker while main fills the inject queue.
    let (go_tx, go_rx) = mpsc::channel::<()>();
    let (busy_tx, busy_rx) = mpsc::channel::<()>();
    rt.spawn(async move {
      busy_tx.send(()).unwrap();
      go_rx.recv().unwrap();
    });
    busy_rx.recv().unwrap();
    let (out_tx, out_rx) = mpsc::channel();
    for i in 0..n {
      let out_tx = out_tx.clone();
      rt.spawn(async move {
        if i == 0 {
          let m = tokio::runtime::Handle::current().metrics();
          out_tx.send(m.global_queue_depth()).unwrap();
        }
      });
    }
    go_tx.send(()).unwrap();
    let depth = out_rx.recv().unwrap();
    println!("n = {n}: first task sees global queue depth {depth}");
  }
}

n = 300: first task sees global queue depth 172
n = 100: first task sees global queue depth 0