chore(ci): add a heaptrack job (#265)

* chore(ci): add a `heaptrack` job

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

* chore(e2e): fix heaptrack file upload

* chore(e2e): fix upload-artifact version

* chore(e2e): `kill` the server after 60s

* chore(e2e/bench): custom bench workflow with optional trigger

Author: Totodore

.github/workflows/bench.yml

@@ -0,0 +1,46 @@
+name: Bench
+
+on:
+  workflow_dispatch:
+    inputs:
+      heaptrack:
+        description: 'Run heaptrack memory benchmark'
+        required: true
+        default: false
+        type: boolean
+
+jobs:
+  heaptrack:
+    if: ${{ github.event.inputs.heaptrack == 'true' }}
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - uses: dtolnay/rust-toolchain@master
+        with:
+          toolchain: stable
+      - uses: actions/cache@v4
+        with:
+          path: |
+            ~/.cargo/bin/
+            ~/.cargo/registry/index/
+            ~/.cargo/registry/cache/
+            ~/.cargo/git/db/
+            target/
+          key: ${{ runner.os }}-cargo-${{ hashFiles('**/Cargo.lock') }}-heaptrack
+      - name: Install heaptrack
+        run: sudo apt-get install -y heaptrack
+      - name: Build server && client
+        run: cargo build -r -p heaptrack && cargo build -r -p heaptrack --bin heaptrack-client
+      - name: Run memory benchmark
+        run: heaptrack target/release/heaptrack > server.txt & ./target/release/heaptrack-client > client.txt
+      - name: Server output
+        if: always()
+        run: cat server.txt
+      - name: Client output
+        if: always()
+        run:  cat client.txt
+      - name: Publish memory benchmark
+        uses: actions/upload-artifact@v4
+        with:
+          name: heaptrack-${{ github.head_ref }}.${{ github.sha }}
+          path: heaptrack.heaptrack.*

.github/workflows/github-ci.yml

@@ -251,4 +251,4 @@ jobs:
         run: cat server.txt
       - name: Client output
         if: always()
-        run:  cat client.txt
+        run:  cat client.txt

.gitignore

@@ -1,4 +1,5 @@
 target
 Cargo.lock
 .env
-.vscode
+.vscode
+*.gz

e2e/heaptrack/.gitignore

@@ -0,0 +1,3 @@
+*.gz
+client/node_modules
+memory_usage.svg

e2e/heaptrack/Cargo.toml

@@ -0,0 +1,19 @@
+[package]
+name = "heaptrack"
+version = "0.1.0"
+edition = "2021"
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+
+[dependencies]
+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"] }
+rust_socketio = { version = "0.4.2", features = ["async"] }
+serde_json = "1.0.68"
+rand = "0.8.4"
+
+[[bin]]
+name = "heaptrack-client"
+path = "src/client.rs"

e2e/heaptrack/README.md

@@ -0,0 +1,7 @@
+# Memory usage benchmark
+## Based on the official implementation : https://socket.io/docs/v4/memory-usage/
+
+The goal of this program is to benchmark the memory usage of the socket.io server under different conditions.
+The server can be configured to generated its own reports with the `custom-report` feature flag.
+
+The best way is still to run the program with (heaptrack)[https://github.com/KDE/heaptrack] and analyze the results with the `heaptrack_gui` tool.

e2e/heaptrack/src/client.rs

@@ -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/src/main.rs

@@ -0,0 +1,41 @@
+use hyper::server::conn::http1;
+use hyper_util::rt::TokioIo;
+use socketioxide::{extract::SocketRef, SocketIo};
+use std::{net::SocketAddr, time::Duration};
+use tokio::net::TcpListener;
+
+fn on_connect(socket: SocketRef) {
+    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();
+
+    io.ns("/", on_connect);
+
+    let addr = SocketAddr::from(([127, 0, 0, 1], 3000));
+    let listener = TcpListener::bind(addr).await?;
+
+    tokio::spawn(async move {
+        tokio::time::sleep(Duration::from_secs(60)).await;
+        std::process::exit(0);
+    });
+
+    loop {
+        let (stream, _) = listener.accept().await?;
+
+        let io = TokioIo::new(stream);
+        let svc = svc.clone();
+
+        tokio::task::spawn(async move {
+            http1::Builder::new()
+                .serve_connection(io, svc)
+                .with_upgrades()
+                .await
+                .ok()
+        });
+    }
+}