double check Reader waker

zonyitoo · May 6, 2022 · 3e23d00 · 3e23d00
1 parent 2c28526
commit 3e23d00
Show file tree

Hide file tree

Showing 3 changed files with 27 additions and 9 deletions.
diff --git a/src/sys/unix/bsd/freebsd.rs b/src/sys/unix/bsd/freebsd.rs
@@ -158,7 +158,9 @@ impl AsyncRead for TcpStream {
             TcpStreamOptionProj::Connected(stream) => stream.poll_read(cx, buf),
             TcpStreamOptionProj::Connecting { reader, .. } => {
                 if let Some(w) = reader.take() {
-                    w.wake();
+                    if !w.will_wake(cx.waker()) {
+                        w.wake();
+                    }
                 }
                 *reader = Some(cx.waker().clone());
                 Poll::Pending
@@ -233,7 +235,9 @@ impl AsyncWrite for TcpStream {
 
                         // Wake up the Future that pending on poll_read
                         if let Some(w) = reader.take() {
-                            w.wake();
+                            if !w.will_wake(cx.waker()) {
+                                w.wake();
+                            }
                         }
 
                         return Ok(ret as usize).into();
@@ -265,7 +269,9 @@ impl AsyncWrite for TcpStream {
 
                             // Wake up the Future that pending on poll_read
                             if let Some(w) = reader.take() {
-                                w.wake();
+                                if !w.will_wake(cx.waker()) {
+                                    w.wake();
+                                }
                             }
                         } else {
                             // Other errors, including EAGAIN, EWOULDBLOCK

diff --git a/src/sys/unix/linux.rs b/src/sys/unix/linux.rs
@@ -184,7 +184,9 @@ impl AsyncRead for TcpStream {
             TcpStreamOptionProj::Connected(stream) => stream.poll_read(cx, buf),
             TcpStreamOptionProj::Connecting { reader, .. } => {
                 if let Some(w) = reader.take() {
-                    w.wake();
+                    if !w.will_wake(cx.waker()) {
+                        w.wake();
+                    }
                 }
                 *reader = Some(cx.waker().clone());
                 Poll::Pending
@@ -258,7 +260,9 @@ impl AsyncWrite for TcpStream {
 
                         // Wake up the Future that pending on poll_read
                         if let Some(w) = reader.take() {
-                            w.wake();
+                            if !w.will_wake(cx.waker()) {
+                                w.wake();
+                            }
                         }
 
                         return Ok(ret as usize).into();
@@ -290,7 +294,9 @@ impl AsyncWrite for TcpStream {
 
                             // Wake up the Future that pending on poll_read
                             if let Some(w) = reader.take() {
-                                w.wake();
+                                if !w.will_wake(cx.waker()) {
+                                    w.wake();
+                                }
                             }
                         } else {
                             // Other errors, including EAGAIN, EWOULDBLOCK

diff --git a/src/sys/windows/mod.rs b/src/sys/windows/mod.rs
@@ -263,7 +263,9 @@ impl AsyncRead for TcpStream {
             TcpStreamOptionProj::Connected(stream) => stream.poll_read(cx, buf),
             TcpStreamOptionProj::Connecting { reader, .. } => {
                 if let Some(w) = reader.take() {
-                    w.wake();
+                    if !w.will_wake(cx.waker()) {
+                        w.wake();
+                    }
                 }
                 *reader = Some(cx.waker().clone());
                 Poll::Pending
@@ -355,7 +357,9 @@ impl AsyncWrite for TcpStream {
 
                             // Wake up the Future that pending on poll_read
                             if let Some(w) = reader.take() {
-                                w.wake();
+                                if !w.will_wake(cx.waker()) {
+                                    w.wake();
+                                }
                             }
 
                             return Ok(bytes_sent as usize).into();
@@ -382,7 +386,9 @@ impl AsyncWrite for TcpStream {
 
                         // Wake up the Future that pending on poll_read
                         if let Some(w) = reader.take() {
-                            w.wake();
+                            if !w.will_wake(cx.waker()) {
+                                w.wake();
+                            }
                         }
                     }
                 }