jobserver: use non-blocking accept

To allow the JobServer to shutdown properly, the accept() loop in
JobServer::run() needs to regularly check whether we're done.  This
change introduces a timed, non-blocking variant of accept() into
SocketComm to achieve this.

Change-Id: Id411096be816c4ed6c7b0b37674410e22152eb22

src/core/comm/SocketComm.cc

@@ -1,6 +1,7 @@
 #include <string>
 #include <errno.h>
 #include <signal.h>
+#include <poll.h>
 
 #include "SocketComm.hpp"
 
@@ -63,6 +64,17 @@ char * SocketComm::getBuf(int sockfd, int *size)
     return buf;
 }
 
+int SocketComm::timedAccept(int sockfd, struct sockaddr *addr, socklen_t *addrlen, int timeout)
+{
+	struct pollfd pfd = {sockfd, POLLIN, 0};
+	int ret = poll(&pfd, 1, timeout);
+	if (ret > 0) {
+		return accept(sockfd, addr, addrlen);
+	}
+	errno = EWOULDBLOCK;
+	return -1;
+}
+
 ssize_t SocketComm::safe_write(int fd, const void *buf, size_t count)
 {
 	ssize_t ret;

src/core/comm/SocketComm.hpp

@@ -45,6 +45,16 @@ public:
 	 */
 	static bool dropMsg(int sockfd);
 
+	/**
+	 * An accept() wrapper that times out (using poll(2))
+	 * @param sockfd listening socket descriptor to accept connections from
+	 * @param addr same as accept()'s
+	 * @param addrlen same as accept()'s
+	 * @param timeout timeout in milliseconds (see poll(2))
+	 * \return < 0 on failure, > 0 for a new socket connection
+	 */
+	static int timedAccept(int sockfd, struct sockaddr *addr, socklen_t *addrlen, int timeout);
+
 private:
 	static char * getBuf(int sockfd, int *size);
 	static ssize_t safe_write(int fd, const void *buf, size_t count);

src/core/cpn/JobServer.cc

@@ -161,11 +161,15 @@ void JobServer::run()
 	boost::thread* th;
 	while (!m_finish){
 		// Accept connection
-		int cs = accept(s, (struct sockaddr*)&clientaddr, &clen);
+		int cs = SocketComm::timedAccept(s, (struct sockaddr*)&clientaddr, &clen, 100);
-		if (cs == -1) {
+		if (cs < 0) {
-			perror("accept");
+			if (errno != EWOULDBLOCK) {
-			// TODO: Log-level?
+				perror("poll/accept");
-			return;
+				// TODO: Log-level?
+				return;
+			} else {
+				continue;
+			}
 		}
 		// Spawn a thread for further communication,
 		// and add this thread to a list threads