changed request handling. @TODO: I still seem to have the problem that the file …

…handles are not closed and freed correctly as the service refuses connections after about a 1000.

+2012-02-08 10:21:04 +0100 Georg Hopp 
+
+	* changed request handling. @TODO: I still seem to have the problem that the file handles are not closed and freed correctly as the service refuses connections after about a 1000. (HEAD, master)
+
+2012-02-07 14:20:00 +0100 Georg Hopp 
+
+	* now stuff seems to work correct even if read does not provide a complete request (tested with telnet)
+
 2012-02-07 13:41:49 +0100 Georg Hopp 
 2012-02-07 13:41:49 +0100 Georg Hopp 
 
 
-	* now each HttpRequestParser initializes its own request queue and enqueus completed requests there. The server now gets the queue and prints completed requests. (HEAD, master)
+	* now each HttpRequestParser initializes its own request queue and enqueus completed requests there. The server now gets the queue and prints completed requests.
 
 
 2012-02-07 11:12:30 +0100 Georg Hopp 
 2012-02-07 11:12:30 +0100 Georg Hopp 
 
 

@@ -21,21 +21,28 @@
 			MOVE_SIZE(sizeof(((server)->conns)[0]),(idx)))
 			MOVE_SIZE(sizeof(((server)->conns)[0]),(idx)))
 
 
 
 
-typedef void (*server_read_hook)(const char *, size_t);
-
 CLASS(Server) {
 CLASS(Server) {
 	Logger        logger;
 	Logger        logger;
 	Sock          sock;
 	Sock          sock;
 	void *        reader;
 	void *        reader;
 
 
+	/**
+	 * loeschen: fds[i].event auf 0
+	 * dann nfds um die anzahl der geloeschten elemente verkleinern.
+	 * die in close pending stehenden socket schliessen.
+	 * vor jedem poll qsort auf fds ueber event.
+	 * nach dem poll qsort auf fds ueber revent und reuckgebewert
+	 * von poll beruecksichtigen.
+	 */
 	nfds_t        nfds;
 	nfds_t        nfds;
+	nfds_t        ndel;
 	struct pollfd fds[POLL_FD_NSIZE];
 	struct pollfd fds[POLL_FD_NSIZE];
 
 
 	struct {
 	struct {
 		Sock         sock;
 		Sock         sock;
 		void *       reader;
 		void *       reader;
 
 
-		char *       wbuf;
+		char         wbuf[2048];
 		char *       rbuf;
 		char *       rbuf;
 		unsigned int rpos;
 		unsigned int rpos;
 		unsigned int wpos;
 		unsigned int wpos;

@@ -142,7 +142,6 @@ httpRequestParserParse(HttpRequestParser this)
 	while(cont) {
 	while(cont) {
 		switch(this->state) {
 		switch(this->state) {
 			case HTTP_REQUEST_GARBAGE:
 			case HTTP_REQUEST_GARBAGE:
-				puts("==skip garbage==");
 				data = this->buffer; // initialize static pointer
 				data = this->buffer; // initialize static pointer
 				httpRequestSkip(&data);
 				httpRequestSkip(&data);
 				request = new(HttpRequest);
 				request = new(HttpRequest);
@@ -151,7 +150,6 @@ httpRequestParserParse(HttpRequestParser this)
 				break;
 				break;
 
 
 			case HTTP_REQUEST_START:
 			case HTTP_REQUEST_START:
-				puts("==request line==");
 				if (NULL == (line = httpRequestLineGet(&data))) {
 				if (NULL == (line = httpRequestLineGet(&data))) {
 					cont = 0;
 					cont = 0;
 					break;
 					break;
@@ -193,7 +191,6 @@ httpRequestParserParse(HttpRequestParser this)
 				break;
 				break;
 
 
 			case HTTP_REQUEST_REQUEST_LINE_DONE:
 			case HTTP_REQUEST_REQUEST_LINE_DONE:
-				puts("==read header==");
 				if (NULL == (line = httpRequestLineGet(&data))) {
 				if (NULL == (line = httpRequestLineGet(&data))) {
 					cont = 0;
 					cont = 0;
 					break;
 					break;
@@ -222,8 +219,6 @@ httpRequestParserParse(HttpRequestParser this)
 				break;
 				break;
 
 
 			case HTTP_REQUEST_HEADERS_DONE:
 			case HTTP_REQUEST_HEADERS_DONE:
-				puts("==headers done==");
-
 				/**
 				/**
 				 * @TODO: here comes the body handling
 				 * @TODO: here comes the body handling
 				 */
 				 */
@@ -231,8 +226,6 @@ httpRequestParserParse(HttpRequestParser this)
 				break;
 				break;
 
 
 			case HTTP_REQUEST_DONE:
 			case HTTP_REQUEST_DONE:
-				puts("==request done==");
-
 				/**
 				/**
 				 * enqueue current request
 				 * enqueue current request
 				 */
 				 */

@@ -6,10 +6,14 @@
 void
 void
 serverCloseConn(Server this, unsigned int i)
 serverCloseConn(Server this, unsigned int i)
 {
 {
-	delete(&((this->conns)[i].sock));
-	delete(&((this->conns)[i].reader));
-	CLEAR_CONN(this, i);
-	this->nfds--;
+	int fd = (this->fds)[i].fd;
+
+	delete(&((this->conns)[fd].sock));
+	delete(&((this->conns)[fd].reader));
+	(this->fds)[i].events = 0;
+	this->ndel++;
+//	CLEAR_CONN(this, i);
+//	this->nfds--;
 }
 }
 
 
 // vim: set ts=4 sw=4:
 // vim: set ts=4 sw=4:

@@ -2,28 +2,26 @@ static
 void
 void
 serverHandleAccept(Server this)
 serverHandleAccept(Server this)
 {
 {
-	if (0 != ((this->fds)[0].revents & POLLIN)) {
-		char remoteAddr[16] = "";
-		Sock acc;
+	char remoteAddr[16] = "";
+	Sock acc;
 
 
-		acc = socketAccept(this->sock, remoteAddr);
+	acc = socketAccept(this->sock, remoteAddr);
 
 
-		if (-1 != acc->handle) {
-			//* save the socket handle
-			(this->conns)[this->nfds].sock   = acc; 
+	if (-1 != acc->handle) {
+		//* save the socket handle
+		(this->conns)[acc->handle].sock   = acc; 
 
 
-			//* clone reader
-			(this->conns)[this->nfds].reader = clone(this->reader);
+		//* clone reader
+		(this->conns)[acc->handle].reader = clone(this->reader);
 
 
-			(this->fds)[this->nfds].fd       = acc->handle;
-			(this->fds)[this->nfds].events   = POLLIN;
-			this->nfds++;
-		} else {
-			delete(&acc);
-		}
-
-		(this->fds)[0].revents |= POLLIN;
+		(this->fds)[this->nfds].fd       = acc->handle;
+		(this->fds)[this->nfds].events   = POLLIN;
+		this->nfds++;
+	} else {
+		delete(&acc);
 	}
 	}
+
+	(this->fds)[0].revents |= POLLIN;
 }
 }
 
 
 // vim: set ts=4 sw=4:
 // vim: set ts=4 sw=4:

+#define POLLFD(ptr)	((struct pollfd *)(ptr))
+
+static
+inline
+int
+sortEvents(const void * a, const void * b)
+{
+	return POLLFD(a)->events > POLLFD(b)->events ?
+		-1 : POLLFD(a)->events < POLLFD(b)->events ?
+		1 : 0;
+}
+
+static
+inline
+int
+sortRevents(const void * a, const void * b)
+{
+	return POLLFD(a)->revents > POLLFD(b)->revents ?
+		-1 : POLLFD(a)->revents < POLLFD(b)->revents ?
+		1 : 0;
+}
+
 static
 static
 int
 int
 serverPoll(Server this) {
 serverPoll(Server this) {
 	int events;
 	int events;
 
 
+	qsort(this->fds, this->nfds, sizeof(struct pollfd), sortEvents);
+	this->nfds -= this->ndel;
+	this->ndel  = 0;
+
 	/*
 	/*
 	 * wait for handles to become ready
 	 * wait for handles to become ready
 	 */
 	 */
@@ -23,6 +49,8 @@ serverPoll(Server this) {
 		}
 		}
 	}
 	}
 
 
+	qsort(this->fds, this->nfds, sizeof(struct pollfd), sortRevents);
+
 	return events;
 	return events;
 }
 }
 
 

 static
 static
 int
 int
-serverRead(Server this)
+serverRead(Server this, unsigned int i)
 {
 {
-	unsigned int i;
+	int fd = (this->fds)[i].fd;
 
 
-	for (i=1; i<this->nfds; i++) {
-		if (0 != ((this->fds)[i].revents & POLLIN)) {
-			if (NULL == (this->conns)[i].reader) {
-				loggerLog(
-						this->logger,
-						LOGGER_INFO,
-						"initialization error: NULL reader");
-				serverCloseConn(this, i);
-			}
+	if (NULL == (this->conns)[fd].reader) {
+		loggerLog(
+				this->logger,
+				LOGGER_INFO,
+				"initialization error: NULL reader");
+		serverCloseConn(this, i);
+	}
 
 
-			switch (streamReaderRead((this->conns)[i].reader, (this->fds)[i].fd)) {
-				case 0:
-					/*
-					 * normal close: write remaining data
-					 * @TODO: actually we have no remaining data here....
-					 */
-					/* DROP-THROUGH */
+	switch (streamReaderRead((this->conns)[fd].reader, fd)) {
+		case 0:
+			/*
+			 * normal close: write remaining data
+			 * @TODO: actually we have no remaining data here....
+			 */
+			/* DROP-THROUGH */
 
 
-				case -1: 
-					/*
-					 * read failure / close connection
-					 */
-					loggerLog(this->logger, LOGGER_INFO, "connection closed...");
-					serverCloseConn(this, i);
-					break;
+		case -1: 
+			/*
+			 * read failure / close connection
+			 */
+			loggerLog(this->logger, LOGGER_INFO, "connection closed...");
+			serverCloseConn(this, i);
+			break;
 
 
-				default:
-					break;
-			}
-		}
+		default:
+			break;
 	}
 	}
-	
+
 	return 0;
 	return 0;
 }
 }
 
 

@@ -3,6 +3,7 @@
 #include <stdlib.h>     /* for exit */
 #include <stdlib.h>     /* for exit */
 #include <errno.h>      /* for errno */
 #include <errno.h>      /* for errno */
 #include <unistd.h>
 #include <unistd.h>
+#include <time.h>
 
 
 #include "server.h"
 #include "server.h"
 #include "socket.h"
 #include "socket.h"
@@ -42,64 +43,100 @@ serverRun(Server this)
 	 */
 	 */
     while (!doShutdown) /* until error or signal  */
     while (!doShutdown) /* until error or signal  */
     {
     {
-		int events;
+		int          events;
+		unsigned int i;
 
 
-		/**
-		 * @TODO take return value of poll into account with
-		 * further handling!
-		 */
 		events = serverPoll(this);
 		events = serverPoll(this);
 		if (doShutdown) break;
 		if (doShutdown) break;
 
 
-        /**
-         * handle accept
-         */
-		serverHandleAccept(this);
-
-        /**
-		 * handle reads 
-		 */
-		serverRead(this);
-
-		/**
-		 * do some other processing
-		 * @TODO: actually this will hard assume that our stream reader
-		 * is a http parser and it has its queue...think about more
-		 * generalizing here.
-		 */
-		{
-			int i;
-
-			for (i=1; i<this->nfds; i++) {
-				int              j;
-				HttpRequestQueue queue =
-					((HttpRequestParser)(this->conns)[i].reader)->request_queue;
-
-				for (j=0; j<queue->nrequests; j++) {
-					int         k;
-					HttpRequest request = queue->requests[j];
-
-					printf("method: %s\n", request->method);
-					printf("uri: %s\n", request->uri);
-					printf("version: %s\n", request->http_version);
-					puts("");
-
-					for (k=0; k<128; k++) {
-						if (NULL == (request->header)[k].name) break;
-						printf("header-name:  %s\n", (request->header)[k].name);
-						printf("header-value: %s\n", (request->header)[k].value);
-					}
+		for (i=0; i < events; i++) {
+			int fd = (this->fds)[i].fd;
+			//int nreads = 0, nwrites = 0;
 
 
-					delete(&request);
+			if (0 != ((this->fds)[i].revents & POLLIN)) {
+				/**
+				 * handle accept
+				 */
+				if (this->sock->handle == (this->fds)[i].fd) {
+					serverHandleAccept(this);
 				}
 				}
-				queue->nrequests = 0;
-			}
-		}
 
 
+				/**
+				 * handle reads 
+				 */
+				else {
+					serverRead(this, i);
+
+					/**
+					 * do some other processing
+					 * @TODO: actually this will hard assume that our stream reader
+					 * is a http parser and it has its queue...think about more
+					 * generalizing here.
+					 */
+					{
+						int              j;
+						HttpRequestQueue queue =
+							((HttpRequestParser)(this->conns)[fd].reader)->request_queue;
+
+						for (j=0; j<queue->nrequests; j++) {
+							HttpRequest request = queue->requests[j];
+
+							/**
+							 * @TODO: for now simply remove request and send not found.
+							 * Make this sane.
+							 */
+							delete(&request);
+
+							/**
+							 * @TODO: the complete response stuff have to be removed here.
+							 */
+							time_t t;
+							struct tm * tmp;
+							char timestr[200];
+
+#define RESP_HEAD "HTTP/1.1 404 Not Found\r\n" \
+							"Content-Type: text/html\r\n" \
+							"Content-Length: %lu\r\n" \
+							"Date: %s\r\n" \
+							"Server: testserver\r\n"
+
+#define RESP_DATA "<?xml version=\"1.0\" encoding=\"iso-8859-1\"?>\n" \
+							"<!DOCTYPE html PUBLIC \"-//W3C//DTD XHTML 1.0 Transitional//EN\"\n" \
+							" \"http://www.w3.org/TR/xhtml1/DTD/xhtml1-transitional.dtd\">\n" \
+							"<html xmlns=\"http://www.w3.org/1999/xhtml\" xml:lang=\"en\" lang=\"en\">\n" \
+							"<head><title>404 - Not Found</title></head>" \
+							"<body><h1>404 - Not Found</h1></body>" \
+							"</html>"
+
+							t = time(NULL);
+							tmp = localtime(&t);
+							strftime(timestr, sizeof(timestr), "%a, %d %b %Y %T %Z", tmp);
+
+							/**
+							 * @TODO: just to send an answer and be able to make some
+							 * apache benchs i do it here...this definetly MUST BE moved
+							 */
+							sprintf((this->conns)[fd].wbuf, RESP_HEAD "\r\n" RESP_DATA, sizeof(RESP_DATA), timestr);
+							(this->fds)[i].events = (this->fds)[i].events | POLLOUT;
+						}
+
+						queue->nrequests = 0;
+					}
+				}
+			}
 
 
-		/**
-		 * handle writes
-		 */
+			/**
+			 * handle writes
+			 */
+			if (0 != ((this->fds)[i].revents & POLLOUT)) {
+				write(
+						(this->fds)[i].fd,
+						(this->conns)[fd].wbuf,
+						strlen((this->conns)[fd].wbuf));
+				(this->fds)[i].events = (this->fds)[i].events & ~POLLOUT;
+				serverCloseConn(this, i);
+			}
+		}
     }
     }
 }
 }
 
 

@@ -21,7 +21,14 @@ ctor(void * _this, va_list * params)
 		loggerLog(this->log, LOGGER_CRIT,
 		loggerLog(this->log, LOGGER_CRIT,
 				"error opening socket: %s - service terminated",
 				"error opening socket: %s - service terminated",
 				strerror(errno));
 				strerror(errno));
-		exit(EXIT_FAILURE);
+		//exit(EXIT_FAILURE);
+		/**
+		 * @TODO: uhhhh, here we are leaking memory the socket is not
+		 * initialized correctly and no one notices...
+		 * Think about a way to prevent this.
+		 * Well maybe we notice in server....
+		 */
+		return;
 	}
 	}
 
 
 	/* Make the socket REUSE a TIME_WAIT socket */
 	/* Make the socket REUSE a TIME_WAIT socket */

@@ -22,7 +22,7 @@ socketAccept(Sock this, char remoteAddr[16])
     sock->handle = accept(this->handle, (struct sockaddr *) &(sock->addr), &len);
     sock->handle = accept(this->handle, (struct sockaddr *) &(sock->addr), &len);
     if (-1 == sock->handle) {
     if (-1 == sock->handle) {
         loggerLog(this->log, LOGGER_WARNING,
         loggerLog(this->log, LOGGER_WARNING,
-                "error acception connection: %s", strerror(errno));
+                "error accepting connection: %s", strerror(errno));
     } else {
     } else {
         loggerLog(this->log, LOGGER_INFO,
         loggerLog(this->log, LOGGER_INFO,
 				"handling client %s\n", inet_ntoa((sock->addr).sin_addr));
 				"handling client %s\n", inet_ntoa((sock->addr).sin_addr));

@@ -16,6 +16,7 @@ main()
 	Logger            logger = new(LoggerStderr, LOGGER_INFO);
 	Logger            logger = new(LoggerStderr, LOGGER_INFO);
 	HttpRequestParser parser = new(HttpRequestParser);
 	HttpRequestParser parser = new(HttpRequestParser);
 	Server            server = new(Server, logger, parser, 11212, SOMAXCONN);
 	Server            server = new(Server, logger, parser, 11212, SOMAXCONN);
+	//Server            server = new(Server, logger, parser, 11212, 20);
 
 
 	init_signals();
 	init_signals();
 	serverRun(server);
 	serverRun(server);