Change socket hashtables for read, accept and write to set

README.md

@@ -3,3 +3,30 @@
 Socket communication layer build upon libtrevent.
 
 ## BUGS
+
+## MULTIPLE WORKER
+
+### Stream sockets
+
+For stream socket some new classes might be neede:
+
+ * A connector derivate that does not create a TR_CON_EVENT_NEW_CON
+   event but instead hands over the socket to one of its worker processes.
+
+ * Another connector no doing TR_socketAccept but instead get the
+   docket via its socket pair. This one then might isse TR_CON_EVENT_NEW_CON
+
+### Datagram sockets
+
+Currently there is no clear plan to parallelize the communication
+on datagram sockets.
+
+As we only have one socket for all communication it makes no sense
+to share it between multiple processes. This neccicarily will lead
+to race condition.
+
+One possible way might be to share the socket with all worker, but
+do the read only in the master worker. When a packet has arraive
+it will be handed over to the worker via shared memory. This might
+be done as an event message. The worker then will process the packet
+and send a respone.

include/tr/comm_manager.h

@@ -35,9 +35,9 @@ TR_CLASS(TR_CommManager) {
 	TR_EXTENDS(TR_EventHandler);
 
 	TR_CommEndPoint * endpoints;
-	TR_Hash           accept;
-	TR_Hash           write;
-	TR_Hash           read;
+	TR_Set            accept;
+	TR_Set            write;
+	TR_Set            read;
 	size_t            n_endpoints;
 	size_t            max_handle;
 };

include/tr/connect_entry_point.h

@@ -41,8 +41,6 @@ TR_CLASSVARS_DECL(TR_ConnEntryPoint) {
 #define TR_CET_EVENT_ACC_READY (TR_CEP_EVENT_MAX + 1)
 #define TR_CET_EVENT_MAX       ((size_t)TR_CET_EVENT_ACC_READY)
 
-TR_TcpSocket TR_cetAccept(TR_ConnEntryPoint);
-
 #endif // __TR_CONNECT_ENTRY_POINT_H__
 
 // vim: set ts=4 sw=4:

src/Makefile.am

@@ -4,8 +4,7 @@ AUTOMAKE_OPTIONS = subdir-objects
 AM_CFLAGS += -I../include/ -std=c99
 AM_LDFLAGS +=
 
-TRCOMM = cet_accept.c \
-	 cep_write_buffered.c \
+TRCOMM = cep_write_buffered.c \
 	 comm_end_point_read.c \
 	 comm_end_point.c \
 	 conn_entry_point.c \

src/cet_accept.c

@@ -1,34 +0,0 @@
-/**
- * \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 "trio.h"
-
-#include "tr/comm_end_point.h"
-#include "tr/connect_entry_point.h"
-
-TR_TcpSocket
-TR_cetAccept(TR_ConnEntryPoint cet)
-{
-	return TR_socketAccept((TR_TcpSocket)((TR_CommEndPoint)cet)->transport);
-}
-
-// vim: set ts=4 sw=4:

src/comm_manager.c

@@ -40,13 +40,13 @@ commManagerCtor(void * _this, va_list * params)
 
 	TR_PARENTCALL(TR_CommManager, _this, TR_Class, ctor, params);
 
-	this->accept = TR_new(TR_Hash);
-	this->write  = TR_new(TR_Hash);
-	this->read   = TR_new(TR_Hash);
+	this->accept = TR_new(TR_Set);
+	this->write  = TR_new(TR_Set);
+	this->read   = TR_new(TR_Set);
 
-	this->accept->cleanup_no_free = TRUE;
-	this->write->cleanup_no_free  = TRUE;
-	this->read->cleanup_no_free   = TRUE;
+	this->accept->free_msgs = 0;
+	this->write->free_msgs  = 0;
+	this->read->free_msgs   = 0;
 
 	this->n_endpoints = sysconf(_SC_OPEN_MAX);
 	this->endpoints   = TR_calloc(sizeof(TR_CommEndPoint), this->n_endpoints);
@@ -76,7 +76,7 @@ TR_commManagerEnableWrite(void * _this, TR_Event event)
 {
 	TR_CommManager this = _this;
 
-	TR_hashAdd(this->write, event->subject);
+	TR_setAdd(this->write, event->subject);
 
 	return TR_EVENT_DONE;
 }

src/comm_manager_epoll.c

@@ -101,10 +101,10 @@ TR_commManagerEpollSelect(void * _this, TR_Event event, unsigned long timeout)
 		if ((events[i].events & EPOLLIN) == EPOLLIN) {
 			if (TR_INSTANCE_OF(TR_TcpSocket, endpoint->transport)
 					&& ((TR_TcpSocket)endpoint->transport)->listen) {
-				TR_hashAdd(cmgr->accept, endpoint);
+				TR_setAdd(cmgr->accept, endpoint);
 			} else {
 				if (! event->subject->fin) {
-					TR_hashAdd(cmgr->read, endpoint);
+					TR_setAdd(cmgr->read, endpoint);
 				}
 			}
 
@@ -116,7 +116,7 @@ TR_commManagerEpollSelect(void * _this, TR_Event event, unsigned long timeout)
 
 		if ((events[i].events & EPOLLOUT) == EPOLLOUT) {
 			if (! event->subject->fin) {
-				TR_hashAdd(cmgr->write, endpoint);
+				TR_setAdd(cmgr->write, endpoint);
 			}
 			this->events[handle] &= ~EPOLLOUT;
 			_event.data.ptr = endpoint;

src/comm_manager_poll.c

@@ -97,10 +97,10 @@ TR_commManagerPollSelect(void * _this, TR_Event event, unsigned long timeout)
 				if ((this->fds[i].revents & POLLIN) == POLLIN) {
 					if (TR_INSTANCE_OF(TR_TcpSocket, endpoint->transport)
 							&& ((TR_TcpSocket)endpoint->transport)->listen) {
-						TR_hashAdd(cmgr->accept, endpoint);
+						TR_setAdd(cmgr->accept, endpoint);
 					} else {
 						if (! event->subject->fin) {
-							TR_hashAdd(cmgr->read, endpoint);
+							TR_setAdd(cmgr->read, endpoint);
 						}
 					}
 					this->fds[endpoint->transport->handle].events &= ~POLLIN;
@@ -108,7 +108,7 @@ TR_commManagerPollSelect(void * _this, TR_Event event, unsigned long timeout)
 
 				if ((this->fds[i].revents & POLLOUT) == POLLOUT) {
 					if (! event->subject->fin) {
-						TR_hashAdd(cmgr->write, endpoint);
+						TR_setAdd(cmgr->write, endpoint);
 					}
 					this->fds[endpoint->transport->handle].events &=
 						~(POLLOUT|POLLHUP);

src/comm_manager_shutdown_read.c

@@ -47,7 +47,7 @@ TR_commManagerShutdownRead(void * _this, TR_Event event)
 		TR_ISSUE_IO_SHUT_WRITE_EVENT(this, event->subject);
 	}
 
-	TR_hashDeleteByVal(this->read, TR_hashableGetHash(event->subject));
+	TR_setDelete(this->read, event->subject);
 
 	return TR_EVENT_DONE;
 }

src/comm_manager_shutdown_write.c

@@ -39,7 +39,7 @@ TR_commManagerShutdownWrite(void * _this, TR_Event event)
 
 	TR_ISSUE_IO_CLOSE_EVENT(this, event->subject);
 
-	TR_hashDeleteByVal(this->write, TR_hashableGetHash(event->subject));
+	TR_setDelete(this->write, event->subject);
 
 	return TR_EVENT_DONE;
 }

src/connector.c

@@ -72,6 +72,16 @@ connectorAccept(void * _this, TR_Event event)
 					(TR_EventSubject)new_con,
 					TR_CON_EVENT_NEW_CON,
 					NULL));
+		/*
+		 * TODO The break is fatal when using edge triggered events...
+		 * so either make sure to only use level trigged events with
+		 * the accept socket or remove the break...
+		 * Currently I think the break can be removed...
+		 *
+		 * TODO Prevent malicious or broken clients from doing a connection
+		 * bomb here... well, doing this might mean that the break needs to be
+		 * here.
+		 */
 		if (++count > 100) break;
 		socket = TR_socketAccept((TR_TcpSocket)connection->transport);
 	}

src/i_comm_manager.c

@@ -52,37 +52,16 @@ TR_commManagerAddEndpoint(void * _this, TR_CommEndPoint endpoint)
 
 	if (TR_INSTANCE_OF(TR_TcpSocket, endpoint->transport)
 			&& ((TR_TcpSocket)endpoint->transport)->listen) {
-		TR_hashAdd(this->accept, endpoint);
+		TR_setAdd(this->accept, endpoint);
 		TR_ISSUE_IO_ACC_EVENT(this, endpoint);
 	} else {
-		TR_hashAdd(this->read, endpoint);
+		TR_setAdd(this->read, endpoint);
 		TR_ISSUE_IO_READ_EVENT(this, endpoint);
 	}
 
 	TR_CALL(_this, TR_CommManager, addEndpoint, endpoint);
 }
 
-static
-void
-commManagerIssueAcceptEvents(const void * endpoint, const void * comm_manager)
-{
-	TR_ISSUE_IO_ACC_EVENT(comm_manager, endpoint);
-}
-
-static
-void
-commManagerIssueWriteEvents(const void * endpoint, const void * comm_manager)
-{
-	TR_ISSUE_IO_WRITE_EVENT(comm_manager, endpoint);
-}
-
-static
-void
-commManagerIssueReadEvents(const void * endpoint, const void * comm_manager)
-{
-	TR_ISSUE_IO_READ_EVENT(comm_manager, endpoint);
-}
-
 TR_EventDone
 TR_commManagerSelect(void * _this, TR_Event event)
 {
@@ -90,13 +69,23 @@ TR_commManagerSelect(void * _this, TR_Event event)
 	TR_Timer           timer      = (TR_Timer)event->data;
 	TR_EventDispatcher dispatcher = (TR_EventDispatcher)event->subject;
 	unsigned long      timeout;  // milliseconds
-	unsigned long      io_triggerd;
 
-	io_triggerd  = TR_hashEach(this->write, this, commManagerIssueWriteEvents);
-	io_triggerd += TR_hashEach(this->accept, this, commManagerIssueAcceptEvents);
-	io_triggerd += TR_hashEach(this->read, this, commManagerIssueReadEvents);
+#define IO_TRIGGERED             \
+	(TR_setSize(this->write) ||  \
+	 TR_setSize(this->accept) || \
+	 TR_setSize(this->read))
+
+	TR_iterableForeach(this->write) {
+		TR_ISSUE_IO_WRITE_EVENT(this, TR_iterableCurrent(this->write));
+	}
+	TR_iterableForeach(this->accept) {
+		TR_ISSUE_IO_ACC_EVENT(this, TR_iterableCurrent(this->accept));
+	}
+	TR_iterableForeach(this->read) {
+		TR_ISSUE_IO_READ_EVENT(this, TR_iterableCurrent(this->read));
+	}
 
-	if (io_triggerd) {
+	if (IO_TRIGGERED) {
 		timeout = 0;
 	} else if (NULL == timer) {
 		timeout = TR_eventDispatcherGetDataWaitTime(dispatcher);
@@ -114,7 +103,7 @@ TR_commManagerPollWrite(void * _this, TR_Event event)
 {
 	TR_CommManager this = _this;
 
-	TR_hashDeleteByVal(this->write, TR_hashableGetHash(event->subject));
+	TR_setDelete(this->write, event->subject);
 	if (! TR_socketFinWr(((TR_CommEndPoint)event->subject)->transport)) {
 		TR_CALL(_this, TR_CommManager, pollWrite, event);
 	}
@@ -130,9 +119,9 @@ TR_commManagerPollRead(void * _this, TR_Event event)
 
 	if (TR_INSTANCE_OF(TR_TcpSocket, endpoint->transport)
 			&& ((TR_TcpSocket)endpoint->transport)->listen) {
-		TR_hashDeleteByVal(this->accept, TR_hashableGetHash(event->subject));
+		TR_setDelete(this->accept, event->subject);
 	} else {
-		TR_hashDeleteByVal(this->read, TR_hashableGetHash(event->subject));
+		TR_setDelete(this->read, event->subject);
 	}
 
 	if (! TR_socketFinRd(endpoint->transport)) {
@@ -146,7 +135,7 @@ TR_EventDone
 TR_commManagerDisableRead(void * _this, TR_Event event)
 {
 	TR_CommManager this = _this;
-	TR_hashDeleteByVal(this->read, TR_hashableGetHash(event->subject));
+	TR_setDelete(this->read, event->subject);
 	TR_CALL(_this, TR_CommManager, disableRead, event);
 
 	return TR_EVENT_DONE;
@@ -157,9 +146,9 @@ TR_commManagerDisableWrite(void * _this, TR_Event event)
 {
 	TR_CommManager this = _this;
 
-	TR_hashDeleteByVal(this->write, TR_hashableGetHash(event->subject));
+	TR_setDelete(this->write, event->subject);
 	if (! event->subject->fin) {
-		TR_hashAdd(this->read, event->subject);
+		TR_setAdd(this->read, event->subject);
 	}
 	TR_CALL(_this, TR_CommManager, disableWrite, event);
 
@@ -186,8 +175,8 @@ TR_commManagerClose(void * _this, TR_Event event)
 		TR_eventSubjectFinalize((TR_EventSubject)this->endpoints[handle]);
 		TR_CALL(_this, TR_CommManager, disableWrite, event);
 		TR_CALL(_this, TR_CommManager, disableRead, event);
-		TR_hashDeleteByVal(this->write, TR_hashableGetHash(endpoint));
-		TR_hashDeleteByVal(this->read, TR_hashableGetHash(endpoint));
+		TR_setDelete(this->write, endpoint);
+		TR_setDelete(this->read, endpoint);
 	}
 
 	TR_CALL(_this, TR_CommManager, close, event);

src/server_bind_tcp.c

@@ -41,7 +41,7 @@ TR_serverBindTcp(
 
 	TR_serverAddEndpoint(
 			this,
-			TR_new(TR_ConnEntryPoint, socket, proto, 2048));
+			TR_new(TR_ConnEntryPoint, socket, proto, CEP_DEFAULT_READ_SIZE));
 }
 
 // vim: set ts=4 sw=4:

src/server_bind_udp.c

@@ -40,7 +40,8 @@ TR_serverBindUdp(
 			port, 0);
 
 	TR_serverAddEndpoint(
-			this, TR_new(TR_DatagramEntryPoint, socket, proto, 2048));
+			this, TR_new(
+				TR_DatagramEntryPoint, socket, proto, CEP_DEFAULT_READ_SIZE));
 }
 
 // vim: set ts=4 sw=4:

testers/Makefile

@@ -14,7 +14,8 @@ LIBS = $(TRLIBS) \
 PROGRAMS = testserver2 \
 	   testtcp \
 	   testudp \
-	   testiterator
+	   testiterator \
+	   testset
 
 all: $(PROGRAMS)
 
@@ -27,11 +28,14 @@ testtcp: testclient.o
 testudp: testclient.o
 	$(CC) $(LDFLAGS) -std=c99 $(LIBS) -o $@ $<
 
+testudp.o: testclient.c
+	$(CC) $(CFLAGS) -DUDP=1 -std=c99 -c -o $@ $<
+
 testiterator: testiterator.o
 	$(CC) $(LDFLAGS) -std=c99 $(LIBS) -o $@ $<
 
-testudp.o: testclient.c
-	$(CC) $(CFLAGS) -DUDP=1 -std=c99 -c -o $@ $<
+testset: testset.o
+	$(CC) $(LDFLAGS) -std=c99 $(LIBS) -o $@ $<
 
 %.o: %.c
 	$(CC) $(CFLAGS) -std=c99 -c -o $@ $<

testers/NOTES.md

@@ -0,0 +1 @@
+The testclient throws a segfault when interruption on a waiting connect.

testers/testclient.c

@@ -34,11 +34,19 @@ main (int argc, char * argv[])
 	protocol   = TR_new(TR_ProtocolRaw);
 #if UDP
 	socket     = TR_new(TR_UdpSocket, TR_logger, "127.0.0.1", 5678, 0);
-	connection = TR_new(TR_DatagramService, socket, protocol, 2048);
+	connection = TR_new(
+			TR_DatagramService,
+			socket,
+			protocol,
+			CEP_DEFAULT_READ_SIZE);
 	TR_socketOpen((TR_Socket)socket);
 #else
 	socket     = TR_new(TR_TcpSocket, TR_logger, "127.0.0.1", 5678, 0);
-	connection = TR_new(TR_Connection, socket, protocol, 2048);
+	connection = TR_new(
+			TR_Connection,
+			socket,
+			protocol,
+			CEP_DEFAULT_READ_SIZE);
 	TR_socketConnect((TR_Socket)socket);
 #endif
 
@@ -56,7 +64,7 @@ main (int argc, char * argv[])
 		message = (TR_ProtoMessageRaw)TR_simpleClientIssue(
 				client,
 				(TR_ProtoMessage)message,
-				1000);
+				100000);
 
 		if (! message) break;
 #if 0

testers/testset.c

@@ -0,0 +1,36 @@
+#include <stdio.h>
+
+#include "trbase.h"
+#include "trdata.h"
+
+int
+main (int argc, char * argv[])
+{
+	TR_Set set = TR_new(TR_Set);
+
+	set->free_msgs = 0;
+
+	TR_setAdd(set, "a");
+	TR_setAdd(set, "b");
+	TR_setAdd(set, "c");
+
+	TR_iterableForeach(set) {
+		printf("%s\n", (char *)TR_iterableCurrent(set));
+	}
+
+	TR_setDelete(set, "a");
+	TR_setDelete(set, "b");
+	TR_setAdd(set, "b");
+	TR_setAdd(set, "a");
+
+	TR_iterableForeach(set) {
+		printf("%s\n", (char *)TR_iterableCurrent(set));
+	}
+
+	TR_delete(set);
+	TR_cleanup();
+
+	return 0;
+}
+
+// vim: set ts=4 sw=4: