checkpoint. udp listener clde split out.

bin/named/Makefile.in

@@ -19,7 +19,7 @@ LIBS =		${DEPLIBS} \
 
 TARGETS =	named
 
-OBJS =		server.o wire_test.o
+OBJS =		server.o udpclient.o wire_test.o
 
 @BIND9_MAKE_RULES@
 
@@ -30,4 +30,4 @@ named: ${OBJS} ${DEPLIBS}
 	${CC} -o $@ ${OBJS} ${LIBS}
 
 clean distclean::
-	rm -f ${TARGETS} $.o
+	rm -f ${TARGETS}

bin/named/server.c

@@ -46,169 +46,20 @@
 
 #include <arpa/inet.h>
 
+#include "udpclient.h"
+
 isc_mem_t *mctx = NULL;
 
-#define INPUT_BUFFER_SIZE (64 * 1024)	/* 64k */
-
-typedef struct {
-	char name[16];		/* socket human-printable name */
-	char *buf;		/* input buffer */
-	isc_mem_t *mctx;	/* memory context used to allocate */
-} client_ctx_t;
-
-static client_ctx_t *client_ctx_allocate(isc_mem_t *mctx);
-static void client_ctx_free(client_ctx_t *ctx);
-
-static void my_send(isc_task_t *task, isc_event_t *event);
-static void udp_recv(isc_task_t *task, isc_event_t *event);
-
-static client_ctx_t *
-client_ctx_allocate(isc_mem_t *mctx)
-{
-	client_ctx_t *ctx;
-
-	ctx = isc_mem_get(mctx, sizeof(client_ctx_t));
-	if (ctx == NULL)
-		return (NULL);
-
-	ctx->buf = isc_mem_get(mctx, INPUT_BUFFER_SIZE);
-	if (ctx->buf == NULL) {
-		isc_mem_put(mctx, ctx, sizeof(client_ctx_t));
-		return (NULL);
-	}
-
-	ctx->name[0] = '\0';
-	ctx->mctx = mctx;
-
-	return (ctx);
-}
-
-static void
-client_ctx_free(client_ctx_t *ctx)
-{
-	isc_mem_put(ctx->mctx, ctx->buf, INPUT_BUFFER_SIZE);
-	isc_mem_put(ctx->mctx, ctx, sizeof(client_ctx_t));
-}
-
-typedef struct dns_message {
-	unsigned int		id;
-	unsigned int		flags;
-	unsigned int		qcount;
-	unsigned int		ancount;
-	unsigned int		aucount;
-	unsigned int		adcount;
-	dns_namelist_t		question;
-	dns_namelist_t		answer;
-	dns_namelist_t		authority;
-	dns_namelist_t		additional;
-} dns_message_t; /* XXX Should be common? */
-
-/*
- * XXX These is in wire_test.c right now.
- */
-void getmessage(dns_message_t *message, isc_buffer_t *source,
-		isc_buffer_t *target);
-dns_result_t printmessage(dns_message_t *message);
-
-static void
-dump_packet(char *buf, u_int len)
-{
-	extern dns_decompress_t dctx;
-	char t[5000]; /* XXX */
-	dns_message_t message;
-	dns_result_t result;
-	isc_buffer_t source, target;
-
-	dctx.allowed = DNS_COMPRESS_GLOBAL14;
-	dns_name_init(&dctx.owner_name, NULL);
-
-	isc_buffer_init(&source, buf, len, ISC_BUFFERTYPE_BINARY);
-	isc_buffer_add(&source, len);
-	isc_buffer_init(&target, t, sizeof(t), ISC_BUFFERTYPE_BINARY);
-
-	getmessage(&message, &source, &target);
-	result = printmessage(&message);
-	if (result != DNS_R_SUCCESS)
-		printf("printmessage() failed: %s\n",
-		       dns_result_totext(result));
-}
-
-static void
-udp_recv(isc_task_t *task, isc_event_t *event)
-{
-	isc_socket_t *sock;
-	isc_socketevent_t *dev;
-	client_ctx_t *ctx;
-
-	sock = event->sender;
-	dev = (isc_socketevent_t *)event;
-	ctx = (client_ctx_t *)(event->arg);
-
-	printf("Task %s (sock %p, base %p, length %d, n %d, result %d)\n",
-	       (char *)(event->arg), sock,
-	       dev->region.base, dev->region.length,
-	       dev->n, dev->result);
-	printf("\tFrom: %s port %d\n",
-	       inet_ntoa(dev->address.type.sin.sin_addr),
-	       ntohs(dev->address.type.sin.sin_port));
-
-	if (dev->result != ISC_R_SUCCESS) {
-		isc_socket_detach(&sock);
-
-		client_ctx_free(ctx);
-
-		isc_event_free(&event);
-
-		/* destroy task */
-
-		return;
-	}
-
-	/*
-	 * Call the dump routine to print this baby out
-	 */
-	dump_packet(ctx->buf, dev->n);
-
-	isc_socket_recv(sock, &dev->region, ISC_FALSE,
-			task, udp_recv, event->arg);
-
-	isc_mem_stats(ctx->mctx, stdout);
-
-	isc_event_free(&event);
-}
-
-static void
-my_send(isc_task_t *task, isc_event_t *event)
-{
-	isc_socket_t *sock;
-	isc_socketevent_t *dev;
-
-	sock = event->sender;
-	dev = (isc_socketevent_t *)event;
-
-	printf("my_send: %s task %p\n\t(sock %p, base %p, length %d, n %d, result %d)\n",
-	       (char *)(event->arg), task, sock,
-	       dev->region.base, dev->region.length,
-	       dev->n, dev->result);
-
-	isc_mem_put(event->mctx, dev->region.base, dev->region.length);
-
-	isc_event_free(&event);
-}
-
 int
 main(int argc, char *argv[])
 {
 	isc_taskmgr_t *manager = NULL;
-	isc_task_t **tasks;
 	unsigned int workers;
 	isc_socketmgr_t *socketmgr;
 	isc_socket_t *so1;
 	isc_sockaddr_t sockaddr;
 	unsigned int addrlen;
-	client_ctx_t **ctxs;
-	unsigned int i;
-	isc_region_t region;
+	udp_listener_t *l;
 
 	memset(&sockaddr, 0, sizeof(sockaddr));
 	sockaddr.type.sin.sin_port = htons(5544);
@@ -222,11 +73,6 @@ main(int argc, char *argv[])
 
 	RUNTIME_CHECK(isc_mem_create(0, 0, &mctx) == ISC_R_SUCCESS);
 
-	tasks = isc_mem_get(mctx, sizeof(isc_task_t *) * workers);
-	RUNTIME_CHECK(tasks != NULL);
-	ctxs = isc_mem_get(mctx, sizeof(client_ctx_t *) * workers);
-	RUNTIME_CHECK(ctxs != NULL);
-
 	RUNTIME_CHECK(isc_taskmgr_create(mctx, workers, 0, &manager) ==
 		      ISC_R_SUCCESS);
 
@@ -246,30 +92,9 @@ main(int argc, char *argv[])
 	RUNTIME_CHECK(isc_socket_bind(so1, &sockaddr,
 				      (int)addrlen) == ISC_R_SUCCESS);
 
-	/*
-	 * Create all the listening tasks and set up the initial read.
-	 */
-	for (i = 0 ; i < workers ; i++) {
-		tasks[i] = NULL;
-		RUNTIME_CHECK(isc_task_create(manager, NULL, 0, &tasks[i])
-			      == ISC_R_SUCCESS);
-
-		/*
-		 * Allocate client context and set its name.
-		 */
-		ctxs[i] = client_ctx_allocate(mctx);
-		RUNTIME_CHECK(ctxs[i] != NULL);
-		region.length = INPUT_BUFFER_SIZE;
-		region.base = ctxs[i]->buf;
-
-		sprintf(ctxs[i]->name, "%u", i);
-
-		printf("recv started for task %s\n", ctxs[i]->name);
-		RUNTIME_CHECK(isc_socket_recv(so1, &region,
-					      ISC_FALSE, tasks[i],
-					      udp_recv, ctxs[i])
-			      == ISC_R_SUCCESS);
-	}
+	l = udp_listener_allocate(mctx, workers);
+	RUNTIME_CHECK(udp_listener_start(l, so1, manager, workers,
+					 workers, 0) == ISC_R_SUCCESS);
 
 	isc_mem_stats(mctx, stdout);
 

bin/named/udpclient.c

@@ -0,0 +1,309 @@
+#include <config.h>
+
+#include <stdio.h>
+#include <stdlib.h>
+#include <unistd.h>
+#include <string.h>
+
+#include <isc/assertions.h>
+#include <isc/error.h>
+#include <isc/mem.h>
+#include <isc/task.h>
+#include <isc/thread.h>
+#include <isc/result.h>
+#include <isc/socket.h>
+#include <isc/timer.h>
+
+#include <dns/types.h>
+#include <dns/result.h>
+#include <dns/name.h>
+#include <dns/rdata.h>
+#include <dns/rdatalist.h>
+#include <dns/rdataset.h>
+#include <dns/compress.h>
+
+#include <sys/types.h>
+#include <sys/socket.h>
+
+#include <netinet/in.h>
+
+#include <arpa/inet.h>
+
+#define LOCK(lp) \
+	RUNTIME_CHECK(isc_mutex_lock((lp)) == ISC_R_SUCCESS)
+#define UNLOCK(lp) \
+	RUNTIME_CHECK(isc_mutex_unlock((lp)) == ISC_R_SUCCESS)
+
+#include "udpclient.h"
+
+/*
+ * XXX see note about debugging below
+ */
+static void dump_packet(char *buf, u_int len);
+
+static udp_cctx_t *udp_cctx_allocate(isc_mem_t *mctx);
+static void udp_cctx_free(udp_cctx_t *ctx);
+
+static void udp_send(isc_task_t *task, isc_event_t *event);
+static void udp_recv(isc_task_t *task, isc_event_t *event);
+
+
+
+static udp_cctx_t *
+udp_cctx_allocate(isc_mem_t *mctx)
+{
+	udp_cctx_t *ctx;
+
+	ctx = isc_mem_get(mctx, sizeof(udp_cctx_t));
+	if (ctx == NULL)
+		return (NULL);
+
+	ctx->buf = isc_mem_get(mctx, UDP_INPUT_BUFFER_SIZE);
+	if (ctx->buf == NULL) {
+		isc_mem_put(mctx, ctx, sizeof(udp_cctx_t));
+		return (NULL);
+	}
+
+	ctx->slot = 0;
+	ctx->mctx = mctx;
+	ctx->count = 0; /* XXX */
+
+	return (ctx);
+}
+
+static void
+udp_cctx_free(udp_cctx_t *ctx)
+{
+	isc_mem_put(ctx->mctx, ctx->buf, UDP_INPUT_BUFFER_SIZE);
+	isc_mem_put(ctx->mctx, ctx, sizeof(udp_cctx_t));
+}
+
+static void
+udp_shutdown(isc_task_t *task, isc_event_t *event)
+{
+	udp_cctx_t *ctx;
+	udp_listener_t *l;
+
+	ctx = (udp_cctx_t *)(event->arg);
+	l = ctx->parent;
+
+	LOCK(&l->lock);
+
+	REQUIRE(l->nwactive > 0);
+
+	/*
+	 * remove our task from the list of tasks that the listener
+	 * maintains by setting things to NULL, then freeing the
+	 * pointers we maintain.
+	 */
+	INSIST(l->tasks[ctx->slot] == task);
+	l->tasks[ctx->slot] = NULL;
+	l->ctxs[ctx->slot] = NULL;
+
+	isc_socket_cancel(l->sock, task, ISC_SOCKCANCEL_ALL);
+
+	l->nwactive--;
+
+	UNLOCK(&l->lock);
+
+	printf("Final shutdown slot %u\n", ctx->slot);
+	udp_cctx_free(ctx);
+
+	isc_event_free(&event);
+}
+
+static void
+udp_recv(isc_task_t *task, isc_event_t *event)
+{
+	isc_socket_t *sock;
+	isc_socketevent_t *dev;
+	udp_cctx_t *ctx;
+
+	sock = event->sender;
+	dev = (isc_socketevent_t *)event;
+	ctx = (udp_cctx_t *)(event->arg);
+
+	printf("Task %u (sock %p, base %p, length %d, n %d, result %d)\n",
+	       ctx->slot, sock,
+	       dev->region.base, dev->region.length,
+	       dev->n, dev->result);
+	printf("\tFrom: %s port %d\n",
+	       inet_ntoa(dev->address.type.sin.sin_addr),
+	       ntohs(dev->address.type.sin.sin_port));
+
+	if (dev->result != ISC_R_SUCCESS) {
+		isc_socket_detach(&sock);
+
+		udp_cctx_free(ctx);
+
+		isc_event_free(&event);
+
+		isc_task_shutdown(task);
+
+		return;
+	}
+
+	/*
+	 * Call the dump routine to print this baby out
+	 */
+	dump_packet(ctx->buf, dev->n);
+
+	isc_socket_recv(sock, &dev->region, ISC_FALSE,
+			task, udp_recv, event->arg);
+
+	/*
+	 * Hack.  Shutdown after we've received (slot + 1) * 2
+	 * queries.
+	 */
+	ctx->count++;
+	if (ctx->count == (ctx->slot + 1) * 2) {
+
+		isc_socket_cancel(ctx->parent->sock, task, ISC_SOCKCANCEL_ALL);
+		printf("Shutting down slot %u\n", ctx->slot);
+	}
+
+	isc_event_free(&event);
+}
+
+static void
+udp_send(isc_task_t *task, isc_event_t *event)
+{
+	isc_socket_t *sock;
+	isc_socketevent_t *dev;
+
+	sock = event->sender;
+	dev = (isc_socketevent_t *)event;
+
+	printf("my_send: %s task %p\n\t(sock %p, base %p, length %d, n %d, result %d)\n",
+	       (char *)(event->arg), task, sock,
+	       dev->region.base, dev->region.length,
+	       dev->n, dev->result);
+
+	isc_mem_put(event->mctx, dev->region.base, dev->region.length);
+
+	isc_event_free(&event);
+}
+
+udp_listener_t *
+udp_listener_allocate(isc_mem_t *mctx, u_int nwmax)
+{
+	udp_listener_t *l;
+
+	l = isc_mem_get(mctx, sizeof(udp_listener_t));
+	if (l == NULL)
+		return (NULL);
+
+	if (isc_mutex_init(&l->lock) != ISC_R_SUCCESS) {
+		isc_mem_put(mctx, l, sizeof(udp_listener_t));
+
+		UNEXPECTED_ERROR(__FILE__, __LINE__,
+				 "isc_mutex_init() failed");
+
+		return (NULL);
+	}
+
+	l->tasks = isc_mem_get(mctx, sizeof(isc_task_t *) * nwmax);
+	RUNTIME_CHECK(l->tasks != NULL); /* XXX should be non-fatal? */
+	l->ctxs = isc_mem_get(mctx, sizeof(udp_cctx_t *) * nwmax);
+	RUNTIME_CHECK(l->ctxs != NULL);  /* XXX should be non-fatal? */
+
+	l->mctx = mctx;
+
+	return (l);
+}
+
+isc_result_t
+udp_listener_start(udp_listener_t *l,
+		   isc_socket_t *sock, isc_taskmgr_t *tmgr,
+		   u_int nwstart, u_int nwkeep, u_int nwtimeout)
+{
+	u_int i;
+	isc_region_t region;
+
+	LOCK(&l->lock);
+	INSIST(l->nwactive == 0);
+
+	l->sock = sock;
+
+	for (i = 0 ; i < nwstart ; i++) {
+		l->tasks[i] = NULL;
+		RUNTIME_CHECK(isc_task_create(tmgr, NULL, 0, &l->tasks[i])
+			      == ISC_R_SUCCESS);
+
+		l->ctxs[i] = udp_cctx_allocate(l->mctx);
+		RUNTIME_CHECK(l->ctxs[i] != NULL);
+
+		l->ctxs[i]->parent = l;
+		l->ctxs[i]->slot = i;
+
+		RUNTIME_CHECK(isc_task_onshutdown(l->tasks[i], udp_shutdown,
+						  l->ctxs[i])
+			      == ISC_R_SUCCESS);
+
+		region.length = UDP_INPUT_BUFFER_SIZE;
+		region.base = l->ctxs[i]->buf;
+
+		RUNTIME_CHECK(isc_socket_recv(sock, &region,
+					      ISC_FALSE, l->tasks[i],
+					      udp_recv, l->ctxs[i])
+			      == ISC_R_SUCCESS);
+
+		l->nwactive++;
+	}
+
+	UNLOCK(&l->lock);
+
+	return (ISC_R_SUCCESS);
+}
+
+/*
+ * XXX All of the following is for debugging only, and will eventually
+ * be in a library or removed when we really answer queries.
+ */
+typedef struct dns_message {
+	unsigned int		id;
+	unsigned int		flags;
+	unsigned int		qcount;
+	unsigned int		ancount;
+	unsigned int		aucount;
+	unsigned int		adcount;
+	dns_namelist_t		question;
+	dns_namelist_t		answer;
+	dns_namelist_t		authority;
+	dns_namelist_t		additional;
+} dns_message_t;
+
+/*
+ * in wire_test.c
+ */
+void getmessage(dns_message_t *message, isc_buffer_t *source,
+		isc_buffer_t *target);
+dns_result_t printmessage(dns_message_t *message);
+
+static void
+dump_packet(char *buf, u_int len)
+{
+	extern dns_decompress_t dctx;
+	extern unsigned int rdcount, rlcount, ncount;
+	char t[5000]; /* XXX */
+	dns_message_t message;
+	dns_result_t result;
+	isc_buffer_t source, target;
+
+	rdcount = 0;
+	rlcount = 0;
+	ncount = 0;
+
+	dctx.allowed = DNS_COMPRESS_GLOBAL14;
+	dns_name_init(&dctx.owner_name, NULL);
+
+	isc_buffer_init(&source, buf, len, ISC_BUFFERTYPE_BINARY);
+	isc_buffer_add(&source, len);
+	isc_buffer_init(&target, t, sizeof(t), ISC_BUFFERTYPE_BINARY);
+
+	getmessage(&message, &source, &target);
+	result = printmessage(&message);
+	if (result != DNS_R_SUCCESS)
+		printf("printmessage() failed: %s\n",
+		       dns_result_totext(result));
+}

bin/named/udpclient.h

@@ -0,0 +1,36 @@
+#include <isc/mutex.h>
+#include <isc/socket.h>
+#include <isc/task.h>
+
+#define UDP_INPUT_BUFFER_SIZE (64 * 1024)	/* 64k */
+
+typedef struct __udp_listener udp_listener_t;
+typedef struct __udp_cctx udp_cctx_t;
+
+struct __udp_cctx {
+	char *buf;		/* input buffer */
+	isc_mem_t *mctx;	/* memory context used to allocate */
+	udp_listener_t *parent;	/* controlling listener */
+	u_int slot;		/* slot # in tasks[] (and ctxs[]) array */
+	u_int count; /* XXX debug */
+};
+
+struct __udp_listener {
+	isc_socket_t *sock;	/* the socket */
+	u_int nwstart;		/* workers to start */
+	u_int nwkeep;		/* workers to keep */
+	u_int nwmax;		/* workers max */
+	isc_mem_t *mctx;
+	isc_mutex_t lock;
+
+	/* locked */
+	isc_task_t **tasks;	/* list of tasks */
+	u_int nwactive;		/* workers active */
+	udp_cctx_t **ctxs;	/* list of contexts */
+};
+
+udp_listener_t *udp_listener_allocate(isc_mem_t *mctx, u_int nwmax);
+
+isc_result_t udp_listener_start(udp_listener_t *l,
+				isc_socket_t *sock, isc_taskmgr_t *tmgr,
+				u_int nwstart, u_int nwkeep, u_int nwtimeout);

lib/isc/unix/socket.c

@@ -2503,6 +2503,7 @@ isc_socket_cancel(isc_socket_t *sock, isc_task_t *task,
 		}
 	}
 
+	/* XXX implement */
 	if (how & ISC_SOCKCANCEL_CONNECT) {
 	}