util: use FIFO ordering in WakeList (#6521)

tokio/src/util/wake_list.rs

@@ -37,12 +37,37 @@ impl WakeList {
     }
 
     pub(crate) fn wake_all(&mut self) {
-        assert!(self.curr <= NUM_WAKERS);
-        while self.curr > 0 {
-            self.curr -= 1;
-            // SAFETY: The first `curr` elements of `WakeList` are initialized, so by decrementing
-            // `curr`, we can take ownership of the last item.
-            let waker = unsafe { ptr::read(self.inner[self.curr].as_mut_ptr()) };
+        struct DropGuard {
+            start: *mut Waker,
+            end: *mut Waker,
+        }
+
+        impl Drop for DropGuard {
+            fn drop(&mut self) {
+                // SAFETY: Both pointers are part of the same object, with `start <= end`.
+                let len = unsafe { self.end.offset_from(self.start) } as usize;
+                let slice = ptr::slice_from_raw_parts_mut(self.start, len);
+                // SAFETY: All elements in `start..len` are initialized, so we can drop them.
+                unsafe { ptr::drop_in_place(slice) };
+            }
+        }
+
+        debug_assert!(self.curr <= NUM_WAKERS);
+
+        let mut guard = {
+            let start = self.inner.as_mut_ptr().cast::<Waker>();
+            // SAFETY: The resulting pointer is in bounds or one after the length of the same object.
+            let end = unsafe { start.add(self.curr) };
+            // Transfer ownership of the wakers in `inner` to `DropGuard`.
+            self.curr = 0;
+            DropGuard { start, end }
+        };
+        while !ptr::eq(guard.start, guard.end) {
+            // SAFETY: `start` is always initialized if `start != end`.
+            let waker = unsafe { ptr::read(guard.start) };
+            // SAFETY: The resulting pointer is in bounds or one after the length of the same object.
+            guard.start = unsafe { guard.start.add(1) };
+            // If this panics, then `guard` will clean up the remaining wakers.
             waker.wake();
         }
     }