Factored out Thread (and Slave, an explicitly signal-driven thread)

git-svn-id: http://svn.drobilla.net/lad/ingen@87 a436a847-0d15-0410-975c-d299462d15a1

1 files changed, 39 insertions, 109 deletions

diff --git a/src/libs/engine/QueuedEventSource.cpp b/src/libs/engine/QueuedEventSource.cpp
index d41b2201..8254f3c8 100644
--- a/src/libs/engine/QueuedEventSource.cpp
+++ b/src/libs/engine/QueuedEventSource.cpp
@@ -25,20 +25,15 @@ namespace Om {
 
 
 QueuedEventSource::QueuedEventSource(size_t size)
-: m_front(0),
-  m_back(0),
-  m_prepared_back(0),
-  m_size(size+1),
-  m_thread_exists(false),
-  m_prepare_thread_exit_flag(false),
-  m_semaphore(0)
+: _front(0),
+  _back(0),
+  _prepared_back(0),
+  _size(size+1),
+  _blocking_semaphore(0)
 {
-	m_events = (QueuedEvent**)calloc(m_size, sizeof(QueuedEvent*));
+	_events = (QueuedEvent**)calloc(_size, sizeof(QueuedEvent*));
 
-	pthread_mutex_init(&m_blocking_mutex, NULL);
-	pthread_cond_init(&m_blocking_cond, NULL);
-
-	mlock(m_events, m_size * sizeof(QueuedEvent*));
+	mlock(_events, _size * sizeof(QueuedEvent*));
 }
 
 
@@ -46,49 +41,7 @@ QueuedEventSource::~QueuedEventSource()
 {
 	stop();
 	
-	free(m_events);
-	pthread_mutex_destroy(&m_blocking_mutex);
-	pthread_cond_destroy(&m_blocking_cond);
-}
-
-
-/** Start the prepare thread.
- */
-void
-QueuedEventSource::start()
-{
-	if (m_thread_exists) {
-		cerr << "[QueuedEventSource] Thread already launched?" << endl;
-		return;
-	} else {
-		cout << "[QueuedEventSource] Launching thread." << endl;
-	}
-
-	m_prepare_thread_exit_flag = false;
-	
-	pthread_attr_t attr;
-	pthread_attr_init(&attr);
-	pthread_attr_setstacksize(&attr, 1500000);
-	
-	pthread_create(&m_prepare_thread, &attr, &QueuedEventSource::prepare_loop, this);
-	pthread_attr_destroy(&attr);
-
-	m_thread_exists = true;
-}
-
-
-/** Destroy the prepare thread.
- */
-void
-QueuedEventSource::stop()
-{
-	if (m_thread_exists) {
-		m_prepare_thread_exit_flag = true;
-		pthread_cancel(m_prepare_thread);
-		pthread_join(m_prepare_thread, NULL);
-		m_thread_exists = false;
-		cout << "[QueuedEventSource] Stopped thread." << endl;
-	}
+	free(_events);
 }
 
 
@@ -99,13 +52,13 @@ QueuedEventSource::push(QueuedEvent* const ev)
 {
 	assert(!ev->is_prepared());
 
-	if (m_events[m_back] != NULL) {
+	if (_events[_back] != NULL) {
 		cerr << "[QueuedEventSource] Error: Queue is full!  Event is lost, please report!" << endl;
 		delete ev;
 	} else {
-		m_events[m_back] = ev;
-		m_back = (m_back + 1) % m_size;
-		m_semaphore.post();
+		_events[_back] = ev;
+		_back = (_back + 1) % _size;
+		signal();
 	}
 }
 
@@ -115,18 +68,18 @@ QueuedEventSource::push(QueuedEvent* const ev)
  * This method will only pop events that have been prepared, and are
  * stamped before the time passed.  In other words, it may return NULL
  * even if there are events pending in the queue.  The events returned are
- * actually QueuedEvent*s, but after this they are "normal" events and the
+ * actually QueuedEvents, but after this they are "normal" events and the
  * engine deals with them just like a realtime in-band event.
  */
 Event*
-QueuedEventSource::pop_earliest_event_before(const samplecount time)
+QueuedEventSource::pop_earliest_before(const samplecount time)
 {
-	QueuedEvent* front_event = m_events[m_front];
+	QueuedEvent* const front_event = _events[_front];
 	
 	// Pop
-	if (front_event != NULL && front_event->time_stamp() < time && front_event->is_prepared()) {
-		m_events[m_front] = NULL;
-		m_front = (m_front + 1) % m_size;
+	if (front_event && front_event->is_prepared() && front_event->time_stamp() < time) {
+		_events[_front] = NULL;
+		_front = (_front + 1) % _size;
 		return front_event;
 	} else {
 		return NULL;
@@ -137,63 +90,40 @@ QueuedEventSource::pop_earliest_event_before(const samplecount time)
 // Private //
 
 
-
 /** Signal that the blocking event is finished.
  *
- * When this is called preparing will resume.  This will be called by
+ * When this is called preparing will resume.  This MUST be called by
  * blocking events in their post_process() method.
  */
 void
 QueuedEventSource::unblock()
 {
-	/* FIXME: Make this a semaphore, and have events signal at the end of their
-	 * execute() methods so the preprocessor can start preparing events immediately
-	 * instead of waiting for the postprocessor to get around to finalizing the event? */
-	pthread_mutex_lock(&m_blocking_mutex);
-	pthread_cond_signal(&m_blocking_cond);
-	pthread_mutex_unlock(&m_blocking_mutex);
+	_blocking_semaphore.post();
 }
 
 
-void*
-QueuedEventSource::m_prepare_loop()
+void
+QueuedEventSource::_signalled()
 {
-	QueuedEvent* ev = NULL;
-
-	while (true) {
-		m_semaphore.wait();
-		
-		if (m_prepare_thread_exit_flag)
-			break; // exit signalled
-
-		ev = m_events[m_prepared_back];
-		assert(ev != NULL);
-		
-		if (ev == NULL) {
-			cerr << "[QueuedEventSource] ERROR: Signalled, but event is NULL." << endl;
-			continue;
-		}
-
-		assert(ev != NULL);
-		assert(!ev->is_prepared());
-
-		if (ev->is_blocking())
-			pthread_mutex_lock(&m_blocking_mutex);
-		
-		ev->pre_process();
-		
-		m_prepared_back = (m_prepared_back+1) % m_size;
-		
-		// If a blocking event, wait for event to finish passing through
-		// the audio cycle before preparing the next event
-		if (ev->is_blocking()) {
-			pthread_cond_wait(&m_blocking_cond, &m_blocking_mutex);
-			pthread_mutex_unlock(&m_blocking_mutex);
-		}
+	QueuedEvent* const ev = _events[_prepared_back];
+	assert(ev != NULL);
+
+	if (ev == NULL) {
+		cerr << "[QueuedEventSource] ERROR: Signalled, but event is NULL." << endl;
+		return;
 	}
 
-	cout << "[QueuedEventSource] Exiting slow event queue thread." << endl;
-	return NULL;
+	assert(ev != NULL);
+	assert(!ev->is_prepared());
+
+	ev->pre_process();
+
+	_prepared_back = (_prepared_back+1) % _size;
+
+	// If event was blocking, wait for event to being run through the
+	// process thread before preparing the next event
+	if (ev->is_blocking())
+		_blocking_semaphore.wait();
 }