ronantakizawa/codereview-bench huggingface dataset:
Repository: tokio-rs/tokio
File: tokio/src/sync/tests/loom_watch.rs (Rust)

Code before:
        block_on(rx5.changed()).unwrap();
        assert_eq!(*rx5.borrow(), 2);

        th.join().unwrap();
    })
}

#[test]
fn wait_for_test() {
    loom::model(move || {
        let (tx, mut rx) = watch::channel(false);

        let tx_arc = Arc::new(tx);
        let tx1 = tx_arc.clone();
        let tx2 = tx_arc.clone();

        let th1 = thread::spawn(move || {
            for _ in 0..3 {
                tx1.send_modify(|x| {
                    // here we avoid unsetting the value in case
                    // the value is already set to true.
                    if !*x {
                        // force the loop to run.
                        *x = false;
                    }
                });
            }
        });

        let th2 = thread::spawn(move || {
            tx2.send(true).unwrap();
        });

        let result = block_on(rx.wait_for(|x| *x));
        assert_eq!(result.unwrap(), true);

        th1.join().unwrap();
        th2.join().unwrap();
    });
}


Review comment (suggestion):
Here you're just doing "if false, then set to false". That's equivalent to this:
```suggestion
                tx1.send_modify(|_x| {});
```

Code after:
        block_on(rx5.changed()).unwrap();
        assert_eq!(*rx5.borrow(), 2);

        th.join().unwrap();
    })
}

#[test]
fn wait_for_test() {
    loom::model(move || {
        let (tx, mut rx) = watch::channel(false);

        let tx_arc = Arc::new(tx);
        let tx1 = tx_arc.clone();
        let tx2 = tx_arc.clone();

        let th1 = thread::spawn(move || {
            for _ in 0..2 {
                tx1.send_modify(|_x| {});
            }
        });

        let th2 = thread::spawn(move || {
            tx2.send(true).unwrap();
        });

        assert_eq!(*block_on(rx.wait_for(|x| *x)).unwrap(), true);

        th1.join().unwrap();
        th2.join().unwrap();
    });
}

#[test]
fn wait_for_returns_correct_value() {
    loom::model(move || {
        let (tx, mut rx) = watch::channel(0);

        let jh = thread::spawn(move || {
            tx.send(1).unwrap();
            tx.send(2).unwrap();
            tx.send(3).unwrap();
        });

        // Stop at the first value we are called at.
        let mut stopped_at = usize::MAX;
        let returned = *block_on(rx.wait_for(|x| {
            stopped_at = *x;
            true
        }))
        .unwrap();
Repository: tokio-rs/tokio
File: tokio/src/sync/tests/loom_watch.rs (Rust)

Code before:
        assert_eq!(*rx4.borrow(), 2);

        block_on(rx5.changed()).unwrap();
        assert_eq!(*rx5.borrow(), 2);

        th.join().unwrap();
    })
}

#[test]
fn wait_for_test() {
    loom::model(move || {
        let (tx, mut rx) = watch::channel(false);

        let tx_arc = Arc::new(tx);
        let tx1 = tx_arc.clone();
        let tx2 = tx_arc.clone();

        let th1 = thread::spawn(move || {
            for _ in 0..3 {
                tx1.send_modify(|_x| {});
            }
        });

        let th2 = thread::spawn(move || {
            tx2.send(true).unwrap();
        });

        block_on(rx.wait_for(|x| *x)).unwrap();

        th1.join().unwrap();
        th2.join().unwrap();
    });
}


Review comment (suggestion):
@Darksonn @debadree25  I think It's not necessary to call `unwrap()` on the result of `tx2.send(true`). Since the test is running in a loom thread, any panics will be caught and reported by loom, so you can leave the `Ok(())` result of `tx2.send(true)` unchecked. That makes sense?

```suggestion
            tx2.send(true).ok();
```

Code after:
        assert_eq!(*rx4.borrow(), 2);

        block_on(rx5.changed()).unwrap();
        assert_eq!(*rx5.borrow(), 2);

        th.join().unwrap();
    })
}

#[test]
fn wait_for_test() {
    loom::model(move || {
        let (tx, mut rx) = watch::channel(false);

        let tx_arc = Arc::new(tx);
        let tx1 = tx_arc.clone();
        let tx2 = tx_arc.clone();

        let th1 = thread::spawn(move || {
            for _ in 0..2 {
                tx1.send_modify(|_x| {});
            }
        });

        let th2 = thread::spawn(move || {
            tx2.send(true).unwrap();
        });

        assert_eq!(*block_on(rx.wait_for(|x| *x)).unwrap(), true);

        th1.join().unwrap();
        th2.join().unwrap();
    });
}

#[test]
fn wait_for_returns_correct_value() {
    loom::model(move || {
        let (tx, mut rx) = watch::channel(0);

        let jh = thread::spawn(move || {
            tx.send(1).unwrap();
            tx.send(2).unwrap();
            tx.send(3).unwrap();
        });

        // Stop at the first value we are called at.
        let mut stopped_at = usize::MAX;
        let returned = *block_on(rx.wait_for(|x| {
            stopped_at = *x;
            true
Repository: tokio-rs/tokio
File: tokio/src/sync/tests/loom_watch.rs (Rust)

Code before:

        block_on(rx3.changed()).unwrap();
        assert_eq!(*rx3.borrow(), 2);

        block_on(rx4.changed()).unwrap();
        assert_eq!(*rx4.borrow(), 2);

        block_on(rx5.changed()).unwrap();
        assert_eq!(*rx5.borrow(), 2);

        th.join().unwrap();
    })
}

#[test]
fn wait_for_test() {
    loom::model(move || {
        let (tx, mut rx) = watch::channel(false);

        let tx_arc = Arc::new(tx);
        let tx1 = tx_arc.clone();
        let tx2 = tx_arc.clone();

        let th1 = thread::spawn(move || {
            for _ in 0..3 {
                tx1.send_modify(|_x| {});
            }
        });

        let th2 = thread::spawn(move || {
            tx2.send(true).unwrap();
        });

        block_on(rx.wait_for(|x| *x)).unwrap();

        th1.join().unwrap();
        th2.join().unwrap();
    });
}


Review comment (suggestion):
@Darksonn @debadree25 In this for loop its calling **send_modify** with an empty closure, which doesn't modify the value being sent. I think It would be more meaningful to actually modify the value being sent, so that the test exercises the wait_for function more thoroughly.  That makes sense, something like?:

 tx1.send_modify(|x| *x = true);
```suggestion
                tx1.send_modify(|x| *x = true);
```


Code after:

        block_on(rx3.changed()).unwrap();
        assert_eq!(*rx3.borrow(), 2);

        block_on(rx4.changed()).unwrap();
        assert_eq!(*rx4.borrow(), 2);

        block_on(rx5.changed()).unwrap();
        assert_eq!(*rx5.borrow(), 2);

        th.join().unwrap();
    })
}

#[test]
fn wait_for_test() {
    loom::model(move || {
        let (tx, mut rx) = watch::channel(false);

        let tx_arc = Arc::new(tx);
        let tx1 = tx_arc.clone();
        let tx2 = tx_arc.clone();

        let th1 = thread::spawn(move || {
            for _ in 0..2 {
                tx1.send_modify(|_x| {});
            }
        });

        let th2 = thread::spawn(move || {
            tx2.send(true).unwrap();
        });

        assert_eq!(*block_on(rx.wait_for(|x| *x)).unwrap(), true);

        th1.join().unwrap();
        th2.join().unwrap();
    });
}

#[test]
fn wait_for_returns_correct_value() {
    loom::model(move || {
        let (tx, mut rx) = watch::channel(0);

        let jh = thread::spawn(move || {
            tx.send(1).unwrap();
            tx.send(2).unwrap();
            tx.send(3).unwrap();
        });
Repository: tokio-rs/tokio
File: tokio/src/sync/tests/loom_watch.rs (Rust)

Code before:
        assert_eq!(*rx2.borrow(), 2);

        block_on(rx3.changed()).unwrap();
        assert_eq!(*rx3.borrow(), 2);

        block_on(rx4.changed()).unwrap();
        assert_eq!(*rx4.borrow(), 2);

        block_on(rx5.changed()).unwrap();
        assert_eq!(*rx5.borrow(), 2);

        th.join().unwrap();
    })
}

#[test]
fn wait_for_test() {
    loom::model(move || {
        let (tx, mut rx) = watch::channel(false);

        let tx_arc = Arc::new(tx);
        let tx1 = tx_arc.clone();
        let tx2 = tx_arc.clone();

        let th1 = thread::spawn(move || {
            for _ in 0..3 {
                tx1.send_modify(|_x| {});
            }
        });

        let th2 = thread::spawn(move || {
            tx2.send(true).unwrap();
        });

        assert_eq!(*block_on(rx.wait_for(|x| *x)).unwrap(), true);

        th1.join().unwrap();
        th2.join().unwrap();
    });
}


Review comment (suggestion):
This test takes quite a long time with `LOOM_MAX_PREEMPTIONS=2`, as configured in CI. This makes it run faster.
```suggestion
            for _ in 0..2 {
```

Code after:
        assert_eq!(*rx2.borrow(), 2);

        block_on(rx3.changed()).unwrap();
        assert_eq!(*rx3.borrow(), 2);

        block_on(rx4.changed()).unwrap();
        assert_eq!(*rx4.borrow(), 2);

        block_on(rx5.changed()).unwrap();
        assert_eq!(*rx5.borrow(), 2);

        th.join().unwrap();
    })
}

#[test]
fn wait_for_test() {
    loom::model(move || {
        let (tx, mut rx) = watch::channel(false);

        let tx_arc = Arc::new(tx);
        let tx1 = tx_arc.clone();
        let tx2 = tx_arc.clone();

        let th1 = thread::spawn(move || {
            for _ in 0..2 {
                tx1.send_modify(|_x| {});
            }
        });

        let th2 = thread::spawn(move || {
            tx2.send(true).unwrap();
        });

        assert_eq!(*block_on(rx.wait_for(|x| *x)).unwrap(), true);

        th1.join().unwrap();
        th2.join().unwrap();
    });
}

#[test]
fn wait_for_returns_correct_value() {
    loom::model(move || {
        let (tx, mut rx) = watch::channel(0);

        let jh = thread::spawn(move || {
            tx.send(1).unwrap();
            tx.send(2).unwrap();
            tx.send(3).unwrap();
        });
Repository: tokio-rs/tokio
File: tokio/src/util/linked_list.rs (Rust)

Code before:
            self.tail = L::pointers(node).as_ref().get_prev();
        }

        L::pointers(node).as_mut().set_next(None);
        L::pointers(node).as_mut().set_prev(None);

        Some(L::from_raw(node))
    }
}

impl<L: Link> fmt::Debug for LinkedList<L, L::Target> {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("LinkedList")
            .field("head", &self.head)
            .field("tail", &self.tail)
            .finish()
    }
}

// ===== impl CountedLinkedList ====

// Delegates operations to the base LinkedList implementation, and adds a counter to the elements
// in the list.
pub(crate) struct CountedLinkedList<L: Link, T> {
    _list: LinkedList<L, T>,
    _count: AtomicUsize,
}

impl<L: Link> CountedLinkedList<L, L::Target> {
    pub(crate) const fn new() -> CountedLinkedList<L, L::Target> {
        CountedLinkedList {
            _list: LinkedList::new(),
            _count: AtomicUsize::new(0),
        }
    }

    pub(crate) fn push_front(&mut self, val: L::Handle) {
        self._list.push_front(val);
        self._count.fetch_add(1, Ordering::Relaxed);
    }

    pub(crate) fn pop_back(&mut self) -> Option<L::Handle> {
        let val = self._list.pop_back();
        if val.is_some() {
            self._count.fetch_sub(1, Ordering::Relaxed);
        }
        val
    }

    pub(crate) fn is_empty(&self) -> bool {
        self._list.is_empty()

Review comment (suggestion):
These fields are not unused, so their names should not start with `_`.
```suggestion
    list: LinkedList<L, T>,
    count: AtomicUsize,
```

Code after:
        }

        L::pointers(node).as_mut().set_next(None);
        L::pointers(node).as_mut().set_prev(None);

        Some(L::from_raw(node))
    }
}

impl<L: Link> fmt::Debug for LinkedList<L, L::Target> {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("LinkedList")
            .field("head", &self.head)
            .field("tail", &self.tail)
            .finish()
    }
}

// ===== impl CountedLinkedList ====

// Delegates operations to the base LinkedList implementation, and adds a counter to the elements
// in the list.
pub(crate) struct CountedLinkedList<L: Link, T> {
    list: LinkedList<L, T>,
    count: usize,
}

impl<L: Link> CountedLinkedList<L, L::Target> {
    pub(crate) fn new() -> CountedLinkedList<L, L::Target> {
        CountedLinkedList {
            list: LinkedList::new(),
            count: 0,
        }
    }

    pub(crate) fn push_front(&mut self, val: L::Handle) {
        self.list.push_front(val);
        self.count += 1;
    }

    pub(crate) fn pop_back(&mut self) -> Option<L::Handle> {
        let val = self.list.pop_back();
        if val.is_some() {
            self.count -= 1;
        }
        val
    }

    pub(crate) fn is_empty(&self) -> bool {
        self.list.is_empty()
    }
Repository: tokio-rs/tokio
File: tokio/src/util/linked_list.rs (Rust)

Code before:
            self.tail = L::pointers(node).as_ref().get_prev();
        }

        L::pointers(node).as_mut().set_next(None);
        L::pointers(node).as_mut().set_prev(None);

        Some(L::from_raw(node))
    }
}

impl<L: Link> fmt::Debug for LinkedList<L, L::Target> {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("LinkedList")
            .field("head", &self.head)
            .field("tail", &self.tail)
            .finish()
    }
}

// ===== impl CountedLinkedList ====

// Delegates operations to the base LinkedList implementation, and adds a counter to the elements
// in the list.
pub(crate) struct CountedLinkedList<L: Link, T> {
    _list: LinkedList<L, T>,
    _count: AtomicUsize,
}

impl<L: Link> CountedLinkedList<L, L::Target> {
    pub(crate) const fn new() -> CountedLinkedList<L, L::Target> {
        CountedLinkedList {
            _list: LinkedList::new(),
            _count: AtomicUsize::new(0),
        }
    }

    pub(crate) fn push_front(&mut self, val: L::Handle) {
        self._list.push_front(val);
        self._count.fetch_add(1, Ordering::Relaxed);
    }

    pub(crate) fn pop_back(&mut self) -> Option<L::Handle> {
        let val = self._list.pop_back();
        if val.is_some() {
            self._count.fetch_sub(1, Ordering::Relaxed);
        }
        val
    }

    pub(crate) fn is_empty(&self) -> bool {
        self._list.is_empty()

Review comment (suggestion):
You have a mutable reference any time you modify it, so you don't need an atomic.
```suggestion
    _list: LinkedList<L, T>,
    _count: usize,
```
If you want an atomic, then you should place it outside of the mutex. In that case, the logic would need to go in `OwnedTasks` rather than in the linked list.

Code after:
        }

        L::pointers(node).as_mut().set_next(None);
        L::pointers(node).as_mut().set_prev(None);

        Some(L::from_raw(node))
    }
}

impl<L: Link> fmt::Debug for LinkedList<L, L::Target> {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("LinkedList")
            .field("head", &self.head)
            .field("tail", &self.tail)
            .finish()
    }
}

// ===== impl CountedLinkedList ====

// Delegates operations to the base LinkedList implementation, and adds a counter to the elements
// in the list.
pub(crate) struct CountedLinkedList<L: Link, T> {
    list: LinkedList<L, T>,
    count: usize,
}

impl<L: Link> CountedLinkedList<L, L::Target> {
    pub(crate) fn new() -> CountedLinkedList<L, L::Target> {
        CountedLinkedList {
            list: LinkedList::new(),
            count: 0,
        }
    }

    pub(crate) fn push_front(&mut self, val: L::Handle) {
        self.list.push_front(val);
        self.count += 1;
    }

    pub(crate) fn pop_back(&mut self) -> Option<L::Handle> {
        let val = self.list.pop_back();
        if val.is_some() {
            self.count -= 1;
        }
        val
    }

    pub(crate) fn is_empty(&self) -> bool {
        self.list.is_empty()
    }
Repository: tokio-rs/tokio
File: tokio/src/runtime/scheduler/current_thread.rs (Rust)

Code before:

            assert!(handle.shared.owned.is_empty());

            // Submit metrics
            core.metrics.submit(&handle.shared.worker_metrics);

            // Shutdown the resource drivers
            if let Some(driver) = core.driver.as_mut() {
                driver.shutdown(&handle.driver);
            }

            (core, ())
        });
    }
}

impl fmt::Debug for CurrentThread {
    fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
        fmt.debug_struct("CurrentThread").finish()
    }
}

// ===== impl Core =====

impl Core {
    fn pop_task(&mut self, handle: &Handle) -> Option<task::Notified<Arc<Handle>>> {
        let ret = self.tasks.pop_front();
        handle
            .shared
            .worker_metrics
            .set_queue_depth(self.tasks.len());
        ret
    }

    fn push_task(&mut self, handle: &Handle, task: task::Notified<Arc<Handle>>) {
        self.tasks.push_back(task);
        self.metrics.inc_local_schedule_count();
        handle
            .shared
            .worker_metrics
            .set_queue_depth(self.tasks.len());
    }
}

fn did_defer_tasks() -> bool {
    context::with_defer(|deferred| !deferred.is_empty()).unwrap()
}

fn wake_deferred_tasks() {
    context::with_defer(|deferred| deferred.wake());
}

Review comment (suggestion):
Renamed `pop_task` -> `next_task_*` to match names in the multi-threaded scheduler.

Code after:
                }
            }

            assert!(handle.shared.owned.is_empty());

            // Submit metrics
            core.submit_metrics(handle);

            // Shutdown the resource drivers
            if let Some(driver) = core.driver.as_mut() {
                driver.shutdown(&handle.driver);
            }

            (core, ())
        });
    }
}

impl fmt::Debug for CurrentThread {
    fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
        fmt.debug_struct("CurrentThread").finish()
    }
}

// ===== impl Core =====

impl Core {
    /// Get and increment the current tick
    fn tick(&mut self) {
        self.tick = self.tick.wrapping_add(1);
    }

    fn next_task(&mut self, handle: &Handle) -> Option<Notified> {
        if self.tick % handle.shared.config.global_queue_interval == 0 {
            handle.pop().or_else(|| self.next_local_task(handle))
        } else {
            self.next_local_task(handle).or_else(|| handle.pop())
        }
    }

    fn next_local_task(&mut self, handle: &Handle) -> Option<Notified> {
        let ret = self.tasks.pop_front();
        handle
            .shared
            .worker_metrics
            .set_queue_depth(self.tasks.len());
        ret
    }

    fn push_task(&mut self, handle: &Handle, task: Notified) {
        self.tasks.push_back(task);
Repository: tokio-rs/tokio
File: tokio/src/runtime/scheduler/current_thread.rs (Rust)

Code before:

/// Data required for executing the scheduler. The struct is passed around to
/// a function that will perform the scheduling work and acts as a capability token.
struct Core {
    /// Scheduler run queue
    tasks: VecDeque<Notified>,

    /// Current tick
    tick: u32,

    /// Runtime driver
    ///
    /// The driver is removed before starting to park the thread
    driver: Option<Driver>,

    /// Metrics batch
    metrics: MetricsBatch,

    /// True if a task panicked without being handled and the runtime is
    /// configured to shutdown on unhandled panic.
    unhandled_panic: bool,
}

/// Scheduler state shared between threads.
struct Shared {
    /// Remote run queue. None if the `Runtime` has been dropped.
    inject: Inject<Arc<Handle>>,

    /// Collection of all active tasks spawned onto this executor.
    owned: OwnedTasks<Arc<Handle>>,

    /// Indicates whether the blocked on thread was woken.
    woken: AtomicBool,

    /// Scheduler configuration options
    config: Config,

    /// Keeps track of various runtime metrics.
    scheduler_metrics: SchedulerMetrics,

    /// This scheduler only has one worker.
    worker_metrics: WorkerMetrics,
}

/// Thread-local context.
struct Context {
    /// Scheduler handle
    handle: Arc<Handle>,

    /// Scheduler core, enabling the holder of `Context` to execute the
    /// scheduler.

Review comment (suggestion):
I think the comment that this is `None` if the runtime has been dropped needs to be removed, as this is no longer an `Option`.

Code after:

/// Data required for executing the scheduler. The struct is passed around to
/// a function that will perform the scheduling work and acts as a capability token.
struct Core {
    /// Scheduler run queue
    tasks: VecDeque<Notified>,

    /// Current tick
    tick: u32,

    /// Runtime driver
    ///
    /// The driver is removed before starting to park the thread
    driver: Option<Driver>,

    /// Metrics batch
    metrics: MetricsBatch,

    /// True if a task panicked without being handled and the runtime is
    /// configured to shutdown on unhandled panic.
    unhandled_panic: bool,
}

/// Scheduler state shared between threads.
struct Shared {
    /// Remote run queue
    inject: Inject<Arc<Handle>>,

    /// Collection of all active tasks spawned onto this executor.
    owned: OwnedTasks<Arc<Handle>>,

    /// Indicates whether the blocked on thread was woken.
    woken: AtomicBool,

    /// Scheduler configuration options
    config: Config,

    /// Keeps track of various runtime metrics.
    scheduler_metrics: SchedulerMetrics,

    /// This scheduler only has one worker.
    worker_metrics: WorkerMetrics,
}

/// Thread-local context.
struct Context {
    /// Scheduler handle
    handle: Arc<Handle>,

    /// Scheduler core, enabling the holder of `Context` to execute the
    /// scheduler.