Subject: Switch to poll() in cxenstored's IO loop
From: Wei Liu wei.liu2@citrix.com Mon Mar 25 11:17:31 2013 +0000
Date: Thu Apr 11 15:42:44 2013 +0100:
Git: edb13df2aece2acf5a50f20aab79599173e55b2e

Poll() can support more file descriptors than select(). We've done this for
xenconsoled, now do this for cxenstored as well.

The code is taken from xenconsoled and modified to adapt to cxenstored.

Note that poll() semantic is a bit different from select(). In Linux, if a fd
is set in IN/OUT fd_set and error occurs inside select(), this fd is still
considered readable / writable, and it is set in the returned IN/OUT fd_set.
So in later handle_input / handle_output, the connection will eventually be
talloc_free'ed(). After switching to poll(), we should take care of any error
right away, making the code clearer.

Signed-off-by: Wei Liu <wei.liu2@citrix.com>
Acked-by: Ian Campbell <ian.campbell@citrix.com>

--- a/tools/xenstore/xenstored_core.c
+++ b/tools/xenstore/xenstored_core.c
@@ -20,7 +20,7 @@
 #include <inttypes.h>
 #include <sys/types.h>
 #include <sys/stat.h>
-#include <sys/select.h>
+#include <sys/poll.h>
 #ifndef NO_SOCKETS
 #include <sys/socket.h>
 #include <sys/un.h>
@@ -55,12 +55,19 @@
 #include "tdb.h"
 
 extern xc_evtchn *xce_handle; /* in xenstored_domain.c */
+static int xce_pollfd_idx = -1;
+static struct pollfd *fds;
+static unsigned int current_array_size;
+static unsigned int nr_fds;
+
+#define ROUNDUP(_x, _w) (((unsigned long)(_x)+(1UL<<(_w))-1) & ~((1UL<<(_w))-1))
 
 static bool verbose = false;
 LIST_HEAD(connections);
 int tracefd = -1;
 static bool recovery = true;
 static int reopen_log_pipe[2];
+static int reopen_log_pipe0_pollfd_idx = -1;
 char *tracefile = NULL;
 TDB_CONTEXT *tdb_ctx = NULL;
 
@@ -158,7 +165,7 @@ void trace_destroy(const void *data, con
 /**
  * Signal handler for SIGHUP, which requests that the trace log is reopened
  * (in the main loop).  A single byte is written to reopen_log_pipe, to awaken
- * the select() in the main loop.
+ * the poll() in the main loop.
  */
 static void trigger_reopen_log(int signal __attribute__((unused)))
 {
@@ -243,15 +250,12 @@ static int destroy_conn(void *_conn)
 
 	/* Flush outgoing if possible, but don't block. */
 	if (!conn->domain) {
-		fd_set set;
-		struct timeval none;
-
-		FD_ZERO(&set);
-		FD_SET(conn->fd, &set);
-		none.tv_sec = none.tv_usec = 0;
+		struct pollfd pfd;
+		pfd.fd = conn->fd;
+		pfd.events = POLLOUT;
 
 		while (!list_empty(&conn->out_list)
-		       && select(conn->fd+1, NULL, &set, NULL, &none) == 1)
+		       && poll(&pfd, 1, 0) == 1)
 			if (!write_messages(conn))
 				break;
 		close(conn->fd);
@@ -264,64 +268,82 @@ static int destroy_conn(void *_conn)
 }
 
 
-static void set_fd(int fd, fd_set *set, int *max)
+/* This function returns index inside the array if succeed, -1 if fail */
+static int set_fd(int fd, short events)
 {
-	if (fd < 0)
-		return;
-	FD_SET(fd, set);
-	if (fd > *max)
-		*max = fd;
-}
+	int ret;
+	if (current_array_size < nr_fds + 1) {
+		struct pollfd *new_fds = NULL;
+		unsigned long newsize;
+
+		/* Round up to 2^8 boundary, in practice this just
+		 * make newsize larger than current_array_size.
+		 */
+		newsize = ROUNDUP(nr_fds + 1, 8);
 
+		new_fds = realloc(fds, sizeof(struct pollfd)*newsize);
+		if (!new_fds)
+			goto fail;
+		fds = new_fds;
+
+		memset(&fds[0] + current_array_size, 0,
+		       sizeof(struct pollfd ) * (newsize-current_array_size));
+		current_array_size = newsize;
+	}
+
+	fds[nr_fds].fd = fd;
+	fds[nr_fds].events = events;
+	ret = nr_fds;
+	nr_fds++;
 
-static int initialize_set(fd_set *inset, fd_set *outset, int sock, int ro_sock,
-			  struct timeval **ptimeout)
+	return ret;
+fail:
+	syslog(LOG_ERR, "realloc failed, ignoring fd %d\n", fd);
+	return -1;
+}
+
+static void initialize_fds(int sock, int *p_sock_pollfd_idx,
+			   int ro_sock, int *p_ro_sock_pollfd_idx,
+			   int *ptimeout)
 {
-	static struct timeval zero_timeout = { 0 };
-	static struct timeval wrl_timeout;
 	struct connection *conn;
 	struct wrl_timestampt now;
-	int timeout = -1;
-	int max = -1;
 
-	*ptimeout = NULL;
+	if (fds)
+		memset(fds, 0, sizeof(struct pollfd) * current_array_size);
+	nr_fds = 0;
 
-	FD_ZERO(inset);
-	FD_ZERO(outset);
+	*ptimeout = -1;
 
 	if (sock != -1)
-		set_fd(sock, inset, &max);
+		*p_sock_pollfd_idx = set_fd(sock, POLLIN|POLLPRI);
 	if (ro_sock != -1)
-		set_fd(ro_sock, inset, &max);
+		*p_ro_sock_pollfd_idx = set_fd(ro_sock, POLLIN|POLLPRI);
 	if (reopen_log_pipe[0] != -1)
-		set_fd(reopen_log_pipe[0], inset, &max);
+		reopen_log_pipe0_pollfd_idx =
+			set_fd(reopen_log_pipe[0], POLLIN|POLLPRI);
 
 	if (xce_handle != NULL)
-		set_fd(xc_evtchn_fd(xce_handle), inset, &max);
+		xce_pollfd_idx = set_fd(xc_evtchn_fd(xce_handle),
+					POLLIN|POLLPRI);
 
 	wrl_gettime_now(&now);
 	wrl_log_periodic(now);
 
 	list_for_each_entry(conn, &connections, list) {
 		if (conn->domain) {
-			wrl_check_timeout(conn->domain, now, &timeout);
+			wrl_check_timeout(conn->domain, now, ptimeout);
 			if (domain_can_read(conn) ||
 			    (domain_can_write(conn) &&
-			     !list_empty(&conn->out_list))) {
-				*ptimeout = &zero_timeout;
-			} else if (timeout > 0) {
-				wrl_timeout.tv_sec = timeout / 1000;
-				wrl_timeout.tv_usec = (timeout % 1000) * 1000;
-				*ptimeout = &wrl_timeout;
-			}
+			     !list_empty(&conn->out_list)))
+				*ptimeout = 0;
 		} else {
-			set_fd(conn->fd, inset, &max);
+			short events = POLLIN|POLLPRI;
 			if (!list_empty(&conn->out_list))
-				FD_SET(conn->fd, outset);
+				events |= POLLOUT;
+			conn->pollfd_idx = set_fd(conn->fd, events);
 		}
 	}
-
-	return max;
 }
 
 /*
@@ -1452,6 +1474,7 @@ struct connection *new_connection(connwr
 		return NULL;
 
 	new->fd = -1;
+	new->pollfd_idx = -1;
 	new->write = write;
 	new->read = read;
 	new->can_write = true;
@@ -1932,14 +1955,13 @@ int priv_domid = 0;
 
 int main(int argc, char *argv[])
 {
-	int opt, *sock, *ro_sock, max;
-	fd_set inset, outset;
+	int opt, *sock, *ro_sock;
+	int sock_pollfd_idx = -1, ro_sock_pollfd_idx = -1;
 	bool dofork = true;
 	bool outputpid = false;
 	bool no_domain_init = false;
 	const char *pidfile = NULL;
-	int evtchn_fd = -1;
-	struct timeval *timeout;
+	int timeout;
 
 	while ((opt = getopt_long(argc, argv, "DE:F:HNPS:t:A:T:RVW:", options,
 				  NULL)) != -1) {
@@ -2046,11 +2068,9 @@ int main(int argc, char *argv[])
 	if (tracefile)
 		tracefile = talloc_strdup(NULL, tracefile);
 
-	if (xce_handle != NULL)
-		evtchn_fd = xc_evtchn_fd(xce_handle);
-
 	/* Get ready to listen to the tools. */
-	max = initialize_set(&inset, &outset, *sock, *ro_sock, &timeout);
+	initialize_fds(*sock, &sock_pollfd_idx, *ro_sock, &ro_sock_pollfd_idx,
+		       &timeout);
 
 	/* Tell the kernel we're up and running. */
 	xenbus_notify_running();
@@ -2059,27 +2079,57 @@ int main(int argc, char *argv[])
 	for (;;) {
 		struct connection *conn, *next;
 
-		if (select(max+1, &inset, &outset, NULL, timeout) < 0) {
+		if (poll(fds, nr_fds, timeout) < 0) {
 			if (errno == EINTR)
 				continue;
-			barf_perror("Select failed");
+			barf_perror("Poll failed");
 		}
 
-		if (reopen_log_pipe[0] != -1 && FD_ISSET(reopen_log_pipe[0], &inset)) {
-			char c;
-			if (read(reopen_log_pipe[0], &c, 1) != 1)
-				barf_perror("read failed");
-			reopen_log();
+		if (reopen_log_pipe0_pollfd_idx != -1) {
+			if (fds[reopen_log_pipe0_pollfd_idx].revents
+			    & ~POLLIN) {
+				close(reopen_log_pipe[0]);
+				close(reopen_log_pipe[1]);
+				init_pipe(reopen_log_pipe);
+			} else if (fds[reopen_log_pipe0_pollfd_idx].revents
+				   & POLLIN) {
+				char c;
+				if (read(reopen_log_pipe[0], &c, 1) != 1)
+					barf_perror("read failed");
+				reopen_log();
+			}
+			reopen_log_pipe0_pollfd_idx = -1;
 		}
 
-		if (*sock != -1 && FD_ISSET(*sock, &inset))
-			accept_connection(*sock, true);
+		if (sock_pollfd_idx != -1) {
+			if (fds[sock_pollfd_idx].revents & ~POLLIN) {
+				barf_perror("sock poll failed");
+				break;
+			} else if (fds[sock_pollfd_idx].revents & POLLIN) {
+				accept_connection(*sock, true);
+				sock_pollfd_idx = -1;
+			}
+		}
 
-		if (*ro_sock != -1 && FD_ISSET(*ro_sock, &inset))
-			accept_connection(*ro_sock, false);
+		if (ro_sock_pollfd_idx != -1) {
+			if (fds[ro_sock_pollfd_idx].revents & ~POLLIN) {
+				barf_perror("ro sock poll failed");
+				break;
+			} else if (fds[ro_sock_pollfd_idx].revents & POLLIN) {
+				accept_connection(*ro_sock, false);
+				ro_sock_pollfd_idx = -1;
+			}
+		}
 
-		if (evtchn_fd != -1 && FD_ISSET(evtchn_fd, &inset))
-			handle_event();
+		if (xce_pollfd_idx != -1) {
+			if (fds[xce_pollfd_idx].revents & ~POLLIN) {
+				barf_perror("xce_handle poll failed");
+				break;
+			} else if (fds[xce_pollfd_idx].revents & POLLIN) {
+				handle_event();
+				xce_pollfd_idx = -1;
+			}
+		}
 
 		next = list_entry(connections.next, typeof(*conn), list);
 		if (&next->list != &connections)
@@ -2105,21 +2155,36 @@ int main(int argc, char *argv[])
 				if (talloc_free(conn) == 0)
 					continue;
 			} else {
-				if (FD_ISSET(conn->fd, &inset))
-					handle_input(conn);
+				if (conn->pollfd_idx != -1) {
+					if (fds[conn->pollfd_idx].revents
+					    & ~(POLLIN|POLLOUT))
+						talloc_free(conn);
+					else if (fds[conn->pollfd_idx].revents
+						 & POLLIN)
+						handle_input(conn);
+				}
 				if (talloc_free(conn) == 0)
 					continue;
 
 				talloc_increase_ref_count(conn);
-				if (FD_ISSET(conn->fd, &outset))
-					handle_output(conn);
+
+				if (conn->pollfd_idx != -1) {
+					if (fds[conn->pollfd_idx].revents
+					    & ~(POLLIN|POLLOUT))
+						talloc_free(conn);
+					else if (fds[conn->pollfd_idx].revents
+						 & POLLOUT)
+						handle_output(conn);
+				}
 				if (talloc_free(conn) == 0)
 					continue;
+
+				conn->pollfd_idx = -1;
 			}
 		}
 
-		max = initialize_set(&inset, &outset, *sock, *ro_sock,
-				     &timeout);
+		initialize_fds(*sock, &sock_pollfd_idx, *ro_sock,
+			       &ro_sock_pollfd_idx, &timeout);
 	}
 }
 
--- a/tools/xenstore/xenstored_core.h
+++ b/tools/xenstore/xenstored_core.h
@@ -71,6 +71,8 @@ struct connection
 
 	/* The file descriptor we came in on. */
 	int fd;
+	/* The index of pollfd in global pollfd array */
+	int pollfd_idx;
 
 	/* Who am I? 0 for socket connections. */
 	unsigned int id;
