Fix race condition resulting in duplicate connections if several identical requests come in rapid succession.

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

8 files changed, 78 insertions, 29 deletions

diff --git a/src/libs/engine/ConnectionImpl.cpp b/src/libs/engine/ConnectionImpl.cpp
index 998b7eb3..a6f573a8 100644
--- a/src/libs/engine/ConnectionImpl.cpp
+++ b/src/libs/engine/ConnectionImpl.cpp
@@ -24,6 +24,9 @@
 #include "AudioBuffer.hpp"
 #include "ProcessContext.hpp"
 
+#include <iostream>
+using namespace std;
+
 namespace Ingen {
 
 
diff --git a/src/libs/engine/InputPort.cpp b/src/libs/engine/InputPort.cpp
index 8aa73e2c..b778579d 100644
--- a/src/libs/engine/InputPort.cpp
+++ b/src/libs/engine/InputPort.cpp
@@ -55,6 +55,19 @@ InputPort::set_buffer_size(size_t size)
 	
 }
 
+	
+bool
+InputPort::apply_poly(Raul::Maid& maid, uint32_t poly)
+{
+	if (!_polyphonic || !_parent->polyphonic())
+		return true;
+	
+	for (Connections::iterator c = _connections.begin(); c != _connections.end(); ++c)
+		((ConnectionImpl*)c->get())->apply_poly(maid, poly);
+
+	return PortImpl::apply_poly(maid, poly);
+}
+
 
 /** Add a connection.  Realtime safe.
  *
@@ -138,7 +151,7 @@ InputPort::remove_connection(const OutputPort* src_port)
 
 /** Returns whether this port is connected to the passed port.
  */
-bool
+/*bool
 InputPort::is_connected_to(const OutputPort* port) const
 {
 	for (Connections::const_iterator i = _connections.begin(); i != _connections.end(); ++i)
@@ -146,7 +159,7 @@ InputPort::is_connected_to(const OutputPort* port) const
 			return true;
 	
 	return false;
-}
+}*/
 
 
 /** Prepare buffer for access, mixing if necessary.  Realtime safe.
diff --git a/src/libs/engine/InputPort.hpp b/src/libs/engine/InputPort.hpp
index c3fa3dad..ffba2a30 100644
--- a/src/libs/engine/InputPort.hpp
+++ b/src/libs/engine/InputPort.hpp
@@ -63,12 +63,14 @@ public:
 	Connections::Node* remove_connection(const OutputPort* src_port);
 
 	const Connections& connections() { return _connections; }
+	
+	bool apply_poly(Raul::Maid& maid, uint32_t poly);
 
 	void pre_process(ProcessContext& context);
 	void post_process(ProcessContext& context);
 	
 	bool is_connected() const { return (_connections.size() > 0); }
-	bool is_connected_to(const OutputPort* port) const;
+	//bool is_connected_to(const OutputPort* port) const;
 	
 	bool is_input()  const { return true; }
 	bool is_output() const { return false; }
diff --git a/src/libs/engine/PatchImpl.cpp b/src/libs/engine/PatchImpl.cpp
index 1625d1c3..2d33dd0d 100644
--- a/src/libs/engine/PatchImpl.cpp
+++ b/src/libs/engine/PatchImpl.cpp
@@ -289,12 +289,12 @@ PatchImpl::remove_node(const string& name)
 
 
 /** Remove a connection.
- * Process thread only.
+ * Preprocessing thread only.
  */
 PatchImpl::Connections::Node*
 PatchImpl::remove_connection(const PortImpl* src_port, const PortImpl* dst_port)
 {
-	assert(ThreadManager::current_thread_id() == THREAD_PROCESS);
+	assert(ThreadManager::current_thread_id() == THREAD_PRE_PROCESS);
 	bool found = false;
 	Connections::Node* connection = NULL;
 	for (Connections::iterator i = _connections.begin(); i != _connections.end(); ++i) {
@@ -311,6 +311,20 @@ PatchImpl::remove_connection(const PortImpl* src_port, const PortImpl* dst_port)
 
 	return connection;
 }
+	
+
+bool
+PatchImpl::has_connection(const PortImpl* src_port, const PortImpl* dst_port) const
+{
+	// FIXME: Doesn't scale
+	for (Connections::const_iterator i = _connections.begin(); i != _connections.end(); ++i) {
+		ConnectionImpl* const c = (ConnectionImpl*)i->get();
+		if (c->src_port() == src_port && c->dst_port() == dst_port)
+			return true;
+	}
+
+	return false;
+}
 
 
 uint32_t
diff --git a/src/libs/engine/PatchImpl.hpp b/src/libs/engine/PatchImpl.hpp
index 209062be..3629f6e5 100644
--- a/src/libs/engine/PatchImpl.hpp
+++ b/src/libs/engine/PatchImpl.hpp
@@ -110,6 +110,8 @@ public:
 	void               add_connection(Connections::Node* c) { _connections.push_back(c); }
 	Connections::Node* remove_connection(const PortImpl* src_port, const PortImpl* dst_port);
 	
+	bool has_connection(const PortImpl* src_port, const PortImpl* dst_port) const;
+	
 	CompiledPatch* compiled_patch()                  { return _compiled_patch; }
 	void           compiled_patch(CompiledPatch* cp) { _compiled_patch = cp; }
 	
@@ -134,7 +136,7 @@ private:
 	Engine&         _engine;
 	uint32_t        _internal_poly;
 	CompiledPatch*  _compiled_patch; ///< Accessed in audio thread only
-	Connections     _connections;    ///< Accessed in audio thread only
+	Connections     _connections;    ///< Accessed in preprocessing thread only
 	List<PortImpl*> _input_ports;    ///< Accessed in preprocessing thread only
 	List<PortImpl*> _output_ports;   ///< Accessed in preprocessing thread only
 	Nodes           _nodes;          ///< Accessed in preprocessing thread only
diff --git a/src/libs/engine/events/ConnectionEvent.cpp b/src/libs/engine/events/ConnectionEvent.cpp
index 893dc9cb..ac04d1cc 100644
--- a/src/libs/engine/events/ConnectionEvent.cpp
+++ b/src/libs/engine/events/ConnectionEvent.cpp
@@ -88,12 +88,6 @@ ConnectionEvent::pre_process()
 		return;
 	}
 
-	if (_dst_input_port->is_connected_to(_src_output_port)) {
-		_error = ALREADY_CONNECTED;
-		QueuedEvent::pre_process();
-		return;
-	}
-
 	NodeImpl* const src_node = _src_port->parent_node();
 	NodeImpl* const dst_node = _dst_port->parent_node();
 
@@ -116,6 +110,13 @@ ConnectionEvent::pre_process()
 	}
 
 	assert(_patch);
+	
+	//if (_dst_input_port->is_connected_to(_src_output_port)) {
+	if (_patch->has_connection(_src_output_port, _dst_input_port)) {
+		_error = ALREADY_CONNECTED;
+		QueuedEvent::pre_process();
+		return;
+	}
 
 	if (src_node == NULL || dst_node == NULL) {
 		_error = PARENTS_NOT_FOUND;
@@ -139,6 +140,8 @@ ConnectionEvent::pre_process()
 		dst_node->providers()->push_back(new Raul::List<NodeImpl*>::Node(src_node));
 		src_node->dependants()->push_back(new Raul::List<NodeImpl*>::Node(dst_node));
 	}
+		
+	_patch->add_connection(_patch_listnode);
 
 	if (_patch->enabled())
 		_compiled_patch = _patch->compile();
@@ -153,9 +156,8 @@ ConnectionEvent::execute(ProcessContext& context)
 	QueuedEvent::execute(context);
 
 	if (_error == NO_ERROR) {
-		// These must be inserted here, since they're actually used by the audio thread
+		// This must be inserted here, since they're actually used by the audio thread
 		_dst_input_port->add_connection(_port_listnode);
-		_patch->add_connection(_patch_listnode);
 		if (_patch->compiled_patch() != NULL)
 			_engine.maid()->push(_patch->compiled_patch());
 		_patch->compiled_patch(_compiled_patch);
diff --git a/src/libs/engine/events/DisconnectionEvent.cpp b/src/libs/engine/events/DisconnectionEvent.cpp
index e85b4fcf..03110f9e 100644
--- a/src/libs/engine/events/DisconnectionEvent.cpp
+++ b/src/libs/engine/events/DisconnectionEvent.cpp
@@ -44,6 +44,7 @@ DisconnectionEvent::DisconnectionEvent(Engine& engine, SharedPtr<Responder> resp
   _src_port(NULL),
   _dst_port(NULL),
   _lookup(true),
+  _patch_connection(NULL),
   _compiled_patch(NULL),
   _error(NO_ERROR)
 {
@@ -97,12 +98,6 @@ DisconnectionEvent::pre_process()
 	assert(_src_output_port);
 	assert(_dst_input_port);
 
-	if (!_dst_input_port->is_connected_to(_src_output_port)) {
-		_error = NOT_CONNECTED;
-		QueuedEvent::pre_process();
-		return;
-	}
-	
 	NodeImpl* const src_node = _src_port->parent_node();
 	NodeImpl* const dst_node = _dst_port->parent_node();
 
@@ -125,6 +120,13 @@ DisconnectionEvent::pre_process()
 	}
 
 	assert(_patch);
+	
+	//if (_dst_input_port->is_connected_to(_src_output_port)) {
+	if (!_patch->has_connection(_src_output_port, _dst_input_port)) {
+		_error = NOT_CONNECTED;
+		QueuedEvent::pre_process();
+		return;
+	}
 
 	if (src_node == NULL || dst_node == NULL) {
 		_error = PARENTS_NOT_FOUND;
@@ -143,6 +145,8 @@ DisconnectionEvent::pre_process()
 			delete src_node->dependants()->erase(i);
 			break;
 		}
+			
+	_patch_connection = _patch->remove_connection(_src_port, _dst_port);
 	
 	if (_patch->enabled())
 		_compiled_patch = _patch->compile();
@@ -161,15 +165,23 @@ DisconnectionEvent::execute(ProcessContext& context)
 			= _dst_input_port->remove_connection(_src_output_port);
 		
 		if (port_connection != NULL) {
-			PatchImpl::Connections::Node* const patch_connection
-				= _patch->remove_connection(_src_port, _dst_port);
-			
-			assert(patch_connection);
-			assert(port_connection->elem() == patch_connection->elem());
+			assert(_patch_connection);
+
+			if (port_connection->elem() != _patch_connection->elem()) {
+				cerr << "ERROR: Corrupt connections:" << endl;
+				cerr << "\t" << port_connection->elem() << ": "
+					<< port_connection->elem()->src_port_path()
+					<< " -> " << port_connection->elem()->dst_port_path() << endl
+					<< "!=" << endl
+					<< "\t" << _patch_connection->elem() << ": "
+					<< _patch_connection->elem()->src_port_path()
+					<< " -> " << _patch_connection->elem()->dst_port_path() << endl;
+			}
+			assert(port_connection->elem() == _patch_connection->elem());
 			
 			// Destroy list node, which will drop reference to connection itself
 			_engine.maid()->push(port_connection);
-			_engine.maid()->push(patch_connection);
+			_engine.maid()->push(_patch_connection);
 	
 			if (_patch->compiled_patch() != NULL)
 				_engine.maid()->push(_patch->compiled_patch());
diff --git a/src/libs/engine/events/DisconnectionEvent.hpp b/src/libs/engine/events/DisconnectionEvent.hpp
index ea4dbbda..700febeb 100644
--- a/src/libs/engine/events/DisconnectionEvent.hpp
+++ b/src/libs/engine/events/DisconnectionEvent.hpp
@@ -22,6 +22,7 @@
 #include <raul/Path.hpp>
 #include "QueuedEvent.hpp"
 #include "types.hpp"
+#include "PatchImpl.hpp"
 using std::string;
 
 namespace Raul {
@@ -31,7 +32,6 @@ namespace Raul {
 
 namespace Ingen {
 	
-class PatchImpl;
 class NodeImpl;
 class ConnectionImpl;
 class MidiMessage;
@@ -76,8 +76,9 @@ private:
 	OutputPort* _src_output_port;
 	InputPort*  _dst_input_port;
 
-	bool   _lookup;
-	
+	bool _lookup;
+			
+	PatchImpl::Connections::Node* _patch_connection;
 	CompiledPatch* _compiled_patch; ///< New process order for Patch
 	
 	ErrorType _error;