Merge remote-tracking branch 'origin/master' into topic/bernhard/input-threads

Conflicts:
	src/CMakeLists.txt
	testing/btest/Baseline/coverage.bare-load-baseline/canonified_loaded_scripts.log
	testing/btest/Baseline/coverage.default-load-baseline/canonified_loaded_scripts.log

src/threading/Manager.h

@@ -81,6 +81,12 @@ public:
 	 */
 	int NumThreads() const { return all_threads.size(); }
 
+	/** Manually triggers processing of any thread input. This can be useful
+	 *  if the main thread is waiting for a specific message from a child.
+	 *  Usually, though, one should avoid using it.
+	 */
+	void ForceProcessing() { Process(); }
+
 protected:
 	friend class BasicThread;
 	friend class MsgThread;

src/threading/MsgThread.cc

@@ -10,13 +10,21 @@ namespace threading  {
 
 ////// Messages.
 
-// Signals child thread to terminate. This is actually a no-op; its only
-// purpose is unblock the current read operation so that the child's Run()
-// methods can check the termination status.
-class TerminateMessage : public InputMessage<MsgThread>
+// Signals child thread to shutdown operation.
+class FinishMessage : public InputMessage<MsgThread>
 {
 public:
-	TerminateMessage(MsgThread* thread) : InputMessage<MsgThread>("Terminate", thread)	{ }
+	FinishMessage(MsgThread* thread) : InputMessage<MsgThread>("Finish", thread)	{ }
+
+	virtual bool Process()	{ return Object()->DoFinish(); }
+};
+
+// A dummy message that's only purpose is unblock the current read operation
+// so that the child's Run() methods can check the termination status.
+class UnblockMessage : public InputMessage<MsgThread>
+{
+public:
+	UnblockMessage(MsgThread* thread) : InputMessage<MsgThread>("Unblock", thread)	{ }
 
 	virtual bool Process()	{ return true; }
 };
@@ -130,13 +138,29 @@ bool ReporterMessage::Process()
 MsgThread::MsgThread() : BasicThread()
 	{
 	cnt_sent_in = cnt_sent_out = 0;
+	finished = false;
 	thread_mgr->AddMsgThread(this);
 	}
 
 void MsgThread::OnStop()
 	{
-	// This is to unblock the current queue read operation.
-	SendIn(new TerminateMessage(this), true);
+	// Signal thread to terminate and wait until it has acknowledged.
+	SendIn(new FinishMessage(this), true);
+
+	int cnt = 0;
+	while ( ! finished )
+		{
+		if ( ++cnt > 1000 ) // Insurance against broken threads ...
+			{
+			reporter->Warning("thread %s didn't finish in time", Name().c_str());
+			break;
+			}
+
+		usleep(1000);
+		}
+
+	// One more message to make sure the current queue read operation unblocks.
+	SendIn(new UnblockMessage(this), true);
 	}
 
 void MsgThread::Heartbeat()
@@ -157,6 +181,14 @@ bool MsgThread::DoHeartbeat(double network_time, double current_time)
 	return true;
 	}
 
+bool MsgThread::DoFinish()
+	{
+	// This is thread-safe "enough", we're the only one ever writing
+	// there.
+	finished = true;
+	return true;
+	}
+
 void MsgThread::Info(const char* msg)
 	{
 	SendOut(new ReporterMessage(ReporterMessage::INFO, this, msg));
@@ -189,7 +221,9 @@ void MsgThread::InternalWarning(const char* msg)
 
 void MsgThread::InternalError(const char* msg)
 	{
-	SendOut(new ReporterMessage(ReporterMessage::INTERNAL_ERROR, this, msg));
+	// This one aborts immediately.
+	fprintf(stderr, "internal error in thread: %s\n", msg);
+	abort();
 	}
 
 #ifdef DEBUG

src/threading/MsgThread.h

@@ -171,6 +171,8 @@ public:
 protected:
 	friend class Manager;
 	friend class HeartbeatMessage;
+	friend class FinishMessage;
+	friend class FinishedMessage;
 
 	/**
 	 * Pops a message sent by the child from the child-to-main queue.
@@ -215,6 +217,12 @@ protected:
 	 */
 	virtual bool DoHeartbeat(double network_time, double current_time);
 
+	/** Triggered for execution in the child thread just before shutting threads down.
+	 *  The child thread should finish its operations and then *must*
+	 *  call this class' implementation.
+	 */
+	virtual bool DoFinish();
+
 private:
 	/**
 	 * Pops a message sent by the main thread from the main-to-chold
@@ -276,6 +284,8 @@ private:
 
 	uint64_t cnt_sent_in;	// Counts message sent to child.
 	uint64_t cnt_sent_out;	// Counts message sent by child.
+
+	bool finished;	// Set to true by Finished message.
 };
 
 /**

src/threading/SerialTypes.cc

@@ -26,6 +26,20 @@ bool Field::Write(SerializationFormat* fmt) const
 			fmt->Write((int)subtype, "subtype"), fmt->Write(optional, "optional"));
 	}
 
+string Field::TypeName() const
+	{
+	string n = type_name(type);
+
+	if ( (type == TYPE_TABLE) || (type == TYPE_VECTOR) )
+		{
+		n += "[";
+		n += type_name(subtype);
+		n += "]";
+		}
+
+	return n;
+	}
+
 Value::~Value()
 	{
 	if ( (type == TYPE_ENUM || type == TYPE_STRING || type == TYPE_FILE || type == TYPE_FUNC)

src/threading/SerialTypes.h

@@ -56,6 +56,12 @@ struct Field {
 	 * @return False if an error occured.
 	 */
 	bool Write(SerializationFormat* fmt) const;
+
+	/**
+	 * Returns a textual description of the field's type. This method is
+	 * thread-safe.
+	 */
+	string TypeName() const;
 };
 
 /**
@@ -135,8 +141,8 @@ struct Value {
 
 	/**
 	 * Returns true if the type can be represented by a Value. If
-	 * `atomic_only` is true, will not permit composite types.
-	 */
+	 * `atomic_only` is true, will not permit composite types. This
+	 * method is thread-safe. */
 	static bool IsCompatibleType(BroType* t, bool atomic_only=false);
 
 private: