Merge remote-tracking branch 'origin/topic/johanna/threads'

Tweaked the new threading code a bit more.

* origin/topic/johanna/threads:
  Move threading to c++11 primitives (mostly).

CHANGES

@@ -1,4 +1,12 @@
 
+2.5-91 | 2017-03-03 10:30:12 -0800
+
+  * Move most threading to C++11 primitives (mostly). (Johanna Amann)
+
+  * Fix a test that sometimes fails on FreeBSD. (Daniel Thayer)
+
+  * Remove build time warnings. (Seth Hall)
+
 2.5-84 | 2017-02-27 15:08:55 -0500
 
   * Change semantics of Broker's remote logging to match old communication 

VERSION

@@ -1 +1 @@
-2.5-84
+2.5-91

src/input/readers/raw/Plugin.cc

@@ -8,7 +8,6 @@ using namespace plugin::Bro_RawReader;
 
 Plugin::Plugin()
 	{
-	init = false;
 	}
 
 plugin::Configuration Plugin::Configure()
@@ -23,21 +22,14 @@ plugin::Configuration Plugin::Configure()
 
 void Plugin::InitPreScript()
 	{
-	if ( pthread_mutex_init(&fork_mutex, 0) != 0 )
-		reporter->FatalError("cannot initialize raw reader's mutex");
-
-	init = true;
 	}
 
 void Plugin::Done()
 	{
-	pthread_mutex_destroy(&fork_mutex);
-	init = false;
 	}
 
-pthread_mutex_t* Plugin::ForkMutex()
+std::unique_lock<std::mutex> Plugin::ForkMutex()
 	{
-	assert(init);
+	return std::move(std::unique_lock<std::mutex>(fork_mutex, std::defer_lock));
-	return &fork_mutex;
 	}
 

src/input/readers/raw/Plugin.h

@@ -1,4 +1,6 @@
-// See the file  in the main distribution directory for copyright.
+// See the file in the main distribution directory for copyright.
+
+#include <mutex>
 
 #include "plugin/Plugin.h"
 
@@ -16,11 +18,10 @@ public:
 	virtual void InitPreScript();
 	virtual void Done();
 
-	pthread_mutex_t * ForkMutex();
+	std::unique_lock<std::mutex> ForkMutex();
 
 private:
-	bool init;
+	std::mutex fork_mutex;
-	pthread_mutex_t fork_mutex;
 
 };
 

src/input/readers/raw/Raw.cc

@@ -96,24 +96,21 @@ bool Raw::SetFDFlags(int fd, int cmd, int flags)
 	}
 
 
-bool Raw::LockForkMutex()
+std::unique_lock<std::mutex> Raw::AcquireForkMutex()
 	{
-	int res = pthread_mutex_lock(plugin::Bro_RawReader::plugin.ForkMutex());
+	auto lock = plugin::Bro_RawReader::plugin.ForkMutex();
-	if ( res == 0 )
-		return true;
 
-	Error(Fmt("cannot lock fork mutex: %d", res));
+	try
-	return false;
+		{
-	}
+		lock.lock();
+		}
 
-bool Raw::UnlockForkMutex()
+	catch ( const std::system_error& e )
-	{
+		{
-	int res = pthread_mutex_unlock(plugin::Bro_RawReader::plugin.ForkMutex());
+		reporter->FatalErrorWithCore("cannot lock fork mutex: %s", e.what());
-	if ( res == 0 )
+		}
-		return true;
 
-	Error(Fmt("cannot unlock fork mutex: %d", res));
+	return lock;
-	return false;
 	}
 
 bool Raw::Execute()
@@ -126,12 +123,10 @@ bool Raw::Execute()
 	// never crops up... ("never" meaning I haven't seen in it in
 	// hundreds of tests using 50+ threads where before I'd see the issue
 	// w/ just 2 threads ~33% of the time).
-	if ( ! LockForkMutex() )
+	auto lock = AcquireForkMutex();
-		return false;
 
 	if ( pipe(pipes) != 0 || pipe(pipes+2) || pipe(pipes+4) )
 		{
-		UnlockForkMutex();
 		Error(Fmt("Could not open pipe: %d", errno));
 		return false;
 		}
@@ -139,7 +134,6 @@ bool Raw::Execute()
 	childpid = fork();
 	if ( childpid < 0 )
 		{
-		UnlockForkMutex();
 		Error(Fmt("Could not create child process: %d", errno));
 		return false;
 		}
@@ -208,8 +202,7 @@ bool Raw::Execute()
 				}
 			}
 
-		if ( ! UnlockForkMutex() )
+		lock.unlock();
-			return false;
 
 		ClosePipeEnd(stdout_out);
 

src/input/readers/raw/Raw.h

@@ -4,8 +4,8 @@
 #define INPUT_READERS_RAW_H
 
 #include <vector>
-#include <pthread.h>
 #include <memory>
+#include <mutex>
 
 #include "input/ReaderBackend.h"
 
@@ -37,8 +37,7 @@ protected:
 private:
 	void ClosePipeEnd(int i);
 	bool SetFDFlags(int fd, int cmd, int flags);
-	bool LockForkMutex();
+	std::unique_lock<std::mutex> AcquireForkMutex();
-	bool UnlockForkMutex();
 
 	bool OpenInput();
 	bool CloseInput();
@@ -87,7 +86,6 @@ private:
 	};
 
 	static const int block_size;
-	static pthread_mutex_t fork_mutex;
 };
 
 }

src/threading/BasicThread.cc

@@ -5,6 +5,7 @@
 #include "bro-config.h"
 #include "BasicThread.h"
 #include "Manager.h"
+#include "pthread.h"
 
 #ifdef HAVE_LINUX
 #include <sys/prctl.h>
@@ -21,7 +22,6 @@ BasicThread::BasicThread()
 	started = false;
 	terminating = false;
 	killed = false;
-	pthread = 0;
 
 	buf_len = STD_FMT_BUF_LEN;
 	buf = (char*) safe_malloc(buf_len);
@@ -50,6 +50,7 @@ void BasicThread::SetName(const char* arg_name)
 
 void BasicThread::SetOSName(const char* arg_name)
 	{
+	static_assert(std::is_same<std::thread::native_handle_type, pthread_t>::value, "libstdc++ doesn't use pthread_t");
 
 #ifdef HAVE_LINUX
 	prctl(PR_SET_NAME, arg_name, 0, 0, 0);
@@ -60,7 +61,7 @@ void BasicThread::SetOSName(const char* arg_name)
 #endif
 
 #ifdef FREEBSD
-	pthread_set_name_np(pthread_self(), arg_name, arg_name);
+	pthread_set_name_np(thread.native_handle(), arg_name, arg_name);
 #endif
 	}
 
@@ -108,9 +109,7 @@ void BasicThread::Start()
 
 	started = true;
 
-	int err = pthread_create(&pthread, 0, BasicThread::launcher, this);
+	thread = std::thread(&BasicThread::launcher, this);
-	if ( err != 0 )
-		reporter->FatalError("Cannot create thread %s: %s", name, Strerror(err));
 
 	DBG_LOG(DBG_THREADING, "Started thread %s", name);
 
@@ -147,17 +146,21 @@ void BasicThread::Join()
 	if ( ! started )
 		return;
 
-	if ( ! pthread )
+	if ( ! thread.joinable() )
 		return;
 
 	assert(terminating);
 
-	if ( pthread_join(pthread, 0) != 0  )
+	try
-		reporter->FatalError("Failure joining thread %s", name);
+		{
+		thread.join();
+		}
+	catch ( const std::system_error& e )
+		{
+		reporter->FatalError("Failure joining thread %s with error %s", name, e.what());
+		}
 
 	DBG_LOG(DBG_THREADING, "Joined with thread %s", name);
-
-	pthread = 0;
 	}
 
 void BasicThread::Kill()
@@ -180,6 +183,7 @@ void BasicThread::Done()
 
 void* BasicThread::launcher(void *arg)
 	{
+	static_assert(std::is_same<std::thread::native_handle_type, pthread_t>::value, "libstdc++ doesn't use pthread_t");
 	BasicThread* thread = (BasicThread *)arg;
 
 	// Block signals in thread. We handle signals only in the main

src/threading/BasicThread.h

@@ -2,8 +2,7 @@
 #ifndef THREADING_BASICTHREAD_H
 #define THREADING_BASICTHREAD_H
 
-#include <pthread.h>
+#include <thread>
-#include <semaphore.h>
 
 #include "util.h"
 
@@ -35,6 +34,9 @@ public:
 	 */
 	BasicThread();
 
+	BasicThread(BasicThread const&) = delete;
+	BasicThread& operator =(BasicThread const&) = delete;
+
 	/**
 	 * Returns a descriptive name for the thread. If not set via
 	 * SetName(). If not set, a default name is choosen automatically.
@@ -192,11 +194,11 @@ protected:
 	void Done();
 
 private:
-	// pthread entry function.
+	// thread entry function.
 	static void* launcher(void *arg);
 
 	const char* name;
-	pthread_t pthread;
+	std::thread thread;
 	bool started; 		// Set to to true once running.
 	bool terminating;	// Set to to true to signal termination.
 	bool killed;	// Set to true once forcefully killed.

src/threading/MsgThread.h

@@ -2,8 +2,6 @@
 #ifndef THREADING_MSGTHREAD_H
 #define THREADING_MSGTHREAD_H
 
-#include <pthread.h>
-
 #include "DebugLogger.h"
 
 #include "BasicThread.h"

src/threading/Queue.h

@@ -1,7 +1,8 @@
 #ifndef THREADING_QUEUE_H
 #define THREADING_QUEUE_H
 
-#include <pthread.h>
+#include <mutex>
+#include <condition_variable>
 #include <queue>
 #include <deque>
 #include <stdint.h>
@@ -22,7 +23,7 @@ namespace threading {
  *
  * All Queue instances must be instantiated by Bro's main thread.
  *
- * TODO: Unclear how critical performance is for this qeueue. We could like;y
+ * TODO: Unclear how critical performance is for this qeueue. We could likely
  * optimize it further if helpful.
  */
 template<typename T>
@@ -71,9 +72,10 @@ public:
 	 */
 	bool MaybeReady() { return (num_reads != num_writes); }
 
-	/** Wake up the reader if it's currently blocked for input. This is
+	/**
-	 primarily to give it a chance to check termination quickly.
+	 * Wake up the reader if it's currently blocked for input. This is
-	**/
+	 * primarily to give it a chance to check termination quickly.
+	 */
 	void WakeUp();
 
 	/**
@@ -94,14 +96,17 @@ public:
 	 * Returns statistics about the queue's usage.
 	 *
 	 * @param stats A pointer to a structure that will be filled with
-	 * current numbers. */
+	 * current numbers.
+	 */
 	void GetStats(Stats* stats);
 
 private:
 	static const int NUM_QUEUES = 8;
 
-	pthread_mutex_t mutex[NUM_QUEUES];	// Mutex protected shared accesses.
+	std::vector<std::unique_lock<std::mutex>> LocksForAllQueues();
-	pthread_cond_t has_data[NUM_QUEUES];	// Signals when data becomes available
+
+	std::mutex mutex[NUM_QUEUES];	// Mutex protected shared accesses.
+	std::condition_variable has_data[NUM_QUEUES];	// Signals when data becomes available
 	std::queue<T> messages[NUM_QUEUES];	// Actually holds the queued messages
 
 	int read_ptr;	// Where the next operation will read from
@@ -115,17 +120,18 @@ private:
 	uint64_t num_writes;
 };
 
-inline static void safe_lock(pthread_mutex_t* mutex)
+inline static std::unique_lock<std::mutex> acquire_lock(std::mutex& m)
 	{
-	int res = pthread_mutex_lock(mutex);
+	try
-	if ( res != 0 )
+		{
-		reporter->FatalErrorWithCore("cannot lock mutex: %d(%s)", res, strerror(res));
+		return std::move(std::unique_lock<std::mutex>(m));
-	}
+		}
-
+	catch ( const std::system_error& e )
-inline static void safe_unlock(pthread_mutex_t* mutex)
+		{
-	{
+		reporter->FatalErrorWithCore("cannot lock mutex: %s", e.what());
-	if ( pthread_mutex_unlock(mutex) != 0 )
+		// Never gets here.
-		reporter->FatalErrorWithCore("cannot unlock mutex");
+		throw std::exception();
+		}
 	}
 
 template<typename T>
@@ -136,50 +142,28 @@ inline Queue<T>::Queue(BasicThread* arg_reader, BasicThread* arg_writer)
 	num_reads = num_writes = 0;
 	reader = arg_reader;
 	writer = arg_writer;
-
-	for( int i = 0; i < NUM_QUEUES; ++i )
-		{
-		if ( pthread_cond_init(&has_data[i], 0) != 0 )
-			reporter->FatalError("cannot init queue condition variable");
-
-		if ( pthread_mutex_init(&mutex[i], 0) != 0 )
-			reporter->FatalError("cannot init queue mutex");
-		}
 	}
 
 template<typename T>
 inline Queue<T>::~Queue()
 	{
-	for( int i = 0; i < NUM_QUEUES; ++i )
-		{
-		pthread_cond_destroy(&has_data[i]);
-		pthread_mutex_destroy(&mutex[i]);
-		}
 	}
 
 template<typename T>
 inline T Queue<T>::Get()
 	{
-	safe_lock(&mutex[read_ptr]);
+	auto lock = acquire_lock(mutex[read_ptr]);
 
 	int old_read_ptr = read_ptr;
 
 	if ( messages[read_ptr].empty() && ! ((reader && reader->Killed()) || (writer && writer->Killed())) )
 		{
-		struct timespec ts;
+		if ( has_data[read_ptr].wait_for(lock, std::chrono::seconds(5)) == std::cv_status::timeout )
-		ts.tv_sec = time(0) + 5;
+			return nullptr;
-		ts.tv_nsec = 0;
-
-		pthread_cond_timedwait(&has_data[read_ptr], &mutex[read_ptr], &ts);
-		safe_unlock(&mutex[read_ptr]);
-		return 0;
 		}
 
-	else if ( messages[read_ptr].empty() )
+	if ( messages[read_ptr].empty() )
-		{
+		return nullptr;
-		safe_unlock(&mutex[read_ptr]);
-		return 0;
-		}
 
 	T data = messages[read_ptr].front();
 	messages[read_ptr].pop();
@@ -187,15 +171,13 @@ inline T Queue<T>::Get()
 	read_ptr = (read_ptr + 1) % NUM_QUEUES;
 	++num_reads;
 
-	safe_unlock(&mutex[old_read_ptr]);
-
 	return data;
 	}
 
 template<typename T>
 inline void Queue<T>::Put(T data)
 	{
-	safe_lock(&mutex[write_ptr]);
+	auto lock = acquire_lock(mutex[write_ptr]);
 
 	int old_write_ptr = write_ptr;
 
@@ -203,43 +185,59 @@ inline void Queue<T>::Put(T data)
 
 	messages[write_ptr].push(data);
 
-	if ( need_signal )
-		pthread_cond_signal(&has_data[write_ptr]);
-
 	write_ptr = (write_ptr + 1) % NUM_QUEUES;
 	++num_writes;
 
-	safe_unlock(&mutex[old_write_ptr]);
+	if ( need_signal )
+		{
+		lock.unlock();
+		has_data[old_write_ptr].notify_one();
+		}
 	}
 
 
 template<typename T>
 inline bool Queue<T>::Ready()
 	{
-	safe_lock(&mutex[read_ptr]);
+	auto lock = acquire_lock(mutex[read_ptr]);
 
 	bool ret = (messages[read_ptr].size());
 
-	safe_unlock(&mutex[read_ptr]);
-
 	return ret;
 	}
 
+template<typename T>
+inline std::vector<std::unique_lock<std::mutex>> Queue<T>::LocksForAllQueues()
+	{
+	std::vector<std::unique_lock<std::mutex>> locks;
+
+	try
+		{
+		for ( int i = 0; i < NUM_QUEUES; i++ )
+			locks.emplace_back(std::unique_lock<std::mutex>(mutex[i]));
+		}
+
+	catch ( const std::system_error& e )
+		{
+		reporter->FatalErrorWithCore("cannot lock all mutexes: %s", e.what());
+		// Never gets here.
+		throw std::exception();
+		}
+
+	return std::move(locks);
+	}
+
 template<typename T>
 inline uint64_t Queue<T>::Size()
 	{
 	// Need to lock all queues.
-	for ( int i = 0; i < NUM_QUEUES; i++ )
+	auto locks = LocksForAllQueues();
-		safe_lock(&mutex[i]);
 
 	uint64_t size = 0;
 
 	for ( int i = 0; i < NUM_QUEUES; i++ )
 		size += messages[i].size();
 
-	for ( int i = 0; i < NUM_QUEUES; i++ )
-		safe_unlock(&mutex[i]);
-
 	return size;
 	}
 
@@ -248,29 +246,19 @@ inline void Queue<T>::GetStats(Stats* stats)
 	{
 	// To be safe, we look all queues. That's probably unneccessary, but
 	// doesn't really hurt.
-	for ( int i = 0; i < NUM_QUEUES; i++ )
+	auto locks = LocksForAllQueues();
-		safe_lock(&mutex[i]);
 
 	stats->num_reads = num_reads;
 	stats->num_writes = num_writes;
-
-	for ( int i = 0; i < NUM_QUEUES; i++ )
-		safe_unlock(&mutex[i]);
 	}
 
 template<typename T>
 inline void Queue<T>::WakeUp()
 	{
 	for ( int i = 0; i < NUM_QUEUES; i++ )
-		{
+		has_data[i].notify_all();
-		safe_lock(&mutex[i]);
-		pthread_cond_signal(&has_data[i]);
-		safe_unlock(&mutex[i]);
-		}
 	}
 
 }
 
-
 #endif
-