fix(net): no spawn_blocking in Incoming (#872)

* fix(net): no spawn_blocking in Incoming

* fix(net): remove IncomingState

Author: Berrysoft

compio-net/src/incoming/windows.rs

@@ -6,29 +6,22 @@ use std::{
 
 use compio_buf::BufResult;
 use compio_driver::{SharedFd, ToSharedFd, op::Accept};
-use compio_runtime::{JoinHandle, Submit};
+use compio_runtime::Submit;
 use futures_util::{FutureExt, Stream, stream::FusedStream};
 use socket2::Socket as Socket2;
 
 use crate::Socket;
 
-#[allow(clippy::large_enum_variant)]
-enum IncomingState {
-    Idle,
-    CreatingSocket(JoinHandle<io::Result<Socket2>>),
-    Accepting(Submit<Accept<SharedFd<Socket2>>>),
-}
-
 pub struct Incoming<'a> {
     listener: &'a Socket,
-    state: IncomingState,
+    state: Option<Submit<Accept<SharedFd<Socket2>>>>,
 }
 
 impl<'a> Incoming<'a> {
     pub fn new(listener: &'a Socket) -> Self {
         Self {
             listener,
-            state: IncomingState::Idle,
+            state: None,
         }
     }
 }
@@ -40,44 +33,26 @@ impl Stream for Incoming<'_> {
         let this = self.get_mut();
         loop {
             match &mut this.state {
-                IncomingState::Idle => {
+                None => {
                     let domain = this.listener.local_addr().map(|addr| addr.domain())?;
                     let ty = this.listener.socket.r#type()?;
                     let protocol = this.listener.socket.protocol()?;
-                    let handle =
-                        compio_runtime::spawn_blocking(move || Socket2::new(domain, ty, protocol));
-                    this.state = IncomingState::CreatingSocket(handle);
+                    let socket = Socket2::new(domain, ty, protocol)?;
+                    let op =
+                        compio_runtime::submit(Accept::new(this.listener.to_shared_fd(), socket));
+                    this.state = Some(op);
                 }
-                IncomingState::CreatingSocket(handle) => match ready!(handle.poll_unpin(cx)) {
-                    Ok(Ok(socket)) => {
-                        let op = compio_runtime::submit(Accept::new(
-                            this.listener.to_shared_fd(),
-                            socket,
-                        ));
-                        this.state = IncomingState::Accepting(op);
-                    }
-                    Ok(Err(e)) => {
-                        this.state = IncomingState::Idle;
-                        return Poll::Ready(Some(Err(e)));
-                    }
-                    Err(e) => {
-                        this.state = IncomingState::Idle;
-                        e.resume_unwind();
-                        // The background task was cancelled; terminate the stream.
-                        return Poll::Ready(None);
-                    }
-                },
-                IncomingState::Accepting(op) => {
+                Some(op) => {
                     let BufResult(res, op) = ready!(op.poll_unpin(cx));
                     match res {
                         Ok(_) => {
-                            this.state = IncomingState::Idle;
+                            this.state = None;
                             op.update_context()?;
                             let (accept_sock, _) = op.into_addr()?;
                             return Poll::Ready(Some(Ok(Socket::from_socket2(accept_sock)?)));
                         }
                         Err(e) => {
-                            this.state = IncomingState::Idle;
+                            this.state = None;
                             return Poll::Ready(Some(Err(e)));
                         }
                     }