Work towards a proper server infrastructure with a thread per connection.

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

12 files changed, 243 insertions, 87 deletions

diff --git a/src/server/Engine.cpp b/src/server/Engine.cpp
index f793ad22..77630d54 100644
--- a/src/server/Engine.cpp
+++ b/src/server/Engine.cpp
@@ -237,8 +237,15 @@ Engine::process_events(ProcessContext& context)
 {
 	ThreadManager::assert_thread(THREAD_PROCESS);
 
-	for (EventSources::iterator i = _event_sources.begin(); i != _event_sources.end(); ++i)
-		(*i)->process(*_post_processor, context);
+	EventSources::iterator i = _event_sources.begin();
+	while (i != _event_sources.end()) {
+		EventSources::iterator next = i;
+		++next;
+		if (!(*i)->process(*_post_processor, context)) {
+			_event_sources.erase(i);
+		}
+		i = next;
+	}
 }
 
 void
diff --git a/src/server/Engine.hpp b/src/server/Engine.hpp
index 9a5d4f76..8e5f381f 100644
--- a/src/server/Engine.hpp
+++ b/src/server/Engine.hpp
@@ -62,24 +62,18 @@ public:
 
 	virtual ~Engine();
 
+	// EngineBase methods
 	virtual bool activate();
-
 	virtual void deactivate();
-
 	virtual void quit();
-
 	virtual bool main_iteration();
-
-	virtual void set_driver(SharedPtr<Driver> driver);
-
-	virtual void add_event_source(SharedPtr<EventSource> source);
-
-	virtual void process_events(ProcessContext& context);
-
 	virtual void register_client(const Raul::URI& uri, Interface* client);
-
 	virtual bool unregister_client(const Raul::URI& uri);
 
+	void set_driver(SharedPtr<Driver> driver);
+	void add_event_source(SharedPtr<EventSource> source);
+	void process_events(ProcessContext& context);
+
 	Ingen::Shared::World* world() const { return _world; }
 
 	ClientBroadcaster*    broadcaster()      const { return _broadcaster; }
diff --git a/src/server/EventQueue.cpp b/src/server/EventQueue.cpp
index 272273cd..585eaa34 100644
--- a/src/server/EventQueue.cpp
+++ b/src/server/EventQueue.cpp
@@ -67,13 +67,13 @@ EventQueue::event(Event* const ev)
  *
  * Executed events will be pushed to @a dest.
  */
-void
+bool
 EventQueue::process(PostProcessor& dest, ProcessContext& context, bool limit)
 {
 	ThreadManager::assert_thread(THREAD_PROCESS);
 
 	if (!_head.get())
-		return;
+		return true;
 
 	/* Limit the maximum number of queued events to process per cycle.  This
 	   makes the process callback (more) realtime-safe by preventing being
@@ -105,6 +105,8 @@ EventQueue::process(PostProcessor& dest, ProcessContext& context, bool limit)
 		if (!next)
 			_tail = NULL;
 	}
+
+	return true;
 }
 
 /** Pre-process a single event */
diff --git a/src/server/EventQueue.hpp b/src/server/EventQueue.hpp
index a3567574..edd6c6b4 100644
--- a/src/server/EventQueue.hpp
+++ b/src/server/EventQueue.hpp
@@ -40,7 +40,7 @@ public:
 	explicit EventQueue();
 	virtual ~EventQueue();
 
-	void process(PostProcessor& dest, ProcessContext& context, bool limit=true);
+	bool process(PostProcessor& dest, ProcessContext& context, bool limit=true);
 
 	inline bool unprepared_events() const { return _prepared_back.get(); }
 	inline bool empty()             const { return !_head.get(); }
diff --git a/src/server/EventSource.hpp b/src/server/EventSource.hpp
index 4425586d..65afece4 100644
--- a/src/server/EventSource.hpp
+++ b/src/server/EventSource.hpp
@@ -35,7 +35,10 @@ class EventSource
 public:
 	virtual ~EventSource() {}
 
-	virtual void process(PostProcessor&  dest,
+	/** Process events for a cycle.
+	 * @return False iff this source is finished and should be removed.
+	 */
+	virtual bool process(PostProcessor&  dest,
 	                     ProcessContext& context,
 	                     bool            limit = true) = 0;
 };
diff --git a/src/server/events/SetMetadata.hpp b/src/server/events/SetMetadata.hpp
index f906193c..e1bc0c55 100644
--- a/src/server/events/SetMetadata.hpp
+++ b/src/server/events/SetMetadata.hpp
@@ -96,7 +96,7 @@ private:
 
 	typedef std::vector<SetPortValue*> SetEvents;
 
-	Event*                 _create_event;
+	Event*                       _create_event;
 	SetEvents                    _set_events;
 	std::vector<SpecialType>     _types;
 	std::vector<SpecialType>     _remove_types;
diff --git a/src/socket/SocketInterface.cpp b/src/socket/SocketInterface.cpp
new file mode 100644
index 00000000..d9994e3e
--- /dev/null
+++ b/src/socket/SocketInterface.cpp
@@ -0,0 +1,149 @@
+/*
+  This file is part of Ingen.
+  Copyright 2007-2012 David Robillard <http://drobilla.net/>
+
+  Ingen is free software: you can redistribute it and/or modify it under the
+  terms of the GNU Affero General Public License as published by the Free
+  Software Foundation, either version 3 of the License, or any later version.
+
+  Ingen is distributed in the hope that it will be useful, but WITHOUT ANY
+  WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR
+  A PARTICULAR PURPOSE.  See the GNU Affero General Public License for details.
+
+  You should have received a copy of the GNU Affero General Public License
+  along with Ingen.  If not, see <http://www.gnu.org/licenses/>.
+*/
+
+#include <errno.h>
+#include <sys/fcntl.h>
+#include <sys/socket.h>
+#include <sys/un.h>
+
+#include "ingen/Interface.hpp"
+#include "ingen/shared/World.hpp"
+#include "ingen/shared/AtomReader.hpp"
+#include "sord/sordmm.hpp"
+#include "sratom/sratom.h"
+#include "SocketInterface.hpp"
+
+#include "../server/Event.hpp"
+#include "../server/PostProcessor.hpp"
+#include "../server/ThreadManager.hpp"
+
+#define LOG(s) s << "[SocketInterface] "
+
+namespace Ingen {
+namespace Socket {
+
+SocketInterface::SocketInterface(Ingen::Shared::World& world, int conn)
+	: _world(world)
+	, _iface(*(Server::Engine*)world.local_engine().get(), *this)
+	, _event(NULL)
+	, _conn(conn)
+{
+	// Set connection to non-blocking so parser can read until EOF
+	// and not block indefinitely waiting for more input
+	fcntl(_conn, F_SETFL, fcntl(_conn, F_GETFL, 0) | O_NONBLOCK);
+
+	set_name("SocketInterface");
+	start();
+}
+
+SocketInterface::~SocketInterface()
+{
+	stop();
+	join();
+	close(_conn);
+}
+
+void
+SocketInterface::event(Server::Event* ev)
+{
+	assert(!_event);
+	ev->pre_process();
+	_event = ev;
+}
+
+bool
+SocketInterface::process(Server::PostProcessor&  dest,
+                         Server::ProcessContext& context,
+                         bool                    limit)
+{
+	if (_event) {
+		_event->execute(context);
+		dest.append(_event, _event);
+		_event = NULL;
+	}
+	if (_conn == -1) {
+		return false;
+	}
+	return true;
+}
+
+void
+SocketInterface::_run()
+{
+	Thread::set_context(Server::THREAD_PRE_PROCESS);
+	while (!_exit_flag) {
+		// Set up a reader to parse the Turtle message into a model
+		Sord::World* world  = _world.rdf_world();
+		SerdEnv*     env    = world->prefixes().c_obj();
+		SordModel*   model  = sord_new(world->c_obj(), SORD_SPO, false);
+		SerdReader*  reader = sord_new_reader(model, env, SERD_TURTLE, NULL);
+		// Set base URI to path: so e.g. </foo/bar> will be a path
+		SordNode* base_uri = sord_new_uri(
+			world->c_obj(), (const uint8_t*)"path:");
+		serd_env_set_base_uri(env, sord_node_to_serd_node(base_uri));
+
+		LV2_URID_Map* map = &_world.lv2_uri_map()->urid_map_feature()->urid_map;
+
+		// Set up sratom to build an LV2_Atom from the model
+		Sratom*        sratom = sratom_new(map);
+		SerdChunk      chunk  = { NULL, 0 };
+		LV2_Atom_Forge forge;
+		lv2_atom_forge_init(&forge, map);
+		lv2_atom_forge_set_sink(
+			&forge, sratom_forge_sink, sratom_forge_deref, &chunk);
+
+		// Read directly from the connection with serd
+		FILE* f = fdopen(_conn, "r");
+		if (!f) {
+			LOG(Raul::error) << "Failed to open connection " << _conn
+			                 << "(" << strerror(errno) << ")" << std::endl;
+			// Connection gone, exit
+			break;
+		}
+
+		serd_reader_read_file_handle(reader, f, (const uint8_t*)"(socket)");
+
+		// FIXME: Sratom needs work to be able to read resources
+		SordNode* msg_node = sord_new_blank(
+			world->c_obj(), (const uint8_t*)"genid1");
+
+		// Build an LV2_Atom at chunk.buf from the message
+		sratom_read(sratom, &forge, world->c_obj(), model, msg_node);
+
+		// Make an AtomReader to read that atom and do Ingen things
+		Shared::AtomReader ar(*_world.lv2_uri_map().get(),
+		                      *_world.uris().get(),
+		                      _world.forge(),
+		                      _iface);
+
+		// Call _iface methods based on atom content
+		ar.write((LV2_Atom*)chunk.buf);
+
+		// Respond and close connection
+		write(_conn, "OK", 2);
+		fclose(f);
+
+		sratom_free(sratom);
+		sord_node_free(world->c_obj(), msg_node);
+		serd_reader_free(reader);
+		sord_free(model);
+	}
+
+	_conn = -1;
+}
+
+}  // namespace Ingen
+}  // namespace Socket
diff --git a/src/socket/SocketInterface.hpp b/src/socket/SocketInterface.hpp
new file mode 100644
index 00000000..2a23cc64
--- /dev/null
+++ b/src/socket/SocketInterface.hpp
@@ -0,0 +1,58 @@
+/*
+  This file is part of Ingen.
+  Copyright 2007-2012 David Robillard <http://drobilla.net/>
+
+  Ingen is free software: you can redistribute it and/or modify it under the
+  terms of the GNU Affero General Public License as published by the Free
+  Software Foundation, either version 3 of the License, or any later version.
+
+  Ingen is distributed in the hope that it will be useful, but WITHOUT ANY
+  WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR
+  A PARTICULAR PURPOSE.  See the GNU Affero General Public License for details.
+
+  You should have received a copy of the GNU Affero General Public License
+  along with Ingen.  If not, see <http://www.gnu.org/licenses/>.
+*/
+
+#include <string>
+
+#include "raul/SharedPtr.hpp"
+#include "raul/Thread.hpp"
+#include "ingen/Interface.hpp"
+
+#include "../server/EventSink.hpp"
+#include "../server/EventSource.hpp"
+#include "../server/EventWriter.hpp"
+
+namespace Ingen {
+
+namespace Shared { class World; }
+
+namespace Socket {
+
+class SocketInterface : public Raul::Thread
+                      , public Server::EventSink
+                      , public Server::EventSource
+{
+public:
+	SocketInterface(Shared::World& world, int conn);
+
+	void event(Server::Event* ev);
+
+	bool process(Server::PostProcessor&  dest,
+	             Server::ProcessContext& context,
+	             bool                    limit = true);
+
+	~SocketInterface();
+
+private:
+	virtual void _run();
+
+	Shared::World&      _world;
+	Server::EventWriter _iface;
+	Server::Event*      _event;
+	int                 _conn;
+};
+
+}  // namespace Ingen
+}  // namespace Socket
diff --git a/src/socket/SocketListener.cpp b/src/socket/SocketListener.cpp
index edd5421b..c2ff1c10 100644
--- a/src/socket/SocketListener.cpp
+++ b/src/socket/SocketListener.cpp
@@ -23,17 +23,18 @@
 #include "ingen/shared/AtomReader.hpp"
 #include "sord/sordmm.hpp"
 #include "sratom/sratom.h"
+
+#include "../server/Engine.hpp"
 #include "SocketListener.hpp"
+#include "SocketInterface.hpp"
 
 #define LOG(s) s << "[SocketListener] "
 
 namespace Ingen {
 namespace Socket {
 
-SocketListener::SocketListener(Ingen::Shared::World& world,
-                               SharedPtr<Interface>  iface)
+SocketListener::SocketListener(Ingen::Shared::World& world)
 	: _world(world)
-	, _iface(iface)
 {
 	// Create server socket
 	_sock = socket(AF_UNIX, SOCK_STREAM, 0);
@@ -88,60 +89,10 @@ SocketListener::_run()
 			continue;
 		}
 
-		// Set connection to non-blocking so parser can read until EOF
-		// and not block indefinitely waiting for more input
-		fcntl(conn, F_SETFL, fcntl(conn, F_GETFL, 0) | O_NONBLOCK);
-
-		// Set up a reader to parse the Turtle message into a model
-		Sord::World* world  = _world.rdf_world();
-		SerdEnv*     env    = world->prefixes().c_obj();
-		SordModel*   model  = sord_new(world->c_obj(), SORD_SPO, false);
-		SerdReader*  reader = sord_new_reader(model, env, SERD_TURTLE, NULL);
-
-		// Set base URI to path: so e.g. </foo/bar> will be a path
-		SordNode* base_uri = sord_new_uri(
-			world->c_obj(), (const uint8_t*)"path:");
-		serd_env_set_base_uri(env, sord_node_to_serd_node(base_uri));
-
-		LV2_URID_Map* map = &_world.lv2_uri_map()->urid_map_feature()->urid_map;
-
-		// Set up sratom to build an LV2_Atom from the model
-		Sratom*        sratom = sratom_new(map);
-		SerdChunk      chunk  = { NULL, 0 };
-		LV2_Atom_Forge forge;
-		lv2_atom_forge_init(&forge, map);
-		lv2_atom_forge_set_sink(
-			&forge, sratom_forge_sink, sratom_forge_deref, &chunk);
-
-		// Read directly from the connection with serd
-		FILE* f = fdopen(conn, "r");
-		serd_reader_read_file_handle(reader, f, (const uint8_t*)"(socket)");
-
-		// FIXME: Sratom needs work to be able to read resources
-		SordNode* msg_node = sord_new_blank(
-			world->c_obj(), (const uint8_t*)"genid1");
-
-		// Build an LV2_Atom at chunk.buf from the message
-		sratom_read(sratom, &forge, world->c_obj(), model, msg_node);
-
-		// Make an AtomReader to read that atom and do Ingen things
-		Shared::AtomReader ar(*_world.lv2_uri_map().get(),
-		                      *_world.uris().get(),
-		                      _world.forge(),
-		                      *_iface.get());
-
-		// Call _iface methods based on atom content
-		ar.write((LV2_Atom*)chunk.buf);
-
-		// Respond and close connection
-		write(conn, "OK", 2);
-		fclose(f);
-		close(conn);
-
-		sratom_free(sratom);
-		sord_node_free(world->c_obj(), msg_node);
-		serd_reader_free(reader);
-		sord_free(model);
+		// Make an new interface/thread to handle the connection
+		Server::Engine* engine = (Server::Engine*)_world.local_engine().get();
+		SharedPtr<SocketInterface> iface(new SocketInterface(_world, conn));
+		engine->add_event_source(iface);
 	}
 }
 
diff --git a/src/socket/SocketListener.hpp b/src/socket/SocketListener.hpp
index 896b08b4..10e9d1d2 100644
--- a/src/socket/SocketListener.hpp
+++ b/src/socket/SocketListener.hpp
@@ -30,16 +30,13 @@ namespace Socket {
 class SocketListener : public Raul::Thread
 {
 public:
-	SocketListener(Ingen::Shared::World& world,
-	               SharedPtr<Interface>  iface);
-
+	SocketListener(Ingen::Shared::World& world);
 	~SocketListener();
 
 private:
 	virtual void _run();
 
 	Ingen::Shared::World& _world;
-	SharedPtr<Interface>  _iface;
 	std::string           _sock_path;
 	int                   _sock;
 };
diff --git a/src/socket/ingen_socket_server.cpp b/src/socket/ingen_socket_server.cpp
index 19257add..71b8ad64 100644
--- a/src/socket/ingen_socket_server.cpp
+++ b/src/socket/ingen_socket_server.cpp
@@ -29,14 +29,8 @@ using namespace Ingen;
 
 struct IngenSocketServerModule : public Ingen::Shared::Module {
 	void load(Ingen::Shared::World* world) {
-		Server::Engine* engine = (Server::Engine*)world->local_engine().get();
-		SharedPtr<Server::EventQueue> queue(new Server::EventQueue());
-		SharedPtr<Server::EventWriter> interface(
-			new Server::EventWriter(*engine, *queue.get()));
 		listener = SharedPtr<Ingen::Socket::SocketListener>(
-			new Ingen::Socket::SocketListener(*world, interface));
-
-		engine->add_event_source(queue);
+			new Ingen::Socket::SocketListener(*world));
 	}
 
 	SharedPtr<Ingen::Socket::SocketListener> listener;
diff --git a/src/socket/wscript b/src/socket/wscript
index 6b104906..6985f377 100644
--- a/src/socket/wscript
+++ b/src/socket/wscript
@@ -4,7 +4,8 @@ from waflib.extras import autowaf as autowaf
 def build(bld):
     if bld.is_defined('HAVE_SOCKET'):
         obj = bld(features        = 'cxx cxxshlib',
-                  source          = ['SocketListener.cpp',
+                  source          = ['SocketInterface.cpp',
+                                     'SocketListener.cpp',
                                      'ingen_socket_server.cpp'],
                   includes        = ['.', '../..'],
                   name            = 'libingen_socket_server',