Repository: tokio-rs/tokio
File: tokio/src/runtime/basic_scheduler.rs (Rust)

Code before:
impl Context {
    /// Execute the closure with the given scheduler core stored in the
    /// thread-local context.
    fn run_task<R>(&self, mut core: Box<Core>, f: impl FnOnce() -> R) -> (Box<Core>, R) {
        core.stats.incr_poll_count();
        self.enter(core, || crate::coop::budget(f))
    }

    fn park(&self, mut core: Box<Core>) -> Box<Core> {
        let mut park = core.park.take().expect("park missing");

        if let Some(f) = &self.spawner.shared.before_park {
            #[allow(clippy::redundant_closure)]
            let (c, _) = self.enter(core, || f());
            core = c;
        }

        // This check will fail if `before_park` spawns a task for us to run
        // instead of parking the thread
        if core.tasks.is_empty() {
            // Park until the thread is signaled
            core.stats.about_to_park();
            core.stats.submit(&core.spawner.shared.stats);

            let (c, _) = self.enter(core, || {
                park.park().expect("failed to park");
            });

            core = c;
            core.stats.returned_from_park();
        }

        if let Some(f) = &self.spawner.shared.after_unpark {
            #[allow(clippy::redundant_closure)]
            let (c, _) = self.enter(core, || f());
            core = c;
        }

        core.park = Some(park);
        core
    }

    fn park_yield(&self, mut core: Box<Core>) -> Box<Core> {
        let mut park = core.park.take().expect("park missing");

        core.stats.submit(&core.spawner.shared.stats);
        let (mut core, _) = self.enter(core, || {
            park.park_timeout(Duration::from_millis(0))
                .expect("failed to park");
        });


Review comment (bug):
if this panics do we fail to reset the core back to allow another thread to make progress? Its been a while since I've looked at this code but not sure if the parker panicking also just means the whole runtime has just gone bust.

Code after:
}

impl Context {
    /// Execute the closure with the given scheduler core stored in the
    /// thread-local context.
    fn run_task<R>(&self, mut core: Box<Core>, f: impl FnOnce() -> R) -> (Box<Core>, R) {
        core.stats.incr_poll_count();
        self.enter(core, || crate::coop::budget(f))
    }

    /// Blocks the current thread until an event is received by the driver,
    /// including I/O events, timer events, ...
    fn park(&self, mut core: Box<Core>) -> Box<Core> {
        let mut driver = core.driver.take().expect("driver missing");

        if let Some(f) = &self.spawner.shared.before_park {
            // Incorrect lint, the closures are actually different types so `f`
            // cannot be passed as an argument to `enter`.
            #[allow(clippy::redundant_closure)]
            let (c, _) = self.enter(core, || f());
            core = c;
        }

        // This check will fail if `before_park` spawns a task for us to run
        // instead of parking the thread
        if core.tasks.is_empty() {
            // Park until the thread is signaled
            core.stats.about_to_park();
            core.stats.submit(&core.spawner.shared.stats);

            let (c, _) = self.enter(core, || {
                driver.park().expect("failed to park");
            });

            core = c;
            core.stats.returned_from_park();
        }

        if let Some(f) = &self.spawner.shared.after_unpark {
            // Incorrect lint, the closures are actually different types so `f`
            // cannot be passed as an argument to `enter`.
            #[allow(clippy::redundant_closure)]
            let (c, _) = self.enter(core, || f());
            core = c;
        }

        core.driver = Some(driver);
        core
    }

    /// Checks the driver for new events without blocking the thread.