X-Git-Url: http://git.ozlabs.org/?a=blobdiff_plain;f=ccan%2Foserver%2Foserver.c;h=2d4cac41b28754fbef15a07f952edf5eb5109b0d;hb=a1ee151190dc9ac9bf17163ab4d31f0491b8bbee;hp=20edfcbc19a79c846d3e2f31d834eebccd64fced;hpb=6dc5af4494849dec8639c57c591c0dcc9bbd7173;p=ccan-lca-2011.git

diff --git a/ccan/oserver/oserver.c b/ccan/oserver/oserver.c
index 20edfcb..2d4cac4 100644
--- a/ccan/oserver/oserver.c
+++ b/ccan/oserver/oserver.c
@@ -1,7 +1,8 @@
 #include <ccan/oserver/oserver.h>
 #include <ccan/read_write_all/read_write_all.h>
 #include <ccan/opt/opt.h>
-#include <ccan/noerr/noerr.h>
+#include <ccan/tevent/tevent.h>
+#include <ccan/array_size/array_size.h>
 #include <sys/types.h>
 #include <sys/socket.h>
 #include <netinet/in.h>
@@ -13,58 +14,328 @@
 #include <stdlib.h>
 #include <string.h>
 #include <errno.h>
+#include <signal.h>
+#include <assert.h>
 
-bool oserver_serve(int fd)
+enum state {
+	SENDING_GREETING,
+	RECEIVING_USER_QUESTION,
+	AWAITING_A_SUBCLIENT,
+	SENDING_OTHER_QUESTION_PREFIX,
+	SENDING_OTHER_QUESTION,
+	RECEIVING_OTHER_ANSWER,
+	AWAITING_OUR_ORACLE,
+	SENDING_ANSWER_PREFIX,
+	SENDING_ANSWER,
+	FINISHED
+};
+
+static uint16_t state_flag_map[] = {
+	[SENDING_GREETING]		= TEVENT_FD_WRITE,
+	[RECEIVING_USER_QUESTION]	= TEVENT_FD_READ,
+	[AWAITING_A_SUBCLIENT]		= 0,
+	[SENDING_OTHER_QUESTION_PREFIX] = TEVENT_FD_WRITE,
+	[SENDING_OTHER_QUESTION]	= TEVENT_FD_WRITE,
+	[RECEIVING_OTHER_ANSWER]	= TEVENT_FD_READ,
+	[AWAITING_OUR_ORACLE]		= 0,
+	[SENDING_ANSWER_PREFIX]		= TEVENT_FD_WRITE,
+	[SENDING_ANSWER]		= TEVENT_FD_WRITE,
+	[FINISHED]			= 0
+};
+
+struct client {
+	/* What are we doing today, brain? */
+	enum state state;
+	/* Our event info, and the file descriptor. */
+	struct tevent_fd *fde;
+	int fd;
+	/* The question we read from client. */
+	char *question;
+	/* The answer to the client. */
+	char *answer;
+	/* How many bytes of the reply we sent so far. */
+	size_t bytes_sent;
+	/* Our server. */
+	struct oserver *oserver;
+	/* Whose question this client is answering. */
+	struct client *subclient;
+	/* Who is answering our question. */
+	struct client *oracle;
+};
+
+struct oserver {
+	/* 5 clients should be enough for anybody! */
+	struct client *clients[5];
+	int fd;
+	struct tevent_fd *fde;
+};
+
+static ssize_t write_string(int fd, const char *str)
 {
-	char buf[1024];
-	unsigned int i;
+	return write(fd, str, strlen(str));
+}
 
-	for (i = 0; i < sizeof(buf)-1; i++) {
-		if (read(fd, &buf[i], 1) != 1) {
-			errno = EIO;
-			return false;
-		}
-		if (buf[i] == '\n' || buf[i] == '\r')
-			break;
-		buf[i] = toupper(buf[i]);
+static ssize_t read_string(int fd, char **buf)
+{
+	ssize_t ret, len, maxlen;
+
+	len = strlen(*buf);
+	maxlen = talloc_array_length(*buf);
+
+	if (maxlen < len + 100) {
+		maxlen += 100;
+		*buf = talloc_realloc(NULL, *buf, char, maxlen);
 	}
-	buf[i] = '\0';
 
-	if (!write_all(fd, "Louder, like this: '",
-		       strlen("Louder, like this: '"))
-	    || !write_all(fd, buf, i)
-	    || !write_all(fd, "'\r\n", strlen("'\r\n")))
+	ret = read(fd, *buf + len, maxlen - len - 1);
+	if (ret >= 0)
+		(*buf)[len + ret] = '\0';
+	return ret;
+}
+
+static bool input_finished(const char *str)
+{
+	return strchr(str, '\n');
+}
+
+/* Update state, and set our READ/WRITE flags appropriately. */
+static void set_state(struct client *c, enum state state)
+{
+	c->state = state;
+	tevent_fd_set_flags(c->fde, state_flag_map[state]);
+}
+
+/* Returns false on error, increments state on finishing string. */
+static bool send_string(struct client *c, const char *str)
+{
+	ssize_t len = write_string(c->fd, str + c->bytes_sent);
+	if (len < 0)
 		return false;
+	c->bytes_sent += len;
+	if (c->bytes_sent == strlen(str)) {
+		c->bytes_sent = 0;
+		set_state(c, c->state+1);
+	}
 	return true;
 }
 
-int oserver_setup(unsigned short port)
+static bool get_subclient(struct client *me)
+{
+	unsigned int i;
+
+	for (i = 0; i < ARRAY_SIZE(me->oserver->clients); i++) {
+		struct client *c = me->oserver->clients[i];
+		if (!c || c == me)
+			continue;
+		if (c->oracle == NULL && input_finished(c->question)) {
+			me->subclient = c;
+			c->oracle = me;
+			return true;
+		}
+	}
+	return false;
+}
+
+static bool get_oracle(struct client *me)
+{
+	unsigned int i;
+
+	for (i = 0; i < ARRAY_SIZE(me->oserver->clients); i++) {
+		struct client *c = me->oserver->clients[i];
+		if (!c || c == me)
+			continue;
+		if (c->subclient == NULL && input_finished(c->question)) {
+			me->oracle = c;
+			c->subclient = me;
+			return true;
+		}
+	}
+	return false;
+}
+
+static void service_client(struct tevent_context *ev,
+			   struct tevent_fd *fde, uint16_t flags, void *_c)
+{
+	struct client *c = _c;
+	ssize_t len;
+
+	switch (c->state) {
+	case SENDING_GREETING:
+		if (!send_string(c, "Welcome.  Please ask your question.\n"))
+			goto fail;
+		break;
+	case RECEIVING_USER_QUESTION:
+		len = read_string(c->fd, &c->question);
+		if (len <= 0)
+			goto fail;
+		if (input_finished(c->question)) {
+			/* Look for someone to be oracle to. */
+			if (get_subclient(c)) {
+				set_state(c, SENDING_OTHER_QUESTION_PREFIX);
+			} else {
+				/* We sit here until someone find_oracles us */
+				set_state(c, AWAITING_A_SUBCLIENT);
+			}
+
+			/* Look for an oracle for ourselves. */
+			if (get_oracle(c)) {
+				assert(c->oracle->state
+				       == AWAITING_A_SUBCLIENT);
+				set_state(c->oracle,
+					  SENDING_OTHER_QUESTION_PREFIX);
+			}
+		}
+		break;
+	case SENDING_OTHER_QUESTION_PREFIX:
+		if (!send_string(c, "While the Oracle ponders,"
+				 " please answer the following question:\n"))
+			goto fail;
+		break;
+	case SENDING_OTHER_QUESTION:
+		if (!send_string(c, c->subclient->question))
+			goto fail;
+		break;
+	case RECEIVING_OTHER_ANSWER:
+		len = read_string(c->fd, &c->subclient->answer);
+		if (len <= 0)
+			goto fail;
+		if (input_finished(c->subclient->answer)) {
+			/* Did our oracle answer for us already? */
+			if (input_finished(c->answer))
+				set_state(c, SENDING_ANSWER_PREFIX);
+			else
+				set_state(c, AWAITING_OUR_ORACLE);
+
+			/* If they were waiting for an answer, move them. */
+			if (c->subclient->state == AWAITING_OUR_ORACLE)
+				set_state(c->subclient, SENDING_ANSWER_PREFIX);
+		}
+		break;
+	case SENDING_ANSWER_PREFIX:
+		if (!send_string(c, "The Oracle spake thus:\n"))
+			goto fail;
+		break;
+	case SENDING_ANSWER:
+		if (!send_string(c, c->answer))
+			goto fail;
+		break;
+	default:
+		goto fail;
+	}
+
+	if (c->state != FINISHED)
+		return;
+
+fail:
+	talloc_free(c);
+}
+
+static int cleanup_client(struct client *client)
+{
+	unsigned int i;
+
+	for (i = 0; i < ARRAY_SIZE(client->oserver->clients); i++) {
+		if (client->oserver->clients[i] == client) {
+			client->oserver->clients[i] = NULL;
+			tevent_fd_set_flags(client->oserver->fde,
+					    TEVENT_FD_READ);
+			return 0;
+		}
+	}
+	abort();
+}
+
+static void add_client(struct tevent_context *ev,
+		       struct tevent_fd *fde, uint16_t flags, void *_oserver)
+{
+	struct oserver *oserver = _oserver;
+	struct client *client;
+	unsigned int i;
+
+	client = talloc(oserver, struct client);
+	client->fd = accept(oserver->fd, NULL, 0);
+	if (client->fd < 0)
+		err(1, "Accepting client connection");
+
+	client->state = SENDING_GREETING;
+	client->bytes_sent = 0;
+	client->question = talloc_strdup(client, "");
+	client->oserver = oserver;
+	client->oracle = NULL;
+	client->subclient = NULL;
+	client->answer = talloc_strdup(client, "");
+	client->fde = tevent_add_fd(ev, client, client->fd,
+				    state_flag_map[client->state],
+				    service_client, client);
+	tevent_fd_set_auto_close(client->fde);
+
+	/* Find empty slot in array for this client. */
+	for (i = 0; oserver->clients[i]; i++);
+	oserver->clients[i] = client;
+	talloc_set_destructor(client, cleanup_client);
+
+	/* Full?  Stop listening... */
+	if (i == ARRAY_SIZE(oserver->clients)-1)
+		tevent_fd_set_flags(oserver->fde, 0);
+}
+
+static void clear_clients(struct oserver *oserver)
+{
+ 	memset(oserver->clients, 0,
+	       ARRAY_SIZE(oserver->clients) * sizeof(oserver->clients[0]));
+}
+
+static int destroy_oserver(struct oserver *oserver)
 {
-	int sockfd;
+	close(oserver->fd);
+	return 0;
+}
+
+struct oserver *oserver_setup(struct tevent_context *ev, unsigned short port)
+{
+	struct oserver *oserver;
 	int one = 1;
 	union {
 		struct sockaddr addr;
 		struct sockaddr_in in;
 	} u;
 
-	sockfd = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
-	if (sockfd < 0)
-		return -1;
+	oserver = talloc(ev, struct oserver);
+	clear_clients(oserver);
+	oserver->fd = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
+	if (oserver->fd < 0) {
+		talloc_free(oserver);
+		return NULL;
+	}
+
+	talloc_set_destructor(oserver, destroy_oserver);
 
-	if (setsockopt(sockfd, SOL_SOCKET, SO_REUSEADDR, &one, sizeof(one)))
+	if (setsockopt(oserver->fd, SOL_SOCKET, SO_REUSEADDR,
+		       &one, sizeof(one)))
 		warn("Setting socket reuse");
 
 	u.in.sin_family = AF_INET;
 	u.in.sin_port = htons(port);
 	u.in.sin_addr.s_addr = INADDR_ANY;
-	if (bind(sockfd, &u.addr, sizeof(u.in)) == -1) {
-		close_noerr(sockfd);
-		return -1;
+	if (bind(oserver->fd, &u.addr, sizeof(u.in)) == -1) {
+		talloc_free(oserver);
+		return NULL;
 	}
 
-	if (listen(sockfd, 0) != 0) {
-		close_noerr(sockfd);
-		return -1;
+	if (listen(oserver->fd, 0) != 0) {
+		talloc_free(oserver);
+		return NULL;
 	}
-	return sockfd;
+
+	oserver->fde = tevent_add_fd(ev, oserver, oserver->fd,
+				     TEVENT_FD_READ, add_client, oserver);
+	if (!oserver->fde) {
+		talloc_free(oserver);
+		return NULL;
+	}
+
+	/* Don't kill us if client dies. */
+	signal(SIGPIPE, SIG_IGN);
+
+	return oserver;
 }