Merge branch '1.0' of /home/carl/git/dvdomatic into 1.0

12 files changed, 344 insertions, 50 deletions

diff --git a/src/lib/config.cc b/src/lib/config.cc
index 0c6aed4a8..02feecce8 100644
--- a/src/lib/config.cc
+++ b/src/lib/config.cc
@@ -54,7 +54,7 @@ Config* Config::_instance = 0;
 /** Construct default configuration */
 Config::Config ()
 	: _num_local_encoding_threads (max (2U, boost::thread::hardware_concurrency()))
-	, _server_port (6192)
+	, _server_port_base (6192)
 	, _tms_path (".")
 	, _sound_processor (SoundProcessor::from_id (N_("dolby_cp750")))
 	, _default_still_length (10)
@@ -95,7 +95,12 @@ Config::read ()
 
 	_num_local_encoding_threads = f.number_child<int> ("NumLocalEncodingThreads");
 	_default_directory = f.string_child ("DefaultDirectory");
-	_server_port = f.number_child<int> ("ServerPort");
+
+	boost::optional<int> b = f.optional_number_child<int> ("ServerPort");
+	if (!b) {
+		b = f.optional_number_child<int> ("ServerPortBase");
+	}
+	_server_port_base = b.get ();
 	
 	list<shared_ptr<cxml::Node> > servers = f.node_children ("Server");
 	for (list<shared_ptr<cxml::Node> >::iterator i = servers.begin(); i != servers.end(); ++i) {
@@ -191,7 +196,7 @@ Config::read_old_metadata ()
 		} else if (k == N_("default_directory")) {
 			_default_directory = v;
 		} else if (k == N_("server_port")) {
-			_server_port = atoi (v.c_str ());
+			_server_port_base = atoi (v.c_str ());
 		} else if (k == N_("server")) {
 			optional<ServerDescription> server = ServerDescription::create_from_metadata (v);
 			if (server) {
@@ -287,7 +292,7 @@ Config::write () const
 	root->add_child("Version")->add_child_text ("1");
 	root->add_child("NumLocalEncodingThreads")->add_child_text (lexical_cast<string> (_num_local_encoding_threads));
 	root->add_child("DefaultDirectory")->add_child_text (_default_directory.string ());
-	root->add_child("ServerPort")->add_child_text (lexical_cast<string> (_server_port));
+	root->add_child("ServerPortBase")->add_child_text (lexical_cast<string> (_server_port_base));
 	
 	for (vector<ServerDescription>::const_iterator i = _servers.begin(); i != _servers.end(); ++i) {
 		i->as_xml (root->add_child ("Server"));
diff --git a/src/lib/config.h b/src/lib/config.h
index 7dd5abd17..07b3d6891 100644
--- a/src/lib/config.h
+++ b/src/lib/config.h
@@ -59,9 +59,9 @@ public:
 
 	boost::filesystem::path default_directory_or (boost::filesystem::path a) const;
 
-	/** @return port to use for J2K encoding servers */
-	int server_port () const {
-		return _server_port;
+	/** @return base port number to use for J2K encoding servers */
+	int server_port_base () const {
+		return _server_port_base;
 	}
 
 	/** @return J2K encoding servers to use */
@@ -156,8 +156,8 @@ public:
 	}
 
 	/** @param p New server port */
-	void set_server_port (int p) {
-		_server_port = p;
+	void set_server_port_base (int p) {
+		_server_port_base = p;
 	}
 
 	/** @param s New list of servers */
@@ -270,8 +270,10 @@ private:
 	int _num_local_encoding_threads;
 	/** default directory to put new films in */
 	boost::filesystem::path _default_directory;
-	/** port to use for J2K encoding servers */
-	int _server_port;
+	/** base port number to use for J2K encoding servers;
+	 *  this port and the one above it will be used.
+	 */
+	int _server_port_base;
 
 	/** J2K encoding servers to use */
 	std::vector<ServerDescription> _servers;
diff --git a/src/lib/content.h b/src/lib/content.h
index c066c61e0..626e270bd 100644
--- a/src/lib/content.h
+++ b/src/lib/content.h
@@ -103,7 +103,7 @@ public:
 	}
 	
 	Time end () const {
-		return position() + length_after_trim();
+		return position() + length_after_trim() - 1;
 	}
 
 	Time length_after_trim () const;
diff --git a/src/lib/cross.cc b/src/lib/cross.cc
index 4b0b440c5..41051ee2e 100644
--- a/src/lib/cross.cc
+++ b/src/lib/cross.cc
@@ -35,6 +35,12 @@
 #include <sys/sysctl.h>
 #include <mach-o/dyld.h>
 #endif
+#ifdef DCPOMATIC_POSIX
+#include <sys/types.h>
+#include <ifaddrs.h>
+#include <netinet/in.h>
+#include <arpa/inet.h>
+#endif
 #include "exceptions.h"
 
 using std::pair;
@@ -45,6 +51,7 @@ using std::wstring;
 using std::make_pair;
 using boost::shared_ptr;
 
+/** @param s Number of seconds to sleep for */
 void
 dcpomatic_sleep (int s)
 {
diff --git a/src/lib/dcp_video_frame.cc b/src/lib/dcp_video_frame.cc
index 464cf672f..25abd6f0d 100644
--- a/src/lib/dcp_video_frame.cc
+++ b/src/lib/dcp_video_frame.cc
@@ -256,7 +256,7 @@ DCPVideoFrame::encode_remotely (ServerDescription serv)
 {
 	boost::asio::io_service io_service;
 	boost::asio::ip::tcp::resolver resolver (io_service);
-	boost::asio::ip::tcp::resolver::query query (serv.host_name(), boost::lexical_cast<string> (Config::instance()->server_port ()));
+	boost::asio::ip::tcp::resolver::query query (serv.host_name(), boost::lexical_cast<string> (Config::instance()->server_port_base ()));
 	boost::asio::ip::tcp::resolver::iterator endpoint_iterator = resolver.resolve (query);
 
 	shared_ptr<Socket> socket (new Socket);
diff --git a/src/lib/encoder.cc b/src/lib/encoder.cc
index 924a91439..a442e64d0 100644
--- a/src/lib/encoder.cc
+++ b/src/lib/encoder.cc
@@ -22,6 +22,8 @@
  */
 
 #include <iostream>
+#include <boost/lambda/lambda.hpp>
+#include <libcxml/cxml.h>
 #include "encoder.h"
 #include "util.h"
 #include "film.h"
@@ -44,6 +46,7 @@ using std::min;
 using std::make_pair;
 using boost::shared_ptr;
 using boost::optional;
+using boost::scoped_array;
 
 int const Encoder::_history_size = 25;
 
@@ -53,6 +56,8 @@ Encoder::Encoder (shared_ptr<const Film> f, shared_ptr<Job> j)
 	, _job (j)
 	, _video_frames_out (0)
 	, _terminate (false)
+	, _broadcast_thread (0)
+	, _listen_thread (0)
 {
 	_have_a_real_frame[EYES_BOTH] = false;
 	_have_a_real_frame[EYES_LEFT] = false;
@@ -67,21 +72,41 @@ Encoder::~Encoder ()
 	}
 }
 
+/** Add a worker thread for a each thread on a remote server.  Caller must hold
+ *  a lock on _mutex, or know that one is not currently required to
+ *  safely modify _threads.
+ */
+void
+Encoder::add_worker_threads (ServerDescription d)
+{
+	for (int i = 0; i < d.threads(); ++i) {
+		_threads.push_back (
+			make_pair (d, new boost::thread (boost::bind (&Encoder::encoder_thread, this, d)))
+			);
+	}
+}
+
 void
 Encoder::process_begin ()
 {
 	for (int i = 0; i < Config::instance()->num_local_encoding_threads (); ++i) {
-		_threads.push_back (new boost::thread (boost::bind (&Encoder::encoder_thread, this, optional<ServerDescription> ())));
+		_threads.push_back (
+			make_pair (
+				optional<ServerDescription> (),
+				new boost::thread (boost::bind (&Encoder::encoder_thread, this, optional<ServerDescription> ()))
+				)
+			);
 	}
 
 	vector<ServerDescription> servers = Config::instance()->servers ();
 
 	for (vector<ServerDescription>::iterator i = servers.begin(); i != servers.end(); ++i) {
-		for (int j = 0; j < i->threads (); ++j) {
-			_threads.push_back (new boost::thread (boost::bind (&Encoder::encoder_thread, this, *i)));
-		}
+		add_worker_threads (*i);
 	}
 
+	_broadcast_thread = new boost::thread (boost::bind (&Encoder::broadcast_thread, this));
+	_listen_thread = new boost::thread (boost::bind (&Encoder::listen_thread, this));
+
 	_writer.reset (new Writer (_film, _job));
 }
 
@@ -228,19 +253,30 @@ Encoder::process_audio (shared_ptr<const AudioBuffers> data)
 void
 Encoder::terminate_threads ()
 {
-	boost::mutex::scoped_lock lock (_mutex);
-	_terminate = true;
-	_condition.notify_all ();
-	lock.unlock ();
+	{
+		boost::mutex::scoped_lock lock (_mutex);
+		_terminate = true;
+		_condition.notify_all ();
+	}
 
-	for (list<boost::thread *>::iterator i = _threads.begin(); i != _threads.end(); ++i) {
-		if ((*i)->joinable ()) {
-			(*i)->join ();
+	for (ThreadList::iterator i = _threads.begin(); i != _threads.end(); ++i) {
+		if (i->second->joinable ()) {
+			i->second->join ();
 		}
-		delete *i;
+		delete i->second;
 	}
 
 	_threads.clear ();
+		     
+	if (_broadcast_thread && _broadcast_thread->joinable ()) {
+		_broadcast_thread->join ();
+	}
+	delete _broadcast_thread;
+
+	if (_listen_thread && _listen_thread->joinable ()) {
+		_listen_thread->join ();
+	}
+	delete _listen_thread;
 }
 
 void
@@ -326,3 +362,78 @@ Encoder::encoder_thread (optional<ServerDescription> server)
 		_condition.notify_all ();
 	}
 }
+
+void
+Encoder::broadcast_thread ()
+{
+	boost::system::error_code error;
+	boost::asio::io_service io_service;
+	boost::asio::ip::udp::socket socket (io_service);
+	socket.open (boost::asio::ip::udp::v4(), error);
+	if (error) {
+		throw NetworkError ("failed to set up broadcast socket");
+	}
+
+        socket.set_option (boost::asio::ip::udp::socket::reuse_address (true));
+        socket.set_option (boost::asio::socket_base::broadcast (true));
+	
+        boost::asio::ip::udp::endpoint end_point (boost::asio::ip::address_v4::broadcast(), Config::instance()->server_port_base() + 1);            
+
+	while (1) {
+		boost::mutex::scoped_lock lm (_mutex);
+		if (_terminate) {
+			socket.close (error);
+			return;
+		}
+		
+		string data = DCPOMATIC_HELLO;
+		socket.send_to (boost::asio::buffer (data.c_str(), data.size() + 1), end_point);
+
+		lm.unlock ();
+		dcpomatic_sleep (10);
+	}
+}
+
+void
+Encoder::listen_thread ()
+{
+	while (1) {
+		{
+			/* See if we need to stop */
+			boost::mutex::scoped_lock lm (_mutex);
+			if (_terminate) {
+				return;
+			}
+		}
+
+		shared_ptr<Socket> sock (new Socket (10));
+
+		try {
+			sock->accept (Config::instance()->server_port_base() + 1);
+		} catch (std::exception& e) {
+			continue;
+		}
+
+		uint32_t length = sock->read_uint32 ();
+		scoped_array<char> buffer (new char[length]);
+		sock->read (reinterpret_cast<uint8_t*> (buffer.get()), length);
+		
+		stringstream s (buffer.get());
+		shared_ptr<cxml::Document> xml (new cxml::Document ("ServerAvailable"));
+		xml->read_stream (s);
+
+		{
+			/* See if we already know about this server */
+			string const ip = sock->socket().remote_endpoint().address().to_string ();
+			boost::mutex::scoped_lock lm (_mutex);
+			ThreadList::iterator i = _threads.begin();
+			while (i != _threads.end() && (!i->first || i->first->host_name() != ip)) {
+				++i;
+			}
+
+			if (i == _threads.end ()) {
+				add_worker_threads (ServerDescription (ip, xml->number_child<int> ("Threads")));
+			}
+		}
+	}
+}
diff --git a/src/lib/encoder.h b/src/lib/encoder.h
index ab3f40762..9fcba560f 100644
--- a/src/lib/encoder.h
+++ b/src/lib/encoder.h
@@ -36,6 +36,7 @@ extern "C" {
 #include <libswresample/swresample.h>
 }
 #include "util.h"
+#include "config.h"
 
 class Image;
 class AudioBuffers;
@@ -83,6 +84,9 @@ private:
 	
 	void encoder_thread (boost::optional<ServerDescription>);
 	void terminate_threads ();
+	void broadcast_thread ();
+	void listen_thread ();
+	void add_worker_threads (ServerDescription);
 
 	/** Film that we are encoding */
 	boost::shared_ptr<const Film> _film;
@@ -103,11 +107,16 @@ private:
 	bool _have_a_real_frame[EYES_COUNT];
 	bool _terminate;
 	std::list<boost::shared_ptr<DCPVideoFrame> > _queue;
-	std::list<boost::thread *> _threads;
+	typedef std::list<std::pair<boost::optional<ServerDescription>, boost::thread *> > ThreadList;
+	ThreadList _threads;
 	mutable boost::mutex _mutex;
 	boost::condition _condition;
 
 	boost::shared_ptr<Writer> _writer;
+
+	/** A thread to periodically issue broadcasts to find encoding servers */
+	boost::thread* _broadcast_thread;
+	boost::thread* _listen_thread;
 };
 
 #endif
diff --git a/src/lib/playlist.cc b/src/lib/playlist.cc
index 621b99dd7..c54b24c1c 100644
--- a/src/lib/playlist.cc
+++ b/src/lib/playlist.cc
@@ -82,14 +82,14 @@ Playlist::maybe_sequence_video ()
 	_sequencing_video = true;
 	
 	ContentList cl = _content;
-	Time last = 0;
+	Time next = 0;
 	for (ContentList::iterator i = _content.begin(); i != _content.end(); ++i) {
 		if (!dynamic_pointer_cast<VideoContent> (*i)) {
 			continue;
 		}
 		
-		(*i)->set_position (last);
-		last = (*i)->end ();
+		(*i)->set_position (next);
+		next = (*i)->end() + 1;
 	}
 
 	/* This won't change order, so it does not need a sort */
@@ -260,7 +260,7 @@ Playlist::length () const
 {
 	Time len = 0;
 	for (ContentList::const_iterator i = _content.begin(); i != _content.end(); ++i) {
-		len = max (len, (*i)->end ());
+		len = max (len, (*i)->end() + 1);
 	}
 
 	return len;
diff --git a/src/lib/server.cc b/src/lib/server.cc
index 0212dbbed..bad7ad893 100644
--- a/src/lib/server.cc
+++ b/src/lib/server.cc
@@ -36,6 +36,7 @@
 #include "image.h"
 #include "dcp_video_frame.h"
 #include "config.h"
+#include "cross.h"
 
 #include "i18n.h"
 
@@ -43,6 +44,11 @@ using std::string;
 using std::stringstream;
 using std::multimap;
 using std::vector;
+using std::list;
+using std::cout;
+using std::cerr;
+using std::setprecision;
+using std::fixed;
 using boost::shared_ptr;
 using boost::algorithm::is_any_of;
 using boost::algorithm::split;
@@ -50,6 +56,7 @@ using boost::thread;
 using boost::bind;
 using boost::scoped_array;
 using boost::optional;
+using boost::lexical_cast;
 using libdcp::Size;
 
 ServerDescription::ServerDescription (shared_ptr<const cxml::Node> node)
@@ -73,7 +80,7 @@ optional<ServerDescription>
 ServerDescription::create_from_metadata (string v)
 {
 	vector<string> b;
-	split (b, v, is_any_of (N_(" ")));
+	split (b, v, is_any_of (" "));
 
 	if (b.size() != 2) {
 		return optional<ServerDescription> ();
@@ -82,14 +89,18 @@ ServerDescription::create_from_metadata (string v)
 	return ServerDescription (b[0], atoi (b[1].c_str ()));
 }
 
-Server::Server (shared_ptr<Log> log)
+Server::Server (shared_ptr<Log> log, bool verbose)
 	: _log (log)
+	, _verbose (verbose)
 {
 
 }
 
+/** @param after_read Filled in with gettimeofday() after reading the input from the network.
+ *  @param after_encode Filled in with gettimeofday() after encoding the image.
+ */
 int
-Server::process (shared_ptr<Socket> socket)
+Server::process (shared_ptr<Socket> socket, struct timeval& after_read, struct timeval& after_encode)
 {
 	uint32_t length = socket->read_uint32 ();
 	scoped_array<char> buffer (new char[length]);
@@ -99,6 +110,7 @@ Server::process (shared_ptr<Socket> socket)
 	shared_ptr<cxml::Document> xml (new cxml::Document ("EncodingRequest"));
 	xml->read_stream (s);
 	if (xml->number_child<int> ("Version") != SERVER_LINK_VERSION) {
+		cerr << "Mismatched server/client versions\n";
 		_log->log ("Mismatched server/client versions");
 		return -1;
 	}
@@ -111,8 +123,13 @@ Server::process (shared_ptr<Socket> socket)
 
 	image->read_from_socket (socket);
 	DCPVideoFrame dcp_video_frame (image, xml, _log);
+
+	gettimeofday (&after_read, 0);
 	
 	shared_ptr<EncodedData> encoded = dcp_video_frame.encode_locally ();
+
+	gettimeofday (&after_encode, 0);
+	
 	try {
 		encoded->send (socket);
 	} catch (std::exception& e) {
@@ -142,16 +159,24 @@ Server::worker_thread ()
 		lock.unlock ();
 
 		int frame = -1;
+		string ip;
 
 		struct timeval start;
+		struct timeval after_read;
+		struct timeval after_encode;
+		struct timeval end;
+		
 		gettimeofday (&start, 0);
 		
 		try {
-			frame = process (socket);
+			frame = process (socket, after_read, after_encode);
+			ip = socket->socket().remote_endpoint().address().to_string();
 		} catch (std::exception& e) {
-			_log->log (String::compose (N_("Error: %1"), e.what()));
+			_log->log (String::compose ("Error: %1", e.what()));
 		}
-		
+
+		gettimeofday (&end, 0);
+
 		socket.reset ();
 		
 		lock.lock ();
@@ -159,7 +184,20 @@ Server::worker_thread ()
 		if (frame >= 0) {
 			struct timeval end;
 			gettimeofday (&end, 0);
-			_log->log (String::compose (N_("Encoded frame %1 in %2"), frame, seconds (end) - seconds (start)));
+
+			stringstream message;
+			message.precision (2);
+			message << fixed
+				<< "Encoded frame " << frame << " from " << ip << ": "
+				<< "receive " << (seconds(after_read) - seconds(start)) << "s "
+				<< "encode " << (seconds(after_encode) - seconds(after_read)) << "s "
+				<< "send " << (seconds(end) - seconds(after_encode)) << "s.";
+						   
+			if (_verbose) {
+				cout << message.str() << "\n";
+			}
+
+			_log->log (message.str ());
 		}
 		
 		_worker_condition.notify_all ();
@@ -169,14 +207,24 @@ Server::worker_thread ()
 void
 Server::run (int num_threads)
 {
-	_log->log (String::compose (N_("Server starting with %1 threads"), num_threads));
+	_log->log (String::compose ("Server starting with %1 threads", num_threads));
+	if (_verbose) {
+		cout << "DCP-o-matic server started with " << num_threads << " threads.\n";
+	}
 	
 	for (int i = 0; i < num_threads; ++i) {
 		_worker_threads.push_back (new thread (bind (&Server::worker_thread, this)));
 	}
 
+	_broadcast.thread = new thread (bind (&Server::broadcast_thread, this));
+	
 	boost::asio::io_service io_service;
-	boost::asio::ip::tcp::acceptor acceptor (io_service, boost::asio::ip::tcp::endpoint (boost::asio::ip::tcp::v4(), Config::instance()->server_port ()));
+
+	boost::asio::ip::tcp::acceptor acceptor (
+		io_service,
+		boost::asio::ip::tcp::endpoint (boost::asio::ip::tcp::v4(), Config::instance()->server_port_base ())
+		);
+	
 	while (1) {
 		shared_ptr<Socket> socket (new Socket);
 		acceptor.accept (socket->socket ());
@@ -192,3 +240,53 @@ Server::run (int num_threads)
 		_worker_condition.notify_all ();
 	}
 }
+
+void
+Server::broadcast_thread ()
+{
+	boost::asio::io_service io_service;
+
+	boost::asio::ip::address address = boost::asio::ip::address_v4::any ();
+	boost::asio::ip::udp::endpoint listen_endpoint (address, Config::instance()->server_port_base() + 1);
+
+	_broadcast.socket = new boost::asio::ip::udp::socket (io_service);
+	_broadcast.socket->open (listen_endpoint.protocol ());
+	_broadcast.socket->bind (listen_endpoint);
+
+	_broadcast.socket->async_receive_from (
+		boost::asio::buffer (_broadcast.buffer, sizeof (_broadcast.buffer)),
+		_broadcast.send_endpoint,
+		boost::bind (&Server::broadcast_received, this)
+		);
+
+	io_service.run ();
+}
+
+void
+Server::broadcast_received ()
+{
+	_broadcast.buffer[sizeof(_broadcast.buffer) - 1] = '\0';
+
+	if (strcmp (_broadcast.buffer, DCPOMATIC_HELLO) == 0) {
+		/* Reply to the client saying what we can do */
+		xmlpp::Document doc;
+		xmlpp::Element* root = doc.create_root_node ("ServerAvailable");
+		root->add_child("Threads")->add_child_text (lexical_cast<string> (_worker_threads.size ()));
+		stringstream xml;
+		doc.write_to_stream (xml, "UTF-8");
+
+		shared_ptr<Socket> socket (new Socket);
+		try {
+			socket->connect (boost::asio::ip::tcp::endpoint (_broadcast.send_endpoint.address(), Config::instance()->server_port_base() + 1));
+			socket->write (xml.str().length() + 1);
+			socket->write ((uint8_t *) xml.str().c_str(), xml.str().length() + 1);
+		} catch (...) {
+
+		}
+	}
+		
+	_broadcast.socket->async_receive_from (
+		boost::asio::buffer (_broadcast.buffer, sizeof (_broadcast.buffer)),
+		_broadcast.send_endpoint, boost::bind (&Server::broadcast_received, this)
+		);
+}
diff --git a/src/lib/server.h b/src/lib/server.h
index 77b51d079..68de3c2f0 100644
--- a/src/lib/server.h
+++ b/src/lib/server.h
@@ -94,19 +94,36 @@ private:
 class Server : public boost::noncopyable
 {
 public:
-	Server (boost::shared_ptr<Log> log);
+	Server (boost::shared_ptr<Log> log, bool verbose);
 
 	void run (int num_threads);
 
 private:
 	void worker_thread ();
-	int process (boost::shared_ptr<Socket> socket);
+	int process (boost::shared_ptr<Socket> socket, struct timeval &, struct timeval &);
+	void broadcast_thread ();
+	void broadcast_received ();
 
 	std::vector<boost::thread *> _worker_threads;
 	std::list<boost::shared_ptr<Socket> > _queue;
 	boost::mutex _worker_mutex;
 	boost::condition _worker_condition;
 	boost::shared_ptr<Log> _log;
+	bool _verbose;
+
+	struct Broadcast {
+
+		Broadcast ()
+			: thread (0)
+			, socket (0)
+		{}
+		
+		boost::thread* thread;
+		boost::asio::ip::udp::socket* socket;
+		char buffer[64];
+		boost::asio::ip::udp::endpoint send_endpoint;
+		
+	} _broadcast;
 };
 
 #endif
diff --git a/src/lib/util.cc b/src/lib/util.cc
index 15efcc099..e2ce94dd9 100644
--- a/src/lib/util.cc
+++ b/src/lib/util.cc
@@ -90,6 +90,7 @@ using std::istream;
 using std::numeric_limits;
 using std::pair;
 using std::ofstream;
+using std::cout;
 using boost::shared_ptr;
 using boost::thread;
 using boost::lexical_cast;
@@ -518,17 +519,27 @@ dcp_audio_frame_rate (int fs)
 Socket::Socket (int timeout)
 	: _deadline (_io_service)
 	, _socket (_io_service)
+	, _acceptor (0)
 	, _timeout (timeout)
 {
 	_deadline.expires_at (boost::posix_time::pos_infin);
 	check ();
 }
 
+Socket::~Socket ()
+{
+	delete _acceptor;
+}
+
 void
 Socket::check ()
 {
 	if (_deadline.expires_at() <= boost::asio::deadline_timer::traits_type::now ()) {
-		_socket.close ();
+		if (_acceptor) {
+			_acceptor->cancel ();
+		} else {
+			_socket.close ();
+		}
 		_deadline.expires_at (boost::posix_time::pos_infin);
 	}
 
@@ -539,7 +550,7 @@ Socket::check ()
  *  @param endpoint End-point to connect to.
  */
 void
-Socket::connect (boost::asio::ip::basic_resolver_entry<boost::asio::ip::tcp> const & endpoint)
+Socket::connect (boost::asio::ip::tcp::endpoint endpoint)
 {
 	_deadline.expires_from_now (boost::posix_time::seconds (_timeout));
 	boost::system::error_code ec = boost::asio::error::would_block;
@@ -548,11 +559,35 @@ Socket::connect (boost::asio::ip::basic_resolver_entry<boost::asio::ip::tcp> con
 		_io_service.run_one();
 	} while (ec == boost::asio::error::would_block);
 
-	if (ec || !_socket.is_open ()) {
+	if (ec) {
+		throw NetworkError (ec.message ());
+	}
+
+	if (!_socket.is_open ()) {
 		throw NetworkError (_("connect timed out"));
 	}
 }
 
+void
+Socket::accept (int port)
+{
+	_acceptor = new boost::asio::ip::tcp::acceptor (_io_service, boost::asio::ip::tcp::endpoint (boost::asio::ip::tcp::v4(), port));
+	
+	_deadline.expires_from_now (boost::posix_time::seconds (_timeout));
+	boost::system::error_code ec = boost::asio::error::would_block;
+	_acceptor->async_accept (_socket, boost::lambda::var(ec) = boost::lambda::_1);
+	do {
+		_io_service.run_one ();
+	} while (ec == boost::asio::error::would_block );
+
+	delete _acceptor;
+	_acceptor = 0;
+	
+	if (ec) {
+		throw NetworkError (ec.message ());
+	}
+}
+
 /** Blocking write.
  *  @param data Buffer to write.
  *  @param size Number of bytes to write.
@@ -763,12 +798,17 @@ FrameRateConversion::FrameRateConversion (float source, int dcp)
 	, repeat (1)
 	, change_speed (false)
 {
-	if (source > (dcp * 2)) {
+	if (fabs (source / 2.0 - dcp) < fabs (source - dcp)) {
+		/* The difference between source and DCP frame rate will be lower
+		   (i.e. better) if we skip.
+		*/
 		skip = true;
-	}
-
-	if (source < dcp) {
-		repeat = floor (dcp / source);
+	} else if (fabs (source * 2 - dcp) < fabs (source - dcp)) {
+		/* The difference between source and DCP frame rate would be better
+		   if we repeated each frame once; it may be better still if we
+		   repeated more than once.  Work out the required repeat.
+		*/
+		repeat = round (dcp / source);
 	}
 
 	change_speed = !about_equal (source * factor(), dcp);
diff --git a/src/lib/util.h b/src/lib/util.h
index 70cb3bb0c..5e568cc27 100644
--- a/src/lib/util.h
+++ b/src/lib/util.h
@@ -51,6 +51,8 @@ extern "C" {
 /** The maximum number of audio channels that we can cope with */
 #define MAX_AUDIO_CHANNELS 6
 
+#define DCPOMATIC_HELLO "Boys, you gotta learn not to talk to nuns that way"
+
 namespace libdcp {
 	class Signer;
 }
@@ -131,13 +133,15 @@ class Socket
 {
 public:
 	Socket (int timeout = 30);
+	~Socket ();
 
 	/** @return Our underlying socket */
 	boost::asio::ip::tcp::socket& socket () {
 		return _socket;
 	}
 
-	void connect (boost::asio::ip::basic_resolver_entry<boost::asio::ip::tcp> const & endpoint);
+	void connect (boost::asio::ip::tcp::endpoint);
+	void accept (int);
 
 	void write (uint32_t n);
 	void write (uint8_t const * data, int size);
@@ -153,6 +157,7 @@ private:
 	boost::asio::io_service _io_service;
 	boost::asio::deadline_timer _deadline;
 	boost::asio::ip::tcp::socket _socket;
+	boost::asio::ip::tcp::acceptor* _acceptor;
 	int _timeout;
 };