Add Broker::max_live_threads and Broker::max_pcap_threads tunables

These may be used to change the number of scheduler threads that the
underlying CAF library creates.  In pcap mode, it's currently hardcoded
to the minimal 4 threads due to potentially significant overhead in CAF.

CHANGES

@@ -1,4 +1,9 @@
 
+2.5-660 | 2018-06-12 13:49:39 -0500
+
+  * Add Broker::max_live_threads and Broker::max_pcap_threads tunables
+    (Corelight)
+
 2.5-658 | 2018-06-08 16:41:07 +0000
 
   * Allow BRO_DEFAULT_LISTEN_ADDRESS to control broker listen address.

VERSION

@@ -1 +1 @@
-2.5-658
+2.5-660

scripts/base/frameworks/broker/main.bro

@@ -51,6 +51,22 @@ export {
 	## all peers.
 	const ssl_keyfile = "" &redef;
 
+	## Max number of threads to use for Broker/CAF functionality when
+	## operating on a live interface.  Using zero will cause this to
+	## be automatically determined based on number of available CPUs.
+	const max_live_threads = 0 &redef;
+
+	## Max number of threads to use for Broker/CAF functionality when
+	## operating on a pcap file.  Using zero will cause this to be
+	## automaticallu determined based on number of available CPUs.
+	# TODO: on systems where number of CPUs starts exceeding ~10,
+	# simply creating a caf::actor_system and not using it incurs
+	# significant performance overhead.  Can CAF be updated to
+	# be more efficient in the case where the application isn't
+	# actually making much use of most of those threads instead
+	# of hardcoding this to the minimal 4 threads?
+	const max_pcap_threads = 4 &redef;
+
 	## Forward all received messages to subscribing peers.
 	const forward_messages = F &redef;
 

src/broker/Manager.cc

@@ -113,21 +113,18 @@ static inline Val* get_option(const char* option)
 	return id->ID_Val();
 	}
 
-class configuration : public broker::configuration {
-public:
-	configuration(broker::broker_options options)
-		: broker::configuration(options)
-		{
-		openssl_cafile = get_option("Broker::ssl_cafile")->AsString()->CheckString();
-		openssl_capath = get_option("Broker::ssl_capath")->AsString()->CheckString();
-		openssl_certificate = get_option("Broker::ssl_certificate")->AsString()->CheckString();
-		openssl_key = get_option("Broker::ssl_keyfile")->AsString()->CheckString();
-		openssl_passphrase = get_option("Broker::ssl_passphrase")->AsString()->CheckString();
-		}
-};
+Manager::BrokerConfig::BrokerConfig(broker::broker_options options)
+	: broker::configuration(options)
+	{
+	openssl_cafile = get_option("Broker::ssl_cafile")->AsString()->CheckString();
+	openssl_capath = get_option("Broker::ssl_capath")->AsString()->CheckString();
+	openssl_certificate = get_option("Broker::ssl_certificate")->AsString()->CheckString();
+	openssl_key = get_option("Broker::ssl_keyfile")->AsString()->CheckString();
+	openssl_passphrase = get_option("Broker::ssl_passphrase")->AsString()->CheckString();
+	}
 
-Manager::BrokerState::BrokerState(broker::broker_options options)
-	: endpoint(configuration(options)),
+Manager::BrokerState::BrokerState(BrokerConfig config)
+	: endpoint(std::move(config)),
 	  subscriber(endpoint.make_subscriber({}, SUBSCRIBER_MAX_QSIZE)),
 	  status_subscriber(endpoint.make_status_subscriber(true))
 	{
@@ -173,7 +170,22 @@ void Manager::InitPostScript()
 	options.forward = get_option("Broker::forward_messages")->AsBool();
 	options.use_real_time = ! reading_pcaps;
 
-	bstate = std::make_shared<BrokerState>(options);
+	BrokerConfig config{std::move(options)};
+	auto max_live_threads = get_option("Broker::max_live_threads")->AsCount();
+	auto max_pcap_threads = get_option("Broker::max_pcap_threads")->AsCount();
+
+	if ( reading_pcaps )
+		{
+		if ( max_pcap_threads )
+			config.scheduler_max_threads = max_pcap_threads;
+		}
+	else
+		{
+		if ( max_live_threads )
+			config.scheduler_max_threads = max_live_threads;
+		}
+
+	bstate = std::make_shared<BrokerState>(std::move(config));
 	}
 
 void Manager::Terminate()

src/broker/Manager.h

@@ -339,9 +339,14 @@ public:
 
 private:
 
+	class BrokerConfig : public broker::configuration {
+	public:
+		BrokerConfig(broker::broker_options options);
+	};
+
 	class BrokerState {
 	public:
-		BrokerState(broker::broker_options options);
+		BrokerState(BrokerConfig config);
 		broker::endpoint endpoint;
 		broker::subscriber subscriber;
 		broker::status_subscriber status_subscriber;