Updated benchmark

added but disabled heartbeat ping
update readme

Author: r2dev2

README.md

@@ -5,15 +5,12 @@ This is derived from the sse in https://github.com/actix/examples which is under
 ## Usage
 
 ```
-cargo build --release
+cargo run --release
 ```
 
-The executable will be in ``target/release``
-
-
-To benchmark, run 
+To benchmark, run multiple instances of
 
 ```
-node benchmark.js
+python3 benchmark.py
 ```
 

benchmark.py

@@ -1,40 +1,48 @@
+import asyncio
 from sseclient import SSEClient
+
+import logging
+from itertools import count
 from time import sleep
 from threading import Lock, Thread
 
-class Counter(Thread):
+logging.basicConfig(filename="sse.log", level=logging.DEBUG)
+
+class Counter:
     def __init__(self):
-        super().__init__()
+        # super().__init__()
         self.daemon = True
         self.counter = 0
-        self.connections = []
+        self._number_of_read = 0
+        self._counter = count()
         self.lock = Lock()
 
-    def increment(self, connection):
+    def increment(self):
         with self.lock:
             self.counter += 1
-            # self.connections.append(connection)
-            print(self.counter, end='\r')
+            print(self.counter, end='\r', flush=True)
+
 
 counter = Counter()
-counter.start()
+# counter.start()
 
-def task(counter):
 
+def task(counter):
     try:
         messages = SSEClient("http://localhost:8080/events/channel1")
-        print("yee")
-    except OSError:
+        counter.increment()
+    except OSError as e:
+        logging.error(str(e))
         return
     for msg in messages:
-        if str(msg) == "connected":
-            counter.increment(messages)
         if "secret" in str(msg):
-            counter.increment(messages)
+            counter.increment()
 
 
 threads = []
-for _ in range(10):
+for i in range(5000):
+#     if i > 4000 and i % 1000 == 0:
+#         sleep(1)
     t = Thread(target=task, args=(counter,), daemon=True)
     t.start()
     threads.append(t)

src/main.rs

@@ -12,6 +12,7 @@ use std::collections::HashMap;
 use std::pin::Pin;
 use std::sync::Mutex;
 use std::task::{Context, Poll};
+use std::time::Duration;
 
 use actix_cors::Cors;
 use actix_web::web::{Bytes, Data, Path};
@@ -20,6 +21,7 @@ use clap::{crate_version, clap_app};
 use futures::Stream;
 use serde::{Deserialize, Serialize};
 use tokio::sync::broadcast::{channel, Receiver, Sender};
+use tokio::time::{interval_at, Instant};
 
 
 #[actix_rt::main]
@@ -52,6 +54,7 @@ async fn main() -> std::io::Result<()> {
     .bind(format!("{}:{}",
             matches.value_of("HOST").unwrap_or("127.0.0.1"),
             matches.value_of("PORT").unwrap_or("8080")))?
+    .maxconn(500000)
     .run()
     .await
 }
@@ -96,6 +99,24 @@ struct SpotifyData {
     msg: String,
 }
 
+struct Broadcaster {
+    senders: Vec<Sender<Bytes>>,
+    num_clients: u32,
+}
+
+impl Broadcaster {
+    fn new() -> Broadcaster {
+        Broadcaster {
+            senders: Vec::new(),
+            num_clients: 1,
+        }
+    }
+
+    // fn send(&msg) {
+    //     
+    // }
+}
+
 struct BroadcasterMap {
     broadcasters: HashMap<String, Sender<Bytes>>,
     channel_num: usize,
@@ -105,20 +126,30 @@ impl BroadcasterMap {
     fn new() -> Self {
         BroadcasterMap {
             broadcasters: HashMap::<String, Sender<Bytes>>::new(),
-            channel_num: 50000
+            channel_num: 500000
         }
     }
 
     fn create() -> Data<Mutex<Self>> {
         Data::new(Mutex::new(BroadcasterMap::new()))
     }
 
+    fn spawn_ping(tx: Sender<Bytes>) {
+        actix_rt::spawn(async move {
+            let mut task = interval_at(Instant::now(), Duration::from_secs(10));
+            while let _ = task.tick().await {
+                tx.send(Bytes::from("data: ping\n\n"));
+            }
+        });
+    }
+
     fn new_client(&mut self, room: &str) -> Client {
         let s = (&room).to_string();
         match self.broadcasters.get_mut(&s) {
             Some(broadcaster) => Client(broadcaster.subscribe()),
             None => {
                 let (tx, rx) = channel(self.channel_num);
+                // BroadcasterMap::spawn_ping(tx.clone());
                 self.broadcasters.insert(s.to_string(), tx);
                 Client(rx)
             }