add ffi & c++ wrapper for reactor and executor (#48280)

Author: jkarneges

src/core/core.pri

@@ -58,6 +58,8 @@ HEADERS += \
 	$$PWD/callback.h \
 	$$PWD/config.h \
 	$$PWD/timerwheel.h \
+	$$PWD/reactor.h \
+	$$PWD/executor.h \
 	$$PWD/jwt.h \
 	$$PWD/timer.h \
 	$$PWD/defercall.h \
@@ -91,6 +93,8 @@ HEADERS += \
 SOURCES += \
 	$$PWD/config.cpp \
 	$$PWD/timerwheel.cpp \
+	$$PWD/reactor.cpp \
+	$$PWD/executor.cpp \
 	$$PWD/jwt.cpp \
 	$$PWD/timer.cpp \
 	$$PWD/defercall.cpp \

src/core/executor.cpp

@@ -0,0 +1,49 @@
+/*
+ * Copyright (C) 2025 Fastly, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#include "executor.h"
+
+Executor::Executor(int tasksMax) :
+    inner_(ffi::executor_create(tasksMax))
+{
+}
+
+Executor::~Executor()
+{
+    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::optional<int> x;
+    if(ms >= 0)
+        x = ms;
+
+    if(!(*park)(x))
+        return -1;
+
+    return 0;
+}
+
+bool Executor::run(std::function<bool (std::optional<int>)> park)
+{
+    if(ffi::executor_run(inner_, park_cb, &park) != 0)
+        return false;
+
+    return true;
+}

src/core/executor.h

@@ -0,0 +1,38 @@
+/*
+ * Copyright (C) 2025 Fastly, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#ifndef EXECUTOR_H
+#define EXECUTOR_H
+
+#include <optional>
+#include <functional>
+#include "rust/bindings.h"
+
+class Executor
+{
+public:
+    Executor(int tasksMax);
+    ~Executor();
+
+    bool run(std::function<bool (std::optional<int>)> park);
+
+private:
+    ffi::Executor *inner_;
+
+    static int park_cb(void *ctx, int ms);
+};
+
+#endif

src/core/executor.rs

@@ -471,6 +471,51 @@ impl Spawner {
     }
 }
 
+mod ffi {
+    use super::*;
+    use std::ffi::c_int;
+
+    #[no_mangle]
+    pub extern "C" fn executor_create(tasks_max: libc::size_t) -> *mut Executor {
+        Box::into_raw(Box::new(Executor::new(tasks_max)))
+    }
+
+    #[no_mangle]
+    pub unsafe extern "C" fn executor_destroy(ex: *mut Executor) {
+        if !ex.is_null() {
+            drop(Box::from_raw(ex));
+        }
+    }
+
+    #[no_mangle]
+    pub unsafe extern "C" fn executor_run(
+        ex: *mut Executor,
+        park_fn: unsafe extern "C" fn(*mut libc::c_void, c_int) -> c_int,
+        park_ctx: *mut libc::c_void,
+    ) -> c_int {
+        let ex = unsafe { ex.as_ref().unwrap() };
+
+        let park = |d: Option<Duration>| {
+            let ms = match d {
+                Some(d) => d.as_millis() as c_int,
+                None => -1,
+            };
+
+            if park_fn(park_ctx, ms) != 0 {
+                return Err(io::Error::from(io::ErrorKind::Other));
+            }
+
+            Ok(())
+        };
+
+        if ex.run(park).is_err() {
+            return -1;
+        }
+
+        0
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;

src/core/reactor.cpp

@@ -0,0 +1,35 @@
+/*
+ * Copyright (C) 2025 Fastly, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#include "reactor.h"
+
+Reactor::Reactor(int registrationsMax) :
+    inner_(ffi::reactor_create(registrationsMax))
+{
+}
+
+Reactor::~Reactor()
+{
+    ffi::reactor_destroy(inner_);
+}
+
+bool Reactor::poll(std::optional<int> ms)
+{
+    if(ffi::reactor_poll(inner_, ms.value_or(-1)) != 0)
+        return false;
+
+    return true;
+}

src/core/reactor.h

@@ -0,0 +1,35 @@
+/*
+ * Copyright (C) 2025 Fastly, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#ifndef REACTOR_H
+#define REACTOR_H
+
+#include <optional>
+#include "rust/bindings.h"
+
+class Reactor
+{
+public:
+    Reactor(int registrationsMax);
+    ~Reactor();
+
+    bool poll(std::optional<int> ms);
+
+private:
+    ffi::Reactor *inner_;
+};
+
+#endif

src/core/reactor.rs

@@ -1,5 +1,6 @@
 /*
  * Copyright (C) 2020-2023 Fanout, Inc.
+ * Copyright (C) 2025 Fastly, Inc.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -1009,6 +1010,37 @@ impl TimerEvented {
     }
 }
 
+mod ffi {
+    use super::*;
+    use std::convert::TryInto;
+    use std::ffi::c_int;
+
+    #[no_mangle]
+    pub extern "C" fn reactor_create(registrations_max: libc::size_t) -> *mut Reactor {
+        Box::into_raw(Box::new(Reactor::new(registrations_max)))
+    }
+
+    #[no_mangle]
+    pub unsafe extern "C" fn reactor_destroy(r: *mut Reactor) {
+        if !r.is_null() {
+            drop(Box::from_raw(r));
+        }
+    }
+
+    #[no_mangle]
+    pub unsafe extern "C" fn reactor_poll(r: *const Reactor, ms: c_int) -> c_int {
+        let r = unsafe { r.as_ref().unwrap() };
+
+        let duration = ms.try_into().ok().map(Duration::from_millis);
+
+        if r.poll(duration).is_err() {
+            return -1;
+        }
+
+        0
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;