remove localwaker from h1::payload (#2051)

actix-http/src/h1/payload.rs

@@ -3,9 +3,8 @@ use std::cell::RefCell;
 use std::collections::VecDeque;
 use std::pin::Pin;
 use std::rc::{Rc, Weak};
-use std::task::{Context, Poll};
+use std::task::{Context, Poll, Waker};
 
-use actix_utils::task::LocalWaker;
 use bytes::Bytes;
 use futures_core::Stream;
 
@@ -134,7 +133,7 @@ impl PayloadSender {
             if shared.borrow().need_read {
                 PayloadStatus::Read
             } else {
-                shared.borrow_mut().io_task.register(cx.waker());
+                shared.borrow_mut().register_io(cx);
                 PayloadStatus::Pause
             }
         } else {
@@ -150,8 +149,8 @@ struct Inner {
     err: Option<PayloadError>,
     need_read: bool,
     items: VecDeque<Bytes>,
-    task: LocalWaker,
+    task: Option<Waker>,
-    io_task: LocalWaker,
+    io_task: Option<Waker>,
 }
 
 impl Inner {
@@ -162,8 +161,48 @@ impl Inner {
             err: None,
             items: VecDeque::new(),
             need_read: true,
-            task: LocalWaker::new(),
+            task: None,
-            io_task: LocalWaker::new(),
+            io_task: None,
+        }
+    }
+
+    /// Wake up future waiting for payload data to be available.
+    fn wake(&mut self) {
+        if let Some(waker) = self.task.take() {
+            waker.wake();
+        }
+    }
+
+    /// Wake up future feeding data to Payload.
+    fn wake_io(&mut self) {
+        if let Some(waker) = self.io_task.take() {
+            waker.wake();
+        }
+    }
+
+    /// Register future waiting data from payload.
+    /// Waker would be used in `Inner::wake`
+    fn register(&mut self, cx: &mut Context<'_>) {
+        if self
+            .task
+            .as_ref()
+            .map(|w| !cx.waker().will_wake(w))
+            .unwrap_or(true)
+        {
+            self.task = Some(cx.waker().clone());
+        }
+    }
+
+    // Register future feeding data to payload.
+    /// Waker would be used in `Inner::wake_io`
+    fn register_io(&mut self, cx: &mut Context<'_>) {
+        if self
+            .io_task
+            .as_ref()
+            .map(|w| !cx.waker().will_wake(w))
+            .unwrap_or(true)
+        {
+            self.io_task = Some(cx.waker().clone());
         }
     }
 
@@ -182,7 +221,7 @@ impl Inner {
         self.len += data.len();
         self.items.push_back(data);
         self.need_read = self.len < MAX_BUFFER_SIZE;
-        self.task.wake();
+        self.wake();
     }
 
     #[cfg(test)]
@@ -199,9 +238,9 @@ impl Inner {
             self.need_read = self.len < MAX_BUFFER_SIZE;
 
             if self.need_read && !self.eof {
-                self.task.register(cx.waker());
+                self.register(cx);
             }
-            self.io_task.wake();
+            self.wake_io();
             Poll::Ready(Some(Ok(data)))
         } else if let Some(err) = self.err.take() {
             Poll::Ready(Some(Err(err)))
@@ -209,8 +248,8 @@ impl Inner {
             Poll::Ready(None)
         } else {
             self.need_read = true;
-            self.task.register(cx.waker());
+            self.register(cx);
-            self.io_task.wake();
+            self.wake_io();
             Poll::Pending
         }
     }