Use Poll instead of ControlFlow.

Berrysoft · Berrysoft · commit efb2fcae3fd6 · 2023-09-22T21:37:34.000+08:00
diff --git a/src/driver/poll/mod.rs b/src/driver/poll/mod.rs
@@ -4,9 +4,9 @@ use std::{
     collections::{HashMap, HashSet, VecDeque},
     io,
     num::NonZeroUsize,
-    ops::ControlFlow,
     os::fd::BorrowedFd,
     pin::Pin,
+    task::Poll,
     time::Duration,
 };
 
@@ -27,7 +27,7 @@ pub trait OpCode {
 
     /// Perform the operation after received corresponding
     /// event.
-    fn on_event(self: Pin<&mut Self>, event: &Event) -> io::Result<ControlFlow<usize>>;
+    fn on_event(self: Pin<&mut Self>, event: &Event) -> Poll<io::Result<usize>>;
 }
 
 /// Result of [`OpCode::pre_submit`].
@@ -218,13 +218,12 @@ impl Driver {
             } else {
                 let op = registry[user_data].as_pin();
                 let res = match op.on_event(&event) {
-                    Ok(ControlFlow::Continue(_)) => {
+                    Poll::Pending => {
                         // The operation should go back to the front.
                         queue.push_front_interest(user_data, interest);
                         None
                     }
-                    Ok(ControlFlow::Break(res)) => Some(Ok(res)),
-                    Err(err) => Some(Err(err)),
+                    Poll::Ready(res) => Some(res),
                 };
                 if let Some(res) = res {
                     let entry = Entry::new(user_data, res);
diff --git a/src/driver/poll/op.rs b/src/driver/poll/op.rs
@@ -1,4 +1,4 @@
-use std::{io, ops::ControlFlow, pin::Pin};
+use std::{io, pin::Pin, task::Poll};
 
 use polling::Event;
 
@@ -30,7 +30,7 @@ impl<T: IoBufMut> OpCode for ReadAt<T> {
         }
     }
 
-    fn on_event(mut self: Pin<&mut Self>, event: &Event) -> std::io::Result<ControlFlow<usize>> {
+    fn on_event(mut self: Pin<&mut Self>, event: &Event) -> Poll<io::Result<usize>> {
         debug_assert!(event.readable);
 
         let fd = self.fd;
@@ -66,7 +66,7 @@ impl<T: IoBuf> OpCode for WriteAt<T> {
         }
     }
 
-    fn on_event(self: Pin<&mut Self>, event: &Event) -> std::io::Result<ControlFlow<usize>> {
+    fn on_event(self: Pin<&mut Self>, event: &Event) -> Poll<io::Result<usize>> {
         debug_assert!(event.writable);
 
         let slice = self.buffer.as_slice();
@@ -87,7 +87,7 @@ impl OpCode for Sync {
         Ok(Decision::Completed(syscall!(fsync(self.fd))? as _))
     }
 
-    fn on_event(self: Pin<&mut Self>, _: &Event) -> std::io::Result<ControlFlow<usize>> {
+    fn on_event(self: Pin<&mut Self>, _: &Event) -> Poll<io::Result<usize>> {
         unreachable!("Sync operation should not be submitted to polling")
     }
 }
@@ -103,18 +103,16 @@ impl OpCode for Accept {
         )
     }
 
-    fn on_event(mut self: Pin<&mut Self>, event: &Event) -> std::io::Result<ControlFlow<usize>> {
+    fn on_event(mut self: Pin<&mut Self>, event: &Event) -> Poll<io::Result<usize>> {
         debug_assert!(event.readable);
 
-        match syscall!(accept(
-            self.fd,
-            &mut self.buffer as *mut _ as *mut _,
-            &mut self.addr_len
-        )) {
-            Ok(fd) => Ok(ControlFlow::Break(fd as _)),
-            Err(e) if e.raw_os_error() == Some(libc::EINPROGRESS) => Ok(ControlFlow::Continue(())),
-            Err(e) => Err(e),
-        }
+        syscall!(
+            break accept(
+                self.fd,
+                &mut self.buffer as *mut _ as *mut _,
+                &mut self.addr_len
+            )
+        )
     }
 }
 
@@ -125,7 +123,7 @@ impl OpCode for Connect {
         )
     }
 
-    fn on_event(self: Pin<&mut Self>, event: &Event) -> std::io::Result<ControlFlow<usize>> {
+    fn on_event(self: Pin<&mut Self>, event: &Event) -> Poll<io::Result<usize>> {
         debug_assert!(event.writable);
 
         let mut err: libc::c_int = 0;
@@ -139,11 +137,12 @@ impl OpCode for Connect {
             &mut err_len
         ))?;
 
-        if err == 0 {
-            Ok(ControlFlow::Break(0))
+        let res = if err == 0 {
+            Ok(0)
         } else {
             Err(io::Error::from_raw_os_error(err))
-        }
+        };
+        Poll::Ready(res)
     }
 }
 
@@ -152,7 +151,7 @@ impl<T: AsIoSlicesMut + Unpin> OpCode for RecvImpl<T> {
         Ok(Decision::wait_readable(self.fd))
     }
 
-    fn on_event(mut self: Pin<&mut Self>, event: &Event) -> std::io::Result<ControlFlow<usize>> {
+    fn on_event(mut self: Pin<&mut Self>, event: &Event) -> Poll<io::Result<usize>> {
         debug_assert!(event.readable);
 
         self.slices = unsafe { self.buffer.as_io_slices_mut() };
@@ -165,7 +164,7 @@ impl<T: AsIoSlices + Unpin> OpCode for SendImpl<T> {
         Ok(Decision::wait_writable(self.fd))
     }
 
-    fn on_event(mut self: Pin<&mut Self>, event: &Event) -> std::io::Result<ControlFlow<usize>> {
+    fn on_event(mut self: Pin<&mut Self>, event: &Event) -> Poll<io::Result<usize>> {
         debug_assert!(event.writable);
 
         self.slices = unsafe { self.buffer.as_io_slices() };
@@ -179,7 +178,7 @@ impl<T: AsIoSlicesMut + Unpin> OpCode for RecvFromImpl<T> {
         syscall!(recvmsg(self.fd, &mut self.msg, 0) or wait_readable(self.fd))
     }
 
-    fn on_event(mut self: Pin<&mut Self>, event: &Event) -> std::io::Result<ControlFlow<usize>> {
+    fn on_event(mut self: Pin<&mut Self>, event: &Event) -> Poll<io::Result<usize>> {
         debug_assert!(event.readable);
 
         syscall!(break recvmsg(self.fd, &mut self.msg, 0))
@@ -192,7 +191,7 @@ impl<T: AsIoSlices + Unpin> OpCode for SendToImpl<T> {
         syscall!(sendmsg(self.fd, &self.msg, 0) or wait_writable(self.fd))
     }
 
-    fn on_event(self: Pin<&mut Self>, event: &Event) -> std::io::Result<ControlFlow<usize>> {
+    fn on_event(self: Pin<&mut Self>, event: &Event) -> Poll<io::Result<usize>> {
         debug_assert!(event.writable);
 
         syscall!(break sendmsg(self.fd, &self.msg, 0))
diff --git a/src/lib.rs b/src/lib.rs
@@ -125,16 +125,18 @@ macro_rules! syscall {
     }};
     // The below branches are used by polling driver.
     (break $fn: ident ( $($arg: expr),* $(,)* )) => {
-        $crate::syscall!( $fn ( $($arg, )* )).map(
-            |res| ::std::ops::ControlFlow::Break(res as usize)
-        )
-    };
-    ($fn: ident ( $($arg: expr),* $(,)* ) or $f:ident($fd:expr)) => {
         match $crate::syscall!( $fn ( $($arg, )* )) {
-            Ok(fd) => Ok($crate::driver::Decision::Completed(fd as usize)),
+            Ok(fd) => ::std::task::Poll::Ready(Ok(fd as usize)),
             Err(e) if e.kind() == ::std::io::ErrorKind::WouldBlock || e.raw_os_error() == Some(::libc::EINPROGRESS)
-                   => Ok($crate::driver::Decision::$f($fd)),
-            Err(e) => Err(e),
+                   => ::std::task::Poll::Pending,
+            Err(e) => ::std::task::Poll::Ready(Err(e)),
+        }
+    };
+    ($fn: ident ( $($arg: expr),* $(,)* ) or $f:ident($fd:expr)) => {
+        match $crate::syscall!( break $fn ( $($arg, )* )) {
+            ::std::task::Poll::Pending => Ok($crate::driver::Decision::$f($fd)),
+            ::std::task::Poll::Ready(Ok(res)) => Ok($crate::driver::Decision::Completed(res)),
+            ::std::task::Poll::Ready(Err(e)) => Err(e),
         }
     };
 }

Original file line number	Diff line number	Diff line change
`@@ -1,4 +1,4 @@`
`1`		`-use std::{io, ops::ControlFlow, pin::Pin};`
	`1`	`+use std::{io, pin::Pin, task::Poll};`
`2`	`2`
`3`	`3`	`use polling::Event;`
`4`	`4`
`@@ -30,7 +30,7 @@ impl<T: IoBufMut> OpCode for ReadAt<T> {`
`30`	`30`	`}`
`31`	`31`	`}`
`32`	`32`
`33`		`- fn on_event(mut self: Pin<&mut Self>, event: &Event) -> std::io::Result<ControlFlow<usize>> {`
	`33`	`+ fn on_event(mut self: Pin<&mut Self>, event: &Event) -> Poll<io::Result<usize>> {`
`34`	`34`	`debug_assert!(event.readable);`
`35`	`35`
`36`	`36`	`let fd = self.fd;`
`@@ -66,7 +66,7 @@ impl<T: IoBuf> OpCode for WriteAt<T> {`
`66`	`66`	`}`
`67`	`67`	`}`
`68`	`68`
`69`		`- fn on_event(self: Pin<&mut Self>, event: &Event) -> std::io::Result<ControlFlow<usize>> {`
	`69`	`+ fn on_event(self: Pin<&mut Self>, event: &Event) -> Poll<io::Result<usize>> {`
`70`	`70`	`debug_assert!(event.writable);`
`71`	`71`
`72`	`72`	`let slice = self.buffer.as_slice();`
`@@ -87,7 +87,7 @@ impl OpCode for Sync {`
`87`	`87`	`Ok(Decision::Completed(syscall!(fsync(self.fd))? as _))`
`88`	`88`	`}`
`89`	`89`
`90`		`- fn on_event(self: Pin<&mut Self>, _: &Event) -> std::io::Result<ControlFlow<usize>> {`
	`90`	`+ fn on_event(self: Pin<&mut Self>, _: &Event) -> Poll<io::Result<usize>> {`
`91`	`91`	`unreachable!("Sync operation should not be submitted to polling")`
`92`	`92`	`}`
`93`	`93`	`}`
`@@ -103,18 +103,16 @@ impl OpCode for Accept {`
`103`	`103`	`)`
`104`	`104`	`}`
`105`	`105`
`106`		`- fn on_event(mut self: Pin<&mut Self>, event: &Event) -> std::io::Result<ControlFlow<usize>> {`
	`106`	`+ fn on_event(mut self: Pin<&mut Self>, event: &Event) -> Poll<io::Result<usize>> {`
`107`	`107`	`debug_assert!(event.readable);`
`108`	`108`
`109`		`- match syscall!(accept(`
`110`		`- self.fd,`
`111`		`- &mut self.buffer as mut _ as mut _,`
`112`		`- &mut self.addr_len`
`113`		`- )) {`
`114`		`- Ok(fd) => Ok(ControlFlow::Break(fd as _)),`
`115`		`- Err(e) if e.raw_os_error() == Some(libc::EINPROGRESS) => Ok(ControlFlow::Continue(())),`
`116`		`- Err(e) => Err(e),`
`117`		`- }`
	`109`	`+ syscall!(`
	`110`	`+ break accept(`
	`111`	`+ self.fd,`
	`112`	`+ &mut self.buffer as mut _ as mut _,`
	`113`	`+ &mut self.addr_len`
	`114`	`+ )`
	`115`	`+ )`
`118`	`116`	`}`
`119`	`117`	`}`
`120`	`118`
`@@ -125,7 +123,7 @@ impl OpCode for Connect {`
`125`	`123`	`)`
`126`	`124`	`}`
`127`	`125`
`128`		`- fn on_event(self: Pin<&mut Self>, event: &Event) -> std::io::Result<ControlFlow<usize>> {`
	`126`	`+ fn on_event(self: Pin<&mut Self>, event: &Event) -> Poll<io::Result<usize>> {`
`129`	`127`	`debug_assert!(event.writable);`
`130`	`128`
`131`	`129`	`let mut err: libc::c_int = 0;`
`@@ -139,11 +137,12 @@ impl OpCode for Connect {`
`139`	`137`	`&mut err_len`
`140`	`138`	`))?;`
`141`	`139`
`142`		`- if err == 0 {`
`143`		`- Ok(ControlFlow::Break(0))`
	`140`	`+ let res = if err == 0 {`
	`141`	`+ Ok(0)`
`144`	`142`	`} else {`
`145`	`143`	`Err(io::Error::from_raw_os_error(err))`
`146`		`- }`
	`144`	`+ };`
	`145`	`+ Poll::Ready(res)`
`147`	`146`	`}`
`148`	`147`	`}`
`149`	`148`
`@@ -152,7 +151,7 @@ impl<T: AsIoSlicesMut + Unpin> OpCode for RecvImpl<T> {`
`152`	`151`	`Ok(Decision::wait_readable(self.fd))`
`153`	`152`	`}`
`154`	`153`
`155`		`- fn on_event(mut self: Pin<&mut Self>, event: &Event) -> std::io::Result<ControlFlow<usize>> {`
	`154`	`+ fn on_event(mut self: Pin<&mut Self>, event: &Event) -> Poll<io::Result<usize>> {`
`156`	`155`	`debug_assert!(event.readable);`
`157`	`156`
`158`	`157`	`self.slices = unsafe { self.buffer.as_io_slices_mut() };`
`@@ -165,7 +164,7 @@ impl<T: AsIoSlices + Unpin> OpCode for SendImpl<T> {`
`165`	`164`	`Ok(Decision::wait_writable(self.fd))`
`166`	`165`	`}`
`167`	`166`
`168`		`- fn on_event(mut self: Pin<&mut Self>, event: &Event) -> std::io::Result<ControlFlow<usize>> {`
	`167`	`+ fn on_event(mut self: Pin<&mut Self>, event: &Event) -> Poll<io::Result<usize>> {`
`169`	`168`	`debug_assert!(event.writable);`
`170`	`169`
`171`	`170`	`self.slices = unsafe { self.buffer.as_io_slices() };`
`@@ -179,7 +178,7 @@ impl<T: AsIoSlicesMut + Unpin> OpCode for RecvFromImpl<T> {`
`179`	`178`	`syscall!(recvmsg(self.fd, &mut self.msg, 0) or wait_readable(self.fd))`
`180`	`179`	`}`
`181`	`180`
`182`		`- fn on_event(mut self: Pin<&mut Self>, event: &Event) -> std::io::Result<ControlFlow<usize>> {`
	`181`	`+ fn on_event(mut self: Pin<&mut Self>, event: &Event) -> Poll<io::Result<usize>> {`
`183`	`182`	`debug_assert!(event.readable);`
`184`	`183`
`185`	`184`	`syscall!(break recvmsg(self.fd, &mut self.msg, 0))`
`@@ -192,7 +191,7 @@ impl<T: AsIoSlices + Unpin> OpCode for SendToImpl<T> {`
`192`	`191`	`syscall!(sendmsg(self.fd, &self.msg, 0) or wait_writable(self.fd))`
`193`	`192`	`}`
`194`	`193`
`195`		`- fn on_event(self: Pin<&mut Self>, event: &Event) -> std::io::Result<ControlFlow<usize>> {`
	`194`	`+ fn on_event(self: Pin<&mut Self>, event: &Event) -> Poll<io::Result<usize>> {`
`196`	`195`	`debug_assert!(event.writable);`
`197`	`196`
`198`	`197`	`syscall!(break sendmsg(self.fd, &self.msg, 0))`