io: make duplex stream cooperative (tokio-rs#4470)

Add coop checks on pipe poll_read and poll_write.

Fixes: tokio-rs#4470
Refs: tokio-rs#4291, tokio-rs#4300

tokio/src/io/util/mem.rs

@@ -183,7 +183,6 @@ impl Pipe {
  cx: &mut task::Context<'_>,
  buf: &mut ReadBuf<'_>,
  ) -> Poll<std::io::Result<()>> {
- ready!(poll_proceed_and_make_progress(cx));
  if self.buffer.has_remaining() {
  let max = self.buffer.remaining().min(buf.remaining());
  buf.put_slice(&self.buffer[..max]);
@@ -209,7 +208,6 @@ impl Pipe {
  cx: &mut task::Context<'_>,
  buf: &[u8],
  ) -> Poll<std::io::Result<usize>> {
- ready!(poll_proceed_and_make_progress(cx));
  if self.is_closed {
  return Poll::Ready(Err(std::io::ErrorKind::BrokenPipe.into()));
  }
@@ -264,6 +262,7 @@ impl AsyncWrite for Pipe {
  buf: &[u8],
  ) -> Poll<std::io::Result<usize>> {
  let coop = ready!(crate::coop::poll_proceed(cx));
+
  let ret = self.poll_write_internal(cx, buf);
  if ret.is_ready() {
  coop.made_progress();
@@ -294,17 +293,3 @@ impl AsyncWrite for Pipe {
  Poll::Ready(Ok(()))
  }
 }
-
-cfg_coop! {
- fn poll_proceed_and_make_progress(cx: &mut task::Context<'_>) -> Poll<()> {
- let coop = ready!(crate::coop::poll_proceed(cx));
- coop.made_progress();
- Poll::Ready(())
- }
-}
-
-cfg_not_coop! {
- fn poll_proceed_and_make_progress(_: &mut Context<'_>) -> Poll<()> {
- Poll::Ready(())
- }
-}