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

Code before:
            // guarantees that an Instant is monotonic, but can happen when running
            // Linux in a VM on a Windows host due to std incorrectly trusting the
            // hardware clock to be monotonic.
            //
            // See <https://github.com/tokio-rs/tokio/issues/3619> for more information.
            now = lock.elapsed();
        }

        while let Some(expiration) = lock.poll(now) {
            // Gets the number of entries in this slot, which will be
            // the maximum number of times we traverse.
            // Limiting the maximum number of iterations is very important,
            // because it's possible that those entries might need to be
            // reinserted into the same slot.

            // This happens on the highest level, when an entry is inserted
            // more than MAX_DURATION into the future. When this happens, we wrap
            // around, and process some entries a multiple of MAX_DURATION before
            // they actually need to be dropped down a level. We then reinsert them
            // back into the same position; we must make sure we don't then process
            // those entries again or we'll end up in an infinite loop.

            // This can also happens when other threads concurrently add entries to
            // this currently traversing slot.
            let count = lock.get_entries_count(&expiration);
            for _ in 0..count {
                if let Some(entry) = lock.get_mut_entries(&expiration).pop_back() {
                    // Try to expire the entry; this is cheap (doesn't synchronize) if
                    // the timer is not expired, and updates cached_when.
                    match unsafe { entry.mark_firing(expiration.deadline) } {
                        Ok(()) => {
                            // Entry was expired.
                            // SAFETY: We hold the driver lock, and just removed the entry from any linked lists.
                            if let Some(waker) = unsafe { entry.fire(Ok(())) } {
                                waker_list.push(waker);

                                if !waker_list.can_push() {
                                    // Wake a batch of wakers. To avoid deadlock,
                                    // we must do this with the lock temporarily dropped.
                                    drop(lock);
                                    waker_list.wake_all();

                                    lock = self.inner.lock_sharded_wheel(id);
                                }
                            }
                        }
                        Err(expiration_tick) => {
                            lock.add_entry(entry, expiration.clone(), expiration_tick);
                        }
                    }
                }

Review comment (performance):
The way we usually handle this kind of thing is to move _all_ entries in the list to another list (using the guarded linked list), and then we wake entries from the guarded list in batches. Can we not do the same here?

Code after:
            // guarantees that an Instant is monotonic, but can happen when running
            // Linux in a VM on a Windows host due to std incorrectly trusting the
            // hardware clock to be monotonic.
            //
            // See <https://github.com/tokio-rs/tokio/issues/3619> for more information.
            now = lock.elapsed();
        }

        while let Some(expiration) = lock.poll(now) {
            lock.set_elapsed(expiration.deadline);
            // It is critical for `GuardedLinkedList` safety that the guard node is
            // pinned in memory and is not dropped until the guarded list is dropped.
            let guard = TimerShared::new(id);
            pin!(guard);
            let guard_handle = guard.as_ref().get_ref().handle();

            // * This list will be still guarded by the lock of the Wheel with the specefied id.
            //   `EntryWaitersList` wrapper makes sure we hold the lock to modify it.
            // * This wrapper will empty the list on drop. It is critical for safety
            //   that we will not leave any list entry with a pointer to the local
            //   guard node after this function returns / panics.
            // Safety: The `TimerShared` inside this `TimerHandle` is pinned in the memory.
            let mut list = unsafe { lock.get_waiters_list(&expiration, guard_handle, id, self) };

            while let Some(entry) = list.pop_back_locked(&mut lock) {
                let deadline = expiration.deadline;
                // Try to expire the entry; this is cheap (doesn't synchronize) if
                // the timer is not expired, and updates cached_when.
                match unsafe { entry.mark_firing(deadline) } {
                    Ok(()) => {
                        // Entry was expired.
                        // SAFETY: We hold the driver lock, and just removed the entry from any linked lists.
                        if let Some(waker) = unsafe { entry.fire(Ok(())) } {
                            waker_list.push(waker);

                            if !waker_list.can_push() {
                                // Wake a batch of wakers. To avoid deadlock,
                                // we must do this with the lock temporarily dropped.
                                drop(lock);
                                waker_list.wake_all();

                                lock = self.inner.lock_sharded_wheel(id);
                            }
                        }
                    }
                    Err(state) if state == STATE_DEREGISTERED => {}
                    Err(state) => {
                        // Safety: This Entry has not expired.
                        unsafe { lock.reinsert_entry(entry, deadline, state) };
                    }
                }