Use lists instead of ringbuffers for event queue - remove upper limit on event queue size and related warnings on big patch load.

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

5 files changed, 63 insertions, 111 deletions

diff --git a/src/engine/EventSource.hpp b/src/engine/EventSource.hpp
index 68532d16..6d03d544 100644
--- a/src/engine/EventSource.hpp
+++ b/src/engine/EventSource.hpp
@@ -25,6 +25,7 @@ namespace Ingen {
 class Event;
 class QueuedEvent;
 class PostProcessor;
+class ProcessContext;
 
 
 /** Source for events to run in the audio thread.
diff --git a/src/engine/PostProcessor.cpp b/src/engine/PostProcessor.cpp
index 5d731fbb..9a8b540a 100644
--- a/src/engine/PostProcessor.cpp
+++ b/src/engine/PostProcessor.cpp
@@ -57,14 +57,16 @@ PostProcessor::process()
 	}
 
 	/* Process normal events */
-	while ( ! _events.empty()) {
-		Event* const ev = _events.front();
-		if (ev->time() > end_time)
+	Raul::List<Event*>::Node* n = _events.head();
+	while (n) {
+		if (n->elem()->time() > end_time)
 			break;
-		_events.pop();
-		assert(ev);
-		ev->post_process();
-        delete ev;
+		Raul::List<Event*>::Node* next = n->next();
+		n->elem()->post_process();
+		_events.erase(_events.begin());
+		delete n->elem();
+		delete n;
+		n = next;
 	}
 }
 
diff --git a/src/engine/PostProcessor.hpp b/src/engine/PostProcessor.hpp
index fe767b60..8f04a75c 100644
--- a/src/engine/PostProcessor.hpp
+++ b/src/engine/PostProcessor.hpp
@@ -21,9 +21,7 @@
 #include <pthread.h>
 #include "types.hpp"
 #include "raul/SRSWQueue.hpp"
-//#include "raul/Slave.hpp"
-
-//namespace Raul { class Maid; }
+#include "raul/List.hpp"
 
 namespace Ingen {
 
@@ -42,13 +40,13 @@ class Engine;
  *
  * \ingroup engine
  */
-class PostProcessor //: public Raul::Slave
+class PostProcessor
 {
 public:
-	PostProcessor(Engine& engine, /*Raul::Maid& maid, */size_t queue_size);
+	PostProcessor(Engine& engine, size_t queue_size);
 
-	/** Push an event on to the process queue, realtime-safe, not thread-safe. */
-	inline void push(Event* const ev) { _events.push(ev); }
+	/** Push a list of events on to the process queue, realtime-safe, not thread-safe. */
+	inline void append(Raul::List<Event*>* l) { _events.append(*l); }
 
     /** Post-process and delete all pending events */
     void process();
@@ -57,14 +55,11 @@ public:
 	void set_end_time(FrameTime time) { _max_time = time; }
 
 private:
-	Engine&                 _engine;
-	Raul::AtomicInt         _max_time;
-	//Raul::Maid&             _maid;
-	Raul::SRSWQueue<Event*> _events;
-	uint32_t                _event_buffer_size;
-	uint8_t*                _event_buffer;
-
-	//virtual void            _whipped();
+	Engine&            _engine;
+	Raul::AtomicInt    _max_time;
+	Raul::List<Event*> _events;
+	uint32_t           _event_buffer_size;
+	uint8_t*           _event_buffer;
 };
 
 
diff --git a/src/engine/QueuedEventSource.cpp b/src/engine/QueuedEventSource.cpp
index e8c6deec..af9b270e 100644
--- a/src/engine/QueuedEventSource.cpp
+++ b/src/engine/QueuedEventSource.cpp
@@ -1,5 +1,5 @@
 /* This file is part of Ingen.
- * Copyright (C) 2007 Dave Robillard <http://drobilla.net>
+ * Copyright (C) 2008 Dave Robillard <http://drobilla.net>
  * 
  * Ingen is free software; you can redistribute it and/or modify it under the
  * terms of the GNU General Public License as published by the Free Software
@@ -29,20 +29,10 @@ namespace Ingen {
 
 
 QueuedEventSource::QueuedEventSource(size_t queue_size)
-	: _front(0)
-	, _back(0)
-	, _prepared_back(0)
-	, _size(queue_size+1)
-	, _blocking_semaphore(0)
-	, _full_semaphore(0)
+	: _blocking_semaphore(0)
 {
-	_events = (QueuedEvent**)calloc(_size, sizeof(QueuedEvent*));
-
-	mlock(_events, _size * sizeof(QueuedEvent*));
-
 	Thread::set_context(THREAD_PRE_PROCESS);
 	assert(context() == THREAD_PRE_PROCESS);
-
 	set_name("QueuedEventSource");
 }
 
@@ -50,8 +40,6 @@ QueuedEventSource::QueuedEventSource(size_t queue_size)
 QueuedEventSource::~QueuedEventSource()
 {
 	Thread::stop();
-	
-	free(_events);
 }
 
 
@@ -61,20 +49,11 @@ void
 QueuedEventSource::push_queued(QueuedEvent* const ev)
 {
 	assert(!ev->is_prepared());
+	Raul::List<Event*>::Node* node = new Raul::List<Event*>::Node(ev);
+	_events.push_back(node);
+	if (_prepared_back.get() == NULL)
+		_prepared_back = node;
 
-	unsigned back = _back.get();
-	bool full = (((_front.get() - back + _size) % _size) == 1);
-	while (full) {
-		whip();
-		cerr << "WARNING: Event queue full.  Waiting..." << endl;
-		_full_semaphore.wait();
-		back = _back.get();
-		full = (((_front.get() - back + _size) % _size) == 1);
-	}
-		
-	assert(_events[back] == NULL);
-	_events[back] = ev;
-	_back = (back + 1) % _size;
 	whip();
 }
 
@@ -87,53 +66,33 @@ void
 QueuedEventSource::process(PostProcessor& dest, ProcessContext& context)
 {
 	assert(ThreadManager::current_thread_id() == THREAD_PROCESS);
-
-	Event* ev = NULL;
+	
+	if (_events.empty())
+		return;
 
 	/* Limit the maximum number of queued events to process per cycle.  This
 	 * makes the process callback (more) realtime-safe by preventing being
 	 * choked by events coming in faster than they can be processed.
 	 * FIXME: test this and figure out a good value */
-	const unsigned int MAX_QUEUED_EVENTS = context.nframes() / 64;
-
-	unsigned int num_events_processed = 0;
+	const size_t MAX_QUEUED_EVENTS = context.nframes() / 32;
 	
-	while ((ev = pop_earliest_queued_before(context.end()))) {
+	size_t num_events_processed = 0;
+
+	QueuedEvent*              ev       = (QueuedEvent*)_events.front();
+	Raul::List<Event*>::Node* new_head = _events.head();
+
+	while (ev && ev->is_prepared() && ev->time() < context.end()) {
 		ev->execute(context);
-		dest.push(ev);
+		new_head = new_head->next();
 		if (++num_events_processed > MAX_QUEUED_EVENTS)
 			break;
+		ev = (new_head ? (QueuedEvent*)new_head->elem() : NULL);
 	}
-	
-	if (_full_semaphore.has_waiter()) {
-		const bool full = (((_front.get() - _back.get() + _size) % _size) == 1);
-		if (!full)
-			_full_semaphore.post();
-	}
-}
-
 
-/** Pop the prepared event at the front of the prepare queue, if it exists.
- *
- * This method will only pop events that are prepared and have time stamp
- * less than @a time. It may return NULL even if there are events pending in
- * the queue, if they are unprepared or stamped in the future.
- */
-Event*
-QueuedEventSource::pop_earliest_queued_before(const SampleCount time)
-{
-	assert(ThreadManager::current_thread_id() == THREAD_PROCESS);
-
-	const unsigned front = _front.get();
-	QueuedEvent* const front_event = _events[front];
-	
-	// Pop
-	if (front_event && front_event->is_prepared() && front_event->time() < time) {
-		_events[front] = NULL;
-		_front = (front + 1) % _size;
-		return front_event;
-	} else {
-		return NULL;
+	if (num_events_processed > 0) {
+		Raul::List<Event*> front;
+		_events.chop_front(front, num_events_processed, new_head);
+		dest.append(&front);
 	}
 }
 
@@ -145,8 +104,12 @@ QueuedEventSource::pop_earliest_queued_before(const SampleCount time)
 void
 QueuedEventSource::_whipped()
 {
-	const unsigned prepared_back = _prepared_back.get();
-	QueuedEvent* const ev = _events[prepared_back];
+	Raul::List<Event*>::Node* pb = _prepared_back.get();
+	if (!pb)
+		return;
+
+	QueuedEvent* const ev = (QueuedEvent*)pb->elem();
+	assert(ev);
 	if (!ev)
 		return;
 	
@@ -154,7 +117,8 @@ QueuedEventSource::_whipped()
 	ev->pre_process();
 	assert(ev->is_prepared());
 
-	_prepared_back = (prepared_back + 1) % _size;
+	assert(_prepared_back.get() == pb);
+	_prepared_back = pb->next();
 
 	// If event was blocking, wait for event to being run through the
 	// process thread before preparing the next event
diff --git a/src/engine/QueuedEventSource.hpp b/src/engine/QueuedEventSource.hpp
index a06de39f..69ac15fe 100644
--- a/src/engine/QueuedEventSource.hpp
+++ b/src/engine/QueuedEventSource.hpp
@@ -1,5 +1,5 @@
 /* This file is part of Ingen.
- * Copyright (C) 2007 Dave Robillard <http://drobilla.net>
+ * Copyright (C) 2008 Dave Robillard <http://drobilla.net>
  * 
  * Ingen is free software; you can redistribute it and/or modify it under the
  * terms of the GNU General Public License as published by the Free Software
@@ -15,23 +15,20 @@
  * 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA
  */
 
-#ifndef QUEUEDEVENTSOURCE_H
-#define QUEUEDEVENTSOURCE_H
+#ifndef INGEN_QUEUED_EVENT_SOURCE_HPP
+#define INGEN_QUEUED_EVENT_SOURCE_HPP
 
 #include <cstdlib>
 #include <pthread.h>
 #include "types.hpp"
 #include "raul/Semaphore.hpp"
-#include "raul/AtomicInt.hpp"
-#include "raul/SRSWQueue.hpp"
 #include "raul/Slave.hpp"
-#include "Event.hpp"
+#include "raul/List.hpp"
 #include "EventSource.hpp"
 
-using Raul::AtomicInt;
-
 namespace Ingen {
 
+class Event;
 class QueuedEvent;
 class PostProcessor;
 
@@ -56,34 +53,27 @@ public:
 	void deactivate() { Slave::stop(); }
 
 	void process(PostProcessor& dest, ProcessContext& context);
-
+	
 	/** Signal that the blocking event is finished.
 	 * When this is called preparing will resume.  This MUST be called by
 	 * blocking events in their post_process() method. */
 	inline void unblock() { _blocking_semaphore.post(); }
 
 protected:
-	void   push_queued(QueuedEvent* const ev);
-	Event* pop_earliest_queued_before(const SampleCount time);
+	void push_queued(QueuedEvent* const ev);
 
-	inline bool unprepared_events() { return (_prepared_back.get() != _back.get()); }
+	inline bool unprepared_events() { return (_prepared_back.get() != NULL); }
 	
 	virtual void _whipped(); ///< Prepare 1 event
 
 private:
-	// Note it's important which functions access which variables for thread safety
-	
-	// 2-part queue for events that require pre-processing: 
-	AtomicInt       _front;         ///< Front of queue
-	AtomicInt       _back;          ///< Back of entire queue (index of back event + 1)
-	AtomicInt       _prepared_back; ///< Back of prepared events (index of back prepared event + 1)
-	const size_t    _size;
-	QueuedEvent**   _events;
-	Raul::Semaphore _blocking_semaphore;
-	Raul::Semaphore _full_semaphore;
+	Raul::List<Event*>                        _events;
+	Raul::AtomicPtr<Raul::List<Event*>::Node> _prepared_back;
+	Raul::Semaphore                           _blocking_semaphore;
 };
 
 
 } // namespace Ingen
 
-#endif // QUEUEDEVENTSOURCE_H
+#endif // INGEN_QUEUED_EVENT_SOURCE_HPP
+