一个任务

struct Inner<T> {
    /// The front index.
    front: AtomicWorkerIndex,
    /// The back index.
    back: AtomicWorkerIndex,
    /// The underlying buffer.
    buffer: CachePadded<Atomic<Buffer<T>>>,
}

pub fn push(&self, task: T) {
    // Load the back index, front index, and buffer.
    let b = self.inner.back.load(Ordering::Relaxed);
    let f = self.inner.front.load(Ordering::Acquire);
    let mut buffer = self.buffer.get();

    // Calculate the length of the queue.
    let len = b.wrapping_sub(f);

    // Is the queue full?
    if len >= buffer.cap as WorkerIndex {
        // Yes. Grow the underlying buffer.
        unsafe {
            self.resize(2 * buffer.cap);
        }
        buffer = self.buffer.get();
    }

    // Write `task` into the slot.
    unsafe {
        buffer.write(b, MaybeUninit::new(task));
    }

    // ThreadSanitizer does not understand fences, so we omit fence and do store with Release ordering.
    #[cfg(not(crossbeam_sanitize_thread))]
    atomic::fence(Ordering::Release);
    let store_order = if cfg!(crossbeam_sanitize_thread) {
        Ordering::Release
    } else {
        Ordering::Relaxed
    };

    // Increment the back index.
    self.inner.back.store(b.wrapping_add(1), store_order);
}

let guard = &epoch::pin();

// Replace the old buffer with the new one.
self.buffer.replace(new);
let old =
    self.inner
        .buffer
        .swap(Owned::new(new).into_shared(guard), Ordering::Release, guard);

// Destroy the old buffer later.
guard.defer_unchecked(move || old.into_owned().into_box().dealloc());

// If the buffer is very large, then flush the thread-local garbage in order to deallocate
// it as soon as possible.
if mem::size_of::<T>() * new_cap >= FLUSH_THRESHOLD_BYTES {
    guard.flush();
}

pub fn steal(&self) -> Steal<T> {
    // Load the front index.
    let f = self.inner.front.load(Ordering::Acquire);

    // A SeqCst fence is needed here.
    //
    // If the current thread is already pinned (reentrantly), we must manually issue the
    // fence. Otherwise, the following pinning will issue the fence anyway, so we don't
    // have to.
    if epoch::is_pinned() {
        atomic::fence(Ordering::SeqCst);
    }

    let guard = &epoch::pin();

    // Load the back index.
    let b = self.inner.back.load(Ordering::Acquire);

    // Is the queue empty?
    if b.wrapping_sub(f) <= 0 {
        return Steal::Empty;
    }

    // Load the buffer and read the task at the front.
    let buffer = self.inner.buffer.load(Ordering::Acquire, guard);
    let task = unsafe { buffer.deref().read(f) };

    // Try incrementing the front index to steal the task.
    // If the buffer has been swapped or the increment fails, we retry.
    if self.inner.buffer.load(Ordering::Acquire, guard) != buffer
        || self
            .inner
            .front
            .compare_exchange(f, f.wrapping_add(1), Ordering::SeqCst, Ordering::Relaxed)
            .is_err()
    {
        // We didn't steal this task, forget it.
        return Steal::Retry;
    }

    // Return the stolen task.
    Steal::Success(unsafe { task.assume_init() })
}