From 7e013dc6986fa9d6dc8616d494d9de5d192c4c69 Mon Sep 17 00:00:00 2001 From: David Robillard Date: Wed, 12 Jul 2006 06:34:30 +0000 Subject: 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 --- src/libs/engine/QueuedEventSource.cpp | 148 +++++++++------------------------- 1 file changed, 39 insertions(+), 109 deletions(-) (limited to 'src/libs/engine/QueuedEventSource.cpp') 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(); } -- cgit v1.2.1