executor: add ffi for spawning (#48283)

Author: jkarneges

src/core/executor.cpp

@@ -17,33 +17,41 @@
 #include "executor.h"
 
 Executor::Executor(int tasksMax) :
-    inner_(ffi::executor_create(tasksMax))
+	inner_(ffi::executor_create(tasksMax))
 {
 }
 
 Executor::~Executor()
 {
-    ffi::executor_destroy(inner_);
+	ffi::executor_destroy(inner_);
 }
 
 int Executor::park_cb(void *ctx, int ms)
 {
-    std::function<bool (std::optional<int>)> *park = (std::function<bool (std::optional<int>)> *)ctx;
+	std::function<bool (std::optional<int>)> *park = (std::function<bool (std::optional<int>)> *)ctx;
 
-    std::optional<int> x;
-    if(ms >= 0)
-        x = ms;
+	std::optional<int> x;
+	if(ms >= 0)
+		x = ms;
 
-    if(!(*park)(x))
-        return -1;
+	if(!(*park)(x))
+		return -1;
 
-    return 0;
+	return 0;
 }
 
 bool Executor::run(std::function<bool (std::optional<int>)> park)
 {
-    if(ffi::executor_run(inner_, park_cb, &park) != 0)
-        return false;
+	if(ffi::executor_run(inner_, park_cb, &park) != 0)
+		return false;
 
-    return true;
+	return true;
+}
+
+bool Executor::currentSpawn(ffi::UnitFuture *fut)
+{
+	if(ffi::executor_current_spawn(fut) != 0)
+		return false;
+
+	return true;
 }

src/core/executor.h

@@ -24,15 +24,21 @@
 class Executor
 {
 public:
-    Executor(int tasksMax);
-    ~Executor();
+	Executor(int tasksMax);
+	~Executor();
 
-    bool run(std::function<bool (std::optional<int>)> park);
+	bool run(std::function<bool (std::optional<int>)> park);
+
+	/// Spawns `fut` on the executor in the current thread. Returns true on
+	/// success or false on error. An error can occur if there is no executor in
+	/// the current thread or if the executor is at capacity. This function takes
+	/// ownership of `fut` regardless of whether spawning is successful.
+	static bool currentSpawn(ffi::UnitFuture *fut);
 
 private:
-    ffi::Executor *inner_;
+	ffi::Executor *inner_;
 
-    static int park_cb(void *ctx, int ms);
+	static int park_cb(void *ctx, int ms);
 };
 
 #endif

src/core/executor.rs

@@ -14,6 +14,7 @@
  * limitations under the License.
  */
 
+use crate::core::future::SizedFuture;
 use crate::core::list;
 use crate::core::waker;
 use log::debug;
@@ -69,7 +70,7 @@ fn poll_fut(fut: &mut BoxFuture, waker: Waker) -> bool {
 }
 
 struct Task {
-    fut: Option<Pin<Box<dyn Future<Output = ()>>>>,
+    fut: Option<BoxFuture>,
     wakeable: bool,
     low: bool,
 }
@@ -124,10 +125,12 @@ impl Tasks {
         !self.data.borrow().next.is_empty() || !self.data.borrow().next_low.is_empty()
     }
 
-    fn add<F>(&self, fut: F) -> Result<(), ()>
+    fn add<T>(&self, get_fut: T, size: usize) -> Result<(), ()>
     where
-        F: Future<Output = ()> + 'static,
+        T: FnOnce() -> BoxFuture,
     {
+        debug!("spawning future with size {size}");
+
         let data = &mut *self.data.borrow_mut();
 
         if data.nodes.len() == data.nodes.capacity() {
@@ -138,7 +141,7 @@ impl Tasks {
         let nkey = entry.key();
 
         let task = Task {
-            fut: Some(Box::pin(fut)),
+            fut: Some(get_fut()),
             wakeable: false,
             low: false,
         };
@@ -340,9 +343,15 @@ impl Executor {
     where
         F: Future<Output = ()> + 'static,
     {
-        debug!("spawning future with size {}", mem::size_of::<F>());
+        self.tasks.add(move || Box::pin(fut), mem::size_of::<F>())
+    }
+
+    #[allow(clippy::result_unit_err)]
+    pub fn spawn_boxed(&self, fut: Pin<Box<dyn SizedFuture<Output = ()>>>) -> Result<(), ()> {
+        let size = (*fut).size();
+        let fut = fut.into_future();
 
-        self.tasks.add(fut)
+        self.tasks.add(move || fut, size)
     }
 
     pub fn set_pre_poll<F>(&self, pre_poll_fn: F)
@@ -473,6 +482,7 @@ impl Spawner {
 
 mod ffi {
     use super::*;
+    use crate::core::future::ffi::UnitFuture;
     use std::ffi::c_int;
 
     #[no_mangle]
@@ -514,6 +524,28 @@ mod ffi {
 
         0
     }
+
+    /// Spawns `fut` on the executor in the current thread. Returns 0 on
+    /// success or non-zero on error. An error can occur if there is no
+    /// executor in the current thread or if the executor is at capacity.
+    /// This function takes ownership of `fut` regardless of whether spawning
+    /// is successful.
+    ///
+    /// SAFETY: `fut` must point to a valid `UnitFuture`.
+    #[no_mangle]
+    pub unsafe extern "C" fn executor_current_spawn(fut: *mut UnitFuture) -> c_int {
+        let Some(executor) = Executor::current() else {
+            return -1;
+        };
+
+        let fut = Box::from_raw(fut).0;
+
+        if executor.spawn_boxed(fut).is_err() {
+            return -1;
+        }
+
+        0
+    }
 }
 
 #[cfg(test)]

src/core/reactor.cpp

@@ -17,19 +17,19 @@
 #include "reactor.h"
 
 Reactor::Reactor(int registrationsMax) :
-    inner_(ffi::reactor_create(registrationsMax))
+	inner_(ffi::reactor_create(registrationsMax))
 {
 }
 
 Reactor::~Reactor()
 {
-    ffi::reactor_destroy(inner_);
+	ffi::reactor_destroy(inner_);
 }
 
 bool Reactor::poll(std::optional<int> ms)
 {
-    if(ffi::reactor_poll(inner_, ms.value_or(-1)) != 0)
-        return false;
+	if(ffi::reactor_poll(inner_, ms.value_or(-1)) != 0)
+		return false;
 
-    return true;
+	return true;
 }

src/core/reactor.h

@@ -23,13 +23,13 @@
 class Reactor
 {
 public:
-    Reactor(int registrationsMax);
-    ~Reactor();
+	Reactor(int registrationsMax);
+	~Reactor();
 
-    bool poll(std::optional<int> ms);
+	bool poll(std::optional<int> ms);
 
 private:
-    ffi::Reactor *inner_;
+	ffi::Reactor *inner_;
 };
 
 #endif