Fix 100% CPU utilization when write closed

ktls/src/ktls_stream.rs

@@ -269,7 +269,7 @@ where
         buf: &[u8],
     ) -> task::Poll<io::Result<usize>> {
         if self.write_closed {
-            return task::Poll::Ready(Ok(0));
+            return task::Poll::Ready(Err(io::ErrorKind::BrokenPipe.into()));
         }
 
         self.project().inner.poll_write(cx, buf)

tests/integration_test.rs → ktls/tests/integration_test.rs

@@ -193,13 +193,18 @@ async fn server_test_inner(cipher_suite: KtlsCipherSuite, flavor: ServerTestFlav
                         stream.read_exact(&mut buf[..1]).await.is_err(),
                         "Session still open?"
                     );
+
+                    debug!("Server trying to write after the peer closed");
+                    let err = stream.write(&PAYLOADS.server).await.unwrap_err();
+                    assert_eq!(err.kind(), io::ErrorKind::BrokenPipe);
                 }
                 ServerTestFlavor::ServerCloses => {
                     debug!("Server sending close notify (5/5)");
                     stream.shutdown().await.unwrap();
 
                     debug!("Server trying to write after closing");
-                    stream.write_all(&PAYLOADS.server).await.unwrap_err();
+                    let err = stream.write(&PAYLOADS.server).await.unwrap_err();
+                    assert_eq!(err.kind(), io::ErrorKind::BrokenPipe);
                 }
             }
 
@@ -425,6 +430,10 @@ async fn client_test_inner(cipher_suite: KtlsCipherSuite, flavor: ClientTestFlav
     );
     assert!(stream.read_exact(buf).await.is_err(), "Session still open?");
 
+    debug!("Client trying to write after the peer closed");
+    let err = stream.write(&PAYLOADS.client).await.unwrap_err();
+    assert_eq!(err.kind(), io::ErrorKind::BrokenPipe);
+
     jh.await.unwrap();
 }