Add first classes for a multi worker server.

include/Makefile.am

@@ -3,8 +3,11 @@ nobase_include_HEADERS = trcomm.h \
 			 tr/comm_manager.h \
 			 tr/comm_manager_poll.h \
 			 tr/comm_manager_epoll.h \
+			 tr/comm_manager_dispatcher.h \
+			 tr/comm_worker.h \
 			 tr/connect_entry_point.h \
 			 tr/connection.h \
+			 tr/connection_getter.h \
 			 tr/connector.h \
 			 tr/datagram_service.h \
 			 tr/datagram_entry_point.h \

include/tr/comm_manager_dispatcher.h

@@ -0,0 +1,52 @@
+/**
+ * \file
+ *
+ * \author	Georg Hopp
+ *
+ * \copyright
+ * Copyright © 2014 Georg Hopp
+ *
+ * This program 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 Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program 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 General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#ifndef __TR_COMM_MANAGER_DISPATCHER_H__
+#define __TR_COMM_MANAGER_DISPATCHER_H__
+
+#include <sys/types.h>
+#include <sys/epoll.h>
+
+#include "trbase.h"
+#include "trdata.h"
+#include "trevent.h"
+
+#include "tr/comm_worker.h"
+
+TR_CLASS(TR_CommManagerDispatcher) {
+	TR_EXTENDS(TR_CommManager);
+
+	TR_Set workers;
+};
+
+#define TR_commManDispAddWorker(this, worker) \
+	(TR_setAdd((this)->workers, (worker))
+
+TR_INSTANCE_INIT(TR_CommManagerDispatcher);
+TR_CLASSVARS_DECL(TR_CommManagerDispatcher) {
+	TR_CV_EXTENDS(TR_EventHandler);
+};
+
+#endif // __TR_COMM_MANAGER_DISPATCHER_H__
+
+// vim: set ts=4 sw=4:
+

include/tr/comm_worker.h

@@ -0,0 +1,52 @@
+/**
+ * \file
+ * This exists only as a base type for all protocols.
+ *
+ * \author	Georg Hopp
+ *
+ * \copyright
+ * Copyright © 2014 Georg Hopp
+ *
+ * This program 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 Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program 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 General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#ifndef __TR_COMM_WORKER_H__
+#define __TR_COMM_WORKER_H__
+
+#include <sys/types.h>
+#include <semaphore.h>
+
+#include "trbase.h"
+#include "trio.h"
+
+typedef struct s_TR_WorkerShm {
+	struct c_TR_Socket socket;
+	sem_t              semaphore;
+} s_TR_WorkerShm;
+typedef s_TR_WorkerShm * sptr_TR_WorkerShm;
+
+TR_CLASS(TR_CommWorker) {
+	const char * const name;
+	pid_t              pid;
+	TR_Socket          socket;
+	sptr_TR_WorkerShm  shm;
+};
+
+TR_INSTANCE_INIT(TR_CommWorker);
+TR_CLASSVARS_DECL(TR_CommWorker) {};
+
+#endif // __TR_COMM_WORKER_H__
+
+// vim: set ts=4 sw=4:
+

include/tr/connection_getter.h

@@ -0,0 +1,46 @@
+/**
+ * \file
+ *
+ * \author	Georg Hopp
+ *
+ * \copyright
+ * Copyright © 2014 Georg Hopp
+ *
+ * This program 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 Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program 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 General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#ifndef __TR_CONNECTION_GETTER_H__
+#define __TR_CONNECTION_GETTER_H__
+
+#include <sys/types.h>
+
+#include "trbase.h"
+#include "trevent.h"
+
+#include "tr/comm_worker.h"
+
+TR_CLASS(TR_ConnectionGetter) {
+	TR_EXTENDS(TR_EventHandler);
+
+	TR_CommWorker worker;
+};
+TR_INSTANCE_INIT(TR_ConnectionGetter);
+TR_CLASSVARS_DECL(TR_ConnectionGetter) {
+	TR_CV_EXTENDS(TR_EventHandler);
+};
+
+#endif // __TR_CONNECTION_GETTER_H__
+
+// vim: set ts=4 sw=4:
+

src/Makefile.am

@@ -9,15 +9,18 @@ TRCOMM = cep_write_buffered.c \
 	 comm_end_point.c \
 	 conn_entry_point.c \
 	 connection.c \
+	 connection_getter.c \
 	 connector.c \
 	 datagram_service.c \
 	 datagram_entry_point.c \
 	 comm_manager.c \
 	 comm_manager_poll.c \
 	 comm_manager_epoll.c \
+	 comm_manager_dispatcher.c \
 	 comm_manager_shutdown.c \
 	 comm_manager_shutdown_read.c \
 	 comm_manager_shutdown_write.c \
+	 comm_worker.c \
 	 io_handler.c \
 	 proto_message.c \
 	 protocol.c \
@@ -38,5 +41,5 @@ lib_LTLIBRARIES = libtrcomm.la
 
 libtrcomm_la_SOURCES = $(TRCOMM)
 libtrcomm_la_CFLAGS  = $(AM_CFLAGS)
-libtrcomm_la_LIBADD  =
+libtrcomm_la_LIBADD  = -lrt -lpthread
 libtrcomm_la_LDFLAGS = -version-info 0:0:0 $(AM_LDFLAGS)

src/comm_manager_dispatcher.c

@@ -0,0 +1,128 @@
+/**
+ * \file
+ *
+ * \author	Georg Hopp
+ *
+ * \copyright
+ * Copyright © 2014 Georg Hopp
+ *
+ * This program 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 Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program 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 General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+#include <sys/types.h>
+#include <sys/wait.h>
+#include <semaphore.h>
+#include <errno.h>
+
+#include "trbase.h"
+#include "trdata.h"
+#include "trevent.h"
+
+#include "tr/comm_manager.h"
+#include "tr/comm_manager_dispatcher.h"
+#include "tr/interface/comm_manager.h"
+#include "tr/comm_end_point.h"
+#include "tr/connection.h"
+#include "tr/connect_entry_point.h"
+#include "tr/_comm_manager.h" // ?? may we don't need this one
+
+
+static
+int
+commManagerDispatcherCtor(void * _this, va_list * params)
+{
+	TR_CommManagerDispatcher this = _this;
+	TR_PARENTCALL(TR_CommManagerDispatcher, _this, TR_Class, ctor, params);
+	this->workers = va_arg(*params, TR_Set);
+
+	return 0;
+}
+
+static
+void
+commManagerDispatcherDtor(void * _this)
+{
+	TR_PARENTCALL(TR_CommManagerDispatcher, _this, TR_Class, dtor);
+}
+
+static
+TR_EventDone
+TR_commManagerDispatcherSendEndpoint(void * _this, TR_CommEndPoint endpoint)
+{
+	TR_CommManagerDispatcher this   = _this;
+	TR_CommWorker            worker = NULL;
+	int                      pid_stat;
+
+	/* code to dispatch stuff.... */
+	while (! TR_setEmpty(this->workers) && ! worker) {
+		if (! TR_iterableValid(this->workers)) {
+			TR_iterableRewind(this->workers);
+		}
+		worker = (TR_CommWorker)TR_iterableCurrent(this->workers);
+		if (0 != waitpid(worker->pid, &pid_stat, WNOHANG)) {
+			TR_setDelete(this->workers, worker);
+			TR_delete(worker);
+			continue;
+		}
+	}
+
+	if (! worker) {
+		TR_delete(endpoint); // no ones there to handle you, sorry...
+		return TR_EVENT_DONE;
+	}
+
+	TR_iterableNext(this->workers);
+
+	if (-1 == sem_trywait(&(worker->shm->semaphore))) {
+		switch(errno) {
+			case EAGAIN:
+			case EINTR:
+				break;
+
+			default:
+				TR_setDelete(this->workers, worker);
+				TR_delete(worker);
+				break;
+		}
+	}
+
+	memcpy(
+			&(worker->shm->socket),
+			endpoint->transport,
+			sizeof(struct c_TR_Socket));
+	TR_socketSendFd(worker->socket, TR_socketHandle(endpoint->transport));
+	TR_delete(endpoint);
+
+	return TR_EVENT_DONE;
+}
+
+
+static
+void
+TR_commManagerDispatcherCvInit(TR_class_ptr cls)
+{
+	TR_INHERIT_CLASSVARS(TR_CommManagerDispatcher, TR_CommManager);
+}
+
+TR_INIT_IFACE(TR_Class, commManagerDispatcherCtor,
+		commManagerDispatcherDtor, NULL);
+TR_INIT_IFACE(TR_CommManager, TR_commManagerDispatcherSendEndpoint,
+		NULL, NULL, NULL, NULL, NULL, NULL);
+TR_CREATE_CLASS(
+		TR_CommManagerDispatcher,
+		TR_CommManager,
+		TR_commManagerDispatcherCvInit,
+		TR_IF(TR_Class),
+		TR_IF(TR_CommManager));
+
+// vim: set ts=4 sw=4:

src/comm_worker.c

@@ -0,0 +1,90 @@
+/**
+ * \file
+ *
+ * \author	Georg Hopp
+ *
+ * \copyright
+ * Copyright © 2014 Georg Hopp
+ *
+ * This program 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 Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program 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 General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#define _GNU_SOURCE
+
+#include <sys/types.h>
+#include <sys/stat.h>
+#include <sys/mman.h>
+#include <sys/wait.h>
+#include <stdarg.h>
+#include <semaphore.h>
+#include <unistd.h>
+#include <fcntl.h>
+
+#include "trbase.h"
+#include "trio.h"
+
+#include "tr/comm_worker.h"
+
+static
+int
+commWorkerCtor(void * _this, va_list * params)
+{
+	TR_CommWorker      this = _this;
+	TR_Socket          socket[2];
+
+	this->shm = mmap(NULL, sizeof(s_TR_WorkerShm),
+			PROT_READ|PROT_WRITE,
+			MAP_ANONYMOUS|MAP_SHARED,
+			0, 0);
+
+	sem_init(&(this->shm->semaphore), 1, 1);
+	TR_socketPair(socket, SOCK_DGRAM);
+
+	/* TODO error handling... */
+	switch (this->pid = fork()) {
+		case -1:
+			TR_delete(socket[0]);
+			TR_delete(socket[1]);
+			break;
+
+		case 0:
+			TR_delete(socket[0]);
+			this->socket = socket[1];
+			break;
+
+		default:
+			TR_delete(socket[1]);
+			this->socket = socket[0];
+			break;
+	}
+
+	return this->pid;
+}
+
+static void commWorkerDtor(void * _this) {
+	TR_CommWorker this = _this;
+
+	TR_delete(this->socket);
+	sem_close(&(this->shm->semaphore));
+	munmap(this->shm, sizeof(s_TR_WorkerShm));
+	if (this->pid != 0) {
+		int state;
+		waitpid(this->pid, &state, 0);
+	}
+}
+
+TR_INIT_IFACE(TR_Class, commWorkerCtor, commWorkerDtor, NULL);
+TR_CREATE_CLASS(TR_CommWorker, NULL, NULL, TR_IF(TR_Class));
+
+// vim: set ts=4 sw=4:

src/connection_getter.c

@@ -0,0 +1,129 @@
+/**
+ * \file
+ *
+ * \author	Georg Hopp
+ *
+ * \copyright
+ * Copyright © 2014 Georg Hopp
+ *
+ * This program 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 Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program 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 General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#include <stdarg.h>
+#include <unistd.h>
+
+#include <sys/types.h>
+#include <sys/stat.h>
+#include <sys/mman.h>
+#include <fcntl.h>
+#include <semaphore.h>
+
+#include "trbase.h"
+#include "trio.h"
+#include "trevent.h"
+
+#include "tr/connection_getter.h"
+#include "tr/connection.h"
+#include "tr/protocol.h"
+#include "tr/connect_entry_point.h"
+
+static
+int
+connectionGetterCtor(void * _this, va_list * params)
+{
+	TR_ConnectionGetter this = _this;
+
+	TR_PARENTCALL(TR_ConnectionGetter, _this, TR_Class, ctor, params);
+	this->worker = va_arg(*params, TR_CommWorker);
+
+	return 0;
+}
+
+static
+void
+connectionGetterDtor(void * _this)
+{
+	TR_PARENTCALL(TR_ConnectionGetter, _this, TR_Class, dtor);
+}
+
+static
+TR_EventDone
+connectionGetterAccept(void * _this, TR_Event event)
+{
+	int                 count      = 0;
+	TR_ConnectionGetter this       = _this;
+	TR_CommEndPoint     connection = (TR_CommEndPoint)event->subject;
+
+	int handle = TR_socketGetFd((TR_Socket)connection->transport);
+
+	while (handle != -1) {
+		TR_Socket socket = TR_new(TR_Socket);
+
+		memcpy(
+				socket,
+				&(this->worker->shm->socket),
+				sizeof(struct c_TR_Socket));
+		sem_post(&(this->worker->shm->semaphore));
+		TR_socketHandle(socket) = handle;
+
+		TR_socketNonblock(socket);
+		TR_Connection new_con = TR_new(
+				TR_Connection,
+				socket,
+				connection->protocol,
+				CEP_DEFAULT_READ_SIZE);
+		TR_eventHandlerIssueEvent(
+				(TR_EventHandler)this,
+				TR_eventSubjectEmit(
+					(TR_EventSubject)new_con,
+					TR_CON_EVENT_NEW_CON,
+					NULL));
+		if (++count > 100) break;
+		handle = TR_socketGetFd((TR_Socket)connection->transport);
+	}
+
+	if (! handle) {
+		TR_eventHandlerIssueEvent(
+				(TR_EventHandler)this,
+				TR_eventSubjectEmit(
+					(TR_EventSubject)connection,
+					TR_CEP_EVENT_READ_BLOCK,
+					NULL));
+	}
+
+	return TR_EVENT_DONE;
+}
+
+static
+void
+connectorCvInit(TR_class_ptr cls)
+{
+    TR_EVENT_HANDLER_SET_METHOD(
+            cls,
+            TR_ConnEntryPoint,
+            TR_CET_EVENT_ACC_READY,
+            connectionGetterAccept);
+}
+
+TR_INIT_HANDLER(TR_ConnectionGetter);
+TR_INIT_IFACE(TR_Class, connectionGetterCtor, connectionGetterDtor, NULL);
+TR_CREATE_CLASS(
+		TR_ConnectionGetter,
+		TR_EventHandler,
+		connectorCvInit,
+		TR_IF(TR_Class)) = {
+	{ TR_HANDLER_CVARS(TR_ConnectionGetter) }
+};
+
+// vim: set ts=4 sw=4: