Immediately apply broker subscriptions made during bro_init()

Otherwise that's begging for unit test failures due to races

CHANGES

@@ -1,4 +1,14 @@
 
+2.5-831 | 2018-08-10 17:12:53 -0500
+
+  * Immediately apply broker subscriptions made during bro_init()
+    (Jon Siwek, Corelight)
+
+  * Update default broker threading configuration to use 4 threads and allow
+    tuning via BRO_BROKER_MAX_THREADS env. variable (Jon Siwek, Corelight)
+
+  * Misc. unit test improvements (Jon Siwek, Corelight)
+
 2.5-826 | 2018-08-08 13:09:27 -0700
 
   * Add support for code coverage statistics for bro source files after running btest

VERSION

@@ -1 +1 @@
-2.5-826
+2.5-831

aux/broker

@@ -1 +1 @@
-Subproject commit 3ab9b647eedf6be674067198551d57b2b17e86d1
+Subproject commit e0f9f6504db9285a48e0be490abddf959999a404

scripts/base/frameworks/broker/main.bro

@@ -261,7 +261,8 @@ export {
 	global publish_id: function(topic: string, id: string): bool;
 
 	## Register interest in all peer event messages that use a certain topic
-	## prefix.
+	## prefix.  Note that subscriptions may not be altered immediately after
+	## calling (except during :bro:see:`bro_init`).
 	##
 	## topic_prefix: a prefix to match against remote message topics.
 	##               e.g. an empty prefix matches everything and "a" matches
@@ -271,6 +272,8 @@ export {
 	global subscribe: function(topic_prefix: string): bool;
 
 	## Unregister interest in all peer event messages that use a topic prefix.
+	## Note that subscriptions may not be altered immediately after calling
+	## (except during :bro:see:`bro_init`).
 	##
 	## topic_prefix: a prefix previously supplied to a successful call to
 	##               :bro:see:`Broker::subscribe`.

src/broker/Manager.cc

@@ -137,6 +137,7 @@ Manager::Manager(bool arg_reading_pcaps)
 	{
 	bound_port = 0;
 	reading_pcaps = arg_reading_pcaps;
+	after_bro_init = false;
 	peer_count = 0;
 	log_topic_func = nullptr;
 	vector_of_data_type = nullptr;
@@ -847,14 +848,14 @@ RecordVal* Manager::MakeEvent(val_list* args, Frame* frame)
 bool Manager::Subscribe(const string& topic_prefix)
 	{
 	DBG_LOG(DBG_BROKER, "Subscribing to topic prefix %s", topic_prefix.c_str());
-	bstate->subscriber.add_topic(topic_prefix);
+	bstate->subscriber.add_topic(topic_prefix, ! after_bro_init);
 	return true;
 	}
 
 bool Manager::Unsubscribe(const string& topic_prefix)
 	{
 	DBG_LOG(DBG_BROKER, "Unsubscribing from topic prefix %s", topic_prefix.c_str());
-	bstate->subscriber.remove_topic(topic_prefix);
+	bstate->subscriber.remove_topic(topic_prefix, ! after_bro_init);
 	return true;
 	}
 

src/broker/Manager.h

@@ -66,6 +66,9 @@ public:
 	 */
 	void InitPostScript();
 
+	void BroInitDone()
+		{ after_bro_init = true; }
+
 	/**
 	 * Shuts Broker down at termination.
 	 */
@@ -404,6 +407,7 @@ private:
 
 	uint16_t bound_port;
 	bool reading_pcaps;
+	bool after_bro_init;
 	int peer_count;
 
 	Func* log_topic_func;

src/main.cc

@@ -1182,6 +1182,7 @@ int main(int argc, char** argv)
 	// Drain the event queue here to support the protocols framework configuring DPM
 	mgr.Drain();
 
+	broker_mgr->BroInitDone();
 	analyzer_mgr->DumpDebug();
 
 	have_pending_timers = ! reading_traces && timer_mgr->Size() > 0;