Merge branch 'poll'

Replace relevant uses of select() by poll(). poll(2) avoids the difficulties
we have with more than 1024 open file descriptors, and seems to be fairly
portable.

Fixes #757.

src/libcharon/plugins/kernel_libipsec/kernel_libipsec_router.c

@@ -130,35 +130,6 @@ static void deliver_plain(private_kernel_libipsec_router_t *this,
 	packet->destroy(packet);
 }
 
-/**
- * Create an FD set covering all TUN devices and the read end of the notify pipe
- */
-static int collect_fds(private_kernel_libipsec_router_t *this, fd_set *fds)
-{
-	enumerator_t *enumerator;
-	tun_entry_t *entry;
-	int maxfd;
-
-	FD_ZERO(fds);
-	FD_SET(this->notify[0], fds);
-	maxfd = this->notify[0];
-
-	FD_SET(this->tun.fd, fds);
-	maxfd = max(maxfd, this->tun.fd);
-
-	this->lock->read_lock(this->lock);
-	enumerator = this->tuns->create_enumerator(this->tuns);
-	while (enumerator->enumerate(enumerator, NULL, &entry))
-	{
-		FD_SET(entry->fd, fds);
-		maxfd = max(maxfd, entry->fd);
-	}
-	enumerator->destroy(enumerator);
-	this->lock->unlock(this->lock);
-
-	return maxfd + 1;
-}
-
 /**
  * Read and process outbound plaintext packet for the given TUN device
  */
@@ -183,14 +154,73 @@ static void process_plain(tun_device_t *tun)
 }
 
 /**
- * Handle waiting data for any TUN device
+ * Find flagged revents in a pollfd set by fd
  */
-static void handle_tuns(private_kernel_libipsec_router_t *this, fd_set *fds)
+static int find_revents(struct pollfd *pfd, int count, int fd)
+{
+	int i;
+
+	for (i = 0; i < count; i++)
+	{
+		if (pfd[i].fd == fd)
+		{
+			return pfd[i].revents;
+		}
+	}
+	return 0;
+}
+
+/**
+ * Job handling outbound plaintext packets
+ */
+static job_requeue_t handle_plain(private_kernel_libipsec_router_t *this)
 {
 	enumerator_t *enumerator;
 	tun_entry_t *entry;
+	bool oldstate;
+	int count = 0;
+	char buf[1];
+	struct pollfd *pfd;
 
-	if (FD_ISSET(this->tun.fd, fds))
+	this->lock->read_lock(this->lock);
+
+	pfd = alloca(sizeof(*pfd) * (this->tuns->get_count(this->tuns) + 2));
+	pfd[count].fd = this->notify[0];
+	pfd[count].events = POLLIN;
+	count++;
+	pfd[count].fd = this->tun.fd;
+	pfd[count].events = POLLIN;
+	count++;
+
+	enumerator = this->tuns->create_enumerator(this->tuns);
+	while (enumerator->enumerate(enumerator, NULL, &entry))
+	{
+		pfd[count].fd = entry->fd;
+		pfd[count].events = POLLIN;
+		count++;
+	}
+	enumerator->destroy(enumerator);
+	this->lock->unlock(this->lock);
+
+	oldstate = thread_cancelability(TRUE);
+	if (poll(pfd, count, -1) <= 0)
+	{
+		thread_cancelability(oldstate);
+		return JOB_REQUEUE_FAIR;
+	}
+	thread_cancelability(oldstate);
+
+	if (pfd[0].revents & POLLIN)
+	{
+		/* list of TUN devices changed, read notification data, rebuild FDs */
+		while (read(this->notify[0], &buf, sizeof(buf)) == sizeof(buf))
+		{
+			/* nop */
+		}
+		return JOB_REQUEUE_DIRECT;
+	}
+
+	if (pfd[1].revents & POLLIN)
 	{
 		process_plain(this->tun.tun);
 	}
@@ -199,42 +229,14 @@ static void handle_tuns(private_kernel_libipsec_router_t *this, fd_set *fds)
 	enumerator = this->tuns->create_enumerator(this->tuns);
 	while (enumerator->enumerate(enumerator, NULL, &entry))
 	{
-		if (FD_ISSET(entry->fd, fds))
+		if (find_revents(pfd, count, entry->fd) & POLLIN)
 		{
 			process_plain(entry->tun);
 		}
 	}
 	enumerator->destroy(enumerator);
 	this->lock->unlock(this->lock);
-}
 
-/**
- * Job handling outbound plaintext packets
- */
-static job_requeue_t handle_plain(private_kernel_libipsec_router_t *this)
-{
-	bool oldstate;
-	fd_set fds;
-	int maxfd;
-
-	maxfd = collect_fds(this, &fds);
-
-	oldstate = thread_cancelability(TRUE);
-	if (select(maxfd, &fds, NULL, NULL, NULL) <= 0)
-	{
-		thread_cancelability(oldstate);
-		return JOB_REQUEUE_FAIR;
-	}
-	thread_cancelability(oldstate);
-
-	if (FD_ISSET(this->notify[0], &fds))
-	{	/* list of TUN devices changed, read notification data, rebuild FDs */
-		char buf[1];
-		while (read(this->notify[0], &buf, sizeof(buf)) == sizeof(buf));
-		return JOB_REQUEUE_DIRECT;
-	}
-
-	handle_tuns(this, &fds);
 	return JOB_REQUEUE_DIRECT;
 }
 

src/libcharon/plugins/socket_default/socket_default_socket.c

@@ -141,6 +141,11 @@ struct private_socket_default_socket_t {
 	 * TRUE if the source address should be set on outbound packets
 	 */
 	bool set_source;
+
+	/**
+	 * A counter to implement round-robin selection of read sockets
+	 */
+	u_int rr_counter;
 };
 
 METHOD(socket_t, receiver, status_t,
@@ -150,66 +155,43 @@ METHOD(socket_t, receiver, status_t,
 	chunk_t data;
 	packet_t *pkt;
 	host_t *source = NULL, *dest = NULL;
-	int bytes_read = 0;
+	int i, rr, index, bytes_read = 0, selected = -1;
 	bool oldstate;
-
-	fd_set rfds;
-	int max_fd = 0, selected = 0;
 	u_int16_t port = 0;
-
-	FD_ZERO(&rfds);
-
-	if (this->ipv4 != -1)
-	{
-		FD_SET(this->ipv4, &rfds);
-		max_fd = max(max_fd, this->ipv4);
-	}
-	if (this->ipv4_natt != -1)
-	{
-		FD_SET(this->ipv4_natt, &rfds);
-		max_fd = max(max_fd, this->ipv4_natt);
-	}
-	if (this->ipv6 != -1)
-	{
-		FD_SET(this->ipv6, &rfds);
-		max_fd = max(max_fd, this->ipv6);
-	}
-	if (this->ipv6_natt != -1)
-	{
-		FD_SET(this->ipv6_natt, &rfds);
-		max_fd = max(max_fd, this->ipv6_natt);
-	}
+	struct pollfd pfd[] = {
+		{ .fd = this->ipv4,			.events = POLLIN },
+		{ .fd = this->ipv4_natt,	.events = POLLIN },
+		{ .fd = this->ipv6,			.events = POLLIN },
+		{ .fd = this->ipv6_natt,	.events = POLLIN },
+	};
+	int ports[] = {
+		/* port numbers assocaited to pollfds */
+		this->port, this->natt, this->port, this->natt,
+	};
 
 	DBG2(DBG_NET, "waiting for data on sockets");
 	oldstate = thread_cancelability(TRUE);
-	if (select(max_fd + 1, &rfds, NULL, NULL, NULL) <= 0)
+	if (poll(pfd, countof(pfd), -1) <= 0)
 	{
 		thread_cancelability(oldstate);
 		return FAILED;
 	}
 	thread_cancelability(oldstate);
 
-	if (this->ipv4 != -1 && FD_ISSET(this->ipv4, &rfds))
+	rr = this->rr_counter++;
+	for (i = 0; i < countof(pfd); i++)
 	{
-		port = this->port;
-		selected = this->ipv4;
+		/* To serve all ports with equal priority, we use a round-robin
+		 * scheme to choose the one to process in this invocation */
+		index = (rr + i) % countof(pfd);
+		if (pfd[index].revents & POLLIN)
+		{
+			selected = pfd[index].fd;
+			port = ports[index];
+			break;
+		}
 	}
-	if (this->ipv4_natt != -1 && FD_ISSET(this->ipv4_natt, &rfds))
-	{
-		port = this->natt;
-		selected = this->ipv4_natt;
-	}
-	if (this->ipv6 != -1 && FD_ISSET(this->ipv6, &rfds))
-	{
-		port = this->port;
-		selected = this->ipv6;
-	}
-	if (this->ipv6_natt != -1 && FD_ISSET(this->ipv6_natt, &rfds))
-	{
-		port = this->natt;
-		selected = this->ipv6_natt;
-	}
-	if (selected)
+	if (selected != -1)
 	{
 		struct msghdr msg;
 		struct cmsghdr *cmsgptr;

src/libradius/radius_socket.c

@@ -129,7 +129,7 @@ METHOD(radius_socket_t, request, radius_message_t*,
 	private_radius_socket_t *this, radius_message_t *request)
 {
 	chunk_t data;
-	int i, *fd;
+	int i, *fd, retransmit = 0;
 	u_int16_t port;
 	rng_t *rng = NULL;
 
@@ -166,64 +166,59 @@ METHOD(radius_socket_t, request, radius_message_t*,
 	for (i = 2; i <= 5; i++)
 	{
 		radius_message_t *response;
-		bool retransmit = FALSE;
-		struct timeval tv;
 		char buf[4096];
-		fd_set fds;
 		int res;
+		struct pollfd pfd = {
+			.fd = *fd,
+			.events = POLLIN,
+		};
 
+		if (retransmit)
+		{
+			DBG1(DBG_CFG, "retransmitting RADIUS %N (attempt %d)",
+				 radius_message_code_names, request->get_code(request),
+				 retransmit);
+		}
 		if (send(*fd, data.ptr, data.len, 0) != data.len)
 		{
 			DBG1(DBG_CFG, "sending RADIUS message failed: %s", strerror(errno));
 			return NULL;
 		}
-		tv.tv_sec = i;
-		tv.tv_usec = 0;
-
-		while (TRUE)
+		res = poll(&pfd, 1, i * 1000);
+		if (res < 0)
 		{
-			FD_ZERO(&fds);
-			FD_SET(*fd, &fds);
-			res = select((*fd) + 1, &fds, NULL, NULL, &tv);
-			/* TODO: updated tv to time not waited. Linux does this for us. */
-			if (res < 0)
-			{	/* failed */
-				DBG1(DBG_CFG, "waiting for RADIUS message failed: %s",
-					 strerror(errno));
-				break;
-			}
-			if (res == 0)
-			{	/* timeout */
-				DBG1(DBG_CFG, "retransmitting RADIUS message");
-				retransmit = TRUE;
-				break;
-			}
-			res = recv(*fd, buf, sizeof(buf), MSG_DONTWAIT);
-			if (res <= 0)
-			{
-				DBG1(DBG_CFG, "receiving RADIUS message failed: %s",
-					 strerror(errno));
-				break;
-			}
-			response = radius_message_parse(chunk_create(buf, res));
-			if (response)
-			{
-				if (response->verify(response,
-							request->get_authenticator(request), this->secret,
-							this->hasher, this->signer))
-				{
-					return response;
-				}
-				response->destroy(response);
-			}
-			DBG1(DBG_CFG, "received invalid RADIUS message, ignored");
+			DBG1(DBG_CFG, "waiting for RADIUS message failed: %s",
+				 strerror(errno));
+			return NULL;
 		}
-		if (!retransmit)
+		if (res == 0)
+		{	/* timeout */
+			retransmit++;
+			continue;
+		}
+		res = recv(*fd, buf, sizeof(buf), MSG_DONTWAIT);
+		if (res <= 0)
 		{
-			break;
+			DBG1(DBG_CFG, "receiving RADIUS message failed: %s",
+				 strerror(errno));
+			return NULL;
 		}
+		response = radius_message_parse(chunk_create(buf, res));
+		if (response)
+		{
+			if (response->verify(response,
+						request->get_authenticator(request), this->secret,
+						this->hasher, this->signer))
+			{
+				return response;
+			}
+			response->destroy(response);
+		}
+		DBG1(DBG_CFG, "received invalid RADIUS message, ignored");
+		return NULL;
 	}
-	DBG1(DBG_CFG, "RADIUS server is not responding");
+	DBG1(DBG_CFG, "RADIUS %N timed out after %d retransmits",
+		 radius_message_code_names, request->get_code(request), retransmit - 1);
 	return NULL;
 }
 

src/libstrongswan/Makefile.am

@@ -101,8 +101,8 @@ utils/utils.h utils/chunk.h utils/debug.h utils/enum.h utils/identification.h \
 utils/lexparser.h utils/optionsfrom.h utils/capabilities.h utils/backtrace.h \
 utils/leak_detective.h utils/printf_hook/printf_hook.h \
 utils/printf_hook/printf_hook_vstr.h utils/printf_hook/printf_hook_builtin.h \
-utils/parser_helper.h utils/test.h utils/integrity_checker.h utils/windows.h \
-utils/process.h utils/utils/strerror.h
+utils/parser_helper.h utils/test.h utils/integrity_checker.h utils/process.h \
+utils/utils/strerror.h utils/compat/windows.h utils/compat/apple.h
 endif
 
 library.lo :	$(top_builddir)/config.status
@@ -133,7 +133,7 @@ if USE_WINDOWS
     threading/windows/rwlock.c \
     threading/windows/spinlock.c \
     threading/windows/semaphore.c \
-    utils/windows.c
+    utils/compat/windows.c
 else
   libstrongswan_la_LIBADD += $(PTHREADLIB)
 endif

src/libstrongswan/library.h

@@ -79,6 +79,9 @@
  *
  * @defgroup utils utils
  * @ingroup libstrongswan
+ *
+ * @defgroup compat compat
+ * @ingroup utils
  */
 
 /**

src/libstrongswan/networking/tun_device.c

@@ -346,40 +346,27 @@ METHOD(tun_device_t, write_packet, bool,
 METHOD(tun_device_t, read_packet, bool,
 	private_tun_device_t *this, chunk_t *packet)
 {
+	chunk_t data;
 	ssize_t len;
-	fd_set set;
 	bool old;
 
-	FD_ZERO(&set);
-	FD_SET(this->tunfd, &set);
+	data = chunk_alloca(get_mtu(this));
 
 	old = thread_cancelability(TRUE);
-	len = select(this->tunfd + 1, &set, NULL, NULL, NULL);
+	len = read(this->tunfd, data.ptr, data.len);
 	thread_cancelability(old);
-
-	if (len < 0)
-	{
-		DBG1(DBG_LIB, "select on TUN device %s failed: %s", this->if_name,
-			 strerror(errno));
-		return FALSE;
-	}
-	/* FIXME: this is quite expensive for lots of small packets, copy from
-	 * local buffer instead? */
-	*packet = chunk_alloc(get_mtu(this));
-	len = read(this->tunfd, packet->ptr, packet->len);
 	if (len < 0)
 	{
 		DBG1(DBG_LIB, "reading from TUN device %s failed: %s", this->if_name,
 			 strerror(errno));
-		chunk_free(packet);
 		return FALSE;
 	}
-	packet->len = len;
+	data.len = len;
 #ifdef __APPLE__
 	/* UTUN's prepend packets with a 32-bit protocol number */
-	packet->len -= sizeof(u_int32_t);
-	memmove(packet->ptr, packet->ptr + sizeof(u_int32_t), packet->len);
+	data = chunk_skip(data, sizeof(u_int32_t));
 #endif
+	*packet = chunk_clone(data);
 	return TRUE;
 }
 

src/libstrongswan/networking/tun_device.h

@@ -31,8 +31,6 @@ typedef struct tun_device_t tun_device_t;
  * Class to create TUN devices
  *
  * Creating such a device requires the CAP_NET_ADMIN capability.
- *
- * @note The implementation is currently very Linux specific
  */
 struct tun_device_t {
 
@@ -42,7 +40,7 @@ struct tun_device_t {
 	 * @note This call blocks until a packet is available. It is a thread
 	 * cancellation point.
 	 *
-	 * @param packet		the packet read from the device
+	 * @param packet		the packet read from the device, allocated
 	 * @return				TRUE if successful
 	 */
 	bool (*read_packet)(tun_device_t *this, chunk_t *packet);

src/libstrongswan/processing/watcher.c

@@ -24,9 +24,6 @@
 
 #include <unistd.h>
 #include <errno.h>
-#ifndef WIN32
-#include <sys/select.h>
-#endif
 #include <fcntl.h>
 
 typedef struct private_watcher_t private_watcher_t;
@@ -121,11 +118,7 @@ static void update(private_watcher_t *this)
 	this->pending = TRUE;
 	if (this->notify[1] != -1)
 	{
-#ifdef WIN32
-		if (send(this->notify[1], buf, sizeof(buf), 0) == -1)
-#else
 		if (write(this->notify[1], buf, sizeof(buf)) == -1)
-#endif
 		{
 			DBG1(DBG_JOB, "notifying watcher failed: %s", strerror(errno));
 		}
@@ -244,6 +237,23 @@ static void activate_all(private_watcher_t *this)
 	this->mutex->unlock(this->mutex);
 }
 
+/**
+ * Find flagged revents in a pollfd set by fd
+ */
+static int find_revents(struct pollfd *pfd, int count, int fd)
+{
+	int i;
+
+	for (i = 0; i < count; i++)
+	{
+		if (pfd[i].fd == fd)
+		{
+			return pfd[i].revents;
+		}
+	}
+	return 0;
+}
+
 /**
  * Dispatching function
  */
@@ -251,17 +261,14 @@ static job_requeue_t watch(private_watcher_t *this)
 {
 	enumerator_t *enumerator;
 	entry_t *entry;
-	fd_set rd, wr, ex;
-	int maxfd = 0, res;
+	struct pollfd *pfd;
+	int count = 0, res;
 	bool rebuild = FALSE;
 
-	FD_ZERO(&rd);
-	FD_ZERO(&wr);
-	FD_ZERO(&ex);
-
 	this->mutex->lock(this->mutex);
 
-	if (this->fds->get_count(this->fds) == 0)
+	count = this->fds->get_count(this->fds);
+	if (count == 0)
 	{
 		this->state = WATCHER_STOPPED;
 		this->mutex->unlock(this->mutex);
@@ -272,33 +279,34 @@ static job_requeue_t watch(private_watcher_t *this)
 		this->state = WATCHER_RUNNING;
 	}
 
-	if (this->notify[0] != -1)
-	{
-		FD_SET(this->notify[0], &rd);
-		maxfd = this->notify[0];
-	}
+	pfd = alloca(sizeof(*pfd) * (count + 1));
+	pfd[0].fd = this->notify[0];
+	pfd[0].events = POLLIN;
+	count = 1;
 
 	enumerator = this->fds->create_enumerator(this->fds);
 	while (enumerator->enumerate(enumerator, &entry))
 	{
 		if (!entry->in_callback)
 		{
+			pfd[count].fd = entry->fd;
+			pfd[count].events = 0;
 			if (entry->events & WATCHER_READ)
 			{
 				DBG3(DBG_JOB, "  watching %d for reading", entry->fd);
-				FD_SET(entry->fd, &rd);
+				pfd[count].events |= POLLIN;
 			}
 			if (entry->events & WATCHER_WRITE)
 			{
 				DBG3(DBG_JOB, "  watching %d for writing", entry->fd);
-				FD_SET(entry->fd, &wr);
+				pfd[count].events |= POLLOUT;
 			}
 			if (entry->events & WATCHER_EXCEPT)
 			{
 				DBG3(DBG_JOB, "  watching %d for exceptions", entry->fd);
-				FD_SET(entry->fd, &ex);
+				pfd[count].events |= POLLERR;
 			}
-			maxfd = max(maxfd, entry->fd);
+			count++;
 		}
 	}
 	enumerator->destroy(enumerator);
@@ -306,6 +314,7 @@ static job_requeue_t watch(private_watcher_t *this)
 
 	while (!rebuild)
 	{
+		int revents;
 		char buf[1];
 		bool old;
 		ssize_t len;
@@ -315,21 +324,17 @@ static job_requeue_t watch(private_watcher_t *this)
 		thread_cleanup_push((void*)activate_all, this);
 		old = thread_cancelability(TRUE);
 
-		res = select(maxfd + 1, &rd, &wr, &ex, NULL);
+		res = poll(pfd, count, -1);
 		thread_cancelability(old);
 		thread_cleanup_pop(FALSE);
 
 		if (res > 0)
 		{
-			if (this->notify[0] != -1 && FD_ISSET(this->notify[0], &rd))
+			if (pfd[0].revents & POLLIN)
 			{
 				while (TRUE)
 				{
-#ifdef WIN32
-					len = recv(this->notify[0], buf, sizeof(buf), 0);
-#else
 					len = read(this->notify[0], buf, sizeof(buf));
-#endif
 					if (len == -1)
 					{
 						if (errno != EAGAIN && errno != EWOULDBLOCK)
@@ -354,17 +359,18 @@ static job_requeue_t watch(private_watcher_t *this)
 					rebuild = TRUE;
 					break;
 				}
-				if (FD_ISSET(entry->fd, &rd) && (entry->events & WATCHER_READ))
+				revents = find_revents(pfd, count, entry->fd);
+				if ((revents & POLLIN) && (entry->events & WATCHER_READ))
 				{
 					DBG2(DBG_JOB, "watched FD %d ready to read", entry->fd);
 					notify(this, entry, WATCHER_READ);
 				}
-				if (FD_ISSET(entry->fd, &wr) && (entry->events & WATCHER_WRITE))
+				if ((revents & POLLOUT) && (entry->events & WATCHER_WRITE))
 				{
 					DBG2(DBG_JOB, "watched FD %d ready to write", entry->fd);
 					notify(this, entry, WATCHER_WRITE);
 				}
-				if (FD_ISSET(entry->fd, &ex) && (entry->events & WATCHER_EXCEPT))
+				if ((revents & POLLERR) && (entry->events & WATCHER_EXCEPT))
 				{
 					DBG2(DBG_JOB, "watched FD %d has exception", entry->fd);
 					notify(this, entry, WATCHER_EXCEPT);

src/libstrongswan/tests/suites/test_threading.c

@@ -1175,6 +1175,191 @@ START_TEST(test_cancel_point)
 }
 END_TEST
 
+static void close_fd_ptr(void *fd)
+{
+	close(*(int*)fd);
+}
+
+static void cancellation_recv()
+{
+	int sv[2];
+	char buf[1];
+
+	ck_assert(socketpair(AF_UNIX, SOCK_STREAM, 0, sv) == 0);
+
+	thread_cleanup_push(close_fd_ptr, &sv[0]);
+	thread_cleanup_push(close_fd_ptr, &sv[1]);
+
+	thread_cancelability(TRUE);
+	while (TRUE)
+	{
+		ck_assert(recv(sv[0], buf, sizeof(buf), 0) == 1);
+	}
+}
+
+static void cancellation_read()
+{
+	int sv[2];
+	char buf[1];
+
+	ck_assert(socketpair(AF_UNIX, SOCK_STREAM, 0, sv) == 0);
+
+	thread_cleanup_push(close_fd_ptr, &sv[0]);
+	thread_cleanup_push(close_fd_ptr, &sv[1]);
+
+	thread_cancelability(TRUE);
+	while (TRUE)
+	{
+		ck_assert(read(sv[0], buf, sizeof(buf)) == 1);
+	}
+}
+
+static void cancellation_select()
+{
+	int sv[2];
+	fd_set set;
+
+	ck_assert(socketpair(AF_UNIX, SOCK_STREAM, 0, sv) == 0);
+
+	thread_cleanup_push(close_fd_ptr, &sv[0]);
+	thread_cleanup_push(close_fd_ptr, &sv[1]);
+
+	FD_ZERO(&set);
+	FD_SET(sv[0], &set);
+	thread_cancelability(TRUE);
+	while (TRUE)
+	{
+		ck_assert(select(sv[0] + 1, &set, NULL, NULL, NULL) == 1);
+	}
+}
+
+static void cancellation_poll()
+{
+	int sv[2];
+	struct pollfd pfd;
+
+	ck_assert(socketpair(AF_UNIX, SOCK_STREAM, 0, sv) == 0);
+
+	thread_cleanup_push(close_fd_ptr, &sv[0]);
+	thread_cleanup_push(close_fd_ptr, &sv[1]);
+
+	pfd.fd = sv[0];
+	pfd.events = POLLIN;
+	thread_cancelability(TRUE);
+	while (TRUE)
+	{
+		ck_assert(poll(&pfd, 1, -1) == 1);
+	}
+}
+
+static void cancellation_accept()
+{
+	host_t *host;
+	int fd, c;
+
+	fd = socket(AF_INET, SOCK_STREAM, 0);
+	ck_assert(fd >= 0);
+	host = host_create_from_string("127.0.0.1", 0);
+	ck_assert_msg(bind(fd, host->get_sockaddr(host),
+					   *host->get_sockaddr_len(host)) == 0, "%m");
+	host->destroy(host);
+	ck_assert(listen(fd, 5) == 0);
+
+	thread_cleanup_push(close_fd_ptr, &fd);
+
+	thread_cancelability(TRUE);
+	while (TRUE)
+	{
+		c = accept(fd, NULL, NULL);
+		ck_assert(c >= 0);
+		close(c);
+	}
+}
+
+static void cancellation_cond()
+{
+	mutex_t *mutex;
+	condvar_t *cond;
+
+	mutex = mutex_create(MUTEX_TYPE_DEFAULT);
+	cond = condvar_create(CONDVAR_TYPE_DEFAULT);
+	mutex->lock(mutex);
+
+	thread_cleanup_push((void*)mutex->destroy, mutex);
+	thread_cleanup_push((void*)cond->destroy, cond);
+
+	thread_cancelability(TRUE);
+	while (TRUE)
+	{
+		cond->wait(cond, mutex);
+	}
+}
+
+static void cancellation_rwcond()
+{
+	rwlock_t *lock;
+	rwlock_condvar_t *cond;
+
+	lock = rwlock_create(RWLOCK_TYPE_DEFAULT);
+	cond = rwlock_condvar_create();
+	lock->write_lock(lock);
+
+	thread_cleanup_push((void*)lock->destroy, lock);
+	thread_cleanup_push((void*)cond->destroy, cond);
+
+	thread_cancelability(TRUE);
+	while (TRUE)
+	{
+		cond->wait(cond, lock);
+	}
+}
+
+static void (*cancellation_points[])() = {
+	cancellation_read,
+	cancellation_recv,
+	cancellation_select,
+	cancellation_poll,
+	cancellation_accept,
+	cancellation_cond,
+	cancellation_rwcond,
+};
+
+static void* run_cancellation_point(void (*fn)())
+{
+	fn();
+	return NULL;
+}
+
+static void* run_cancellation_point_pre(void (*fn)())
+{
+	usleep(5000);
+	fn();
+	return NULL;
+}
+
+START_TEST(test_cancellation_point)
+{
+	thread_t *thread;
+
+	thread = thread_create((void*)run_cancellation_point,
+						   cancellation_points[_i]);
+	usleep(5000);
+	thread->cancel(thread);
+	thread->join(thread);
+}
+END_TEST
+
+START_TEST(test_cancellation_point_pre)
+{
+	thread_t *thread;
+
+	thread = thread_create((void*)run_cancellation_point_pre,
+						   cancellation_points[_i]);
+	thread->cancel(thread);
+	thread->join(thread);
+}
+END_TEST
+
 static void cleanup1(void *data)
 {
 	uintptr_t *value = (uintptr_t*)data;
@@ -1500,6 +1685,13 @@ Suite *threading_suite_create()
 	tcase_add_test(tc, test_cancel_point);
 	suite_add_tcase(s, tc);
 
+	tc = tcase_create("thread cancellation point");
+	tcase_add_loop_test(tc, test_cancellation_point,
+						0, countof(cancellation_points));
+	tcase_add_loop_test(tc, test_cancellation_point_pre,
+						0, countof(cancellation_points));
+	suite_add_tcase(s, tc);
+
 	tc = tcase_create("thread cleanup");
 	tcase_add_test(tc, test_cleanup);
 	tcase_add_test(tc, test_cleanup_exit);

src/libstrongswan/threading/semaphore.h

@@ -21,10 +21,7 @@
 #ifndef THREADING_SEMAPHORE_H_
 #define THREADING_SEMAPHORE_H_
 
-#ifdef __APPLE__
-/* Mach uses a semaphore_create() call, use a different name for ours */
-#define semaphore_create(x) strongswan_semaphore_create(x)
-#endif /* __APPLE__ */
+#include <utils/utils.h>
 
 typedef struct semaphore_t semaphore_t;
 
@@ -87,4 +84,3 @@ struct semaphore_t {
 semaphore_t *semaphore_create(u_int value);
 
 #endif /** THREADING_SEMAPHORE_H_ @} */
-

src/libstrongswan/threading/thread.h

@@ -21,41 +21,10 @@
 #ifndef THREADING_THREAD_H_
 #define THREADING_THREAD_H_
 
+#include <utils/utils.h>
+
 typedef struct thread_t thread_t;
 
-#ifdef __APPLE__
-/* thread_create is a syscall used to create Mach kernel threads and although
- * there are no errors or warnings during compilation or linkage the dynamic
- * linker does not use our implementation, therefore we rename it here
- */
-#define thread_create(main, arg) strongswan_thread_create(main, arg)
-
-/* on Mac OS X 10.5 several system calls we use are no cancellation points.
- * fortunately, select isn't one of them, so we wrap some of the others with
- * calls to select(2).
- */
-#include <sys/socket.h>
-#include <sys/select.h>
-
-#define WRAP_WITH_SELECT(func, socket, ...)\
-	fd_set rfds; FD_ZERO(&rfds); FD_SET(socket, &rfds);\
-	if (select(socket + 1, &rfds, NULL, NULL, NULL) <= 0) { return -1; }\
-	return func(socket, __VA_ARGS__)
-
-static inline int cancellable_accept(int socket, struct sockaddr *address,
-									 socklen_t *address_len)
-{
-	WRAP_WITH_SELECT(accept, socket, address, address_len);
-}
-#define accept cancellable_accept
-static inline int cancellable_recvfrom(int socket, void *buffer, size_t length,
-				int flags, struct sockaddr *address, socklen_t *address_len)
-{
-	WRAP_WITH_SELECT(recvfrom, socket, buffer, length, flags, address, address_len);
-}
-#define recvfrom cancellable_recvfrom
-#endif /* __APPLE__ */
-
 /**
  * Main function of a thread.
  *
@@ -189,32 +158,4 @@ void threads_init();
  */
 void threads_deinit();
 
-
-#ifdef __APPLE__
-
-/*
- * While select() is a cancellation point, it seems that OS X does not honor
- * pending cancellation points when entering the function. We manually test for
- * and honor pending cancellation requests, but this obviously can't prevent
- * some race conditions where the the cancellation happens after the check,
- * but before the select.
- */
-static inline int precancellable_select(int nfds, fd_set *restrict readfds,
-						fd_set *restrict writefds, fd_set *restrict errorfds,
-						struct timeval *restrict timeout)
-{
-	if (thread_cancelability(TRUE))
-	{
-		thread_cancellation_point();
-	}
-	else
-	{
-		thread_cancelability(FALSE);
-	}
-	return select(nfds, readfds, writefds, errorfds, timeout);
-}
-#define select precancellable_select
-
-#endif /* __APPLE__ */
-
 #endif /** THREADING_THREAD_H_ @} */

src/libstrongswan/utils/compat/apple.h

@@ -0,0 +1,109 @@
+/*
+ * Copyright (C) 2014 Martin Willi
+ * Copyright (C) 2014 revosec AG
+ *
+ * This program is free software; you can redistribute it and/or modify it
+ * under the terms of the GNU General Public License as published by the
+ * Free Software Foundation; either version 2 of the License, or (at your
+ * option) any later version.  See <http://www.fsf.org/copyleft/gpl.txt>.
+ *
+ * This program is distributed in the hope that it will be useful, but
+ * WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY
+ * or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License
+ * for more details.
+ */
+
+/**
+ * @defgroup apple apple
+ * @{ @ingroup compat
+ */
+
+#ifndef APPLE_H_
+#define APPLE_H_
+
+#include <poll.h>
+#include <sys/socket.h>
+#include <sys/select.h>
+
+/* thread_create is a syscall used to create Mach kernel threads and although
+ * there are no errors or warnings during compilation or linkage the dynamic
+ * linker does not use our implementation, therefore we rename it here
+ */
+#define thread_create(main, arg) strongswan_thread_create(main, arg)
+
+/* Mach uses a semaphore_create() call, use a different name for ours */
+#define semaphore_create(x) strongswan_semaphore_create(x)
+
+/* on Mac OS X 10.5 several system calls we use are no cancellation points.
+ * fortunately, select isn't one of them, so we wrap some of the others with
+ * calls to select(2).
+ */
+
+#define WRAP_WITH_POLL(func, socket, ...) \
+	struct pollfd pfd = { \
+		.fd = socket, \
+		.events = POLLIN, \
+	}; \
+	if (poll(&pfd, 1, -1) <= 0) \
+	{\
+		return -1; \
+	}\
+	return func(socket, __VA_ARGS__)
+
+static inline int cancellable_accept(int socket, struct sockaddr *address,
+									 socklen_t *address_len)
+{
+	WRAP_WITH_POLL(accept, socket, address, address_len);
+}
+#define accept cancellable_accept
+static inline int cancellable_recvfrom(int socket, void *buffer, size_t length,
+				int flags, struct sockaddr *address, socklen_t *address_len)
+{
+	WRAP_WITH_POLL(recvfrom, socket, buffer, length, flags, address, address_len);
+}
+#define recvfrom cancellable_recvfrom
+
+#include <threading/thread.h>
+
+/*
+ * While select() is a cancellation point, it seems that OS X does not honor
+ * pending cancellation points when entering the function. We manually test for
+ * and honor pending cancellation requests, but this obviously can't prevent
+ * some race conditions where the the cancellation happens after the check,
+ * but before the select.
+ */
+static inline int precancellable_select(int nfds, fd_set *restrict readfds,
+						fd_set *restrict writefds, fd_set *restrict errorfds,
+						struct timeval *restrict timeout)
+{
+	if (thread_cancelability(TRUE))
+	{
+		thread_cancellation_point();
+	}
+	else
+	{
+		thread_cancelability(FALSE);
+	}
+	return select(nfds, readfds, writefds, errorfds, timeout);
+}
+#define select precancellable_select
+
+/*
+ * The same as to select(2) applies to poll(2)
+ */
+static inline int precancellable_poll(struct pollfd fds[], nfds_t nfds,
+									  int timeout)
+{
+	if (thread_cancelability(TRUE))
+	{
+		thread_cancellation_point();
+	}
+	else
+	{
+		thread_cancelability(FALSE);
+	}
+	return poll(fds, nfds, timeout);
+}
+#define poll precancellable_poll
+
+#endif /** APPLE_H_ @}*/

src/libstrongswan/utils/compat/windows.c

@@ -13,7 +13,10 @@
  * for more details.
  */
 
-#include "utils.h"
+/* WSAPoll() */
+#define _WIN32_WINNT 0x0600
+
+#include <utils/utils.h>
 
 #include <errno.h>
 
@@ -639,3 +642,43 @@ ssize_t windows_sendto(int sockfd, const void *buf, size_t len, int flags,
 	}
 	return outlen;
 }
+
+/**
+ * See header
+ */
+#undef read
+ssize_t windows_read(int fd, void *buf, size_t count)
+{
+	ssize_t ret;
+
+	ret = recv(fd, buf, count, 0);
+	if (ret == -1 && WSAGetLastError() == WSAENOTSOCK)
+	{
+		ret = read(fd, buf, count);
+	}
+	return ret;
+}
+
+/**
+ * See header
+ */
+#undef write
+ssize_t windows_write(int fd, void *buf, size_t count)
+{
+	ssize_t ret;
+
+	ret = send(fd, buf, count, 0);
+	if (ret == -1 && WSAGetLastError() == WSAENOTSOCK)
+	{
+		ret = write(fd, buf, count);
+	}
+	return ret;
+}
+
+/**
+ * See header
+ */
+int poll(struct pollfd *fds, int nfds, int timeout)
+{
+	return wserr(WSAPoll(fds, nfds, timeout));
+}

src/libstrongswan/utils/compat/windows.h

@@ -15,7 +15,7 @@
 
 /**
  * @defgroup windows windows
- * @{ @ingroup utils
+ * @{ @ingroup compat
  */
 
 #ifndef WINDOWS_H_
@@ -362,6 +362,49 @@ ssize_t windows_send(int sockfd, const void *buf, size_t len, int flags);
 ssize_t windows_sendto(int sockfd, const void *buf, size_t len, int flags,
 					   const struct sockaddr *dest_addr, socklen_t addrlen);
 
+/**
+ * read(2) working on files and sockets, cancellable on sockets only
+ *
+ * On Windows, there does not seem to be a way how a cancellable read can
+ * be implemented on Low level I/O functions for files, _pipe()s or stdio.
+ */
+#define read windows_read
+ssize_t windows_read(int fd, void *buf, size_t count);
+
+/**
+ * write(2) working on files and sockets
+ */
+#define write windows_write
+ssize_t windows_write(int fd, void *buf, size_t count);
+
+#if _WIN32_WINNT < 0x0600
+/**
+ * Define pollfd and flags on our own if not specified
+ */
+struct pollfd {
+	SOCKET fd;
+	short events;
+	short revents;
+};
+enum {
+	POLLERR =		0x0001,
+	POLLHUP =		0x0002,
+	POLLNVAL =		0x0004,
+	POLLWRNORM =	0x0010,
+	POLLWRBAND =	0x0020,
+	POLLPRI =		0x0400,
+	POLLRDNORM =	0x0100,
+	POLLRDBAND =	0x0200,
+	POLLIN =		POLLRDNORM | POLLRDBAND,
+	POLLOUT =		POLLWRNORM,
+};
+#endif /* _WIN32_WINNT < 0x0600 */
+
+/**
+ * poll(2), implemented using Winsock2 WSAPoll()
+ */
+int poll(struct pollfd *fds, int nfds, int timeout);
+
 /**
  * Declaration missing on older WinGW
  */

src/libstrongswan/utils/utils.h

@@ -29,7 +29,7 @@
 #include <string.h>
 
 #ifdef WIN32
-# include "windows.h"
+# include "compat/windows.h"
 #else
 # define _GNU_SOURCE
 # include <arpa/inet.h>
@@ -37,6 +37,7 @@
 # include <netdb.h>
 # include <netinet/in.h>
 # include <sched.h>
+# include <poll.h>
 #endif
 
 /**
@@ -96,6 +97,9 @@
 
 #include "enum.h"
 #include "utils/strerror.h"
+#ifdef __APPLE__
+# include "compat/apple.h"
+#endif
 
 /**
  * Directory separator character in paths on this platform

src/libtls/tls_socket.c

@@ -291,25 +291,24 @@ METHOD(tls_socket_t, splice, bool,
 	private_tls_socket_t *this, int rfd, int wfd)
 {
 	char buf[PLAIN_BUF_SIZE], *pos;
-	fd_set set;
 	ssize_t in, out;
 	bool old, plain_eof = FALSE, crypto_eof = FALSE;
+	struct pollfd pfd[] = {
+		{ .fd = this->fd,	.events = POLLIN, },
+		{ .fd = rfd,		.events = POLLIN, },
+	};
 
 	while (!plain_eof && !crypto_eof)
 	{
-		FD_ZERO(&set);
-		FD_SET(rfd, &set);
-		FD_SET(this->fd, &set);
-
 		old = thread_cancelability(TRUE);
-		in = select(max(rfd, this->fd) + 1, &set, NULL, NULL, NULL);
+		in = poll(pfd, countof(pfd), -1);
 		thread_cancelability(old);
 		if (in == -1)
 		{
 			DBG1(DBG_TLS, "TLS select error: %s", strerror(errno));
 			return FALSE;
 		}
-		while (!plain_eof && FD_ISSET(this->fd, &set))
+		while (!plain_eof && pfd[0].revents & POLLIN)
 		{
 			in = read_(this, buf, sizeof(buf), FALSE);
 			switch (in)
@@ -342,7 +341,7 @@ METHOD(tls_socket_t, splice, bool,
 			}
 			break;
 		}
-		if (!crypto_eof && FD_ISSET(rfd, &set))
+		if (!crypto_eof && pfd[1].revents & POLLIN)
 		{
 			in = read(rfd, buf, sizeof(buf));
 			switch (in)