feat: add actix_max_connections/workers settings (#515)

and print the available_parallelism on startup

SYNC-4005

autoconnect/autoconnect-settings/src/lib.rs

@@ -61,8 +61,6 @@ pub struct Settings {
     /// How long to wait for a response Pong before being timed out and connection drop
     #[serde(deserialize_with = "deserialize_f64_to_duration")]
     pub auto_ping_timeout: Duration,
-    /// Max number of websocket connections to allow
-    pub max_connections: u32,
     /// How long to wait for the initial connection handshake.
     #[serde(deserialize_with = "deserialize_u32_to_duration")]
     pub open_handshake_timeout: Duration,
@@ -99,10 +97,15 @@ pub struct Settings {
     /// Maximum allowed number of backlogged messages. Exceeding this number will
     /// trigger a user reset because the user may have been offline way too long.
     pub msg_limit: u32,
-    /// Maximum number of pending notifications for individual UserAgent handlers.
-    /// (if a given [autoconnect-common::RegisteredClient] receives more than this number, the calling
-    /// thread will lock.)
-    pub max_pending_notification_queue: u32,
+    /// Sets the maximum number of concurrent connections per actix-web worker.
+    ///
+    /// All socket listeners will stop accepting connections when this limit is
+    /// reached for each worker.
+    pub actix_max_connections: Option<usize>,
+    /// Sets number of actix-web workers to start (per bind address).
+    ///
+    /// By default, the number of available physical CPUs is used as the worker count.
+    pub actix_workers: Option<usize>,
 }
 
 impl Default for Settings {
@@ -115,7 +118,6 @@ impl Default for Settings {
             router_hostname: None,
             auto_ping_interval: Duration::from_secs(300),
             auto_ping_timeout: Duration::from_secs(4),
-            max_connections: 0,
             open_handshake_timeout: Duration::from_secs(5),
             close_handshake_timeout: Duration::from_secs(0),
             endpoint_scheme: "http".to_owned(),
@@ -133,7 +135,8 @@ impl Default for Settings {
             megaphone_poll_interval: Duration::from_secs(30),
             human_logs: false,
             msg_limit: 100,
-            max_pending_notification_queue: 10,
+            actix_max_connections: None,
+            actix_workers: None,
         }
     }
 }

autoconnect/src/main.rs

@@ -75,16 +75,20 @@ async fn main() -> Result<()> {
 
     let port = settings.port;
     let router_port = settings.router_port;
+    let actix_max_connections = settings.actix_max_connections;
+    let actix_workers = settings.actix_workers;
     let app_state = AppState::from_settings(settings)?;
     app_state.init_and_spawn_megaphone_updater().await?;
 
     info!(
-        "Starting autoconnect on port {} (router_port: {})",
-        port, router_port
+        "Starting autoconnect on port: {} router_port: {} (available_parallelism: {:?})",
+        port,
+        router_port,
+        std::thread::available_parallelism()
     );
 
     let router_app_state = app_state.clone();
-    Server::build()
+    let mut builder = Server::build()
         .bind("autoconnect", ("0.0.0.0", port), move || {
             let app = build_app!(app_state, config);
             HttpService::build()
@@ -99,12 +103,15 @@ async fn main() -> Result<()> {
                 // XXX:
                 .finish(map_config(app, |_| AppConfig::default()))
                 .tcp()
-        })?
-        .run()
-        .await
-        .map_err(|e| e.into())
-        .map(|v| {
-            info!("Shutting down autoconnect");
-            v
-        })
+        })?;
+    if let Some(max_connections) = actix_max_connections {
+        builder = builder.max_concurrent_connections(max_connections);
+    }
+    if let Some(workers) = actix_workers {
+        builder = builder.workers(workers);
+    }
+    builder.run().await?;
+
+    info!("Shutting down autoconnect");
+    Ok(())
 }

autoendpoint/src/main.rs

@@ -58,11 +58,15 @@ async fn main() -> Result<(), Box<dyn Error>> {
     let server = server::Server::with_settings(settings)
         .await
         .expect("Could not start server");
-    info!("Server started: {}", host_port);
+    info!(
+        "Starting autoendpoint on port: {} (available_parallelism: {:?})",
+        host_port,
+        std::thread::available_parallelism()
+    );
     server.await?;
 
     // Shutdown
-    info!("Server closing");
+    info!("Shutting down autoendpoint");
     logging::reset_logging();
     Ok(())
 }