Xthread: blocking read + non-blocking write mode.

Needed for switching the butler to use Crossthreads.

libs/pbd/crossthread.posix.cc

@@ -1,3 +1,5 @@
+#include <poll.h>
+
 CrossThreadChannel::CrossThreadChannel (bool non_blocking)
         : receive_channel (0)
 {
@@ -61,8 +63,37 @@ CrossThreadChannel::deliver (char msg)
         return ::write (fds[1], &msg, 1);
 }
 
-int 
-CrossThreadChannel::receive (char& msg)
+bool
+CrossThreadChannel::poll_for_request()
 {
+	struct pollfd pfd[1];
+	pfd[0].fd = fds[0];
+	pfd[0].events = POLLIN|POLLERR|POLLHUP;
+	while(true) {
+		if (poll (pfd, 1, -1) < 0) {
+			if (errno == EINTR) {
+				continue;
+			}
+			break;
+		}
+		if (pfd[0].revents & ~POLLIN) {
+			break;
+		}
+
+		if (pfd[0].revents & POLLIN) {
+			return true;
+		}
+	}
+	return false;
+}
+
+int 
+CrossThreadChannel::receive (char& msg, bool wait)
+{
+	if (wait) {
+		if (!poll_for_request ()) {
+			return -1;
+		}
+	}
         return ::read (fds[0], &msg, 1);
 }

libs/pbd/crossthread.win.cc

@@ -152,12 +152,39 @@ CrossThreadChannel::deliver (char msg)
 	return status;
 }
 
+bool
+CrossThreadChannel::poll_for_request()
+{
+	// windows before Vista has no poll
+	while(true) {
+		fd_set rfds;
+		FD_ZERO(&rfds);
+		FD_SET(receive_socket, &rfds);
+		if ((select(receive_socket+1, &rfds, NULL, NULL, NULL)) < 0) {
+			if (errno == EINTR) {
+				continue;
+			}
+			break;
+		}
+		if(FD_ISSET(receive_socket, &rfds)) {
+			return true;
+		}
+	}
+	return false;
+}
+
 int 
-CrossThreadChannel::receive (char& msg)
+CrossThreadChannel::receive (char& msg, bool wait)
 {
 	gsize read = 0;
 	GError *g_error = 0;
 
+	if (wait) {
+		if (!poll_for_request ()) {
+			return -1;
+		}
+	}
+	
 	// fetch the message from the channel.
 	GIOStatus g_status = g_io_channel_read_chars (receive_channel, &msg, sizeof(msg), &read, &g_error);
 

libs/pbd/pbd/crossthread.h

@@ -67,8 +67,11 @@ class LIBPBD_API CrossThreadChannel {
 	/** if using ::deliver() to wakeup the listening thread, then
 	 * the listener should call ::receive() to fetch the message
 	 * type from the channel.
+	 *
+	 * wait = true only make sense for non_blocking channels,
+	 * it polls for data to become available.
 	 */
-     int receive (char& msg);
+	int receive (char& msg, bool wait = false);
 
 	/** empty the channel of all requests.
 	 * Typically this is done as soon as input 
@@ -89,6 +92,8 @@ private:
 	GSource*    receive_source;
 	sigc::slot<bool,Glib::IOCondition> receive_slot;
 
+	bool poll_for_request();
+
 #ifndef PLATFORM_WINDOWS
 	int fds[2]; // current implementation uses a pipe/fifo
 #else