refactor: merge some methods and remove OpFuture struct

merge pop and poll_task

Author: Sherlock-Holo

compio-driver/src/lib.rs

@@ -273,7 +273,8 @@ impl Proactor {
     }
 
     /// Poll the driver and get completed entries.
-    /// You need to call [`Proactor::pop`] to get the pushed operations.
+    /// You need to call [`Proactor::pop`] to get the pushed
+    /// operations.
     pub fn poll(
         &mut self,
         timeout: Option<Duration>,
@@ -290,25 +291,7 @@ impl Proactor {
     /// # Panics
     /// This function will panic if the requested operation has not been
     /// completed.
-    pub fn pop<T>(&mut self, op: Key<T>) -> PushEntry<Key<T>, BufResult<usize, T>> {
-        instrument!(compio_log::Level::DEBUG, "pop", ?op);
-        if op.has_result() {
-            // SAFETY: completed.
-            PushEntry::Ready(unsafe { op.into_inner() })
-        } else {
-            PushEntry::Pending(op)
-        }
-    }
-
-    /// Get the pushed operations from the completion entries.
-    ///
-    /// # Panics
-    /// This function will panic if the requested operation has not been
-    /// completed.
-    pub fn pop_with_flags<T>(
-        &mut self,
-        op: Key<T>,
-    ) -> PushEntry<Key<T>, (BufResult<usize, T>, u32)> {
+    pub fn pop<T>(&mut self, op: Key<T>) -> PushEntry<Key<T>, (BufResult<usize, T>, u32)> {
         instrument!(compio_log::Level::DEBUG, "pop_flags", ?op);
         if op.has_result() {
             let flags = op.flags();

compio-driver/tests/file.rs

@@ -53,7 +53,11 @@ fn push_and_wait<O: OpCode + 'static>(driver: &mut Proactor, op: O) -> BufResult
                 driver.poll(None, &mut entries).unwrap();
             }
             assert_eq!(entries[0], user_data.user_data());
-            driver.pop(user_data).take_ready().unwrap()
+            driver
+                .pop(user_data)
+                .map_ready(|(res, _)| res)
+                .take_ready()
+                .unwrap()
         }
     }
 }

compio-runtime/src/runtime/mod.rs

@@ -31,10 +31,7 @@ use send_wrapper::SendWrapper;
 
 #[cfg(feature = "time")]
 use crate::runtime::time::{TimerFuture, TimerRuntime};
-use crate::{
-    runtime::op::{OpFlagsFuture, OpFuture},
-    BufResult,
-};
+use crate::{runtime::op::OpFlagsFuture, BufResult};
 
 scoped_tls::scoped_thread_local!(static CURRENT_RUNTIME: Runtime);
 
@@ -238,10 +235,7 @@ impl Runtime {
     ///
     /// You only need this when authoring your own [`OpCode`].
     pub fn submit<T: OpCode + 'static>(&self, op: T) -> impl Future<Output = BufResult<usize, T>> {
-        match self.submit_raw(op) {
-            PushEntry::Pending(user_data) => Either::Left(OpFuture::new(user_data)),
-            PushEntry::Ready(res) => Either::Right(ready(res)),
-        }
+        self.submit_with_flags(op).map(|(res, _)| res)
     }
 
     /// Submit an operation to the runtime.
@@ -287,7 +281,7 @@ impl Runtime {
         &self,
         cx: &mut Context,
         op: Key<T>,
-    ) -> PushEntry<Key<T>, BufResult<usize, T>> {
+    ) -> PushEntry<Key<T>, (BufResult<usize, T>, u32)> {
         instrument!(compio_log::Level::DEBUG, "poll_task", ?op);
         let mut driver = self.driver.borrow_mut();
         driver.pop(op).map_pending(|mut k| {
@@ -296,19 +290,6 @@ impl Runtime {
         })
     }
 
-    pub(crate) fn poll_task_with_flags<T: OpCode>(
-        &self,
-        cx: &mut Context,
-        op: Key<T>,
-    ) -> PushEntry<Key<T>, (BufResult<usize, T>, u32)> {
-        instrument!(compio_log::Level::DEBUG, "poll_task_flags", ?op);
-        let mut driver = self.driver.borrow_mut();
-        driver.pop_with_flags(op).map_pending(|mut k| {
-            driver.update_waker(&mut k, cx.waker().clone());
-            k
-        })
-    }
-
     #[cfg(feature = "time")]
     pub(crate) fn poll_timer(&self, cx: &mut Context, key: usize) -> Poll<()> {
         instrument!(compio_log::Level::DEBUG, "poll_timer", ?cx, ?key);

compio-runtime/src/runtime/op.rs

@@ -9,40 +9,6 @@ use compio_driver::{Key, OpCode, PushEntry};
 
 use crate::runtime::Runtime;
 
-#[derive(Debug)]
-pub struct OpFuture<T: OpCode> {
-    key: Option<Key<T>>,
-}
-
-impl<T: OpCode> OpFuture<T> {
-    pub fn new(key: Key<T>) -> Self {
-        Self { key: Some(key) }
-    }
-}
-
-impl<T: OpCode> Future for OpFuture<T> {
-    type Output = BufResult<usize, T>;
-
-    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
-        let res = Runtime::with_current(|r| r.poll_task(cx, self.key.take().unwrap()));
-        match res {
-            PushEntry::Pending(key) => {
-                self.key = Some(key);
-                Poll::Pending
-            }
-            PushEntry::Ready(res) => Poll::Ready(res),
-        }
-    }
-}
-
-impl<T: OpCode> Drop for OpFuture<T> {
-    fn drop(&mut self) {
-        if let Some(key) = self.key.take() {
-            Runtime::with_current(|r| r.cancel_op(key))
-        }
-    }
-}
-
 #[derive(Debug)]
 pub struct OpFlagsFuture<T: OpCode> {
     key: Option<Key<T>>,
@@ -58,7 +24,7 @@ impl<T: OpCode> Future for OpFlagsFuture<T> {
     type Output = (BufResult<usize, T>, u32);
 
     fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
-        let res = Runtime::with_current(|r| r.poll_task_with_flags(cx, self.key.take().unwrap()));
+        let res = Runtime::with_current(|r| r.poll_task(cx, self.key.take().unwrap()));
         match res {
             PushEntry::Pending(key) => {
                 self.key = Some(key);

compio/examples/driver.rs

@@ -54,7 +54,12 @@ fn push_and_wait<O: OpCode + 'static>(driver: &mut Proactor, op: O) -> (usize, O
                 driver.poll(None, &mut entries).unwrap();
             }
             assert_eq!(entries[0], user_data.user_data());
-            driver.pop(user_data).take_ready().unwrap().unwrap()
+            driver
+                .pop(user_data)
+                .map_ready(|(res, _)| res)
+                .take_ready()
+                .unwrap()
+                .unwrap()
         }
     }
 }