Make the number of threads that the application service uses configur…

…able This helps too keep the number of open files low when running the tests
exul · Apr 2, 2017 · 5e759d6 · 5e759d6
1 parent 0601573
commit 5e759d6
Show file tree

Hide file tree

Showing 6 changed files with 28 additions and 32 deletions.
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/Cargo.toml b/Cargo.toml
@@ -15,6 +15,7 @@ diesel_codegen = { version = "0.9", features = ["sqlite"] }
 error-chain = "0.7"
 iron = "0.5"
 lazy_static = "0.2"
+num_cpus = "1.2"
 persistent = "0.3"
 pulldown-cmark = "0.0"
 r2d2 = "0.7"

diff --git a/src/bin/matrix-rocketchat.rs b/src/bin/matrix-rocketchat.rs
@@ -6,6 +6,7 @@
 extern crate clap;
 extern crate iron;
 extern crate matrix_rocketchat;
+extern crate num_cpus;
 #[macro_use]
 extern crate slog;
 extern crate slog_json;
@@ -40,7 +41,8 @@ fn run() -> Result<Listening> {
     let log_file_path =
         matches.value_of("log_file").expect("Log file path is always present (default value is set)").to_string();
     let log = build_logger(&log_file_path);
-    Server::new(&config, log).run()
+    let threads = num_cpus::get() * 8;
+    Server::new(&config, log).run(threads)
 }
 
 fn build_logger(log_file_path: &str) -> slog::Logger {

diff --git a/src/matrix-rocketchat/server.rs b/src/matrix-rocketchat/server.rs
@@ -33,7 +33,7 @@ impl<'a> Server<'a> {
     }
 
     /// Runs the application service bridge.
-    pub fn run(&self) -> Result<Listening> {
+    pub fn run(&self, threads: usize) -> Result<Listening> {
         self.prepare_database()?;
         let connection_pool = ConnectionPool::create(&self.config.database_url)?;
         let connection = connection_pool.get().chain_err(|| ErrorKind::ConnectionPoolExtractionError)?;
@@ -47,7 +47,9 @@ impl<'a> Server<'a> {
         chain.link_before(State::<IronLogger>::one::<Logger>(self.logger.clone()));
 
         info!(self.logger, "Starting server"; "address" => format!("{:?}", self.config.as_address));
-        Iron::new(chain).http(self.config.as_address).chain_err(|| ErrorKind::ServerStartupError).map_err(Error::from)
+        let mut server = Iron::new(chain);
+        server.threads = threads;
+        server.http(self.config.as_address).chain_err(|| ErrorKind::ServerStartupError).map_err(Error::from)
     }
 
     fn setup_routes(&self, matrix_api: Box<MatrixApi>) -> Router {

diff --git a/tests/matrix-rocketchat-test/lib.rs b/tests/matrix-rocketchat-test/lib.rs
@@ -328,7 +328,7 @@ impl Test {
                                          o!("version" => env!("CARGO_PKG_VERSION"),
                                             "place" => file_line_logger_format));
             debug!(DEFAULT_LOGGER, "config: {:?}", server_config);
-            let listening = match Server::new(&server_config, log).run() {
+            let listening = match Server::new(&server_config, log).run(IRON_THREADS) {
                 Ok(listening) => listening,
                 Err(err) => {
                     error!(DEFAULT_LOGGER, "error: {}", err);

diff --git a/tests/startup.rs b/tests/startup.rs
@@ -30,9 +30,7 @@ fn starup_fails_when_server_cannot_bind_to_address() {
 
     let (homeserver_mock_tx, homeserver_mock_rx) = channel::<Listening>();
     let homeserver_mock_socket_addr = matrix_rocketchat_test::get_free_socket_addr();
-    config.hs_url = format!("http://{}:{}",
-                            homeserver_mock_socket_addr.ip(),
-                            homeserver_mock_socket_addr.port());
+    config.hs_url = format!("http://{}:{}", homeserver_mock_socket_addr.ip(), homeserver_mock_socket_addr.port());
 
     thread::spawn(move || {
         let mut router = Router::new();
@@ -51,18 +49,18 @@ fn starup_fails_when_server_cannot_bind_to_address() {
     let running_server_log = log.clone();
     let (running_server_tx, running_server_rx) = channel::<Result<Listening>>();
     thread::spawn(move || {
-        let running_server_result = Server::new(&running_server_config, running_server_log).run();
-        homeserver_mock_listen.close().unwrap();
-        running_server_tx.send(running_server_result).unwrap();
-    });
+                      let running_server_result = Server::new(&running_server_config, running_server_log).run(IRON_THREADS);
+                      homeserver_mock_listen.close().unwrap();
+                      running_server_tx.send(running_server_result).unwrap();
+                  });
     let running_server_result = running_server_rx.recv_timeout(matrix_rocketchat_test::default_timeout()).unwrap();
     assert!(running_server_result.is_ok());
 
     let (failed_server_tx, failed_server_rx) = channel::<Result<Listening>>();
     thread::spawn(move || {
-        let failed_server_result = Server::new(&config, log).run();
-        failed_server_tx.send(failed_server_result).unwrap();
-    });
+                      let failed_server_result = Server::new(&config, log).run(IRON_THREADS);
+                      failed_server_tx.send(failed_server_result).unwrap();
+                  });
     let failed_server_result = failed_server_rx.recv_timeout(matrix_rocketchat_test::default_timeout()).unwrap();
     assert!(failed_server_result.is_err());
     running_server_result.unwrap().close().unwrap();
@@ -101,9 +99,7 @@ fn startup_fails_when_querying_the_api_version_is_not_successful_and_returns_an_
 #[test]
 fn startup_fails_when_the_server_can_query_the_matrix_api_version_but_gets_an_invalid_response() {
     let mut router = Router::new();
-    router.get("/_matrix/client/versions",
-               handlers::InvalidJsonResponse { status: status::Ok },
-               "get_versions");
+    router.get("/_matrix/client/versions", handlers::InvalidJsonResponse { status: status::Ok }, "get_versions");
 
     let server_result = start_servers(router);
 
@@ -115,9 +111,7 @@ fn startup_fails_when_the_server_can_query_the_matrix_api_version_but_gets_an_in
 #[test]
 fn startup_failes_when_the_server_cannot_find_a_compatible_matrix_api_version() {
     let mut router = Router::new();
-    router.get("/_matrix/client/versions",
-               handlers::MatrixVersion { versions: vec!["9999"] },
-               "get_versions");
+    router.get("/_matrix/client/versions", handlers::MatrixVersion { versions: vec!["9999"] }, "get_versions");
 
     let server_result = start_servers(router);
 
@@ -129,9 +123,7 @@ fn startup_failes_when_the_server_cannot_find_a_compatible_matrix_api_version()
 #[test]
 fn startup_failes_when_the_bot_user_registration_failes() {
     let mut router = Router::new();
-    router.get("/_matrix/client/versions",
-               handlers::MatrixVersion { versions: default_matrix_api_versions() },
-               "get_versions");
+    router.get("/_matrix/client/versions", handlers::MatrixVersion { versions: default_matrix_api_versions() }, "get_versions");
     let error_responder = handlers::MatrixErrorResponder {
         status: status::InternalServerError,
         message: "Could not register user".to_string(),
@@ -148,9 +140,7 @@ fn startup_failes_when_the_bot_user_registration_failes() {
 #[test]
 fn startup_failes_when_the_bot_user_registration_returns_invalid_json() {
     let mut router = Router::new();
-    router.get("/_matrix/client/versions",
-               handlers::MatrixVersion { versions: default_matrix_api_versions() },
-               "get_versions");
+    router.get("/_matrix/client/versions", handlers::MatrixVersion { versions: default_matrix_api_versions() }, "get_versions");
     router.post(RegisterEndpoint::router_path(),
                 handlers::InvalidJsonResponse { status: status::InternalServerError },
                 "register");
@@ -167,11 +157,11 @@ fn start_servers(matrix_router: Router) -> Result<Listening> {
 
     let (homeserver_mock_tx, homeserver_mock_rx) = channel::<Listening>();
     thread::spawn(move || {
-        let mut server = Iron::new(matrix_router);
-        server.threads = IRON_THREADS;
-        let listening = server.http(homeserver_mock_socket_addr).unwrap();
-        homeserver_mock_tx.send(listening).unwrap();
-    });
+                      let mut server = Iron::new(matrix_router);
+                      server.threads = IRON_THREADS;
+                      let listening = server.http(homeserver_mock_socket_addr).unwrap();
+                      homeserver_mock_tx.send(listening).unwrap();
+                  });
 
     let (server_tx, server_rx) = channel::<Result<Listening>>();
     thread::spawn(move || {
@@ -181,7 +171,7 @@ fn start_servers(matrix_router: Router) -> Result<Listening> {
                             homeserver_mock_socket_addr.ip(),
                             homeserver_mock_socket_addr.port());
         let log = DEFAULT_LOGGER.clone();
-        let server_result = Server::new(&config, log).run();
+        let server_result = Server::new(&config, log).run(IRON_THREADS);
         server_tx.send(server_result).unwrap();
     });