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

Code before:
        if !core.is_searching && core.run_queue.is_stealable() {
            self.worker.handle.notify_parked_local();
        }

        core
    }
}

impl Core {
    /// Increment the tick
    fn tick(&mut self) {
        self.tick = self.tick.wrapping_add(1);
    }

    /// Return the next notified task available to this worker.
    fn next_task(&mut self, worker: &Worker) -> Option<Notified> {
        if self.tick % worker.handle.shared.config.global_queue_interval == 0 {
            worker.inject().pop().or_else(|| self.next_local_task())
        } else {
            let maybe_task = self.next_local_task();

            if maybe_task.is_some() {
                return maybe_task;
            }

            let cap = self.run_queue.capacity();

            // The worker is currently idle, pull a batch of work from the
            // injection queue. We don't want to pull *all* the work so other
            // workers can also get some.
            let n = cmp::min(
                worker.inject().len() / worker.handle.shared.remotes.len() + 1,
                cap,
            );

            let n = cmp::min(n, self.run_queue.max_capacity() / 2);

            let mut tasks = worker.inject().pop_n(n);

            // Pop the first task to return immedietly
            let ret = tasks.next();

            // Push the rest of the on the run queue
            self.run_queue.push_back(tasks);

            ret
        }
    }

    fn next_local_task(&mut self) -> Option<Notified> {
        self.lifo_slot.take().or_else(|| self.run_queue.pop())

Review comment (bug):
The documentation of this method should probably call out that you can only _decrease_ the amount of available slots by accessing the `run_queue`, to help make it clear why `push_back` can't fail.

Code after:
            self.worker.handle.notify_parked_local();
        }

        core
    }
}

impl Core {
    /// Increment the tick
    fn tick(&mut self) {
        self.tick = self.tick.wrapping_add(1);
    }

    /// Return the next notified task available to this worker.
    fn next_task(&mut self, worker: &Worker) -> Option<Notified> {
        if self.tick % worker.handle.shared.config.global_queue_interval == 0 {
            worker.inject().pop().or_else(|| self.next_local_task())
        } else {
            let maybe_task = self.next_local_task();

            if maybe_task.is_some() {
                return maybe_task;
            }

            // Other threads can only **remove** tasks from the current worker's
            // `run_queue`. So, we can be confident that by the time we call
            // `run_queue.push_back` below, there will be *at least* `cap`
            // available slots in the queue.
            let cap = usize::min(
                self.run_queue.remaining_slots(),
                self.run_queue.max_capacity() / 2,
            );

            // The worker is currently idle, pull a batch of work from the
            // injection queue. We don't want to pull *all* the work so other
            // workers can also get some.
            let n = usize::min(
                worker.inject().len() / worker.handle.shared.remotes.len() + 1,
                cap,
            );

            let mut tasks = worker.inject().pop_n(n);

            // Pop the first task to return immedietly
            let ret = tasks.next();

            // Push the rest of the on the run queue
            self.run_queue.push_back(tasks);

            ret
        }