improved http sample code

qse/configure

@@ -16680,7 +16680,7 @@ QSE_PROJECT_AUTHOR="${PACKAGE_BUGREPORT}"
 QSE_PROJECT_URL="${PACKAGE_URL}"
 
 
-ac_config_files="$ac_config_files Makefile README include/Makefile include/qse/Makefile include/qse/cmn/Makefile include/qse/awk/Makefile include/qse/cut/Makefile include/qse/sed/Makefile include/qse/stx/Makefile lib/Makefile lib/cmn/Makefile lib/awk/Makefile lib/cut/Makefile lib/sed/Makefile lib/stx/Makefile cmd/Makefile cmd/awk/Makefile cmd/cut/Makefile cmd/sed/Makefile cmd/stx/Makefile samples/Makefile samples/cmn/Makefile samples/awk/Makefile samples/cut/Makefile samples/sed/Makefile regress/Makefile regress/awk/Makefile regress/sed/Makefile doc/Makefile doc/page/Makefile doc/Doxyfile"
+ac_config_files="$ac_config_files Makefile README include/Makefile include/qse/Makefile include/qse/cmn/Makefile include/qse/awk/Makefile include/qse/cut/Makefile include/qse/sed/Makefile include/qse/stx/Makefile lib/Makefile lib/cmn/Makefile lib/awk/Makefile lib/cut/Makefile lib/sed/Makefile lib/stx/Makefile lib/http/Makefile cmd/Makefile cmd/awk/Makefile cmd/cut/Makefile cmd/sed/Makefile cmd/stx/Makefile samples/Makefile samples/cmn/Makefile samples/awk/Makefile samples/cut/Makefile samples/sed/Makefile samples/http/Makefile regress/Makefile regress/awk/Makefile regress/sed/Makefile doc/Makefile doc/page/Makefile doc/Doxyfile"
 
 cat >confcache <<\_ACEOF
 # This file is a shell script that caches the results of configure
@@ -17786,6 +17786,7 @@ do
     "lib/cut/Makefile") CONFIG_FILES="$CONFIG_FILES lib/cut/Makefile" ;;
     "lib/sed/Makefile") CONFIG_FILES="$CONFIG_FILES lib/sed/Makefile" ;;
     "lib/stx/Makefile") CONFIG_FILES="$CONFIG_FILES lib/stx/Makefile" ;;
+    "lib/http/Makefile") CONFIG_FILES="$CONFIG_FILES lib/http/Makefile" ;;
     "cmd/Makefile") CONFIG_FILES="$CONFIG_FILES cmd/Makefile" ;;
     "cmd/awk/Makefile") CONFIG_FILES="$CONFIG_FILES cmd/awk/Makefile" ;;
     "cmd/cut/Makefile") CONFIG_FILES="$CONFIG_FILES cmd/cut/Makefile" ;;
@@ -17796,6 +17797,7 @@ do
     "samples/awk/Makefile") CONFIG_FILES="$CONFIG_FILES samples/awk/Makefile" ;;
     "samples/cut/Makefile") CONFIG_FILES="$CONFIG_FILES samples/cut/Makefile" ;;
     "samples/sed/Makefile") CONFIG_FILES="$CONFIG_FILES samples/sed/Makefile" ;;
+    "samples/http/Makefile") CONFIG_FILES="$CONFIG_FILES samples/http/Makefile" ;;
     "regress/Makefile") CONFIG_FILES="$CONFIG_FILES regress/Makefile" ;;
     "regress/awk/Makefile") CONFIG_FILES="$CONFIG_FILES regress/awk/Makefile" ;;
     "regress/sed/Makefile") CONFIG_FILES="$CONFIG_FILES regress/sed/Makefile" ;;

qse/configure.ac

@@ -242,6 +242,7 @@ AC_CONFIG_FILES([
 	lib/cut/Makefile 
 	lib/sed/Makefile 
 	lib/stx/Makefile 
+	lib/http/Makefile 
 	cmd/Makefile 
 	cmd/awk/Makefile 
 	cmd/cut/Makefile 
@@ -252,6 +253,7 @@ AC_CONFIG_FILES([
 	samples/awk/Makefile 
 	samples/cut/Makefile 
 	samples/sed/Makefile 
+	samples/http/Makefile
 	regress/Makefile
 	regress/awk/Makefile
 	regress/sed/Makefile

qse/lib/http/http.c

@@ -989,7 +989,6 @@ done:
 	return ptr;
 }
 
-
 /* feed the percent encoded string */
 int qse_http_feed (qse_http_t* http, const qse_byte_t* req, qse_size_t len)
 {
@@ -1193,8 +1192,11 @@ int qse_http_feed (qse_http_t* http, const qse_byte_t* req, qse_size_t len)
 						}
 					}
 
-					QSE_ASSERTX (http->reqcbs->request != QSE_NULL,
+					QSE_ASSERTX (
-						"set request callbacks before feeding");
+						http->reqcbs->request != QSE_NULL,
+						"set request callbacks before feeding"
+					);
+
 					http->errnum = QSE_HTTP_ENOERR;
 					if (http->reqcbs->request (http, &http->req) <= -1)
 					{
@@ -1248,19 +1250,3 @@ feedme_more:
 	return 0;
 }
 
-int qse_http_addtext (qse_http_t* http, const qse_byte_t* ptr, qse_size_t len)
-{
-}
-
-int qse_http_addresource (qse_http_t* http, const void* ptr, qse_size_t len)
-{
-}
-
-int qse_http_addheader (
-	qse_http_t* http, const qse_byte_t* key, const qse_byte_t* val)
-{
-}
-
-int qse_http_emit (qse_http_t* http)
-{
-}

qse/samples/http/Makefile.am

@@ -8,7 +8,7 @@ AM_CPPFLAGS = \
 bin_PROGRAMS = http01 
 
 LDFLAGS = -L../../lib/cmn -L../../lib/http
-LDADD = -lqsehttp -lqsecmn $(LIBM)
+LDADD = -lqsehttp -lqsecmn $(LIBM) -lpthread
 
 http01_SOURCES = http01.c
 

qse/samples/http/Makefile.in

@@ -216,7 +216,7 @@ AM_CPPFLAGS = \
 	-I$(top_srcdir)/include \
 	-I$(includedir)
 
-LDADD = -lqsehttp -lqsecmn $(LIBM)
+LDADD = -lqsehttp -lqsecmn $(LIBM) -lpthread
 http01_SOURCES = http01.c
 all: all-am
 

qse/samples/http/http01.c

@@ -12,18 +12,51 @@
 #include <unistd.h>
 #include <errno.h>
 #include <signal.h>
+#include <pthread.h>
+#include <sys/time.h>
 
 #include <sys/sendfile.h>
 #include <sys/stat.h>
 
+#define MAX_SENDFILE_SIZE 4096
+//#define MAX_SENDFILE_SIZE 64
+
 typedef struct client_t client_t;
 typedef struct client_array_t client_array_t;
 
+typedef struct client_action_t client_action_t;
+struct client_action_t
+{
+	enum
+	{
+		ACTION_SENDTEXT,
+		ACTION_SENDFILE
+	} type;
+
+	union
+	{
+		struct
+		{
+			int   fd;
+			off_t left;
+			off_t offset;
+		} sendfile;
+	} u;
+};
+
 struct client_t
 {
 	int                     fd;
 	struct sockaddr_storage addr;
 	qse_http_t*             http;
+
+	pthread_mutex_t action_mutex;
+	struct
+	{
+		int             offset;
+		int             count;
+		client_action_t target[10];
+	} action;
 };
 
 struct client_array_t
@@ -31,6 +64,16 @@ struct client_array_t
 	int       capa;
 	int       size;
 	client_t* data;
+	pthread_cond_t cond;
+};
+
+typedef struct appdata_t appdata_t;
+
+struct appdata_t
+{
+	client_array_t ca;
+	pthread_mutex_t camutex;
+	pthread_cond_t cacond;
 };
 
 typedef struct http_xtn_t http_xtn_t;
@@ -41,85 +84,131 @@ struct http_xtn_t
 	qse_size_t      index; 
 };
 
-int handle_request (qse_http_t* http, qse_http_req_t* req)
+static int enqueue_client_action_unlocked (client_t* client, const client_action_t* action)
+{
+	int index;
+
+	if (client->action.count >= QSE_COUNTOF(client->action.target)) return -1;
+	
+	index = (client->action.offset + client->action.count) % 
+	        QSE_COUNTOF(client->action.target);
+	client->action.target[index] = *action;
+	client->action.count++;
+	return 0;
+}
+
+static int enqueue_client_action_locked (client_t* client, const client_action_t* action)
+{
+	int ret;
+	pthread_mutex_lock (&client->action_mutex);
+	ret = enqueue_client_action_unlocked (client, action);
+	pthread_mutex_unlock (&client->action_mutex);
+	return ret;
+}
+
+static int dequeue_client_action_unlocked (client_t* client, client_action_t* action)
+{
+	if (client->action.count <= 0) return -1;
+
+	if (action) *action = client->action.target[client->action.offset];
+	client->action.offset = (client->action.offset + 1) % QSE_COUNTOF(client->action.target);
+	client->action.count--;
+	return 0;
+}
+
+static int dequeue_client_action_locked (client_t* client, client_action_t* action)
+{
+	int ret;
+	pthread_mutex_lock (&client->action_mutex);
+	ret = dequeue_client_action_unlocked (client, action);
+	pthread_mutex_unlock (&client->action_mutex);
+	return ret;
+}
+
+static void purge_client_actions_locked (client_t* client)
+{
+	client_action_t action;
+	pthread_mutex_lock (&client->action_mutex);
+	while (dequeue_client_action_unlocked (client, &action) == 0)
+	{
+		if (action.type == ACTION_SENDFILE) close (action.u.sendfile.fd);
+	}
+	pthread_mutex_unlock (&client->action_mutex);
+}
+
+static int handle_request (qse_http_t* http, qse_http_req_t* req)
 {
 	http_xtn_t* xtn = (http_xtn_t*) qse_http_getxtn (http);
 	qse_printf (QSE_T("got a request... %S\n"), req->path.ptr);
 
 	if (req->method == QSE_HTTP_REQ_GET)
 	{
-#if 0
-		qse_http_rep_t* rep;
-
 		int fd = open (req->path.ptr, O_RDONLY);
 		if (fd <= -1)
 		{
+qse_printf (QSE_T("open failure....\n"));
+			/*
 			qse_http_addtext (http, 
 				"<html><title>FILE NOT FOUND</title><body></body></html>");
+			*/
 		}
 		else
 		{
-		#if 0
-			qse_http_addheader (
-				http, "Content-Type", detect_file_type(req->path.ptr)
-			);
-
-			qse_http_addtext (http, .....);
-			qse_http_adddata (http, 
-
-			qse_http_addoutputdata (http, fd);
-		#endif
-
-			/*
 			struct stat st;
-
+			if (fstat (fd, &st) <= -1)
-			fstat (fd, &st);
+			{
-			sendfile (xtn->array->data[xtn->index].fd, fd, NULL, st.st_size);
 				close (fd);
-			*/
-		}	
 
-		rep = qse_http_emit (http);
+qse_printf (QSE_T("fstat failure....\n"));
-		if (rep == NULL)
+			}
+			else if (st.st_size <= 0)
 			{
-			/* ERROR */
+				close (fd);
-		}
-#endif
-	}
-
-	return 0;
-}
 
+qse_printf (QSE_T("empty file....\n"));
 #if 0
-int handle_action (qse_http_t* http, qse_http_req_t* req)
+				qse_http_req_t* rep = qse_http_newreply (http);
+				if (req == QSE_NULL)
 				{
-	http_xtn_t* xtn = (http_xtn_t*) qse_http_getxtn (http);
+					/* hard failure... can't answer */
+					/* arrange to close connection */
+				}
 
-	switch (xtn->action.type)
-	{
-		case SENDFILE:
-		{
-			off_t offset = 0;	
 
-			fstat (fd, &st);
+				ptr = qse_http_emitreply (http, rep, &len);
-			sendfile (
+				if (ptr == QSE_NULL)
-				xtn->array->data[xtn->index].fd,  /* socket */
-				fd, /* input file descriptor */
-				&offset, 
-				st.st_size - offset
-			);
-
-			if (offset >= st.st_size)
 				{
-				/* done */
+					/* hard failure... can't answer */
-				xtn->action.type = NONE;
+					/* arrange to close connection */
+				}
+
+				action.type = ACTION_SENDTEXT;
+				action.u.sendtext.ptr = ptr;
+				action.u.sendtext.len = len;
+#endif
+			}
+			else
+			{
+				client_t* client = &xtn->array->data[xtn->index];
+				client_action_t action;
+
+				memset (&action, 0, sizeof(action));
+				action.type = ACTION_SENDFILE;
+				action.u.sendfile.fd = fd;
+				action.u.sendfile.left = st.st_size;;
+
+				if (enqueue_client_action_locked (client, &action) <= -1)
+				{
+	/* TODO: close??? send error page ...*/
+qse_printf (QSE_T("failed to push action....\n"));
+				}
 			}
 		}	
 	}
 
+	pthread_cond_signal (&xtn->array->cond);
 	return 0;
 }
-#endif
 
 qse_http_reqcbs_t http_reqcbs =
 {
@@ -160,18 +249,21 @@ int mkserver (const char* portstr)
 	return s;
 }
 
-
 static void init_client_array (client_array_t* array)
 {
 	array->capa = 0;
 	array->size = 0;
 	array->data = QSE_NULL;
+	pthread_cond_init (&array->cond, NULL);
 }
 
 static void delete_from_client_array (client_array_t* array, int fd)
 {
 	if (array->data[fd].http)
 	{
+		purge_client_actions_locked (&array->data[fd]);
+		pthread_mutex_destroy (&array->data[fd].action_mutex);
+
 		qse_http_close (array->data[fd].http);
 		array->data[fd].http = QSE_NULL;	
 		close (array->data[fd].fd);
@@ -193,6 +285,7 @@ static void fini_client_array (client_array_t* array)
 		array->size = 0;
 		array->data = QSE_NULL;
 	}
+	pthread_cond_destroy (&array->cond);
 }
 
 static client_t* insert_into_client_array (
@@ -222,6 +315,7 @@ static client_t* insert_into_client_array (
 	array->data[fd].addr = *addr;
 	array->data[fd].http = qse_http_open (QSE_MMGR_GETDFL(), QSE_SIZEOF(*xtn));
 	if (array->data[fd].http == QSE_NULL) return QSE_NULL;
+	pthread_mutex_init (&array->data[fd].action_mutex, NULL);
 
 	xtn = (http_xtn_t*)qse_http_getxtn (array->data[fd].http);	
 	xtn->array = array;
@@ -251,10 +345,18 @@ static int make_fd_set_from_client_array (
 	{
 		if (ca->data[fd].http) 
 		{
-			if (r) FD_SET (ca->data[fd].fd, r);
+			if (r) 
-			if (w) FD_SET (ca->data[fd].fd, w);
+			{
+				FD_SET (ca->data[fd].fd, r);
 				if (ca->data[fd].fd > max) max = ca->data[fd].fd;
 			}
+			if (w && ca->data[fd].action.count > 0)
+			{
+				/* add it to the set if it has a reply to send */
+				FD_SET (ca->data[fd].fd, w);
+				if (ca->data[fd].fd > max) max = ca->data[fd].fd;
+			}
+		}
 	}
 
 	return max;
@@ -262,15 +364,133 @@ static int make_fd_set_from_client_array (
 
 static int quit = 0;
 
-void sigint (int sig)
+static void sigint (int sig)
 {
 	quit = 1;
 }
 
+static int take_client_action (client_t* client)
+{
+	client_action_t* action;
+
+	action = &client->action.target[client->action.offset];
+
+	switch (action->type)
+	{
+		case ACTION_SENDTEXT:
+		{
+			break;
+		}
+
+		case ACTION_SENDFILE:
+		{
+			ssize_t n;
+			size_t count;
+
+			count = MAX_SENDFILE_SIZE;
+			if (count >= action->u.sendfile.left)
+				count = action->u.sendfile.left;
+
+			n = sendfile (
+				client->fd, 
+				action->u.sendfile.fd,
+				&action->u.sendfile.offset,
+				count
+			);
+
+			if (n <= -1) 
+			{
+qse_printf (QSE_T("sendfile failure... arrange to close this connection....\n"));
+				close (action->u.sendfile.fd);
+				dequeue_client_action_locked (client, NULL);
+			}
+			else
+			{
+				action->u.sendfile.left -= n;
+
+				if (action->u.sendfile.left <= 0)
+				{
+qse_printf (QSE_T("finished sending...\n"));
+					close (action->u.sendfile.fd);
+					dequeue_client_action_locked (client, NULL);
+				}
+			}
+
+			break;
+		}
+	}
+
+	return 0;
+}
+
+static void* replier_thread (void* arg)
+{
+	appdata_t* appdata = (appdata_t*)arg;
+
+	while (!quit)
+	{
+		int n, max, fd;
+		fd_set w;
+		struct timeval tv;
+
+		tv.tv_sec = 1;
+		tv.tv_usec = 0;
+
+		pthread_mutex_lock (&appdata->camutex);
+		max = make_fd_set_from_client_array (&appdata->ca, -1, NULL, &w);
+		pthread_mutex_unlock (&appdata->camutex);
+
+		while (max == -1 && !quit)
+		{
+			struct timeval now;
+			struct timespec timeout;
+
+			pthread_mutex_lock (&appdata->camutex);
+
+			gettimeofday (&now, NULL);
+			timeout.tv_sec = now.tv_sec + 2;
+			timeout.tv_nsec = now.tv_usec * 1000;
+
+			pthread_cond_timedwait (&appdata->ca.cond, &appdata->camutex, &timeout);
+			max = make_fd_set_from_client_array (&appdata->ca, -1, NULL, &w);
+
+			pthread_mutex_unlock (&appdata->camutex);
+		}
+
+		if (quit) break;
+
+		n = select (max + 1, NULL, &w, NULL, &tv);
+		if (n <= -1)
+		{
+			if (errno == EINTR) continue;
+			qse_fprintf (QSE_STDERR, QSE_T("Error: select returned failure\n"));
+			break;
+		}
+		if (n == 0) continue;
+
+		for (fd = 0; fd < appdata->ca.capa; fd++)
+		{
+			client_t* client = &appdata->ca.data[fd];
+
+			if (!client->http) continue;
+
+			if (FD_ISSET(client->fd, &w)) 
+			{
+				if (client->action.count > 0) take_client_action (client);
+			}
+		
+		}
+	}
+
+	pthread_exit (NULL);
+	return NULL;
+}
+
 int main (int argc, char* argv[])
 {
 	int s;
-	client_array_t ca;
+	pthread_t replier_thread_id;
+	appdata_t appdata;
 
 	if (argc != 2)
 	{
@@ -291,19 +511,24 @@ int main (int argc, char* argv[])
 		return -1;
 	}
 
-	init_client_array (&ca);
+	/* data receiver main logic */
+	init_client_array (&appdata.ca);
+	pthread_mutex_init (&appdata.camutex, NULL);
+
+	/* start the reply sender as a thread */
+	pthread_create (&replier_thread_id, NULL, replier_thread, &appdata);
 
 	while (!quit)
 	{
 		int n, max, fd;
-		fd_set r, w;
+		fd_set r;
 		struct timeval tv;
 
 		tv.tv_sec = 1;
 		tv.tv_usec = 0;
 
-		max = make_fd_set_from_client_array (&ca, s, &r, &w);
+		max = make_fd_set_from_client_array (&appdata.ca, s, &r, NULL);
-		n = select (max + 1, &r, &w, NULL, &tv);
+		n = select (max + 1, &r, NULL, NULL, &tv);
 		if (n <= -1)
 		{
 			if (errno == EINTR) continue;
@@ -314,10 +539,12 @@ int main (int argc, char* argv[])
 
 		if (FD_ISSET(s, &r))
 		{
-			int flag;
+			int flag, c;
 			struct sockaddr_storage addr;
 			socklen_t addrlen = sizeof(addr);
-			int c = accept (s, (struct sockaddr*)&addr, &addrlen);
+			client_t* client;
+
+			c = accept (s, (struct sockaddr*)&addr, &addrlen);
 			if (c <= -1)
 			{
 				if (errno != EINTR) 
@@ -340,17 +567,21 @@ int main (int argc, char* argv[])
 			flag = fcntl (c, F_GETFL);
 			if (flag >= 0) fcntl (c, F_SETFL, flag | O_NONBLOCK);
 
-			if (insert_into_client_array (&ca, c, &addr) == QSE_NULL)
+			pthread_mutex_lock (&appdata.camutex);
+			client = insert_into_client_array (&appdata.ca, c, &addr);
+			pthread_mutex_unlock (&appdata.camutex);
+			if (client == QSE_NULL)
 			{
 				close (c);
 				qse_fprintf (QSE_STDERR, QSE_T("Error: failed to add a client\n"));
 				continue;
 			}
+qse_printf (QSE_T("connection %d accepted\n"), c);
 		}
 
-		for (fd = 0; fd < ca.capa; fd++)
+		for (fd = 0; fd < appdata.ca.capa; fd++)
 		{
-			client_t* client = &ca.data[fd];
+			client_t* client = &appdata.ca.data[fd];
 
 			if (!client->http) continue;
 
@@ -367,14 +598,18 @@ int main (int argc, char* argv[])
 				{
 					if (errno != EINTR)
 					{
-						delete_from_client_array (&ca, fd);	
+						pthread_mutex_lock (&appdata.camutex);
+						delete_from_client_array (&appdata.ca, fd);	
+						pthread_mutex_unlock (&appdata.camutex);
 						qse_fprintf (QSE_STDERR, QSE_T("Error: failed to read from a client %d\n"), fd);
 					}
 					goto reread;
 				}
 				else if (m == 0)
 				{
-					delete_from_client_array (&ca, fd);	
+					pthread_mutex_lock (&appdata.camutex);
+					delete_from_client_array (&appdata.ca, fd);	
+					pthread_mutex_unlock (&appdata.camutex);
 					qse_fprintf (QSE_STDERR, QSE_T("Debug: connection closed %d\n"), fd);
 					continue;
 				}
@@ -390,27 +625,20 @@ int main (int argc, char* argv[])
 						/* TODO: write a reply to indicate bad request... */	
 					}
 
-					delete_from_client_array (&ca, fd);	
+					pthread_mutex_lock (&appdata.camutex);
+					delete_from_client_array (&appdata.ca, fd);	
+					pthread_mutex_unlock (&appdata.camutex);
 					qse_fprintf (QSE_STDERR, QSE_T("Error: http error while processing \n"));
 					continue;
 				}
 			}
-
-		#if 0
-			if (FD_ISSET(client->fd, &w) && client->rep)
-			{
-				/* ready to send output */
-				if (handle_output (xxxxx) <= -1)
-				{
-					delete_from_client_array (&ca, fd);	
-					qse_fprintf (QSE_STDERR, QSE_T("Error: output error\n"));
-				}
-			}
-		#endif
 		}
 	}
 
-	fini_client_array (&ca);
+	pthread_join (replier_thread_id, NULL);
+
+	fini_client_array (&appdata.ca);
+	pthread_mutex_destroy (&appdata.camutex);
 
 	close (s);
 	return 0;