fix(io): avoid livelock between flush and shutdown (#912)

* Try fix deadlock

* Avoid allocation

---------

Co-authored-by: Dude so hot <djohn@fbi.gov>

Author: revonateB0T

compio-io/src/compat/async_stream.rs

@@ -141,6 +141,7 @@ pin_project! {
         write_waker: Option<Waker>,
         flush_waker: Option<Waker>,
         close_waker: Option<Waker>,
+        closed: bool,
         #[pin]
         _p: PhantomPinned,
     }
@@ -169,6 +170,7 @@ impl<S> AsyncWriteStream<S> {
             write_waker: None,
             flush_waker: None,
             close_waker: None,
+            closed: false,
             _p: PhantomPinned,
         }
     }
@@ -402,6 +404,9 @@ impl<S: AsyncWrite + Unpin + 'static> AsyncWriteStream<S> {
     }
 
     fn poll_close_impl(self: Pin<&mut Self>) -> Poll<io::Result<()>> {
+        if self.closed {
+            return Poll::Ready(Ok(()));
+        }
         let this = self.project();
         // SAFETY:
         // - The future won't live longer than the stream.
@@ -415,7 +420,7 @@ impl<S: AsyncWrite + Unpin + 'static> AsyncWriteStream<S> {
         arr.with(|waker| {
             let cx = &mut Context::from_waker(waker);
             let res = poll_future!(this.shutdown_future, cx, inner.get_mut().shutdown());
-            Poll::Ready(res)
+            Poll::Ready(res.inspect(|_| *this.closed = true))
         })
     }
 }