Socket not connected (os error 107)

mio = { version = "=1.2.4", features = ["os-poll", "net", "os-ext"] }
libc = "=0.2.191"

use std::io::{self, Read, Write};
use std::net::SocketAddr;
use std::time::Duration;

use mio::net::TcpStream;
use mio::{Events, Interest, Poll, Token};

fn kind<T: std::fmt::Debug>(r: io::Result<T>) -> String {
  match r {
    Ok(v) => format!("Ok({v:?})"),
    Err(e) => format!("Err({e})"),
  }
}

fn writable(poll: &mut Poll, events: &mut Events) -> io::Result<usize> {
  poll.poll(events, Some(Duration::from_millis(200)))?;
  Ok(events.iter().filter(|e| e.is_writable()).count())
}

fn targets() -> io::Result<(std::net::TcpListener, SocketAddr)> {
  let listening = std::net::TcpListener::bind("127.0.0.1:0")?;
  // bind then drop: nobody listens on this port any more
  let closed = std::net::TcpListener::bind("127.0.0.1:0")?.local_addr()?;
  Ok((listening, closed))
}

fn main() -> io::Result<()> {
  println!("os={}", std::env::consts::OS);
  let (listening, closed) = targets()?;
  let cases = [("listening", listening.local_addr()?), ("closed", closed)];

  println!("-- A. write right after connect, no event yet");
  for (name, addr) in cases {
    let mut s = TcpStream::connect(addr)?;
    println!("{name:<9} write = {}", kind(s.write(b"hi")));
  }

  println!("-- B. writable event, then write without take_error");
  for (name, addr) in cases {
    let mut poll = Poll::new()?;
    let mut events = Events::with_capacity(8);
    let mut s = TcpStream::connect(addr)?;
    poll.registry().register(&mut s, Token(0), Interest::WRITABLE)?;
    println!("{name:<9} writable events = {}", writable(&mut poll, &mut events)?);
    println!("{name:<9} write #1 = {}", kind(s.write(b"hi")));
    println!("{name:<9} write #2 = {}", kind(s.write(b"hi")));
    let mut buf = [0u8; 8];
    println!("{name:<9} read     = {}", kind(s.read(&mut buf)));
    println!("{name:<9} take_error = {}", kind(s.take_error()));
  }

  println!("-- C. step 5 before step 4: peer_addr first, NotConnected means wait again");
  for (name, addr) in cases {
    let mut poll = Poll::new()?;
    let mut events = Events::with_capacity(8);
    let mut s = TcpStream::connect(addr)?;
    poll.registry().register(&mut s, Token(0), Interest::WRITABLE)?;
    for round in 1..=3 {
      let n = writable(&mut poll, &mut events)?;
      let p = s.peer_addr();
      let not_yet = matches!(&p, Err(e) if e.kind() == io::ErrorKind::NotConnected);
      println!("{name:<9} round {round}: writable events = {n}, peer_addr = {}", kind(p));
      if !not_yet {
        break;
      }
    }
    println!("{name:<9} take_error = {}", kind(s.take_error()));
  }

  println!("-- D. the documented order: writable event, take_error, then peer_addr");
  for (name, addr) in cases {
    let mut poll = Poll::new()?;
    let mut events = Events::with_capacity(8);
    let mut s = TcpStream::connect(addr)?;
    poll.registry().register(&mut s, Token(0), Interest::WRITABLE)?;
    println!("{name:<9} {}", kind(wait_connected(&mut poll, &mut events, &s)));
  }
  Ok(())
}

// steps 3 to 6 of the TcpStream::connect docs
fn wait_connected(poll: &mut Poll, events: &mut Events, s: &TcpStream) -> io::Result<SocketAddr> {
  loop {
    poll.poll(events, None)?;
    if !events.iter().any(|e| e.token() == Token(0) && e.is_writable()) {
      continue;
    }
    if let Some(e) = s.take_error()? {
      return Err(e); // step 4: the connect failed
    }
    match s.peer_addr() {
      Ok(addr) => return Ok(addr), // step 6: connected
      Err(e) if e.kind() == io::ErrorKind::NotConnected => continue, // step 5: not yet
      Err(e) if e.raw_os_error() == Some(libc::EINPROGRESS) => continue,
      Err(e) => return Err(e),
    }
  }
}

os=macos
-- A. write right after connect, no event yet
listening write = Err(Socket is not connected (os error 57))
closed    write = Err(Socket is not connected (os error 57))
-- B. writable event, then write without take_error
listening writable events = 1
listening write #1 = Ok(2)
listening write #2 = Ok(2)
listening read     = Err(Resource temporarily unavailable (os error 35))
listening take_error = Ok(None)
closed    writable events = 1
closed    write #1 = Err(Broken pipe (os error 32))
closed    write #2 = Err(Broken pipe (os error 32))
closed    read     = Err(Connection refused (os error 61))
closed    take_error = Ok(None)
-- C. step 5 before step 4: peer_addr first, NotConnected means wait again
listening round 1: writable events = 1, peer_addr = Ok(127.0.0.1:56389)
listening take_error = Ok(None)
closed    round 1: writable events = 1, peer_addr = Err(Invalid argument (os error 22))
closed    take_error = Ok(Some(Os { code: 61, kind: ConnectionRefused, message: "Connection refused" }))
-- D. the documented order: writable event, take_error, then peer_addr
listening Ok(127.0.0.1:56389)
closed    Err(Connection refused (os error 61))

os=linux
-- A. write right after connect, no event yet
listening write = Ok(2)
closed    write = Err(Connection refused (os error 111))
-- B. writable event, then write without take_error
listening writable events = 1
listening write #1 = Ok(2)
listening write #2 = Ok(2)
listening read     = Err(Resource temporarily unavailable (os error 11))
listening take_error = Ok(None)
closed    writable events = 1
closed    write #1 = Err(Connection refused (os error 111))
closed    write #2 = Err(Broken pipe (os error 32))
closed    read     = Ok(0)
closed    take_error = Ok(None)
-- C. step 5 before step 4: peer_addr first, NotConnected means wait again
listening round 1: writable events = 1, peer_addr = Ok(127.0.0.1:43173)
listening take_error = Ok(None)
closed    round 1: writable events = 1, peer_addr = Err(Socket not connected (os error 107))
closed    round 2: writable events = 0, peer_addr = Err(Socket not connected (os error 107))
closed    round 3: writable events = 0, peer_addr = Err(Socket not connected (os error 107))
closed    take_error = Ok(Some(Os { code: 111, kind: ConnectionRefused, message: "Connection refused" }))
-- D. the documented order: writable event, take_error, then peer_addr
listening Ok(127.0.0.1:43173)
closed    Err(Connection refused (os error 111))

    match syscall!(connect(
        socket.as_raw_fd(),
        raw_addr.as_ptr(),
        raw_addr_length
    )) {
        Err(err) if err.raw_os_error() != Some(libc::EINPROGRESS) => Err(err),
        _ => Ok(()),
    }

            poll_fn(|cx| stream.io.registration().poll_write_ready(cx)).await?;

            if let Some(e) = stream.io.take_error()? {
                return Err(e);
            }

            Ok(stream)