implementation of queuing of messages to send.

miniupnpd/.gitignore

@@ -9,6 +9,7 @@ testgetifstats
 testupnpdescgen
 testupnppermissions
 testgetroute
+testasyncsendto
 netfilter/testiptcrdr
 netfilter/testiptcrdr_dscp
 netfilter/testiptcrdr_peer

miniupnpd/Makefile

@@ -116,10 +116,11 @@ TESTUPNPDESCGENOBJS = testupnpdescgen.o upnpdescgen.o
 TESTUPNPPERMISSIONSOBJS = testupnppermissions.o upnppermissions.o
 TESTGETIFADDROBJS = testgetifaddr.o getifaddr.o
 MINIUPNPDCTLOBJS = miniupnpdctl.o
+TESTASYNCSENDTOOBJS = testasyncsendto.o asyncsendto.o upnputils.o
 
 EXECUTABLES = miniupnpd testupnpdescgen testgetifstats \
               testupnppermissions miniupnpdctl \
-              testgetifaddr testgetroute
+              testgetifaddr testgetroute testasyncsendto
 .if $(OSNAME) == "Darwin"
 LIBS =
 .else
@@ -142,7 +143,7 @@ clean:
 	$(RM) $(STDOBJS) $(BSDOBJS) $(SUNOSOBJS) $(MACOBJS) $(EXECUTABLES) \
 	testupnpdescgen.o \
 	$(MISCOBJS) config.h testgetifstats.o testupnppermissions.o \
-	miniupnpdctl.o testgetifaddr.o testgetroute.o \
+	miniupnpdctl.o testgetifaddr.o testgetroute.o testasyncsendto.o \
 	$(PFOBJS) $(IPFOBJS) $(IPFWOBJS)
 
 install:	miniupnpd genuuid
@@ -199,6 +200,9 @@ testupnppermissions:	config.h $(TESTUPNPPERMISSIONSOBJS)
 testgetroute:	config.h $(TESTGETROUTEOBJS)
 	$(CC) $(CFLAGS) -o $@ $(TESTGETROUTEOBJS)
 
+testasyncsendto:	config.h $(TESTASYNCSENDTOOBJS)
+	$(CC) $(CFLAGS) -o $@ $(TESTASYNCSENDTOOBJS)
+
 # gmake :
 #	$(CC) $(CFLAGS) -o $@ $^
 # BSDmake :

miniupnpd/Makefile.linux

@@ -150,7 +150,7 @@ TESTUPNPDESCGENOBJS = testupnpdescgen.o upnpdescgen.o
 
 EXECUTABLES = miniupnpd testupnpdescgen testgetifstats \
               testupnppermissions miniupnpdctl testgetifaddr \
-              testgetroute
+              testgetroute testasyncsendto
 
 .PHONY:	all clean install depend genuuid
 
@@ -161,7 +161,7 @@ clean:
 	$(RM) $(EXECUTABLES)
 	$(RM) testupnpdescgen.o testgetifstats.o
 	$(RM) testupnppermissions.o testgetifaddr.o
-	$(RM) testgetroute.o
+	$(RM) testgetroute.o testasyncsendto.o
 	$(RM) miniupnpdctl.o
 
 install:	miniupnpd miniupnpd.8 miniupnpd.conf genuuid \
@@ -205,6 +205,8 @@ testgetifaddr:	testgetifaddr.o getifaddr.o
 
 testgetroute:	testgetroute.o linux/getroute.o upnputils.o -lnfnetlink
 
+testasyncsendto:	testasyncsendto.o asyncsendto.o upnputils.o
+
 miniupnpdctl:	miniupnpdctl.o
 
 config.h:	genconfig.sh VERSION
@@ -214,7 +216,7 @@ depend:	config.h
 	makedepend -f$(MAKEFILE_LIST) -Y \
 	$(ALLOBJS:.o=.c) $(TESTUPNPDESCGENOBJS:.o=.c) \
 	testgetifstats.c testupnppermissions.c testgetifaddr.c \
-	testgetroute.c miniupnpdctl.c 2>/dev/null
+	testgetroute.c testasyncsendto.c miniupnpdctl.c 2>/dev/null
 
 # DO NOT DELETE
 

miniupnpd/asyncsendto.c

@@ -0,0 +1,153 @@
+/* $Id: $ */
+/* MiniUPnP project
+ * http://miniupnp.free.fr/ or http://miniupnp.tuxfamily.org/
+ * (c) 2006-2014 Thomas Bernard
+ * This software is subject to the conditions detailed
+ * in the LICENCE file provided within the distribution */
+
+#include <sys/types.h>
+#include <sys/socket.h>
+#include <sys/time.h>
+#include <sys/queue.h>
+#include <stdlib.h>
+#include <string.h>
+#include <syslog.h>
+#include <errno.h>
+
+#include "asyncsendto.h"
+
+struct scheduled_send {
+	LIST_ENTRY(scheduled_send) entries;
+	struct timeval ts;
+	int sockfd;
+	const void * buf;
+	size_t len;
+	int flags;
+	const struct sockaddr *dest_addr;
+	socklen_t addrlen;
+	char data[];
+};
+
+static LIST_HEAD(listhead, scheduled_send) send_list = { NULL };
+
+/*
+ * ssize_t sendto(int sockfd, const void *buf, size_t len, int flags,
+ *                const struct sockaddr *dest_addr, socklen_t addrlen);
+ */
+
+/* delay = milli seconds */
+ssize_t
+sendto_schedule(int sockfd, const void *buf, size_t len, int flags,
+                const struct sockaddr *dest_addr, socklen_t addrlen,
+                unsigned int delay)
+{
+	ssize_t n;
+	struct timeval tv;
+	struct scheduled_send * elt;
+
+	if(delay == 0) {
+		/* first try to send at once */
+		n = sendto(sockfd, buf, len, flags, dest_addr, addrlen);
+		if((n >= 0) || (errno != EINTR && errno != EAGAIN && errno != EWOULDBLOCK))
+			return n;
+	}
+	/* schedule */
+	if(gettimeofday(&tv, 0) < 0) {
+		return -1;
+	}
+	/* allocate enough space for structure + buffers */
+	elt = malloc(sizeof(struct scheduled_send) + len + addrlen);
+	if(elt == NULL) {
+		syslog(LOG_ERR, "malloc failed to allocate %u bytes",
+		       (unsigned)(sizeof(struct scheduled_send) + len + addrlen));
+		return -1;
+	}
+	/* time the packet should be sent */
+	elt->ts.tv_sec = tv.tv_sec + (delay / 1000);
+	elt->ts.tv_usec = tv.tv_usec + (delay % 1000) * 1000;
+	if(elt->ts.tv_usec > 1000000) {
+		elt->ts.tv_sec++;
+		elt->ts.tv_usec -= 1000000;
+	}
+	elt->sockfd = sockfd;
+	elt->flags = flags;
+	memcpy(elt->data, dest_addr, addrlen);
+	elt->dest_addr = (struct sockaddr *)elt->data;
+	elt->addrlen = addrlen;
+	memcpy(elt->data + addrlen, buf, len);
+	elt->buf = (void *)(elt->data + addrlen);
+	elt->len = len;
+	/* insert */
+	LIST_INSERT_HEAD( &send_list, elt, entries);
+	return 0;
+}
+
+
+ssize_t
+sendto_or_schedule(int sockfd, const void *buf, size_t len, int flags,
+                   const struct sockaddr *dest_addr, socklen_t addrlen)
+{
+	return sendto_schedule(sockfd, buf, len, flags, dest_addr, addrlen, 0);
+}
+
+/* get_next_scheduled_send() return number of scheduled send in list */
+int get_next_scheduled_send(struct timeval * next_send)
+{
+	int n = 0;
+	struct scheduled_send * elt;
+	if(next_send == NULL)
+		return -1;
+	for(elt = send_list.lh_first; elt != NULL; elt = elt->entries.le_next) {
+		if(n == 0 || (elt->ts.tv_sec < next_send->tv_sec) ||
+		   (elt->ts.tv_sec == next_send->tv_sec && elt->ts.tv_usec < next_send->tv_usec)) {
+			next_send->tv_sec = elt->ts.tv_sec;
+			next_send->tv_usec = elt->ts.tv_usec;
+		}
+		n++;
+	}
+	return n;
+}
+
+int get_sendto_fds(fd_set * writefds, int * max_fd, const struct timeval * now)
+{
+	int n = 0;
+	struct scheduled_send * elt;
+	for(elt = send_list.lh_first; elt != NULL; elt = elt->entries.le_next) {
+		if((elt->ts.tv_sec < now->tv_sec) ||
+		   (elt->ts.tv_sec == now->tv_sec && elt->ts.tv_usec <= now->tv_usec)) {
+			FD_SET(elt->sockfd, writefds);
+			if(elt->sockfd > *max_fd)
+				*max_fd = elt->sockfd;
+			n++;
+		}
+	}
+syslog(LOG_DEBUG, "%x", (int)writefds->fds_bits[0]);
+	return n;
+}
+
+int try_sendto(fd_set * writefds)
+{
+	ssize_t n;
+	struct scheduled_send * elt;
+	struct scheduled_send * next;
+	for(elt = send_list.lh_first; elt != NULL; elt = next) {
+		next = elt->entries.le_next;
+syslog(LOG_DEBUG, "s=%d fds=%x", elt->sockfd, (int)writefds->fds_bits[0]);
+		if(FD_ISSET(elt->sockfd, writefds)) {
+			syslog(LOG_DEBUG, "sending %d bytes", (int)elt->len);
+			n = sendto(elt->sockfd, elt->buf, elt->len, elt->flags,
+			           elt->dest_addr, elt->addrlen);
+			if(n < 0) {
+				syslog(LOG_DEBUG, "sendto: %m");
+				if(errno == EINTR || errno == EAGAIN || errno == EWOULDBLOCK)
+					continue;
+				return n;
+			} else {
+				LIST_REMOVE(elt, entries);
+				free(elt);
+			}
+		}
+	}
+	return 0;
+}
+

miniupnpd/asyncsendto.h

@@ -0,0 +1,26 @@
+/* $Id: $ */
+/* MiniUPnP project
+ * http://miniupnp.free.fr/ or http://miniupnp.tuxfamily.org/
+ * (c) 2006-2014 Thomas Bernard
+ * This software is subject to the conditions detailed
+ * in the LICENCE file provided within the distribution */
+
+#ifndef ASYNCSENDTO_H_INCLUDED
+#define ASYNCSENDTO_H_INCLUDED
+
+ssize_t
+sendto_schedule(int sockfd, const void *buf, size_t len, int flags,
+                const struct sockaddr *dest_addr, socklen_t addrlen,
+                unsigned int delay);
+
+ssize_t
+sendto_or_schedule(int sockfd, const void *buf, size_t len, int flags,
+                   const struct sockaddr *dest_addr, socklen_t addrlen);
+
+int get_next_scheduled_send(struct timeval * next_send);
+
+int try_sendto(fd_set * writefds);
+
+int get_sendto_fds(fd_set * writefds, int * max_fd, const struct timeval * now);
+
+#endif

miniupnpd/testasyncsendto.c

@@ -0,0 +1,117 @@
+/* $Id: $ */
+/* MiniUPnP project
+ * http://miniupnp.free.fr/ or http://miniupnp.tuxfamily.org/
+ * (c) 2006-2014 Thomas Bernard
+ * This software is subject to the conditions detailed
+ * in the LICENCE file provided within the distribution */
+
+#include <sys/types.h>
+#include <sys/socket.h>
+#include <netinet/in.h>
+#include <arpa/inet.h>
+#include <sys/time.h>
+#include <string.h>
+#include <unistd.h>
+#include <syslog.h>
+#include <errno.h>
+
+#include "miniupnpdtypes.h"
+#include "upnputils.h"
+#include "asyncsendto.h"
+
+struct lan_addr_list lan_addrs;
+
+#define DEST_IP "239.255.255.250"
+#define DEST_PORT 1900
+/*
+ssize_t
+sendto_schedule(int sockfd, const void *buf, size_t len, int flags,
+                const struct sockaddr *dest_addr, socklen_t addrlen,
+                unsigned int delay)
+*/
+
+int test(void)
+{
+	int s;
+	ssize_t n;
+	int i;
+	struct sockaddr_in addr;
+	struct sockaddr_in dest_addr;
+	struct timeval next_send;
+	if( (s = socket(PF_INET, SOCK_DGRAM, 0)) < 0) {
+		syslog(LOG_ERR, "socket(): %m");
+		return 1;
+	}
+	set_non_blocking(s);
+	memset(&addr, 0, sizeof(struct sockaddr_in));
+	addr.sin_family = AF_INET;
+	addr.sin_addr.s_addr = INADDR_ANY;
+	if(bind(s, &addr, sizeof(addr)) < 0) {
+		syslog(LOG_ERR, "bind(): %m");
+		close(s);
+		return 1;
+	}
+	memset(&dest_addr, 0, sizeof(struct sockaddr_in));
+	dest_addr.sin_family = AF_INET;
+	dest_addr.sin_addr.s_addr = inet_addr(DEST_IP);
+	dest_addr.sin_port =  htons(DEST_PORT);
+	n = sendto_or_schedule(s, "1234", 4, 0,
+	                       (struct sockaddr *)&dest_addr, sizeof(dest_addr));
+	syslog(LOG_DEBUG, "sendto_or_schedule : %d", (int)n);
+	n = sendto_schedule(s, "1234", 4, 0,
+	                    (struct sockaddr *)&dest_addr, sizeof(dest_addr),
+	                    3000);
+	syslog(LOG_DEBUG, "sendto_schedule : %d", (int)n);
+	while ((i = get_next_scheduled_send(&next_send)) > 0) {
+		fd_set writefds;
+		int max_fd;
+		struct timeval timeout;
+		struct timeval now;
+		syslog(LOG_DEBUG, "get_next_scheduled_send : %d next_send=%ld.%06ld",
+		       i, next_send.tv_sec, next_send.tv_usec);
+		FD_ZERO(&writefds);
+		max_fd = 0;
+		gettimeofday(&now, NULL);
+		if(now.tv_sec > next_send.tv_sec ||
+		   (now.tv_sec == next_send.tv_sec && now.tv_usec >= next_send.tv_usec)) {
+			/* wait 10sec :) */
+			timeout.tv_sec = 10;
+			timeout.tv_usec = 0;
+		} else {
+			/* ... */
+			timeout.tv_sec = (next_send.tv_sec - now.tv_sec);
+			timeout.tv_usec = (next_send.tv_usec - now.tv_usec);
+			if(timeout.tv_usec < 0) {
+				timeout.tv_usec += 1000000;
+				timeout.tv_sec--;
+			}
+		}
+		i = get_sendto_fds(&writefds, &max_fd, &now);
+		syslog(LOG_DEBUG, "get_sendto_fds() returned %d", i);
+		syslog(LOG_DEBUG, "select(%d, NULL, xx, NULL, %ld.%06ld)",
+		       max_fd, timeout.tv_sec, timeout.tv_usec);
+		i = select(max_fd, NULL, &writefds, NULL, &timeout);
+		if(i < 0) {
+			syslog(LOG_ERR, "select: %m");
+			if(errno != EINTR)
+				break;
+		} else if(try_sendto(&writefds) < 0) {
+			syslog(LOG_ERR, "try_sendto: %m");
+			break;
+		}
+	}
+	close(s);
+	return 0;
+}
+
+int main(int argc, char * * argv)
+{
+	int r;
+	(void)argc;
+	(void)argv;
+	openlog("testasyncsendto", LOG_CONS|LOG_PERROR, LOG_USER);
+	r = test();
+	closelog();
+	return r;
+}
+