Port to smol async runtime

* Use smol instead of tokio
* Use flume instead of futures::mpsc

Signed-off-by: Teddy Astie <[email protected]>

Author: TSnake41

Cargo.lock

@@ -6,6 +6,6 @@ repository.workspace = true
 categories.workspace = true
 
 [dependencies]
-futures = "0.3"
+flume = "0.11.1"
 uuid = "1.11"
 os_info = "3"

guest-metrics/Cargo.toml

@@ -1,13 +1,11 @@
 use std::future::Future;
 
-use futures::channel::mpsc;
-
 use crate::GuestMetric;
 
 pub trait GuestAgentPlugin {
-    fn run(self, channel: mpsc::Sender<GuestMetric>) -> impl Future<Output = ()> + Send;
+    fn run(self, channel: flume::Sender<GuestMetric>) -> impl Future<Output = ()> + Send;
 }
 
 pub trait GuestAgentPublisher {
-    fn run(self, channel: mpsc::Receiver<GuestMetric>) -> impl Future<Output = ()> + Send;
+    fn run(self, channel: flume::Receiver<GuestMetric>) -> impl Future<Output = ()> + Send;
 }

guest-metrics/src/plugin.rs

@@ -6,7 +6,8 @@ edition = "2021"
 [dependencies]
 guest-metrics = { path = "../../guest-metrics" }
 futures = "0.3"
-tokio = { version = "1.25.0", features = ["time"] }
+flume = "0.11.1"
+smol = "2.0.2"
 
 [target.'cfg(target_os = "windows")'.dependencies]
 windows = { version = "0.58", features = ["Win32_System_SystemInformation"] }

providers/provider-memory/Cargo.toml

@@ -1,6 +1,6 @@
 use std::{io, time::Duration};
 
-use futures::{channel::mpsc, SinkExt};
+use futures::StreamExt;
 use guest_metrics::{plugin::GuestAgentPlugin, MemoryInfo};
 
 #[cfg(target_os = "freebsd")]
@@ -30,16 +30,16 @@ pub type PlatformMemorySource = windows::WindowsMemorySource;
 pub struct MemoryPlugin;
 
 impl GuestAgentPlugin for MemoryPlugin {
-    async fn run(self, mut channel: mpsc::Sender<guest_metrics::GuestMetric>) {
-        let mut timer = tokio::time::interval(Duration::from_secs_f32(5.0));
+    async fn run(self, channel: flume::Sender<guest_metrics::GuestMetric>) {
+        let mut timer = smol::Timer::interval(Duration::from_secs_f32(5.0));
         let mut memory_source =
             PlatformMemorySource::new().expect("Unable to get memory information");
 
         loop {
-            timer.tick().await;
+            timer.next().await;
 
             if channel
-                .send(guest_metrics::GuestMetric::Memory(MemoryInfo {
+                .send_async(guest_metrics::GuestMetric::Memory(MemoryInfo {
                     mem_free: memory_source.get_available_kb().unwrap(),
                     mem_total: memory_source.get_total_kb().unwrap(),
                 }))

providers/provider-memory/src/lib.rs

@@ -7,6 +7,8 @@ edition = "2021"
 futures = "0.3"
 guest-metrics = { path = "../../guest-metrics" }
 tokio = { version = "1", features = ["rt"] }
+smol = "2.0.2"
+flume = "0.11.1"
 uuid = "1.11"
 anyhow = "*"
 log = "*"

providers/provider-netlink/Cargo.toml

@@ -3,10 +3,7 @@ use std::{collections::HashMap, io};
 use guest_metrics::{plugin::GuestAgentPlugin, GuestMetric, NetEvent, NetEventOp, NetInterface};
 use vif_detect::VifDetector;
 
-use futures::{
-    channel::mpsc::{self, UnboundedReceiver},
-    SinkExt, StreamExt,
-};
+use futures::{channel::mpsc::UnboundedReceiver, StreamExt};
 use uuid::Uuid;
 
 use netlink_packet_core::{
@@ -49,7 +46,7 @@ impl NetlinkConnection {
 pub struct NetlinkPlugin;
 
 impl GuestAgentPlugin for NetlinkPlugin {
-    async fn run(self, mut channel: mpsc::Sender<GuestMetric>) {
+    async fn run(self, channel: flume::Sender<GuestMetric>) {
         let connection = NetlinkConnection::new().unwrap();
         let vif_identify = vif_detect::PlatformVifDetector::default();
         let mut interfaces = HashMap::new();
@@ -91,7 +88,7 @@ impl GuestAgentPlugin for NetlinkPlugin {
         while let Some(msg) = stream.next().await {
             if let NetlinkPayload::InnerMessage(inner_msg) = msg.payload {
                 if let Err(e) =
-                    process_message(inner_msg, &mut channel, &vif_identify, &mut interfaces).await
+                    process_message(inner_msg, &channel, &vif_identify, &mut interfaces).await
                 {
                     log::error!("Unable to process netlink message: {e}");
                 }
@@ -102,7 +99,7 @@ impl GuestAgentPlugin for NetlinkPlugin {
 
 async fn process_message(
     inner_msg: RouteNetlinkMessage,
-    channel: &mut mpsc::Sender<GuestMetric>,
+    channel: &flume::Sender<GuestMetric>,
     vif_identify: &impl VifDetector,
     interfaces: &mut HashMap<u32, Uuid>,
 ) -> anyhow::Result<()> {
@@ -145,7 +142,7 @@ async fn process_message(
 
             interfaces.insert(link_message.header.index, uuid);
             channel
-                .send(GuestMetric::AddIface(NetInterface {
+                .send_async(GuestMetric::AddIface(NetInterface {
                     uuid,
                     index: link_message.header.index,
                     name: ifname.clone(),
@@ -158,7 +155,7 @@ async fn process_message(
                 return Ok(());
             };
 
-            channel.send(GuestMetric::RmIface(uuid)).await.ok();
+            channel.send_async(GuestMetric::RmIface(uuid)).await.ok();
         }
         RouteNetlinkMessage::NewAddress(address_message)
         | RouteNetlinkMessage::GetAddress(address_message) => {
@@ -184,7 +181,7 @@ async fn process_message(
             };
 
             channel
-                .send(GuestMetric::Network(NetEvent {
+                .send_async(GuestMetric::Network(NetEvent {
                     iface_id,
                     op: NetEventOp::AddIp(addr),
                 }))
@@ -213,7 +210,7 @@ async fn process_message(
             };
 
             channel
-                .send(GuestMetric::Network(NetEvent {
+                .send_async(GuestMetric::Network(NetEvent {
                     iface_id,
                     op: NetEventOp::RmIp(addr),
                 }))

providers/provider-netlink/src/lib.rs

@@ -7,7 +7,8 @@ categories.workspace = true
 
 [dependencies]
 guest-metrics = { path = "../../guest-metrics" }
-futures = "0.3"
+smol = "2.0.2"
+flume = "0.11.1"
 
 [target.'cfg(unix)'.dependencies]
 uname = "0.1.1"

providers/provider-os/Cargo.toml

@@ -1,4 +1,3 @@
-use futures::{channel::mpsc, SinkExt};
 use guest_metrics::{os_info, plugin::GuestAgentPlugin, GuestMetric, KernelInfo, OsInfo};
 use std::io;
 
@@ -20,11 +19,11 @@ pub fn collect_kernel() -> io::Result<Option<KernelInfo>> {
 pub struct OsInfoPlugin;
 
 impl GuestAgentPlugin for OsInfoPlugin {
-    async fn run(self, mut channel: mpsc::Sender<guest_metrics::GuestMetric>) {
+    async fn run(self, channel: flume::Sender<guest_metrics::GuestMetric>) {
         let kernel_info = collect_kernel().expect("Unable to fetch kernel information");
 
         channel
-            .send(GuestMetric::OperatingSystem(OsInfo {
+            .send_async(GuestMetric::OperatingSystem(OsInfo {
                 os_info: os_info::get(),
                 kernel_info,
             }))

providers/provider-os/src/lib.rs

@@ -7,7 +7,8 @@ edition = "2021"
 futures = "0.3"
 guest-metrics = { path = "../../guest-metrics" }
 uuid = { version = "1.11", features = ["v4"] }
-tokio = { version = "1", features = ["time"] }
+smol = "2.0.2"
+flume = "0.11.1"
 
 vif-detect = { path = "../vif-detect" }
 network-interface = "2.0"