chore(e2e): switch the client from JS to rust

Author: Totodore

.github/workflows/github-ci.yml

@@ -268,18 +268,13 @@ jobs:
             ~/.cargo/registry/cache/
             ~/.cargo/git/db/
             target/
-          key: ${{ runner.os }}-cargo-${{ hashFiles('**/Cargo.lock') }}
+          key: ${{ runner.os }}-cargo-${{ hashFiles('**/Cargo.lock') }}-heaptrack
       - name: Install heaptrack
         run: sudo apt-get install -y heaptrack
-      - uses: actions/setup-node@v4
-        with:
-          node-version: 16
-      - name: Install client deps
-        run: cd e2e/heaptrack/client && npm install && cd ../../..
-      - name: Build server
+      - name: Build server && client
         run: cargo build -r -p heaptrack
       - name: Run memory benchmark
-        run: heaptrack target/release/heaptrack & npm --prefix e2e/heaptrack/client start
+        run: heaptrack target/release/heaptrack & sleep 1 && ./target/release/heaptrack-client && kill -SIGINT $!
       - name: Publish memory benchmark
         uses: actions/upload-artifact@v2
         with:

e2e/heaptrack/Cargo.toml

@@ -10,9 +10,10 @@ socketioxide = { path = "../../socketioxide" }
 hyper = { workspace = true, features = ["server", "http1", "http2"] }
 hyper-util = { workspace = true, features = ["tokio"] }
 tokio = { workspace = true, features = ["rt-multi-thread", "macros", "signal"] }
-tracing-subscriber = { workspace = true, optional = true }
-tracing = { workspace = true, optional = true }
-charts-rs = { version = "0.3.3", optional = true }
+rust_socketio = { version = "0.4.2", features = ["async"] }
+serde_json = "1.0.68"
+rand = "0.8.4"
 
-[features]
-custom-report = ["socketioxide/tracing", "tracing", "tracing-subscriber"]
+[[bin]]
+name = "heaptrack-client"
+path = "src/client.rs"

e2e/heaptrack/client/main.js

@@ -0,0 +1,52 @@
+use std::{pin::Pin, time::Duration};
+
+use rust_socketio::{
+    asynchronous::{Client, ClientBuilder},
+    Payload,
+};
+
+const PING_INTERVAL: Duration = Duration::from_millis(1000);
+const POLLING_PERCENTAGE: f32 = 0.05;
+const MAX_CLIENT: usize = 200;
+
+fn cb(_: Payload, socket: Client) -> Pin<Box<dyn std::future::Future<Output = ()> + Send>> {
+    Box::pin(async move {
+        tokio::spawn(async move {
+            let mut inter = tokio::time::interval(PING_INTERVAL);
+            loop {
+                inter.tick().await;
+                let _ = socket.emit("ping", serde_json::Value::Null).await;
+                let _ = socket
+                    .emit("ping", (0..u8::MAX).into_iter().collect::<Vec<u8>>())
+                    .await;
+            }
+        });
+    })
+}
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn std::error::Error>> {
+    tokio::spawn(async move {
+        for _ in 0..MAX_CLIENT {
+            let random: f32 = rand::random();
+            let transport_type = if POLLING_PERCENTAGE > random {
+                rust_socketio::TransportType::Polling
+            } else {
+                rust_socketio::TransportType::WebsocketUpgrade
+            };
+            // get a socket that is connected to the admin namespace
+            ClientBuilder::new("http://localhost:3000/")
+                .transport_type(transport_type)
+                .namespace("/")
+                .on("open", cb)
+                .on("error", |err, _| {
+                    Box::pin(async move { eprintln!("Error: {:#?}", err) })
+                })
+                .connect()
+                .await
+                .expect("Connection failed");
+        }
+    });
+    tokio::time::sleep(Duration::from_secs(60)).await;
+
+    Ok(())
+}

e2e/heaptrack/client/package-lock.json

@@ -1,110 +1,24 @@
-#[cfg(feature = "custom-report")]
-use charts_rs::LineChart;
 use hyper::server::conn::http1;
 use hyper_util::rt::TokioIo;
-use socketioxide::{
-    extract::{AckSender, SocketRef},
-    SocketIo,
-};
+use socketioxide::{extract::SocketRef, SocketIo};
 use std::net::SocketAddr;
-#[cfg(feature = "custom-report")]
-use std::sync::{atomic::AtomicUsize, Mutex};
-#[cfg(feature = "custom-report")]
-use std::time::Instant;
 use tokio::net::TcpListener;
-#[cfg(feature = "custom-report")]
-use tracing::Level;
-#[cfg(feature = "custom-report")]
-use tracing_subscriber::FmtSubscriber;
-
-use std::alloc::{GlobalAlloc, Layout, System};
-
-struct MyAllocator;
-
-#[cfg(feature = "custom-report")]
-static CURRENT: AtomicUsize = AtomicUsize::new(0);
-#[cfg(feature = "custom-report")]
-static MEMORY: Mutex<Vec<(Instant, usize, usize)>> = Mutex::new(Vec::new());
-#[cfg(feature = "custom-report")]
-static SOCKETS_CNT: AtomicUsize = AtomicUsize::new(0);
-
-unsafe impl GlobalAlloc for MyAllocator {
-    unsafe fn alloc(&self, layout: Layout) -> *mut u8 {
-        #[cfg(feature = "custom-report")]
-        CURRENT.fetch_add(layout.size(), std::sync::atomic::Ordering::Relaxed);
-        System.alloc(layout)
-    }
-
-    unsafe fn dealloc(&self, ptr: *mut u8, layout: Layout) {
-        #[cfg(feature = "custom-report")]
-        CURRENT.fetch_sub(layout.size(), std::sync::atomic::Ordering::Relaxed);
-        System.dealloc(ptr, layout)
-    }
-}
-
-#[global_allocator]
-static GLOBAL: MyAllocator = MyAllocator;
 
 fn on_connect(socket: SocketRef) {
-    #[cfg(feature = "custom-report")]
-    SOCKETS_CNT.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
-    socket.on("ping", |ack: AckSender| {
-        ack.send("pong").ok();
-    });
-    socket.on_disconnect(|| {
-        #[cfg(feature = "custom-report")]
-        SOCKETS_CNT.fetch_sub(1, std::sync::atomic::Ordering::Relaxed);
+    socket.on("ping", |s: SocketRef| {
+        s.emit("pong", ()).ok();
     });
 }
 
 #[tokio::main]
 async fn main() -> Result<(), Box<dyn std::error::Error>> {
     let (svc, io) = SocketIo::new_svc();
-    #[cfg(feature = "custom-report")]
-    {
-        let subscriber = FmtSubscriber::builder()
-            .with_line_number(true)
-            .with_max_level(Level::TRACE)
-            .finish();
-        tracing::subscriber::set_global_default(subscriber)?;
-    }
 
     io.ns("/", on_connect);
 
     let addr = SocketAddr::from(([127, 0, 0, 1], 3000));
     let listener = TcpListener::bind(addr).await?;
 
-    #[cfg(feature = "custom-report")]
-    tokio::task::spawn(async move {
-        loop {
-            tokio::time::sleep(tokio::time::Duration::from_millis(1000)).await;
-            MEMORY.lock().unwrap().push((
-                Instant::now(),
-                SOCKETS_CNT.load(std::sync::atomic::Ordering::Relaxed),
-                CURRENT.load(std::sync::atomic::Ordering::Relaxed),
-            ));
-        }
-    });
-
-    #[cfg(feature = "custom-report")]
-    tokio::task::spawn(async move {
-        tokio::signal::ctrl_c().await.unwrap();
-        println!("saving charts...");
-        let points = MEMORY.lock().unwrap();
-        let mut time: Vec<String> = Vec::with_capacity(points.len());
-        let mut sockets: Vec<String> = Vec::with_capacity(points.len());
-        let mut memory: Vec<f32> = Vec::with_capacity(points.len());
-        for (t, s, m) in points.iter() {
-            time.push(t.elapsed().as_secs_f64().to_string());
-            sockets.push(s.to_string());
-            memory.push(*m as f32);
-        }
-        let line_chart = LineChart::new(vec![("Memory", memory).into()], sockets);
-        let svg = line_chart.svg().unwrap();
-        std::fs::write("memory_usage.svg", svg).unwrap();
-        std::process::exit(0);
-    });
-
     loop {
         let (stream, _) = listener.accept().await?;