ronantakizawa/codereview-bench huggingface dataset:
Repository: tokio-rs/tokio
File: tokio/src/runtime/blocking/pool.rs (Rust)

Code before:
    num_threads: MetricAtomicUsize,
    num_idle_threads: MetricAtomicUsize,
    queue_depth: MetricAtomicUsize,
}

impl SpawnerMetrics {
    fn num_threads(&self) -> usize {
        self.num_threads.load(Ordering::Relaxed)
    }

    fn num_idle_threads(&self) -> usize {
        self.num_idle_threads.load(Ordering::Relaxed)
    }

    cfg_unstable_metrics! {
        fn queue_depth(&self) -> usize {
            self.queue_depth.load(Ordering::Relaxed)
        }
    }

    fn inc_num_threads(&self) {
        self.num_threads.increment();
    }

    fn dec_num_threads(&self) {
        self.num_threads.increment();
    }

    fn inc_num_idle_threads(&self) {
        self.num_idle_threads.increment();
    }

    fn dec_num_idle_threads(&self) -> usize {
        self.num_idle_threads.decrement()
    }

    fn inc_queue_depth(&self) {
        self.queue_depth.increment();
    }

    fn dec_queue_depth(&self) {
        self.queue_depth.decrement();
    }
}

struct Inner {
    /// State shared between worker threads.
    shared: Mutex<Shared>,

    /// Pool threads wait on this.
    condvar: Condvar,

Review comment (suggestion):
```suggestion
        self.num_threads.decrement();
```

Code after:
    num_threads: MetricAtomicUsize,
    num_idle_threads: MetricAtomicUsize,
    queue_depth: MetricAtomicUsize,
}

impl SpawnerMetrics {
    fn num_threads(&self) -> usize {
        self.num_threads.load(Ordering::Relaxed)
    }

    fn num_idle_threads(&self) -> usize {
        self.num_idle_threads.load(Ordering::Relaxed)
    }

    cfg_unstable_metrics! {
        fn queue_depth(&self) -> usize {
            self.queue_depth.load(Ordering::Relaxed)
        }
    }

    fn inc_num_threads(&self) {
        self.num_threads.increment();
    }

    fn dec_num_threads(&self) {
        self.num_threads.decrement();
    }

    fn inc_num_idle_threads(&self) {
        self.num_idle_threads.increment();
    }

    fn dec_num_idle_threads(&self) -> usize {
        self.num_idle_threads.decrement()
    }

    fn inc_queue_depth(&self) {
        self.queue_depth.increment();
    }

    fn dec_queue_depth(&self) {
        self.queue_depth.decrement();
    }
}

struct Inner {
    /// State shared between worker threads.
    shared: Mutex<Shared>,

    /// Pool threads wait on this.
    condvar: Condvar,
Repository: tokio-rs/tokio
File: tokio/src/sync/broadcast.rs (Rust)

Code before:
/// use tokio::sync::broadcast;
///
/// #[tokio::main]
/// async fn main() {
///     let (tx, mut rx1) = broadcast::channel(16);
///     let mut rx2 = tx.subscribe();
///
///     tokio::spawn(async move {
///         assert_eq!(rx1.recv().await.unwrap(), 10);
///         assert_eq!(rx1.recv().await.unwrap(), 20);
///     });
///
///     tokio::spawn(async move {
///         assert_eq!(rx2.recv().await.unwrap(), 10);
///         assert_eq!(rx2.recv().await.unwrap(), 20);
///     });
///
///     tx.send(10).unwrap();
///     tx.send(20).unwrap();
/// }
/// ```
///
/// [`broadcast`]: crate::sync::broadcast
pub struct Sender<T> {
    shared: Arc<Shared<T>>,
    notify_rx_closed: Arc<Notify>,
}

/// Receiving-half of the [`broadcast`] channel.
///
/// Must not be used concurrently. Messages may be retrieved using
/// [`recv`][Receiver::recv].
///
/// To turn this receiver into a `Stream`, you can use the [`BroadcastStream`]
/// wrapper.
///
/// [`BroadcastStream`]: https://docs.rs/tokio-stream/0.1/tokio_stream/wrappers/struct.BroadcastStream.html
///
/// # Examples
///
/// ```
/// use tokio::sync::broadcast;
///
/// #[tokio::main]
/// async fn main() {
///     let (tx, mut rx1) = broadcast::channel(16);
///     let mut rx2 = tx.subscribe();
///
///     tokio::spawn(async move {
///         assert_eq!(rx1.recv().await.unwrap(), 10);
///         assert_eq!(rx1.recv().await.unwrap(), 20);

Review comment (suggestion):
It looks like this is being added both here and in `Shared`. I think we just want one of them. (And I'd like it to be inside `Shared`.)

Code after:
/// use tokio::sync::broadcast;
///
/// #[tokio::main]
/// async fn main() {
///     let (tx, mut rx1) = broadcast::channel(16);
///     let mut rx2 = tx.subscribe();
///
///     tokio::spawn(async move {
///         assert_eq!(rx1.recv().await.unwrap(), 10);
///         assert_eq!(rx1.recv().await.unwrap(), 20);
///     });
///
///     tokio::spawn(async move {
///         assert_eq!(rx2.recv().await.unwrap(), 10);
///         assert_eq!(rx2.recv().await.unwrap(), 20);
///     });
///
///     tx.send(10).unwrap();
///     tx.send(20).unwrap();
/// }
/// ```
///
/// [`broadcast`]: crate::sync::broadcast
pub struct Sender<T> {
    shared: Arc<Shared<T>>,
}

/// Receiving-half of the [`broadcast`] channel.
///
/// Must not be used concurrently. Messages may be retrieved using
/// [`recv`][Receiver::recv].
///
/// To turn this receiver into a `Stream`, you can use the [`BroadcastStream`]
/// wrapper.
///
/// [`BroadcastStream`]: https://docs.rs/tokio-stream/0.1/tokio_stream/wrappers/struct.BroadcastStream.html
///
/// # Examples
///
/// ```
/// use tokio::sync::broadcast;
///
/// #[tokio::main]
/// async fn main() {
///     let (tx, mut rx1) = broadcast::channel(16);
///     let mut rx2 = tx.subscribe();
///
///     tokio::spawn(async move {
///         assert_eq!(rx1.recv().await.unwrap(), 10);
///         assert_eq!(rx1.recv().await.unwrap(), 20);
///     });
Repository: tokio-rs/tokio
File: tokio/src/sync/broadcast.rs (Rust)

Code before:
    ///     });
    ///
    ///     let _ = tx.send(10);
    ///     assert!(tx.closed().now_or_never().is_none());
    ///
    ///     let _ = tokio::spawn(async move {
    ///         assert_eq!(rx2.recv().await.unwrap(), 10);
    ///     }).await;
    ///
    ///     assert!(tx.closed().now_or_never().is_some());
    /// }
    /// ```
    pub async fn closed(&self) {
        self.shared.notify_rx_drop.notified().await;

        poll_fn(|_| {
            let tail = self.shared.tail.lock();

            if tail.closed || tail.rx_cnt == 0 {
                return Poll::Ready(());
            }

            drop(tail);
            return Poll::Pending;
        })
        .await;
    }

    fn close_channel(&self) {
        let mut tail = self.shared.tail.lock();
        tail.closed = true;

        self.shared.notify_rx(tail);
    }
}

/// Create a new `Receiver` which reads starting from the tail.
fn new_receiver<T>(shared: Arc<Shared<T>>) -> Receiver<T> {
    let mut tail = shared.tail.lock();

    assert!(tail.rx_cnt != MAX_RECEIVERS, "max receivers");

    tail.rx_cnt = tail.rx_cnt.checked_add(1).expect("overflow");

    let next = tail.pos;

    drop(tail);

    Receiver { shared, next }
}


Review comment (suggestion):
The poll fn needs to interact with the `notified()` future for this to work. Check out this for inspiration:

https://github.com/tokio-rs/tokio/blob/5b9a290acd25d88416e70c73c49a54106fcf02b5/tokio-util/src/sync/cancellation_token.rs#L303-L319

You should be able to do something similar without creating a whole future struct if you use `tokio::pin!` on the `notified()` future.

Code after:
    ///         assert_eq!(rx2.recv().await.unwrap(), 10);
    ///     }).await;
    ///
    ///     assert!(tx.closed().now_or_never().is_some());
    /// }
    /// ```
    pub async fn closed(&self) {
        loop {
            let notified = self.shared.notify_last_rx_drop.notified();

            {
                // Ensure the lock drops if the channel isn't closed
                let tail = self.shared.tail.lock();
                if tail.closed {
                    return;
                }
            }

            notified.await;
        }
    }

    fn close_channel(&self) {
        let mut tail = self.shared.tail.lock();
        tail.closed = true;

        self.shared.notify_rx(tail);
    }
}

/// Create a new `Receiver` which reads starting from the tail.
fn new_receiver<T>(shared: Arc<Shared<T>>) -> Receiver<T> {
    let mut tail = shared.tail.lock();

    assert!(tail.rx_cnt != MAX_RECEIVERS, "max receivers");

    if tail.rx_cnt == 0 {
        // Potentially need to re-open the channel, if a new receiver has been added between calls
        // to poll(). Note that we use rx_cnt == 0 instead of is_closed since is_closed also
        // applies if the sender has been dropped
        tail.closed = false;
    }

    tail.rx_cnt = tail.rx_cnt.checked_add(1).expect("overflow");
    let next = tail.pos;

    drop(tail);

    Receiver { shared, next }
}
Repository: tokio-rs/tokio
File: tokio/src/runtime/mod.rs (Rust)

Code before:

    mod builder;
    pub use self::builder::Builder;
    cfg_unstable! {
        mod id;
        #[cfg_attr(not(tokio_unstable), allow(unreachable_pub))]
        pub use id::Id;

        pub use self::builder::UnhandledPanic;
        pub use crate::util::rand::RngSeed;
    }

    cfg_taskdump! {
        pub mod dump;
        pub use dump::Dump;
    }

    mod handle;
    pub use handle::{EnterGuard, Handle, TryCurrentError};

    mod runtime;
    pub use runtime::{Runtime, RuntimeFlavor};

    /// Boundary value to prevent stack overflow caused by a large-sized
    /// Future being placed in the stack.
    pub(crate) const MAX_FUTURE_SIZE: usize = 2048;

    mod thread_id;
    pub(crate) use thread_id::ThreadId;

    pub(crate) mod metrics;
    pub use metrics::RuntimeMetrics;

    cfg_unstable_metrics! {
        pub use metrics::HistogramScale;

        cfg_net! {
            pub(crate) use metrics::IoDriverMetrics;
        }
    }

    pub(crate) use metrics::{MetricsBatch, SchedulerMetrics, WorkerMetrics, HistogramBuilder};

    /// After thread starts / before thread stops
    type Callback = std::sync::Arc<dyn Fn() + Send + Sync>;
}


Review comment (bug):
It's not really a maximum for the future size.
```suggestion
    /// Boundary value to prevent stack overflow caused by a large-sized
    /// Future being placed in the stack.
    pub(crate) const BOX_FUTURE_THRESHOLD: usize = 2048;
```

Code after:

    mod builder;
    pub use self::builder::Builder;
    cfg_unstable! {
        mod id;
        #[cfg_attr(not(tokio_unstable), allow(unreachable_pub))]
        pub use id::Id;

        pub use self::builder::UnhandledPanic;
        pub use crate::util::rand::RngSeed;
    }

    cfg_taskdump! {
        pub mod dump;
        pub use dump::Dump;
    }

    mod handle;
    pub use handle::{EnterGuard, Handle, TryCurrentError};

    mod runtime;
    pub use runtime::{Runtime, RuntimeFlavor};

    /// Boundary value to prevent stack overflow caused by a large-sized
    /// Future being placed in the stack.
    pub(crate) const BOX_FUTURE_THRESHOLD: usize = 2048;

    mod thread_id;
    pub(crate) use thread_id::ThreadId;

    pub(crate) mod metrics;
    pub use metrics::RuntimeMetrics;

    cfg_unstable_metrics! {
        pub use metrics::HistogramScale;

        cfg_net! {
            pub(crate) use metrics::IoDriverMetrics;
        }
    }

    pub(crate) use metrics::{MetricsBatch, SchedulerMetrics, WorkerMetrics, HistogramBuilder};

    /// After thread starts / before thread stops
    type Callback = std::sync::Arc<dyn Fn() + Send + Sync>;
}
Repository: tokio-rs/tokio
File: tokio/tests/rt_unstable_metrics.rs (Rust)

Code before:
        })
    })
    .join()
    .unwrap();

    rt.block_on(task).unwrap();

    assert_eq!(1, rt.metrics().remote_schedule_count());

    let rt = threaded();
    let handle = rt.handle().clone();
    let task = thread::spawn(move || {
        handle.spawn(async {
            // DO nothing
        })
    })
    .join()
    .unwrap();

    rt.block_on(task).unwrap();

    assert_eq!(1, rt.metrics().remote_schedule_count());
}

#[test]
fn worker_thread_id() {
    let rt = current_thread();
    let metrics = rt.metrics();
    rt.block_on(async {
        time::sleep(Duration::from_millis(1)).await;
    });
    drop(rt);
    assert_eq!(Some(thread::current().id()), metrics.worker_thread_id(0));

    let rt = threaded();
    let metrics = rt.metrics();
    rt.block_on(async {
        time::sleep(Duration::from_millis(1)).await;
    });
    drop(rt);
    assert!(metrics.worker_thread_id(0).is_some());
    assert!(metrics.worker_thread_id(1).is_some());
    assert_ne!(Some(thread::current().id()), metrics.worker_thread_id(0));
    assert_ne!(Some(thread::current().id()), metrics.worker_thread_id(1));
}

#[test]
fn worker_park_count() {
    let rt = current_thread();
    let metrics = rt.metrics();
    rt.block_on(async {

Review comment (suggestion):
(No need to put everything you test related to thread ids in a single test. You can make more than one.)

Code after:
        })
    })
    .join()
    .unwrap();

    rt.block_on(task).unwrap();

    assert_eq!(1, rt.metrics().remote_schedule_count());

    let rt = threaded();
    let handle = rt.handle().clone();
    let task = thread::spawn(move || {
        handle.spawn(async {
            // DO nothing
        })
    })
    .join()
    .unwrap();

    rt.block_on(task).unwrap();

    assert_eq!(1, rt.metrics().remote_schedule_count());
}

#[test]
fn worker_thread_id_current_thread() {
    let rt = current_thread();
    let metrics = rt.metrics();

    // Check that runtime is on this thread.
    rt.block_on(async {});
    assert_eq!(Some(thread::current().id()), metrics.worker_thread_id(0));

    // Move runtime to another thread.
    let thread_id = std::thread::scope(|scope| {
        let join_handle = scope.spawn(|| {
            rt.block_on(async {});
        });
        join_handle.thread().id()
    });
    assert_eq!(Some(thread_id), metrics.worker_thread_id(0));

    // Move runtime back to this thread.
    rt.block_on(async {});
    assert_eq!(Some(thread::current().id()), metrics.worker_thread_id(0));
}

#[test]
fn worker_thread_id_threaded() {
    let rt = threaded();
    let metrics = rt.metrics();
Repository: tokio-rs/tokio
File: tokio/tests/io_write_all_buf.rs (Rust)

Code before:
            Ok(()).into()
        }
        fn poll_shutdown(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<io::Result<()>> {
            Ok(()).into()
        }
        fn poll_write_vectored(
            mut self: Pin<&mut Self>,
            _cx: &mut Context<'_>,
            bufs: &[io::IoSlice<'_>],
        ) -> Poll<Result<usize, io::Error>> {
            for buf in bufs {
                self.buf.extend_from_slice(buf);
            }
            let n = self.buf.len();
            Ok(n).into()
        }
        fn is_write_vectored(&self) -> bool {
            // Enable vectored write.
            true
        }
    }

    let mut wr = Wr {
        buf: BytesMut::with_capacity(64),
    };
    let mut buf = Bytes::from_static(b"hello world");

    wr.write_all_buf(&mut buf).await.unwrap();
    assert_eq!(&wr.buf[..], b"hello world");
}


Review comment (suggestion):
Maybe we should use a buffer for the test that actually returns multiple chunks when calling `.chunks_vectored()` so that we actually utilize the multiple `IoSlices`?

```suggestion
    let mut buf = Bytes::from_static(b"hello")
        .chain(Bytes::from_static(b" "))
        .chain(Bytes::from_static(b"world"));
```

Code after:
            Ok(()).into()
        }
        fn poll_shutdown(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<io::Result<()>> {
            Ok(()).into()
        }
        fn poll_write_vectored(
            mut self: Pin<&mut Self>,
            _cx: &mut Context<'_>,
            bufs: &[io::IoSlice<'_>],
        ) -> Poll<Result<usize, io::Error>> {
            for buf in bufs {
                self.buf.extend_from_slice(buf);
            }
            let n = self.buf.len();
            Ok(n).into()
        }
        fn is_write_vectored(&self) -> bool {
            // Enable vectored write.
            true
        }
    }

    let mut wr = Wr {
        buf: BytesMut::with_capacity(64),
    };
    let mut buf = Bytes::from_static(b"hello")
        .chain(Bytes::from_static(b" "))
        .chain(Bytes::from_static(b"world"));

    wr.write_all_buf(&mut buf).await.unwrap();
    assert_eq!(&wr.buf[..], b"hello world");
}
Repository: tokio-rs/tokio
File: tokio/src/runtime/task/id.rs (Rust)

Code before:
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        self.0.fmt(f)
    }
}

impl Id {
    pub(crate) fn next() -> Self {
        use crate::loom::sync::atomic::Ordering::Relaxed;
        use crate::loom::sync::atomic::StaticAtomicU64;

        #[cfg(all(test, loom))]
        {
            crate::loom::lazy_static! {
                static ref NEXT_ID: StaticAtomicU64 = StaticAtomicU64::new(1);
            }
            let id = NonZeroU64::new(NEXT_ID.fetch_add(1, Relaxed))
                .unwrap_or_else(|| NonZeroU64::new(1).unwrap());

            Self(id)
        }

        #[cfg(not(all(test, loom)))]
        {
            static NEXT_ID: StaticAtomicU64 = StaticAtomicU64::new(1);
            let id = NonZeroU64::new(NEXT_ID.fetch_add(1, Relaxed))
                .unwrap_or_else(|| NonZeroU64::new(1).unwrap());

            Self(id)
        }
    }

    pub(crate) fn as_u64(&self) -> u64 {
        self.0.get()
    }
}


Review comment (suggestion):
Please use a loop like this one:

https://github.com/tokio-rs/tokio/blob/0cbf1a5adae81e8ff86ca6d040aeee67cf888262/tokio/src/runtime/task/list.rs#L33-L40

Otherwise it's going to create two threads with id 1 in a row.

Code after:
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        self.0.fmt(f)
    }
}

impl Id {
    pub(crate) fn next() -> Self {
        use crate::loom::sync::atomic::Ordering::Relaxed;
        use crate::loom::sync::atomic::StaticAtomicU64;

        #[cfg(all(test, loom))]
        crate::loom::lazy_static! {
            static ref NEXT_ID: StaticAtomicU64 = StaticAtomicU64::new(1);
        }

        #[cfg(not(all(test, loom)))]
        static NEXT_ID: StaticAtomicU64 = StaticAtomicU64::new(1);

        loop {
            let id = NEXT_ID.fetch_add(1, Relaxed);
            if let Some(id) = NonZeroU64::new(id) {
                return Self(id);
            }
        }
    }

    pub(crate) fn as_u64(&self) -> u64 {
        self.0.get()
    }
}
Repository: tokio-rs/tokio
File: tokio/src/runtime/mod.rs (Rust)

Code before:
    cfg_trace! {
        pub(crate) use blocking::Mandatory;
    }

    cfg_fs! {
        pub(crate) use blocking::spawn_mandatory_blocking;
    }

    mod builder;
    pub use self::builder::Builder;
    cfg_unstable! {
        mod id;
        #[cfg_attr(not(tokio_unstable), allow(unreachable_pub))]
        pub use id::Id;

        pub use self::builder::UnhandledPanic;
        pub use crate::util::rand::RngSeed;
    }

    cfg_taskdump! {
        pub mod dump;
        pub use dump::Dump;
    }

    mod task_hooks;
    pub (crate) use task_hooks::{TaskHooks, TaskCallback};
    #[cfg(tokio_unstable)]
    pub use task_hooks::TaskMeta;
    #[cfg(not(tokio_unstable))]
    pub(crate) use task_hooks::TaskMeta;

    mod handle;
    pub use handle::{EnterGuard, Handle, TryCurrentError};

    mod runtime;
    pub use runtime::{Runtime, RuntimeFlavor};

    /// Boundary value to prevent stack overflow caused by a large-sized
    /// Future being placed in the stack.
    pub(crate) const BOX_FUTURE_THRESHOLD: usize = 2048;

    mod thread_id;
    pub(crate) use thread_id::ThreadId;

    pub(crate) mod metrics;
    pub use metrics::RuntimeMetrics;

    cfg_unstable_metrics! {
        pub use metrics::HistogramScale;

        cfg_net! {

Review comment (suggestion):
Formatting issue here.
```suggestion
    pub(crate) use task_hooks::{TaskHooks, TaskCallback};
```

Code after:
    cfg_trace! {
        pub(crate) use blocking::Mandatory;
    }

    cfg_fs! {
        pub(crate) use blocking::spawn_mandatory_blocking;
    }

    mod builder;
    pub use self::builder::Builder;
    cfg_unstable! {
        mod id;
        #[cfg_attr(not(tokio_unstable), allow(unreachable_pub))]
        pub use id::Id;

        pub use self::builder::UnhandledPanic;
        pub use crate::util::rand::RngSeed;
    }

    cfg_taskdump! {
        pub mod dump;
        pub use dump::Dump;
    }

    mod task_hooks;
    pub(crate) use task_hooks::{TaskHooks, TaskCallback};
    #[cfg(tokio_unstable)]
    pub use task_hooks::TaskMeta;
    #[cfg(not(tokio_unstable))]
    pub(crate) use task_hooks::TaskMeta;

    mod handle;
    pub use handle::{EnterGuard, Handle, TryCurrentError};

    mod runtime;
    pub use runtime::{Runtime, RuntimeFlavor};

    /// Boundary value to prevent stack overflow caused by a large-sized
    /// Future being placed in the stack.
    pub(crate) const BOX_FUTURE_THRESHOLD: usize = 2048;

    mod thread_id;
    pub(crate) use thread_id::ThreadId;

    pub(crate) mod metrics;
    pub use metrics::RuntimeMetrics;

    cfg_unstable_metrics! {
        pub use metrics::HistogramScale;

        cfg_net! {
Repository: tokio-rs/tokio
File: tokio/src/runtime/scheduler/multi_thread/worker.rs (Rust)

Code before:
            .shared
            .owned
            .close_and_shutdown_all(start as usize);

        self.stats
            .submit(&worker.handle.shared.worker_metrics[worker.index]);
    }

    /// Shuts down the core.
    fn shutdown(&mut self, handle: &Handle) {
        // Take the core
        let mut park = self.park.take().expect("park missing");

        // Drain the queue
        while self.next_local_task().is_some() {}

        park.shutdown(&handle.driver);
    }

    fn tune_global_queue_interval(&mut self, worker: &Worker) {
        let next = self
            .stats
            .tuned_global_queue_interval(&worker.handle.shared.config);

        // Smooth out jitter
        if self.global_queue_interval.abs_diff(next) > 2 {
            self.global_queue_interval = next;
        }
    }
}

impl Worker {
    /// Returns a reference to the scheduler's injection queue.
    fn inject(&self) -> &inject::Shared<Arc<Handle>> {
        &self.handle.shared.inject
    }
}

// TODO: Move `Handle` impls into handle.rs
impl task::Schedule for Arc<Handle> {
    fn release(&self, task: &Task) -> Option<Task> {
        self.shared.owned.remove(task)
    }

    fn schedule(&self, task: Notified) {
        self.schedule_task(task, false);
    }

    fn yield_now(&self, task: Notified) {
        self.schedule_task(task, true);
    }

Review comment (suggestion):
Hmm, I think it reads better like this:
```suggestion
        if u32::abs_diff(self.global_queue_interval, next) > 2 {
```

Code after:
            .shared
            .owned
            .close_and_shutdown_all(start as usize);

        self.stats
            .submit(&worker.handle.shared.worker_metrics[worker.index]);
    }

    /// Shuts down the core.
    fn shutdown(&mut self, handle: &Handle) {
        // Take the core
        let mut park = self.park.take().expect("park missing");

        // Drain the queue
        while self.next_local_task().is_some() {}

        park.shutdown(&handle.driver);
    }

    fn tune_global_queue_interval(&mut self, worker: &Worker) {
        let next = self
            .stats
            .tuned_global_queue_interval(&worker.handle.shared.config);

        // Smooth out jitter
        if u32::abs_diff(self.global_queue_interval, next) > 2 {
            self.global_queue_interval = next;
        }
    }
}

impl Worker {
    /// Returns a reference to the scheduler's injection queue.
    fn inject(&self) -> &inject::Shared<Arc<Handle>> {
        &self.handle.shared.inject
    }
}

// TODO: Move `Handle` impls into handle.rs
impl task::Schedule for Arc<Handle> {
    fn release(&self, task: &Task) -> Option<Task> {
        self.shared.owned.remove(task)
    }

    fn schedule(&self, task: Notified) {
        self.schedule_task(task, false);
    }

    fn yield_now(&self, task: Notified) {
        self.schedule_task(task, true);
    }