enhanced qse_upxd_t

qse/lib/cmn/rex.c

@@ -1753,7 +1753,7 @@ static qse_lda_walk_t walk_cands_for_match (
 			{
 				int equal;
 
-				equal =(e->rex->option & QSE_REX_IGNORECASE)?
+				equal = (e->rex->option & QSE_REX_IGNORECASE)?
 					(QSE_TOUPPER(node->u.c) == QSE_TOUPPER(*cand->mptr)):
 					(node->u.c == *cand->mptr) ;
 

qse/lib/net/Makefile.am

@@ -7,10 +7,11 @@ AM_CPPFLAGS = \
 
 lib_LTLIBRARIES = libqsenet.la
 libqsenet_la_SOURCES = \
+	httpd.h \
+	upxd.h \
 	http.c \
 	htre.c \
 	htrd.c \
-	httpd.h \
 	httpd.c \
 	httpd-cgi.c \
 	httpd-proxy.c \

qse/lib/net/Makefile.in

@@ -79,7 +79,8 @@ am__installdirs = "$(DESTDIR)$(libdir)"
 LTLIBRARIES = $(lib_LTLIBRARIES)
 libqsenet_la_DEPENDENCIES =
 am_libqsenet_la_OBJECTS = http.lo htre.lo htrd.lo httpd.lo \
-	httpd-cgi.lo httpd-proxy.lo httpd-resol.lo httpd-task.lo
+	httpd-cgi.lo httpd-proxy.lo httpd-resol.lo httpd-task.lo \
+	upxd.lo
 libqsenet_la_OBJECTS = $(am_libqsenet_la_OBJECTS)
 libqsenet_la_LINK = $(LIBTOOL) --tag=CC $(AM_LIBTOOLFLAGS) \
 	$(LIBTOOLFLAGS) --mode=link $(CCLD) $(AM_CFLAGS) $(CFLAGS) \
@@ -256,15 +257,17 @@ AM_CPPFLAGS = \
 
 lib_LTLIBRARIES = libqsenet.la
 libqsenet_la_SOURCES = \
+	httpd.h \
+	upxd.h \
 	http.c \
 	htre.c \
 	htrd.c \
-	httpd.h \
 	httpd.c \
 	httpd-cgi.c \
 	httpd-proxy.c \
 	httpd-resol.c \
-	httpd-task.c
+	httpd-task.c \
+	upxd.c
 
 libqsenet_la_LDFLAGS = -version-info 1:0:0 -no-undefined -L../cmn -L$(libdir)
 libqsenet_la_LIBADD = -lqsecmn
@@ -350,6 +353,7 @@ distclean-compile:
 @AMDEP_TRUE@@am__include@ @am__quote@./$(DEPDIR)/httpd-resol.Plo@am__quote@
 @AMDEP_TRUE@@am__include@ @am__quote@./$(DEPDIR)/httpd-task.Plo@am__quote@
 @AMDEP_TRUE@@am__include@ @am__quote@./$(DEPDIR)/httpd.Plo@am__quote@
+@AMDEP_TRUE@@am__include@ @am__quote@./$(DEPDIR)/upxd.Plo@am__quote@
 
 .c.o:
 @am__fastdepCC_TRUE@	$(COMPILE) -MT $@ -MD -MP -MF $(DEPDIR)/$*.Tpo -c -o $@ $<

qse/lib/net/upxd.c

@@ -5,8 +5,8 @@
     This file is part of QSE.
 
     QSE is free software: you can redistribute it and/or modify
-    it under the terms of the GNU Lesser General Public License as 
-    published by the Free Software Foundation, either version 3 of 
+    it under the terms of the GNU Lesser General Public License as
+    published by the Free Software Foundation, either version 3 of
     the License, or (at your option) any later version.
 
     QSE is distributed in the hope that it will be useful,
@@ -14,11 +14,21 @@
     MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
     GNU Lesser General Public License for more details.
 
-    You should have received a copy of the GNU Lesser General Public 
+    You should have received a copy of the GNU Lesser General Public
     License along with QSE. If not, see <htrd://www.gnu.org/licenses/>.
  */
 
 #include "upxd.h"
+#include <qse/cmn/str.h>
+
+QSE_IMPLEMENT_COMMON_FUNCTIONS (upxd)
+
+static void free_all_servers (qse_upxd_t* upxd);
+static qse_upxd_server_session_t* find_server_session (
+	qse_upxd_t* upxd, qse_upxd_server_t* server, qse_nwad_t* from);
+static void release_session (
+	qse_upxd_t* upxd, qse_upxd_server_session_t* session);
+
 
 qse_upxd_t* qse_upxd_open (qse_mmgr_t* mmgr, qse_size_t xtnsize)
 {
@@ -53,16 +63,11 @@ int qse_upxd_init (qse_upxd_t* upxd, qse_mmgr_t* mmgr)
 
 void qse_upxd_fini (qse_upxd_t* upxd)
 {
-	free_server_list (upxd, upxd->server.list);
-	QSE_ASSERT (upxd->server.navail == 0);
-	upxd->server.list = QSE_NULL;
+	QSE_ASSERTX (upxd->server.nactive == 0,
+		"Deactivate all servers before destroying me");
+	free_all_servers (upxd);
 }
 
-void qse_upxd_stop (qse_upxd_t* upxd)
-{
-	upxd->stopreq = 1;
-}
- 
 qse_upxd_errnum_t qse_upxd_geterrnum (qse_upxd_t* upxd)
 {
 	return upxd->errnum;
@@ -73,37 +78,461 @@ void qse_upxd_seterrnum (qse_upxd_t* upxd, qse_upxd_errnum_t errnum)
 	upxd->errnum = errnum;
 }
 
-int qse_upxd_addserver (qse_upxd_t* upxd, const qse_nwad_t* nwad)
+QSE_INLINE void* qse_upxd_allocmem (qse_upxd_t* upxd, qse_size_t size)
+{
+	void* ptr = QSE_MMGR_ALLOC (upxd->mmgr, size);
+	if (ptr == QSE_NULL) upxd->errnum = QSE_UPXD_ENOMEM;
+	return ptr;
+}
+
+QSE_INLINE void* qse_upxd_reallocmem (
+	qse_upxd_t* upxd, void* ptr, qse_size_t size)
+{
+	void* nptr = QSE_MMGR_REALLOC (upxd->mmgr, ptr, size);
+	if (nptr == QSE_NULL) upxd->errnum = QSE_UPXD_ENOMEM;
+	return nptr;
+}
+
+QSE_INLINE void qse_upxd_freemem (qse_upxd_t* upxd, void* ptr)
+{
+	QSE_MMGR_FREE (upxd->mmgr, ptr);
+}
+
+void qse_upxd_stop (qse_upxd_t* upxd)
+{
+	upxd->stopreq = 1;
+}
+
+static int perform_session_task (
+	qse_upxd_t* upxd, void* mux, qse_ubi_t handle, void* cbarg)
+{
+	qse_upxd_server_session_t* session;
+	qse_upxd_server_t* server;
+	qse_ssize_t n;
+
+	session = (qse_upxd_server_session_t*)cbarg;
+	server = session->inner.server;
+
+	qse_gettime (&session->modified);
+
+	/* this handler should set the 'from' field of server->scok */
+	n = upxd->cbs->sock.recv (
+		upxd, &session->peer, upxd->rbuf, QSE_SIZEOF(upxd->rbuf));
+	if (n <= -1)
+	{
+		upxd->cbs->session.error (upxd, &session->inner);
+		release_session (upxd, session);
+		return -1;
+	}
+
+	/* TODO: inspect if session->inner.to matches session->sock.from.
+	         drop it if they don't match if a certain option (QSE_UPXD_STRICT)
+			 is set??? */
+
+	/* send the peer's packet back to the client */ 
+	server->local.to = session->inner.client;
+	n = upxd->cbs->sock.send (upxd, &server->local, upxd->rbuf, n);
+	if (n <= -1)
+	{
+		upxd->cbs->session.error (upxd, &session->inner);
+		release_session (upxd, session);
+		return -1;
+	}
+
+	return 0;
+}
+
+static int perform_server_task (
+	qse_upxd_t* upxd, void* mux, qse_ubi_t handle, void* cbarg)
 {
 	qse_upxd_server_t* server;
+	qse_upxd_server_session_t* session;
+	qse_ssize_t n;
+
+	server = (qse_upxd_server_t*)cbarg;
+
+	/* this handler should set the 'from' field of server->scok */
+	n = upxd->cbs->sock.recv (
+		upxd, &server->local, upxd->rbuf, QSE_SIZEOF(upxd->rbuf));
+	if (n <= -1) return -1;
+
+	/* get the existing session or create a new session based on
+	 * server->local->from */
+	session = find_server_session (upxd, server, &server->local.from);
+	if (session == QSE_NULL)
+	{
+		qse_upxd_session_t interim;
+		QSE_MEMSET (&interim, 0, QSE_SIZEOF(interim));
+		interim.client = server->local.from;
+		upxd->cbs->session.error (upxd, &interim);
+		return -1;
+	}
+
+	n = upxd->cbs->sock.send (upxd, &session->peer, upxd->rbuf, n);
+	if (n <= -1)
+	{
+		upxd->cbs->session.error (upxd, &session->inner);
+		release_session (upxd, session);
+		return -1;
+	}
+
+	return 0;
+}
+
+static qse_upxd_server_session_t* find_server_session (
+	qse_upxd_t* upxd, qse_upxd_server_t* server, qse_nwad_t* from)
+{
+	qse_upxd_server_session_t* session;
+
+	/* TODO: make it indexable or hashable with 'from'
+	 *       don't perform linear search */
+	/* find an existing session made for the source address 'from' */
+	for (session = server->session.list; session; session = session->next)
+	{
+		if (QSE_MEMCMP (&session->inner.client, from, QSE_SIZEOF(*from)) == 0)
+		{
+			qse_gettime (&session->modified);
+			return session;
+		}
+	}
+
+	/* there is no session found for the source address 'from'.
+	 * let's create a new session. */
+	session = qse_upxd_allocmem (upxd, QSE_SIZEOF(*session));
+	if (session == QSE_NULL) return QSE_NULL;
+
+	QSE_MEMSET (session, 0, QSE_SIZEOF(*session));
+
+
+	if (qse_gettime (&session->created) <= -1)
+	{
+		qse_upxd_freemem (upxd, session);
+		upxd->errnum = QSE_UPXD_ESYSERR;
+		return QSE_NULL;
+	}
+	session->modified = session->created;
+	
+	session->inner.server = server;
+	session->inner.client = *from;
+
+	/* set the default dormancy */
+	session->inner.config.dormancy = QSE_UPXD_SESSION_DORMANCY;
+
+	/* call the configurationc callback for configuration data */
+	if (upxd->cbs->session.config (upxd, &session->inner) <= -1)
+	{
+		qse_upxd_freemem (upxd, session);
+		return QSE_NULL;
+	}
+
+	/* set up the peer socket with the configuration data */ 
+	session->peer.bind = session->inner.config.bind;
+	session->peer.to = session->inner.config.peer;
+	if (session->inner.config.dev[0] != QSE_T('\0'))
+		session->peer.dev = session->inner.config.dev;
+
+	if (upxd->cbs->sock.open (upxd, &session->peer) <= -1)
+	{
+		qse_upxd_freemem (upxd, session);
+		return QSE_NULL;
+	}
+
+	if (upxd->cbs->mux.addhnd (
+		upxd, upxd->mux, session->peer.handle,
+		perform_session_task, session) <= -1)
+	{
+		upxd->cbs->sock.close (upxd, &session->peer);
+		qse_upxd_freemem (upxd, session);
+		return QSE_NULL;
+	}
+
+	/* insert the session into the head of the session list */
+	if (server->session.list)
+		server->session.list->prev = session;
+	session->next = server->session.list;
+	server->session.list = session;
+
+	return session;
+}
+
+static void release_session (
+	qse_upxd_t* upxd, qse_upxd_server_session_t* session)
+{
+	qse_upxd_server_t* server;
+
+	server = session->inner.server;
+	QSE_ASSERT (server != QSE_NULL);
+
+	upxd->cbs->mux.delhnd (upxd, upxd->mux, session->peer.handle);
+	upxd->cbs->sock.close (upxd, &session->peer);
+
+	/* remove the session from the session list */
+	if (session->next) session->next->prev = session->prev;
+	if (session->prev) session->prev->next = session->next;
+	else server->session.list = session->next;
+
+	/* destroy the session */
+	qse_upxd_freemem (upxd, session);
+}
+
+static int activate_server (qse_upxd_t* upxd, qse_upxd_server_t* server)
+{
+	QSE_ASSERT (upxd->cbs != QSE_NULL);
+	QSE_ASSERT (!(server->flags & QSE_UPXD_SERVER_ACTIVE));
+	
+	if (upxd->cbs->sock.open (upxd, &server->local) <= -1)
+	{
+		return -1;
+	}
+
+	if (upxd->cbs->mux.addhnd (
+		upxd, upxd->mux, server->local.handle,
+		perform_server_task, server) <= -1)
+	{
+		upxd->cbs->sock.close (upxd, &server->local);
+		return -1;
+	}
+
+	server->flags |= QSE_UPXD_SERVER_ACTIVE;
+	upxd->server.nactive++;
+	return 0;
+}
+
+static void deactivate_server (qse_upxd_t* upxd, qse_upxd_server_t* server)
+{
+	qse_upxd_server_session_t* session;
+
+	QSE_ASSERT (upxd->cbs != QSE_NULL);
+	QSE_ASSERT (server->flags & QSE_UPXD_SERVER_ACTIVE);
+
+	session = server->session.list;
+	while (session)
+	{
+		qse_upxd_server_session_t* next = session->next;
+		release_session (upxd, session);
+		session = next;
+	}
+
+	upxd->cbs->mux.delhnd (upxd, upxd->mux, server->local.handle);
+	upxd->cbs->sock.close (upxd, &server->local);
+
+	server->flags &= ~QSE_UPXD_SERVER_ACTIVE;
+	upxd->server.nactive--;
+}
+
+static void activate_all_servers (qse_upxd_t* upxd)
+{
+	qse_upxd_server_t* server;
+
+	for (server = upxd->server.list; server; server = server->next)
+	{
+		if (!(server->flags & QSE_UPXD_SERVER_ACTIVE))
+		{
+			activate_server (upxd, server);
+		}
+	}
+}
+
+static void deactivate_all_servers (qse_upxd_t* upxd)
+{
+	qse_upxd_server_t* server;
+
+	server = upxd->server.list;
+	while (server)
+	{
+		if (server->flags & QSE_UPXD_SERVER_ACTIVE)
+			deactivate_server (upxd, server);
+
+		server = server->next;
+	}
+}
+
+static void free_all_servers (qse_upxd_t* upxd)
+{
+	qse_upxd_server_t* server;
+	qse_upxd_server_t* next;
+
+	server = upxd->server.list;
+	while (server)
+	{
+		next = server->next;
+		QSE_MMGR_FREE (upxd->mmgr, server);
+		server = next;
+	}
+	upxd->server.list = QSE_NULL;
+}
+
+static void purge_deleted_servers (qse_upxd_t* upxd)
+{
+	qse_upxd_server_t* server;
+	qse_upxd_server_t* next;
+
+	server = upxd->server.list;
+	while (server)
+	{
+		next = server->next;
+
+		if (server->flags & QSE_UPXD_SERVER_DELETED)
+		{
+			if (server->flags & QSE_UPXD_SERVER_ACTIVE)
+				deactivate_server (upxd, server);
+
+			if (server == upxd->server.list) upxd->server.list = next;
+			
+			QSE_MMGR_FREE (upxd->mmgr, server);
+		}
+		server = next;
+	}
+}
+
+qse_upxd_server_t* qse_upxd_addserver (
+	qse_upxd_t* upxd, const qse_nwad_t* nwad, const qse_char_t* dev)
+{
+	qse_upxd_server_t* server;
+
+	if (dev && qse_strlen(dev) >= QSE_COUNTOF(server->dev))
+	{
+		upxd->errnum = QSE_UPXD_EINVAL;
+		return QSE_NULL;
+	}
 
 	server = QSE_MMGR_ALLOC (upxd->mmgr, QSE_SIZEOF(*server));
 	if (server == QSE_NULL)
 	{
 		upxd->errnum = QSE_UPXD_ENOMEM;
-		return -1;
+		return QSE_NULL;
 	}
 
 	QSE_MEMSET (server, 0, QSE_SIZEOF(*server));
-	server->nwad = *nwad;
-
-	if (upxd->cbs->server.open (upxd, server) <= -1)
+	if (dev)
 	{
-		QSE_MMGR_FREE (upxd->mmgr, server);
-		return -1;
+		qse_strxcpy (server->dev, QSE_COUNTOF(server->dev), dev);
+		server->local.dev = server->dev;
 	}
+	server->local.bind = *nwad;
 
-	server->next = upxd->server.list;	
+	upxd->cbs->lock.acquire (upxd);
+	server->next = upxd->server.list;
 	upxd->server.list = server;
+	upxd->cbs->lock.release (upxd);
+
+	return server;
 }
 
-int qse_upxd_loop (qse_upxd_t* upxd, qse_upxd_cbls_t* cbs)
+void qse_upxd_delserver (
+	qse_upxd_t* upxd, qse_upxd_server_t* server)
 {
+	server->flags |= QSE_UPXD_SERVER_DELETED;
+}
+
+void* qse_upxd_getserverctx (
+	qse_upxd_t* upxd, qse_upxd_server_t* server)
+{
+	return server->ctx;
+}
+
+void qse_upxd_setserverctx (
+	qse_upxd_t* upxd, qse_upxd_server_t* server, void* ctx)
+{
+	server->ctx = ctx;
+}
+
+qse_upxd_cbs_t* qse_upxd_getcbs (qse_upxd_t* upxd)
+{
+	return upxd->cbs;
+}
+
+void qse_upxd_setcbs (qse_upxd_t* upxd, qse_upxd_cbs_t* cbs)
+{
+	upxd->cbs = cbs;
+}
+
+static QSE_INLINE void purge_idle_sessions_in_server (
+	qse_upxd_t* upxd, qse_upxd_server_t* server)
+{
+	qse_upxd_server_session_t* session;
+	qse_upxd_server_session_t* next;
+	qse_ntime_t now;
+
+	qse_gettime (&now);
+
+	session = server->session.list;
+	while (session)
+	{
+		next = session->next;
+
+		if (session->inner.config.dormancy > 0 &&
+		    now > session->modified &&
+		    now - session->modified > session->inner.config.dormancy)
+		{
+			release_session (upxd, session);
+		}
+
+		session = next;
+	}
+}
+
+static void purge_idle_sessions (qse_upxd_t* upxd)
+{
+	qse_upxd_server_t* server;
+
+	for (server = upxd->server.list; server; server = server->next)
+	{
+		if (server->flags & QSE_UPXD_SERVER_ACTIVE)
+		{
+			purge_idle_sessions_in_server (upxd, server);
+		}
+	}
+}
+
+int qse_upxd_loop (qse_upxd_t* upxd, qse_ntime_t timeout)
+{
+	int retv = -1;
+
+	QSE_ASSERTX (upxd->cbs != QSE_NULL,
+		"Call qse_upxd_setcbs() before calling qse_upxd_loop()");
+	QSE_ASSERT (upxd->mux == QSE_NULL);
+
+	if (upxd->cbs == QSE_NULL || upxd->mux /*||
+	    upxd->server.list == QSE_NULL*/)
+	{
+		upxd->errnum = QSE_UPXD_EINVAL;
+		goto oops;
+	}
+
 	upxd->stopreq = 0;
+	upxd->mux = upxd->cbs->	mux.open (upxd);
+	if (upxd->mux == QSE_NULL) goto oops;
+
+	activate_all_servers (upxd);
+	if (upxd->server.nactive == 0)
+	{
+		/* at least 1 server must be activated here */
+		upxd->errnum = QSE_UPXD_EINVAL;
+		goto oops;
+	}
 
 	while (!upxd->stopreq)
 	{
+		int count;
+
+		count = upxd->cbs->mux.poll (upxd, upxd->mux, timeout);
+		if (count <= -1)
+		{
+			/* TODO: anything? */
+		}
+
+		upxd->cbs->lock.acquire (upxd);
+		purge_idle_sessions (upxd);
+		purge_deleted_servers (upxd);
+		activate_all_servers (upxd);
+		upxd->cbs->lock.release (upxd);
 	}
 
-	return 0;
+	retv = 0;
+
+oops:
+	if (upxd->server.nactive > 0) deactivate_all_servers (upxd);
+	if (upxd->mux) upxd->cbs->mux.close (upxd, upxd->mux);
+	return retv;
 }

qse/lib/net/upxd.h

@@ -5,8 +5,8 @@
     This file is part of QSE.
 
     QSE is free software: you can redistribute it and/or modify
-    it under the terms of the GNU Lesser General Public License as 
-    published by the Free Software Foundation, either version 3 of 
+    it under the terms of the GNU Lesser General Public License as
+    published by the Free Software Foundation, either version 3 of
     the License, or (at your option) any later version.
 
     QSE is distributed in the hope that it will be useful,
@@ -14,7 +14,7 @@
     MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
     GNU Lesser General Public License for more details.
 
-    You should have received a copy of the GNU Lesser General Public 
+    You should have received a copy of the GNU Lesser General Public
     License along with QSE. If not, see <htrd://www.gnu.org/licenses/>.
  */
 
@@ -22,15 +22,80 @@
 #define _QSE_LIB_NET_UPXD_H_
 
 #include <qse/net/upxd.h>
+#include "../cmn/mem.h"
+
+typedef struct qse_upxd_server_session_t qse_upxd_server_session_t;
+
+struct qse_upxd_server_t
+{
+	qse_upxd_server_t* next;
+
+#define QSE_UPXD_SERVER_ACTIVE (1 << 0)
+#define QSE_UPXD_SERVER_DELETED (1 << 1)
+	int flags;
+	
+	/* the socket can be bound to this interface.
+	 * sock->dev points to this buffer when necessary. */
+	qse_char_t dev[QSE_UPXD_SESSION_DEV_LEN + 1];
+
+	/* list of sessions beloning to this server */
+	struct
+	{
+		qse_upxd_server_session_t* list;
+		qse_size_t count;
+	} session;
+
+	qse_upxd_sock_t local;
+
+	/* user-defined context data that can be set
+	 * with qse_upxd_setserverctx() */
+	void* ctx;
+};
+
+struct qse_upxd_server_session_t
+{
+	/* internal fields */
+	qse_upxd_server_session_t* next;
+	qse_upxd_server_session_t* prev;
+
+	/* timestamps for housekeeping */
+	qse_ntime_t created;
+	qse_ntime_t modified;
+
+	/* socket used to talk with a peer */
+	qse_upxd_sock_t peer;
+
+	/* exposed to a caller via callbacks */
+	qse_upxd_session_t inner;
+};
 
 struct qse_upxd_t
 {
+	QSE_DEFINE_COMMON_FIELDS (upxd)
+	qse_upxd_errnum_t errnum;
+
 	int stopreq;
+	qse_upxd_cbs_t* cbs;
 
 	struct
 	{
 		qse_upxd_server_t* list;
+		qse_size_t nactive;
 	} server;
+
+	void* mux;
+	qse_uint8_t rbuf[65535];
 };
 
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+int qse_upxd_init (qse_upxd_t* upxd, qse_mmgr_t* mmgr);
+void qse_upxd_fini (qse_upxd_t* upxd);
+
+#ifdef __cplusplus
+}
+#endif
+
 #endif