Merge 29b5eef4b7 into ccf6c4707f

Cargo.toml

@@ -64,6 +64,7 @@ rust-version = "1.85.0"
 anyhow = { version = "1", default-features = false }
 assert_matches = { version = "1.5.0", default-features = false }
 async-io = { version = "2.0", default-features = false }
+async-global-executor = { version = "3.1.0", default-features = false }
 base64 = { version = "0.22.1", default-features = false }
 bindgen = { version = "0.71", default-features = false }
 bitflags = { version = "2.2.1", default-features = false }
@@ -71,6 +72,7 @@ bytes = { version = "1", default-features = false }
 cargo_metadata = { version = "0.19.0", default-features = false }
 clap = { version = "4", default-features = false }
 const-assert = { version = "1.0.1", default-features = false }
+compio = { version = "0.14.0", default-features = false }
 dialoguer = { version = "0.11", default-features = false }
 diff = { version = "0.1.13", default-features = false }
 env_logger = { version = "0.11", default-features = false }

aya-log/Cargo.toml

@@ -16,13 +16,20 @@ rust-version.workspace = true
 [lints]
 workspace = true
 
+[features]
+async_tokio = ["aya/async_tokio", "dep:tokio"]
+async_std = ["aya/async_std", "dep:async-global-executor"]
+async_compio = ["aya/async_compio", "dep:compio"]
+
 [dependencies]
-aya = { path = "../aya", version = "^0.13.1", features = ["async_tokio"] }
+aya = { path = "../aya", version = "^0.13.1" }
 aya-log-common = { path = "../aya-log-common", version = "^0.1.15", default-features = false }
 bytes = { workspace = true }
 log = { workspace = true }
 thiserror = { workspace = true }
-tokio = { workspace = true, features = ["rt"] }
+tokio = { workspace = true, features = ["rt"], optional = true }
+async-global-executor = { workspace = true, optional = true }
+compio = { workspace = true, optional = true }
 
 [dev-dependencies]
 env_logger = { workspace = true }

aya-log/src/lib.rs

@@ -59,6 +59,23 @@ use std::{
 
 const MAP_NAME: &str = "AYA_LOGS";
 
+// check async feature mutual exclusivity
+#[cfg(all(feature = "async_tokio", feature = "async_std"))]
+compile_error!("Cannot enable both async_tokio and async_std features at the same time");
+
+#[cfg(all(feature = "async_tokio", feature = "async_compio"))]
+compile_error!("Cannot enable both async_tokio and async_compio features at the same time");
+
+#[cfg(all(feature = "async_std", feature = "async_compio"))]
+compile_error!("Cannot enable both async_std and async_compio features at the same time");
+
+#[cfg(not(any(
+    feature = "async_tokio",
+    feature = "async_std",
+    feature = "async_compio"
+)))]
+compile_error!("Must enable at least one async feature: async_tokio, async_std, or async_compio");
+
 use aya::{
     Ebpf, Pod,
     maps::{
@@ -165,7 +182,8 @@ impl EbpfLogger {
             let mut buf = logs.open(cpu_id, None)?;
 
             let log = logger.clone();
-            tokio::spawn(async move {
+
+            let fut = async move {
                 let mut buffers = vec![BytesMut::with_capacity(LOG_BUF_CAPACITY); 10];
 
                 loop {
@@ -175,7 +193,22 @@ impl EbpfLogger {
                         log_buf(buf.as_ref(), &*log).unwrap();
                     }
                 }
-            });
+            };
+
+            #[cfg(feature = "async_tokio")]
+            {
+                tokio::spawn(fut);
+            }
+
+            #[cfg(feature = "async_std")]
+            {
+                async_global_executor::spawn(fut).detach();
+            }
+
+            #[cfg(feature = "async_compio")]
+            {
+                compio::runtime::spawn(fut).detach();
+            }
         }
         Ok(())
     }

aya/Cargo.toml

@@ -22,6 +22,7 @@ async-io = { workspace = true, optional = true }
 aya-obj = { path = "../aya-obj", version = "^0.2.1", features = ["std"] }
 bitflags = { workspace = true }
 bytes = { workspace = true }
+compio = { workspace = true, optional = true, features = ["default"] }
 hashbrown = { workspace = true }
 libc = { workspace = true }
 log = { workspace = true }
@@ -36,6 +37,7 @@ tempfile = { workspace = true }
 [features]
 async_std = ["dep:async-io"]
 async_tokio = ["tokio/net"]
+async_compio = ["dep:compio"]
 default = []
 
 [package.metadata.docs.rs]

aya/src/lib.rs

@@ -24,13 +24,14 @@
 //! * Support for function call relocation and global data maps, which
 //!   allows eBPF programs to make **function calls** and use **global variables
 //!   and initializers**.
-//! * **Async support** with both [tokio] and [async-std].
+//! * **Async support** with [tokio], [async-std] and [compio].
 //! * Easy to deploy and fast to build: aya doesn't require a kernel build or
 //!   compiled headers, and not even a C toolchain; a release build completes in a matter
 //!   of seconds.
 //!
 //! [tokio]: https://docs.rs/tokio
 //! [async-std]: https://docs.rs/async-std
+//! [compio]: https://docs.rs/compio
 
 #![doc(
     html_logo_url = "https://aya-rs.dev/assets/images/crabby.svg",
@@ -68,7 +69,11 @@
     //unused_results,
 )]
 #![cfg_attr(
-    all(feature = "async_tokio", feature = "async_std"),
+    all(
+        feature = "async_tokio",
+        feature = "async_std",
+        feature = "async_compio"
+    ),
     expect(unused_crate_dependencies)
 )]
 

aya/src/maps/mod.rs

@@ -92,8 +92,19 @@ pub use bloom_filter::BloomFilter;
 pub use hash_map::{HashMap, PerCpuHashMap};
 pub use info::{MapInfo, MapType, loaded_maps};
 pub use lpm_trie::LpmTrie;
-#[cfg(any(feature = "async_tokio", feature = "async_std"))]
-#[cfg_attr(docsrs, doc(cfg(any(feature = "async_tokio", feature = "async_std"))))]
+#[cfg(any(
+    feature = "async_tokio",
+    feature = "async_std",
+    feature = "async_compio"
+))]
+#[cfg_attr(
+    docsrs,
+    doc(cfg(any(
+        feature = "async_tokio",
+        feature = "async_std",
+        feature = "async_compio"
+    )))
+)]
 pub use perf::AsyncPerfEventArray;
 pub use perf::PerfEventArray;
 pub use queue::Queue;
@@ -487,8 +498,8 @@ impl_try_from_map!(() {
     SockMap,
     StackTraceMap,
     XskMap,
-    #[cfg(any(feature = "async_tokio", feature = "async_std"))]
-    #[cfg_attr(docsrs, doc(cfg(any(feature = "async_tokio", feature = "async_std"))))]
+    #[cfg(any(feature = "async_tokio", feature = "async_std", feature = "async_compio"))]
+    #[cfg_attr(docsrs, doc(cfg(any(feature = "async_tokio", feature = "async_std", feature = "async_compio"))))]
     AsyncPerfEventArray from PerfEventArray,
 });
 

aya/src/maps/perf/async_perf_event_array.rs

@@ -1,3 +1,15 @@
+// check async feature mutual exclusivity
+#[cfg(all(feature = "async_tokio", feature = "async_std"))]
+compile_error!("Cannot enable both async_tokio and async_std features at the same time");
+
+#[cfg(all(feature = "async_tokio", feature = "async_compio"))]
+compile_error!("Cannot enable both async_tokio and async_compio features at the same time");
+
+#[cfg(all(feature = "async_std", feature = "async_compio"))]
+compile_error!("Cannot enable both async_std and async_compio features at the same time");
+
+#[cfg(feature = "async_compio")]
+use std::os::fd::AsRawFd as _;
 use std::{
     borrow::{Borrow, BorrowMut},
     path::Path,
@@ -7,9 +19,17 @@ use std::{
 //
 // We should eventually split async functionality out into separate crates "aya-async-tokio" and
 // "async-async-std". Presently we arbitrarily choose tokio over async-std when both are requested.
-#[cfg(all(not(feature = "async_tokio"), feature = "async_std"))]
+#[cfg(feature = "async_std")]
 use async_io::Async;
 use bytes::BytesMut;
+#[cfg(feature = "async_compio")]
+use compio::{
+    driver::{
+        SharedFd,
+        op::{Interest, PollOnce},
+    },
+    runtime,
+};
 #[cfg(feature = "async_tokio")]
 use tokio::io::unix::AsyncFd;
 
@@ -103,7 +123,7 @@ impl<T: BorrowMut<MapData>> AsyncPerfEventArray<T> {
         let buf = perf_map.open(index, page_count)?;
         #[cfg(feature = "async_tokio")]
         let buf = AsyncFd::new(buf)?;
-        #[cfg(all(not(feature = "async_tokio"), feature = "async_std"))]
+        #[cfg(feature = "async_std")]
         let buf = Async::new(buf)?;
         Ok(AsyncPerfEventArrayBuffer { buf })
     }
@@ -139,7 +159,7 @@ pub struct AsyncPerfEventArrayBuffer<T: BorrowMut<MapData>> {
     #[cfg(feature = "async_tokio")]
     buf: AsyncFd<PerfEventArrayBuffer<T>>,
 
-    #[cfg(all(not(feature = "async_tokio"), feature = "async_std"))]
+    #[cfg(feature = "async_std")]
     buf: Async<PerfEventArrayBuffer<T>>,
 }
 
@@ -164,7 +184,7 @@ impl<T: BorrowMut<MapData>> AsyncPerfEventArrayBuffer<T> {
             #[cfg(feature = "async_tokio")]
             let buf = guard.get_inner_mut();
 
-            #[cfg(all(not(feature = "async_tokio"), feature = "async_std"))]
+            #[cfg(feature = "async_std")]
             let buf = {
                 if !buf.get_ref().readable() {
                     buf.readable().await?;
@@ -172,6 +192,12 @@ impl<T: BorrowMut<MapData>> AsyncPerfEventArrayBuffer<T> {
                 unsafe { buf.get_mut() }
             };
 
+            #[cfg(feature = "async_compio")]
+            {
+                let poll_once = PollOnce::new(SharedFd::new(buf.as_raw_fd()), Interest::Readable);
+                runtime::submit(poll_once).await.0?;
+            }
+
             let events = buf.read_events(buffers)?;
             const EMPTY: Events = Events { read: 0, lost: 0 };
             if events != EMPTY {

aya/src/maps/perf/mod.rs

@@ -2,14 +2,36 @@
 //! `perf` API.
 //!
 //! See [`PerfEventArray`] and [`AsyncPerfEventArray`].
-#[cfg(any(feature = "async_tokio", feature = "async_std"))]
-#[cfg_attr(docsrs, doc(cfg(any(feature = "async_tokio", feature = "async_std"))))]
+#[cfg(any(
+    feature = "async_tokio",
+    feature = "async_std",
+    feature = "async_compio"
+))]
+#[cfg_attr(
+    docsrs,
+    doc(cfg(any(
+        feature = "async_tokio",
+        feature = "async_std",
+        feature = "async_compio"
+    )))
+)]
 mod async_perf_event_array;
 mod perf_buffer;
 mod perf_event_array;
 
-#[cfg(any(feature = "async_tokio", feature = "async_std"))]
-#[cfg_attr(docsrs, doc(cfg(any(feature = "async_tokio", feature = "async_std"))))]
+#[cfg(any(
+    feature = "async_tokio",
+    feature = "async_std",
+    feature = "async_compio"
+))]
+#[cfg_attr(
+    docsrs,
+    doc(cfg(any(
+        feature = "async_tokio",
+        feature = "async_std",
+        feature = "async_compio"
+    )))
+)]
 pub use async_perf_event_array::*;
 pub use perf_buffer::*;
 pub use perf_event_array::*;