ronantakizawa/codereview-bench huggingface dataset:
Repository: tokio-rs/tokio
File: tokio/src/io/driver/mod.rs (Rust)

Code before:

    /// The system event queue.
    poll: mio::Poll,

    /// State shared between the reactor and the handles.
    inner: Arc<Inner>,
}

/// A reference to an I/O driver.
#[derive(Clone)]
pub(crate) struct Handle {
    inner: Arc<Inner>,
}

#[derive(Debug)]
pub(crate) struct ReadyEvent {
    tick: u8,
    pub(crate) ready: Ready,
}

struct IoDispatcher {
    allocator: slab::Allocator<ScheduledIo>,
    is_shutdown: bool,
}

impl IoDispatcher {
    fn new(allocator: slab::Allocator<ScheduledIo>) -> Self {
        Self {
            allocator,
            is_shutdown: false,
        }
    }
}

pub(super) struct Inner {
    /// Registers I/O resources.
    registry: mio::Registry,

    /// Allocates `ScheduledIo` handles when creating new resources.
    io_dispatch: RwLock<IoDispatcher>,

    /// Used to wake up the reactor from a call to `turn`.
    waker: mio::Waker,

    metrics: IoDriverMetrics,
}

#[derive(Debug, Eq, PartialEq, Clone, Copy)]
enum Direction {
    Read,
    Write,

Review comment (refactor):
Can we move this below with the other impl blocks, generally tokio's style is to have enum/struct defs above impl blocks.

Code after:

    /// The system event queue.
    poll: mio::Poll,

    /// State shared between the reactor and the handles.
    inner: Arc<Inner>,
}

/// A reference to an I/O driver.
#[derive(Clone)]
pub(crate) struct Handle {
    pub(super) inner: Arc<Inner>,
}

#[derive(Debug)]
pub(crate) struct ReadyEvent {
    tick: u8,
    pub(crate) ready: Ready,
}

struct IoDispatcher {
    allocator: slab::Allocator<ScheduledIo>,
    is_shutdown: bool,
}

pub(super) struct Inner {
    /// Registers I/O resources.
    registry: mio::Registry,

    /// Allocates `ScheduledIo` handles when creating new resources.
    io_dispatch: RwLock<IoDispatcher>,

    /// Used to wake up the reactor from a call to `turn`.
    waker: mio::Waker,

    metrics: IoDriverMetrics,
}

#[derive(Debug, Eq, PartialEq, Clone, Copy)]
enum Direction {
    Read,
    Write,
}

enum Tick {
    Set(u8),
    Clear(u8),
}

// TODO: Don't use a fake token. Instead, reserve a slot entry for the wakeup
// token.
Repository: tokio-rs/tokio
File: tokio-util/src/sync/reusable_box.rs (Rust)

Code before:
        // - `raw` points to a valid instance of `U` (we just wrote it in).
        let boxed = unsafe { Box::from_raw(raw) };

        callback(boxed)
    });

    // Drop the old value.
    unsafe { ptr::drop_in_place(raw) };

    // Run the rest of the code.
    Ok(guard.into_inner()())
}

struct CallOnDrop<O, F: FnOnce() -> O> {
    f: ManuallyDrop<F>,
}

impl<O, F: FnOnce() -> O> CallOnDrop<O, F> {
    fn new(f: F) -> Self {
        let f = ManuallyDrop::new(f);
        Self { f }
    }
    fn into_inner(self) -> F {
        let mut this = ManuallyDrop::new(self);
        unsafe { ManuallyDrop::take(&mut this.f) }
    }
}

impl<O, F: FnOnce() -> O> Drop for CallOnDrop<O, F> {
    fn drop(&mut self) {
        let f = unsafe { ManuallyDrop::take(&mut self.f) };
        f();
    }
}

/// The same as `std::future::Pending<T>`; we can't use that type directly because on rustc
/// versions <1.60 it didn't unconditionally implement `Send`.
// FIXME: use `std::future::Pending<T>` once the MSRV is >=1.60
struct Pending<T>(PhantomData<fn() -> T>);

impl<T> Future for Pending<T> {
    type Output = T;

    fn poll(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Self::Output> {
        Poll::Pending
    }
}


Review comment (suggestion):
I think it reads better if we replace this with a function that actually calls the closure.
```suggestion
    fn call(self) -> O {
        let mut this = ManuallyDrop::new(self);
        let f = unsafe { ManuallyDrop::take(&mut this.f) };
        f()
    }
```

Code after:
        // - `raw` points to a valid instance of `U` (we just wrote it in).
        let boxed = unsafe { Box::from_raw(raw) };

        callback(boxed)
    });

    // Drop the old value.
    unsafe { ptr::drop_in_place(raw) };

    // Run the rest of the code.
    Ok(guard.call())
}

struct CallOnDrop<O, F: FnOnce() -> O> {
    f: ManuallyDrop<F>,
}

impl<O, F: FnOnce() -> O> CallOnDrop<O, F> {
    fn new(f: F) -> Self {
        let f = ManuallyDrop::new(f);
        Self { f }
    }
    fn call(self) -> O {
        let mut this = ManuallyDrop::new(self);
        let f = unsafe { ManuallyDrop::take(&mut this.f) };
        f()
    }
}

impl<O, F: FnOnce() -> O> Drop for CallOnDrop<O, F> {
    fn drop(&mut self) {
        let f = unsafe { ManuallyDrop::take(&mut self.f) };
        f();
    }
}

/// The same as `std::future::Pending<T>`; we can't use that type directly because on rustc
/// versions <1.60 it didn't unconditionally implement `Send`.
// FIXME: use `std::future::Pending<T>` once the MSRV is >=1.60
struct Pending<T>(PhantomData<fn() -> T>);

impl<T> Future for Pending<T> {
    type Output = T;

    fn poll(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Self::Output> {
        Poll::Pending
    }
}
Repository: tokio-rs/tokio
File: tokio-util/tests/spawn_pinned.rs (Rust)

Code before:
    assert_eq!(loads[2], 0);

    let thread_id1 = handle1.unwrap().await.unwrap();
    let thread_id2 = handle2.unwrap().await.unwrap();

    assert_ne!(thread_id1, thread_id2);
}

#[tokio::test]
async fn spawn_on_all_workers() {
    let pool = task::LocalPoolHandle::new(3);

    let _ = pool.spawn_pinned_on_all_workers(|| {
        // Rc is !Send + !Sync
        let local_data = Rc::new("test");

        // This future holds an Rc, so it is !Send
        async move { local_data.to_string() }
    });

    let loads = pool.get_task_loads_for_each_worker();
    assert_eq!(loads[0], 1);
    assert_eq!(loads[1], 1);
    assert_eq!(loads[2], 1);
}


Review comment (suggestion):
This test is fragile. The tasks could complete before you check the load. Consider using a `tokio::sync::Barrier` to prevent them from stopping before checking these counts.

Code after:
        || async move {
            barrier1.wait().await;
            std::thread::current().id()
        },
        0,
    );
    let _ = pool.spawn_pinned_by_idx(
        || async move {
            barrier2.wait().await;
            std::thread::current().id()
        },
        0,
    );
    let handle2 = pool.spawn_pinned_by_idx(
        || async move {
            barrier3.wait().await;
            std::thread::current().id()
        },
        1,
    );

    let loads = pool.get_task_loads_for_each_worker();
    barrier.wait().await;
    assert_eq!(loads[0], 2);
    assert_eq!(loads[1], 1);
    assert_eq!(loads[2], 0);

    let thread_id1 = handle1.await.unwrap();
    let thread_id2 = handle2.await.unwrap();

    assert_ne!(thread_id1, thread_id2);
}
Repository: tokio-rs/tokio
File: tokio-util/src/task/spawn_pinned.rs (Rust)

Code before:
    /// # Examples
    ///
    /// ```
    /// use std::rc::Rc;
    /// use tokio_util::task::LocalPoolHandle;
    ///
    /// #[tokio::main]
    /// async fn main() {
    ///     let pool = LocalPoolHandle::new(3);
    ///
    ///     let _ = pool.spawn_pinned_on_all_workers(|| {
    ///         // Rc is !Send + !Sync
    ///         let local_data = Rc::new("test");
    ///
    ///         // This future holds an Rc, so it is !Send
    ///         async move { local_data.to_string() }
    ///     });
    /// }
    /// ```
    pub fn spawn_pinned_on_all_workers<F, Fut>(
        &self,
        create_task: F,
    ) -> Vec<JoinHandle<Fut::Output>>
    where
        F: FnOnce() -> Fut,
        F: Send + Clone + 'static,
        Fut: Future + 'static,
        Fut::Output: Send + 'static,
    {
        (0..self.pool.workers.len())
            .map(|idx| self.spawn_pinned_by_idx(create_task.clone(), idx))
            .collect::<Vec<_>>()
    }
}

impl Debug for LocalPoolHandle {
    fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
        f.write_str("LocalPoolHandle")
    }
}

enum WorkerChoice {
    LeastBurdened,
    ByIdx(usize),
}

struct LocalPool {
    workers: Vec<LocalWorkerHandle>,
}

impl LocalPool {

Review comment (suggestion):
I just now noticed that this is cloning the closure. I generally find that bad practice, and would prefer that this method takes an `Fn` and calls it multiple times rather than cloning it.

Code after:
    ///     let handles = (0..pool.num_threads())
    ///         .map(|worker_idx| {
    ///             pool.spawn_pinned_by_idx(
    ///                 || {
    ///                     async {
    ///                         "test"
    ///                     }
    ///                 },
    ///                 worker_idx,
    ///             )
    ///         })
    ///         .collect::<Vec<_>>();
    ///
    ///     for handle in handles {
    ///         handle.await.unwrap();
    ///     }
    /// }
    /// ```
    ///
    pub fn spawn_pinned_by_idx<F, Fut>(&self, create_task: F, idx: usize) -> JoinHandle<Fut::Output>
    where
        F: FnOnce() -> Fut,
        F: Send + 'static,
        Fut: Future + 'static,
        Fut::Output: Send + 'static,
    {
        self.pool
            .spawn_pinned(create_task, WorkerChoice::ByIdx(idx))
    }
}

impl Debug for LocalPoolHandle {
    fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
        f.write_str("LocalPoolHandle")
    }
}

enum WorkerChoice {
    LeastBurdened,
    ByIdx(usize),
}

struct LocalPool {
    workers: Vec<LocalWorkerHandle>,
}

impl LocalPool {
    /// Spawn a `?Send` future onto a worker
    fn spawn_pinned<F, Fut>(
        &self,
        create_task: F,
Repository: tokio-rs/tokio
File: tokio-util/src/task/spawn_pinned.rs (Rust)

Code before:
    ///
    /// This method can be used to spawn a task on all worker threads of the pool:
    ///
    /// ```
    /// use tokio_util::task::LocalPoolHandle;
    ///
    /// #[tokio::main]
    /// async fn main() {
    ///     const NUM_WORKERS: usize = 3;
    ///     let pool = LocalPoolHandle::new(NUM_WORKERS);
    ///     let handles = (0..pool.num_threads())
    ///         .map(|worker_idx| {
    ///             pool.spawn_pinned_by_idx(
    ///                 || {
    ///                     async {
    ///                         "test"
    ///                     }
    ///                 },
    ///                 worker_idx,
    ///             )
    ///         })
    ///         .collect::<Vec<_>>();
    ///
    ///     let _ = handles
    ///         .into_iter()
    ///         .map(|handle| async { handle.await.unwrap() });
    /// }
    /// ```
    ///
    pub fn spawn_pinned_by_idx<F, Fut>(&self, create_task: F, idx: usize) -> JoinHandle<Fut::Output>
    where
        F: FnOnce() -> Fut,
        F: Send + 'static,
        Fut: Future + 'static,
        Fut::Output: Send + 'static,
    {
        self.pool
            .spawn_pinned(create_task, WorkerChoice::ByIdx(idx))
    }
}

impl Debug for LocalPoolHandle {
    fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
        f.write_str("LocalPoolHandle")
    }
}

enum WorkerChoice {
    LeastBurdened,
    ByIdx(usize),
}

Review comment (suggestion):
This doesn't actually await them because iterators are lazy. Just use a for loop over `handles` instead.
```suggestion
    ///     for handle in handles {
    ///         handle.await.unwrap();
    ///     }
```

Code after:
    ///
    /// This method can be used to spawn a task on all worker threads of the pool:
    ///
    /// ```
    /// use tokio_util::task::LocalPoolHandle;
    ///
    /// #[tokio::main]
    /// async fn main() {
    ///     const NUM_WORKERS: usize = 3;
    ///     let pool = LocalPoolHandle::new(NUM_WORKERS);
    ///     let handles = (0..pool.num_threads())
    ///         .map(|worker_idx| {
    ///             pool.spawn_pinned_by_idx(
    ///                 || {
    ///                     async {
    ///                         "test"
    ///                     }
    ///                 },
    ///                 worker_idx,
    ///             )
    ///         })
    ///         .collect::<Vec<_>>();
    ///
    ///     for handle in handles {
    ///         handle.await.unwrap();
    ///     }
    /// }
    /// ```
    ///
    pub fn spawn_pinned_by_idx<F, Fut>(&self, create_task: F, idx: usize) -> JoinHandle<Fut::Output>
    where
        F: FnOnce() -> Fut,
        F: Send + 'static,
        Fut: Future + 'static,
        Fut::Output: Send + 'static,
    {
        self.pool
            .spawn_pinned(create_task, WorkerChoice::ByIdx(idx))
    }
}

impl Debug for LocalPoolHandle {
    fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
        f.write_str("LocalPoolHandle")
    }
}

enum WorkerChoice {
    LeastBurdened,
    ByIdx(usize),
}
Repository: tokio-rs/tokio
File: tokio/src/task/join_set.rs (Rust)

Code before:
///     let mut seen = [false; 10];
///     while let Some(res) = set.join_one().await.unwrap() {
///         seen[res] = true;
///     }
///
///     for i in 0..10 {
///         assert!(seen[i]);
///     }
/// }
/// ```
///
/// [unstable]: crate#unstable-features
#[cfg_attr(docsrs, doc(cfg(all(feature = "rt", tokio_unstable))))]
pub struct JoinSet<T> {
    inner: IdleNotifiedSet<JoinHandle<T>>,
}

/// A variant of [`task::Builder`] that spawns tasks on a [`JoinSet`] rather
/// than on the current default runtime.
/// 
/// [`task::Builder`]: crate::task::Builder
#[cfg(all(tokio_unstable, feature = "tracing"))]
#[cfg_attr(docsrs, doc(cfg(all(tokio_unstable, feature = "tracing"))))]
#[must_use = "builders do nothing unless used to spawn a task"]
#[derive(Debug)]
pub struct Builder<'a, T> {
    joinset: &'a mut JoinSet<T>,
    builder: super::Builder<'a>,
}

impl<T> JoinSet<T> {
    /// Create a new `JoinSet`.
    pub fn new() -> Self {
        Self {
            inner: IdleNotifiedSet::new(),
        }
    }

    /// Returns the number of tasks currently in the `JoinSet`.
    pub fn len(&self) -> usize {
        self.inner.len()
    }

    /// Returns whether the `JoinSet` is empty.
    pub fn is_empty(&self) -> bool {
        self.inner.is_empty()
    }
}

impl<T: 'static> JoinSet<T> {
    /// Returns a [`Builder`] that can be used to configure a task prior to

Review comment (bug):
This `#[derive(Debug)]` will put a `T: Debug` constraint on the impl, but it isn't actually necessary as `JoinSet` is `Debug` for non-Debug return types too.

Code after:
///     let mut seen = [false; 10];
///     while let Some(res) = set.join_one().await.unwrap() {
///         seen[res] = true;
///     }
///
///     for i in 0..10 {
///         assert!(seen[i]);
///     }
/// }
/// ```
///
/// [unstable]: crate#unstable-features
#[cfg_attr(docsrs, doc(cfg(all(feature = "rt", tokio_unstable))))]
pub struct JoinSet<T> {
    inner: IdleNotifiedSet<JoinHandle<T>>,
}

/// A variant of [`task::Builder`] that spawns tasks on a [`JoinSet`] rather
/// than on the current default runtime.
///
/// [`task::Builder`]: crate::task::Builder
#[cfg(all(tokio_unstable, feature = "tracing"))]
#[cfg_attr(docsrs, doc(cfg(all(tokio_unstable, feature = "tracing"))))]
#[must_use = "builders do nothing unless used to spawn a task"]
pub struct Builder<'a, T> {
    joinset: &'a mut JoinSet<T>,
    builder: super::Builder<'a>,
}

impl<T> JoinSet<T> {
    /// Create a new `JoinSet`.
    pub fn new() -> Self {
        Self {
            inner: IdleNotifiedSet::new(),
        }
    }

    /// Returns the number of tasks currently in the `JoinSet`.
    pub fn len(&self) -> usize {
        self.inner.len()
    }

    /// Returns whether the `JoinSet` is empty.
    pub fn is_empty(&self) -> bool {
        self.inner.is_empty()
    }
}

impl<T: 'static> JoinSet<T> {
    /// Returns a [`Builder`] that can be used to configure a task prior to
    /// spawning it on this `JoinSet`.
Repository: tokio-rs/tokio
File: tokio/src/task/join_set.rs (Rust)

Code before:
    pub fn is_empty(&self) -> bool {
        self.inner.is_empty()
    }
}

impl<T: 'static> JoinSet<T> {
    /// Returns a [`Builder`] that can be used to configure a task prior to
    /// spawning it on this `JoinSet`.
    ///
    /// # Examples
    ///
    /// ```
    /// use tokio::task::JoinSet;
    ///
    /// #[tokio::main]
    /// async fn main() {
    ///     let mut set = JoinSet::new();
    ///
    ///     // Use the builder to configure a task's name before spawning it.
    ///     set.build_task()
    ///         .name("my_task")
    ///         .spawn(async { /* ... */ });
    /// }
    /// ```
    #[cfg(all(tokio_unstable, feature = "tracing"))]
    #[cfg_attr(docsrs, doc(cfg(all(tokio_unstable, feature = "tracing"))))]
    pub fn build_task(&mut self) -> Builder<'_, T> {
        Builder {
            builder: super::Builder::new(),
            joinset: self,
        }
    }

    /// Spawn the provided task on the `JoinSet`, returning an [`AbortHandle`]
    /// that can be used to remotely cancel the task.
    ///
    /// # Panics
    ///
    /// This method panics if called outside of a Tokio runtime.
    ///
    /// [`AbortHandle`]: crate::task::AbortHandle
    pub fn spawn<F>(&mut self, task: F) -> AbortHandle
    where
        F: Future<Output = T>,
        F: Send + 'static,
        T: Send,
    {
        self.insert(crate::spawn(task))
    }

    /// Spawn the provided task on the provided runtime and store it in this

Review comment (suggestion):
Idiom wise, Tokio favors initializing the builder (vs a `build` method on the type being built).

So, this would be:

```
Builder::new(&mut join_set)`
```

Code after:
        self.inner.is_empty()
    }
}

impl<T: 'static> JoinSet<T> {
    /// Returns a [`Builder`] that can be used to configure a task prior to
    /// spawning it on this `JoinSet`.
    ///
    /// # Examples
    ///
    /// ```
    /// use tokio::task::JoinSet;
    ///
    /// #[tokio::main]
    /// async fn main() {
    ///     let mut set = JoinSet::new();
    ///
    ///     // Use the builder to configure a task's name before spawning it.
    ///     set.build_task()
    ///         .name("my_task")
    ///         .spawn(async { /* ... */ });
    /// }
    /// ```
    #[cfg(all(tokio_unstable, feature = "tracing"))]
    #[cfg_attr(docsrs, doc(cfg(all(tokio_unstable, feature = "tracing"))))]
    pub fn build_task(&mut self) -> Builder<'_, T> {
        Builder {
            builder: super::Builder::new(),
            joinset: self,
        }
    }

    /// Spawn the provided task on the `JoinSet`, returning an [`AbortHandle`]
    /// that can be used to remotely cancel the task.
    ///
    /// # Panics
    ///
    /// This method panics if called outside of a Tokio runtime.
    ///
    /// [`AbortHandle`]: crate::task::AbortHandle
    #[track_caller]
    pub fn spawn<F>(&mut self, task: F) -> AbortHandle
    where
        F: Future<Output = T>,
        F: Send + 'static,
        T: Send,
    {
        self.insert(crate::spawn(task))
    }

    /// Spawn the provided task on the provided runtime and store it in this
Repository: tokio-rs/tokio
File: tokio-stream/src/stream_ext/chunks_timeout.rs (Rust)

Code before:
                if me.items.len() >= *me.cap {
                    return Poll::Ready(Some(self.take()));
                }
            }

            Poll::Ready(None) => {
                let last = if me.items.is_empty() {
                    None
                } else {
                    let full_buf = std::mem::take(me.items);
                    Some(full_buf)
                };

                return Poll::Ready(last);
            }
        }

        match me.deadline.as_mut().poll(cx) {
            Poll::Pending => {}
            Poll::Ready(()) => {
                return Poll::Ready(Some(self.take()));
            }
        }

        cx.waker().wake_by_ref();
        Poll::Pending
    }

    fn size_hint(&self) -> (usize, Option<usize>) {
        let chunk_len = if self.items.is_empty() { 0 } else { 1 };
        let (lower, upper) = self.stream.size_hint();
        let lower = lower.saturating_add(chunk_len);
        let upper = upper.and_then(|x| x.checked_add(chunk_len));
        (lower, upper)
    }
}


Review comment (suggestion):
```suggestion
        match me.deadline.as_mut().poll(cx) {
            Poll::Pending => Poll::Pending,
            Poll::Ready(()) => Poll::Ready(Some(self.take())),
        }
```

Code after:
                    // Returning Some here is only correct because we fuse the inner stream.
                    let last = if me.items.is_empty() {
                        None
                    } else {
                        Some(std::mem::take(me.items))
                    };

                    return Poll::Ready(last);
                }
            }
        }

        if !me.items.is_empty() {
            ready!(me.deadline.poll(cx));
            return Poll::Ready(Some(std::mem::take(me.items)));
        }

        Poll::Pending
    }

    fn size_hint(&self) -> (usize, Option<usize>) {
        let chunk_len = if self.items.is_empty() { 0 } else { 1 };
        let (lower, upper) = self.stream.size_hint();
        let lower = (lower / self.cap).saturating_add(chunk_len);
        let upper = upper.and_then(|x| x.checked_add(chunk_len));
        (lower, upper)
    }
}
Repository: tokio-rs/tokio
File: tokio-stream/src/stream_ext/chunks_timeout.rs (Rust)

Code before:
                    let last = if me.items.is_empty() {
                        None
                    } else {
                        Some(std::mem::take(me.items))
                    };

                    return Poll::Ready(last);
                }
            }
        }

        if !me.items.is_empty() {
            ready!(me.deadline.poll(cx));
            return Poll::Ready(Some(std::mem::take(me.items)));
        }

        Poll::Pending
    }

    fn size_hint(&self) -> (usize, Option<usize>) {
        let chunk_len = if self.items.is_empty() { 0 } else { 1 };
        let (lower, upper) = self.stream.size_hint();
        let lower = lower.saturating_add(chunk_len);
        let upper = upper.and_then(|x| x.checked_add(chunk_len));
        (lower, upper)
    }
}


Review comment (bug):
I just realized that the lower bound of this size hint is wrong. Even if `self.items` has 200 more items, we might only return 20 items if our `max_size` is 10. (The upper bound is correct because we might yield every item separately.)

The lower bound in this suggestion is not completely precise - it might sometimes be less than the fully precise size hint. However, computing the precise size hint is somewhat difficult.
```suggestion
    fn size_hint(&self) -> (usize, Option<usize>) {
        let chunk_len = if self.items.is_empty() { 0 } else { 1 };
        let (lower, upper) = self.stream.size_hint();
        let lower = (lower / self.cap).saturating_add(chunk_len);
        let upper = upper.and_then(|x| x.checked_add(chunk_len));
        (lower, upper)
    }
```
The same issue exists in the futures crate. I opened https://github.com/rust-lang/futures-rs/issues/2610 for that.

Code after:
                    let last = if me.items.is_empty() {
                        None
                    } else {
                        Some(std::mem::take(me.items))
                    };

                    return Poll::Ready(last);
                }
            }
        }

        if !me.items.is_empty() {
            ready!(me.deadline.poll(cx));
            return Poll::Ready(Some(std::mem::take(me.items)));
        }

        Poll::Pending
    }

    fn size_hint(&self) -> (usize, Option<usize>) {
        let chunk_len = if self.items.is_empty() { 0 } else { 1 };
        let (lower, upper) = self.stream.size_hint();
        let lower = (lower / self.cap).saturating_add(chunk_len);
        let upper = upper.and_then(|x| x.checked_add(chunk_len));
        (lower, upper)
    }
}