本地队列

pub(crate) struct Inner<T: 'static> {
    /// Concurrently updated by many threads.
    ///
    /// Contains two `UnsignedShort` values. The `LSB` byte is the "real" head of
    /// the queue. The `UnsignedShort` in the `MSB` is set by a stealer in process
    /// of stealing values. It represents the first value being stolen in the
    /// batch. The `UnsignedShort` indices are intentionally wider than strictly
    /// required for buffer indexing in order to provide ABA mitigation and make
    /// it possible to distinguish between full and empty buffers.
    ///
    /// When both `UnsignedShort` values are the same, there is no active
    /// stealer.
    ///
    /// Tracking an in-progress stealer prevents a wrapping scenario.
    head: AtomicUnsignedLong,

    /// Only updated by producer thread but read by many threads.
    tail: AtomicUnsignedShort,

    /// Elements
    buffer: Box<[UnsafeCell<MaybeUninit<task::Notified<T>>>; LOCAL_QUEUE_CAPACITY]>,
}

fn unpack(n: UnsignedLong) -> (UnsignedShort, UnsignedShort) {
    let real = n & UnsignedShort::MAX as UnsignedLong;
    let steal = n >> (mem::size_of::<UnsignedShort>() * 8);

    (steal as UnsignedShort, real as UnsignedShort)
}

fn pack(steal: UnsignedShort, real: UnsignedShort) -> UnsignedLong {
    (real as UnsignedLong) | ((steal as UnsignedLong) << (mem::size_of::<UnsignedShort>() * 8))
}

pub(crate) fn pop(&mut self) -> Option<task::Notified<T>> {
    let mut head = self.inner.head.load(Acquire);

    let idx = loop {
        let (steal, real) = unpack(head);

        // safety: this is the **only** thread that updates this cell.
        let tail = unsafe { self.inner.tail.unsync_load() };

        if real == tail {
            // queue is empty
            return None;
        }

        let next_real = real.wrapping_add(1);

        // If `steal == real` there are no concurrent stealers. Both `steal`
        // and `real` are updated.
        let next = if steal == real {
            pack(next_real, next_real)
        } else {
            assert_ne!(steal, next_real);
            pack(steal, next_real)
        };

        // Attempt to claim a task.
        let res = self
            .inner
            .head
            .compare_exchange_weak(head, next, AcqRel, Acquire);

        match res {
            Ok(_) => break real as usize & MASK,
            Err(actual) => head = actual,
        }
    };

    Some(self.inner.buffer[idx].with(|ptr| unsafe { ptr::read(ptr).assume_init() }))
}

let tail = loop {
    let head = self.inner.head.load(Acquire);
    let (steal, real) = unpack(head);

    // safety: this is the **only** thread that updates this cell.
    let tail = unsafe { self.inner.tail.unsync_load() };

    if tail.wrapping_sub(steal) < LOCAL_QUEUE_CAPACITY as UnsignedShort {
        // There is capacity for the task
        break tail;
    } else if steal != real {
        // Concurrently stealing, this will free up capacity, so only
        // push the task onto the inject queue
        overflow.push(task);
        return;
    } else {
        // Push the current task and half of the queue into the
        // inject queue.
        match self.push_overflow(task, real, tail, overflow, stats) {
            Ok(_) => return,
            // Lost the race, try again
            Err(v) => {
                task = v;
            }
        }
    }
};

self.push_back_finish(task, tail);

const NUM_TASKS_TAKEN: UnsignedShort = (LOCAL_QUEUE_CAPACITY / 2) as UnsignedShort;

// ...

// Claim all tasks.
if self
    .inner
    .head
    .compare_exchange_weak(pack(head, head), pack(tail, tail), Release, Relaxed)
    .is_err()
{
    // We failed to claim the tasks, losing the race. Return out of
    // this function and try the full `push` routine again. The queue
    // may not be full anymore.
    return Err(task);
}

// Add back the first half of tasks.
self.inner
    .tail
    .store(tail.wrapping_add(NUM_TASKS_TAKEN), Release);

#[tokio::main(flavor = "multi_thread", worker_threads = 1)]
async fn main() {
  for n in [257, 258, 300] {
    let depth = tokio::spawn(async move {
      for _ in 0..n {
        tokio::spawn(async {});
      }
      tokio::runtime::Handle::current().metrics().global_queue_depth()
    })
    .await
    .unwrap();
    println!("spawned {n} on a worker -> global queue depth {depth}");
    tokio::time::sleep(std::time::Duration::from_millis(50)).await;
  }
}

spawned 257 on a worker -> global queue depth 0
spawned 258 on a worker -> global queue depth 129
spawned 300 on a worker -> global queue depth 129

let (src_head_steal, src_head_real) = unpack(prev_packed);
let src_tail = self.0.tail.load(Acquire);

// If these two do not match, another thread is concurrently
// stealing from the queue.
if src_head_steal != src_head_real {
    return 0;
}

// Number of available tasks to steal
let n = src_tail.wrapping_sub(src_head_real);
let n = n - n / 2;

// ...

// Update the real head index to acquire the tasks.
let steal_to = src_head_real.wrapping_add(n);
assert_ne!(src_head_steal, steal_to);
next_packed = pack(src_head_steal, steal_to);

// Claim all those tasks. This is done by incrementing the "real"
// head but not the steal. By doing this, no other thread is able to
// steal from this queue until the current thread completes.
let res = self
    .0
    .head
    .compare_exchange_weak(prev_packed, next_packed, AcqRel, Acquire);

spawned 386 on a worker -> global queue depth 129
spawned 387 on a worker -> global queue depth 258
spawned 513 on a worker -> global queue depth 258