[Summary] Restored Asynchronous Midi input processing on Windows by providing Windows specific implementation of CrossThreadChannel synchronization class which serves Event Loops

[Details] Restored SceneIn and MMC ports processing
Fixed crash during exit on both platforms
[Reviewed by] Paul Davis

libs/ardour/ardour/async_midi_port.h

@@ -62,18 +62,12 @@ class LIBARDOUR_API AsyncMIDIPort : public ARDOUR::MidiPort, public MIDI::Port {
 
 	/* clears async request communication channel */
 	void clear () {
-#ifndef PLATFORM_WINDOWS
 		xthread.drain ();
-#endif
 	}
 	/* Not selectable; use ios() */
 	int selectable() const { return -1; }
 	Glib::RefPtr<Glib::IOSource> ios() {
-#ifndef PLATFORM_WINDOWS
 		return xthread.ios();
-#else
-		Glib::RefPtr<Glib::IOSource>();
-#endif
 	}
 	void set_timer (boost::function<framecnt_t (void)>&);
 
@@ -87,11 +81,10 @@ class LIBARDOUR_API AsyncMIDIPort : public ARDOUR::MidiPort, public MIDI::Port {
 	bool                    have_timer;
 	boost::function<framecnt_t (void)> timer;
 	RingBuffer< Evoral::Event<double> > output_fifo;
-        Evoral::EventRingBuffer<MIDI::timestamp_t> input_fifo;
-        Glib::Threads::Mutex output_fifo_lock;
-#ifndef PLATFORM_WINDOWS
+    Evoral::EventRingBuffer<MIDI::timestamp_t> input_fifo;
+    Glib::Threads::Mutex output_fifo_lock;
 	CrossThreadChannel xthread;
-#endif
+
 
 	int create_port ();
 
@@ -105,7 +98,7 @@ class LIBARDOUR_API AsyncMIDIPort : public ARDOUR::MidiPort, public MIDI::Port {
 	void make_connections ();
 	void init (std::string const &, Flags);
 
-        void flush_output_fifo (pframes_t);
+    void flush_output_fifo (pframes_t);
 
 	static pthread_t _process_thread;
 };

libs/ardour/async_midi_port.cc

@@ -53,9 +53,7 @@ AsyncMIDIPort::AsyncMIDIPort (string const & name, PortFlags flags)
 	, have_timer (false)
 	, output_fifo (512)
 	, input_fifo (1024)
-#ifndef PLATFORM_WINDOWS
 	, xthread (true)
-#endif
 {
 }
 
@@ -136,11 +134,9 @@ AsyncMIDIPort::cycle_start (MIDI::pframes_t nframes)
 			input_fifo.write (when, (Evoral::EventType) 0, (*b).size(), (*b).buffer());
 		}
 
-#ifndef PLATFORM_WINDOWS
 		if (!mb.empty()) {
 			xthread.wakeup ();
 		}
-#endif
 	}
 }
 

libs/pbd/crossthread.cc

@@ -17,6 +17,8 @@
 
 */
 
+#ifndef PLATFORM_WINDOWS
+
 #include <cstdlib>
 #include <cerrno>
 #include <cstring>
@@ -81,26 +83,19 @@ RefPtr<IOSource>
 CrossThreadChannel::ios () 
 {
 	if (!_ios) {
-		_ios = new RefPtr<IOSource> (IOSource::create (fds[0], IOCondition(IO_IN|IO_PRI|IO_ERR|IO_HUP|IO_NVAL)));
+		_ios.reset (IOSource::create (fds[0], IOCondition(IO_IN|IO_PRI|IO_ERR|IO_HUP|IO_NVAL)));
 	}
-	return *_ios;
+	return _ios;
 }
 
 void
 CrossThreadChannel::drop_ios ()
 {
-	delete _ios;
-	_ios = 0;
+	_ios.clear ();
 }
 
 void
 CrossThreadChannel::drain ()
-{
-	drain (fds[0]);
-}
-
-void
-CrossThreadChannel::drain (int fd)
 {
 	/* drain selectable fd */
 	char buf[64];
@@ -118,3 +113,5 @@ CrossThreadChannel::receive (char& msg)
 {
         return ::read (fds[0], &msg, 1);
 }
+
+#endif

libs/pbd/crossthread.win.cc

@@ -0,0 +1,204 @@
+/*
+    Copyright (C) 2009 Paul Davis 
+
+    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.
+
+    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.
+
+    You should have received a copy of the GNU General Public License
+    along with this program; if not, write to the Free Software
+    Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
+
+*/
+
+#ifdef PLATFORM_WINDOWS
+
+#include <cstdlib>
+#include <cerrno>
+#include <cstring>
+#include <fcntl.h>
+#include <unistd.h>
+
+#include <csignal> // or signal.h if C code
+
+#include <winsock2.h>
+#include <ws2tcpip.h>
+
+#include "pbd/error.h"
+#include "pbd/crossthread.h"
+
+using namespace std;
+using namespace PBD;
+using namespace Glib;
+
+CrossThreadChannel::CrossThreadChannel (bool non_blocking)
+	: _ios()
+	, _send_socket()
+	, _receive_socket()
+	, _p_recv_channel(0)
+{
+	WSADATA	wsaData;
+
+    if(WSAStartup(MAKEWORD(1,1),&wsaData) != 0)
+    {
+		std::cerr << "CrossThreadChannel::CrossThreadChannel() Winsock initialization failed with error: " << WSAGetLastError() << std::endl;
+        return;
+    }
+
+	struct sockaddr_in send_address;
+
+	// Create Send Socket
+    _send_socket = socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP);
+    send_address.sin_family = AF_INET;
+    send_address.sin_addr.s_addr = INADDR_ANY;
+    send_address.sin_port = htons(0);
+    int status = bind(_send_socket, (SOCKADDR*)&send_address,   
+                  sizeof(send_address));
+
+	if (status != 0) {
+		std::cerr << "CrossThreadChannel::CrossThreadChannel() Send socket binding failed with error: " << WSAGetLastError() << std::endl;
+        return;
+	}
+
+	// make the socket non-blockable if required
+	u_long mode = (u_long)non_blocking;
+	int otp_result = 0;
+	
+	otp_result = ioctlsocket(_send_socket, FIONBIO, &mode);
+	if (otp_result != NO_ERROR) {
+		std::cerr << "CrossThreadChannel::CrossThreadChannel() Send socket cannot be set to non blocking mode with error: " << WSAGetLastError() << std::endl;
+	}
+
+	// Create Receive Socket, this socket will be set to unblockable mode by IO channel
+    _receive_socket = socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP);
+    _recv_address.sin_family = AF_INET;
+    _recv_address.sin_addr.s_addr = inet_addr("127.0.0.1");
+    _recv_address.sin_port = htons(0);
+    status = bind(_receive_socket, (SOCKADDR*)&_recv_address, 
+                    sizeof(_recv_address));
+    
+	if (status != 0) {
+		std::cerr << "CrossThreadChannel::CrossThreadChannel() Receive socket binding failed with error: " << WSAGetLastError() << std::endl;
+        return;
+	}
+
+	// make the socket non-blockable if required
+	mode = (u_long)non_blocking;
+	otp_result = 0;
+	
+	otp_result = ioctlsocket(_receive_socket, FIONBIO, &mode);
+	if (otp_result != NO_ERROR) {
+		std::cerr << "CrossThreadChannel::CrossThreadChannel() Receive socket cannot be set to non blocking mode with error: " << WSAGetLastError() << std::endl;
+	}
+
+	// get assigned port number for Receive Socket
+	int recv_addr_len = sizeof(_recv_address);
+    status = getsockname(_receive_socket, (SOCKADDR*)&_recv_address, &recv_addr_len);
+    
+	if (status != 0) {
+		std::cerr << "CrossThreadChannel::CrossThreadChannel() Setting receive socket address to local failed with error: " << WSAGetLastError() << std::endl;
+        return;
+	}
+	
+	// construct IOChannel
+	_p_recv_channel = g_io_channel_win32_new_socket((gint)_receive_socket);
+	
+	int flags = G_IO_FLAG_APPEND;
+	if (non_blocking) {
+		flags |= G_IO_FLAG_NONBLOCK;
+	}
+	
+	GIOStatus g_status = g_io_channel_set_flags(_p_recv_channel, (GIOFlags)flags,
+                           NULL);
+
+	if (G_IO_STATUS_NORMAL != g_status ) {
+		std::cerr << "CrossThreadChannel::CrossThreadChannel() Cannot set IOChannel flags " << std::endl;
+        return;
+	}
+}
+
+CrossThreadChannel::~CrossThreadChannel ()
+{
+	/* glibmm hack */
+	drop_ios ();
+	delete _p_recv_channel;
+	closesocket(_send_socket);
+	closesocket(_receive_socket);
+	WSACleanup();
+}
+
+void
+CrossThreadChannel::wakeup ()
+{
+	char c = 0;
+
+	// write one byte to wake up a thread which is listening our IOS	
+	sendto(_send_socket, &c, sizeof(c), 0, (SOCKADDR*)&_recv_address, sizeof(_recv_address) );
+}
+
+RefPtr<IOSource>
+CrossThreadChannel::ios () 
+{
+	if (!_ios) {
+		_ios = IOSource::create (wrap(_p_recv_channel), IOCondition(IO_IN|IO_PRI|IO_ERR|IO_HUP|IO_NVAL));
+	}
+
+	return _ios;
+}
+
+void
+CrossThreadChannel::drop_ios ()
+{
+	_ios.reset ();
+}
+
+void
+CrossThreadChannel::drain ()
+{
+	/* flush the buffer - empty the channel from all requests */
+	GError *g_error = 0;
+	gchar* buffer;
+	gsize read = 0;
+
+	g_io_channel_read_to_end (_p_recv_channel, &buffer, &read, &g_error);
+	g_free(buffer);
+}
+
+
+int
+CrossThreadChannel::deliver (char msg)
+{
+
+	// write one particular byte to wake up the thread which is listening our IOS
+	int status = sendto(_send_socket, &msg, sizeof(msg), 0, (SOCKADDR*)&_recv_address, sizeof(_recv_address) );
+
+	if (SOCKET_ERROR  == status) {
+		return -1;
+	}
+
+	return status;
+}
+
+int 
+CrossThreadChannel::receive (char& msg)
+{
+	gsize read = 0;
+	GError *g_error = 0;
+	
+	// fetch the message from the channel.
+	GIOStatus g_status = g_io_channel_read_chars (_p_recv_channel, &msg, sizeof(msg), &read, &g_error);
+
+	if (G_IO_STATUS_NORMAL != g_status) {
+		read = -1;
+	}
+
+	return read;
+}
+
+#endif

libs/pbd/pbd/crossthread.h

@@ -28,6 +28,11 @@
 
 #include "pbd/libpbd_visibility.h"
 
+#ifdef PLATFORM_WINDOWS
+#include <Windows.h>
+#endif // PLATFORM_WINDOWS
+
+
 /** A simple abstraction of a mechanism of signalling one thread from another.
  * The signaller calls ::wakeup() to tell the signalled thread to check for
  * work to be done. 
@@ -57,13 +62,13 @@ class LIBPBD_API CrossThreadChannel {
 	 * because there is no way to know which byte value will be used
 	 * for ::wakeup()
 	 */
-        int deliver (char msg);
+     int deliver (char msg);
 
 	/** if using ::deliver() to wakeup the listening thread, then
 	 * the listener should call ::receive() to fetch the message
 	 * type from the channel.
 	 */
-        int receive (char& msg);
+     int receive (char& msg);
 
 	/** empty the channel of all requests.
 	 * Typically this is done as soon as input 
@@ -73,14 +78,6 @@ class LIBPBD_API CrossThreadChannel {
 	 * in the channel will not be important.
 	 */
 	void drain ();
-	static void drain (int fd);
-
-	/** File descriptor that can be used with poll/select to
-	 * detect when wakeup() has been called on this channel.
-	 * It be marked as readable/input-ready when this condition
-	 * is true. It has already been marked non-blocking.
-	 */
-	int selectable() const { return fds[0]; }
 
 	/* glibmm 2.22 and earlier has a terrifying bug that will
 	   cause crashes whenever a Source is removed from
@@ -90,16 +87,20 @@ class LIBPBD_API CrossThreadChannel {
 	   but in the meantime, we need a hack to get around the issue.
 	*/
 	Glib::RefPtr<Glib::IOSource> ios();
-	void drop_ios ();
-
-	/** returns true if the CrossThreadChannel was
-	 * correctly constructed.
-	 */
-	bool ok() const { return fds[0] >= 0 && fds[1] >= 0; }
 
   private:
-	Glib::RefPtr<Glib::IOSource>* _ios; // lazily constructed
+	void drop_ios ();
+	Glib::RefPtr<Glib::IOSource> _ios; // lazily constructed
+
+#ifndef PLATFORM_WINDOWS
 	int fds[2]; // current implementation uses a pipe/fifo
+#else
+	SOCKET _send_socket;
+	SOCKET _receive_socket;
+	GIOChannel* _p_recv_channel;
+	struct sockaddr_in _recv_address;
+#endif
+
 };
 
 #endif /* __pbd__crossthread_h__ */

libs/pbd/wscript

@@ -133,6 +133,8 @@ def build(bld):
 
     if bld.env['build_target'] != 'mingw':
         obj.source += [ 'crossthread.cc' ]
+    else:
+        obj.source += [ 'crossthread.win.cc' ]
 
     obj.export_includes = ['.']
     obj.includes     = ['.']