Tidy up nanomsg class API; add unmounting for Linux.

9 files changed, 124 insertions, 70 deletions

diff --git a/src/lib/copy_to_drive_job.cc b/src/lib/copy_to_drive_job.cc
index fe35daba6..fcac7e41b 100644
--- a/src/lib/copy_to_drive_job.cc
+++ b/src/lib/copy_to_drive_job.cc
@@ -37,6 +37,7 @@ using std::string;
 using std::cout;
 using std::min;
 using boost::shared_ptr;
+using boost::optional;
 using dcp::raw_convert;
 
 CopyToDriveJob::CopyToDriveJob (boost::filesystem::path dcp, Drive drive, Nanomsg& nanomsg)
@@ -63,30 +64,36 @@ CopyToDriveJob::json_name () const
 void
 CopyToDriveJob::run ()
 {
-	if (!_nanomsg.nonblocking_send(String::compose(DISK_WRITER_WRITE "\n%1\n%2\n", _dcp.string(), _drive.internal_name()))) {
-		throw CopyError ("Could not communicate with writer process", 0);
+	if (!_nanomsg.send(String::compose(DISK_WRITER_WRITE "\n%1\n%2\n", _dcp.string(), _drive.internal_name()), 2000)) {
+		throw CommunicationFailedError ();
 	}
 
 	bool formatting = false;
 	while (true) {
-		string s = _nanomsg.blocking_get ();
-		if (s == DISK_WRITER_OK) {
+		optional<string> s = _nanomsg.receive (10000);
+		if (!s) {
+			continue;
+		}
+		if (*s == DISK_WRITER_OK) {
 			set_state (FINISHED_OK);
 			return;
-		} else if (s == DISK_WRITER_ERROR) {
-			string const m = _nanomsg.blocking_get ();
-			string const n = _nanomsg.blocking_get ();
-			throw CopyError (m, raw_convert<int>(n));
-		} else if (s == DISK_WRITER_FORMATTING) {
+		} else if (*s == DISK_WRITER_ERROR) {
+			optional<string> const m = _nanomsg.receive (500);
+			optional<string> const n = _nanomsg.receive (500);
+			throw CopyError (m.get_value_or("Unknown"), raw_convert<int>(n.get_value_or("0")));
+		} else if (*s == DISK_WRITER_FORMATTING) {
 			sub ("Formatting drive");
 			set_progress_unknown ();
 			formatting = true;
-		} else if (s == DISK_WRITER_PROGRESS) {
+		} else if (*s == DISK_WRITER_PROGRESS) {
 			if (formatting) {
 				sub ("Copying DCP");
 				formatting = false;
 			}
-			set_progress (raw_convert<float>(_nanomsg.blocking_get()));
+			optional<string> progress = _nanomsg.receive (500);
+			if (progress) {
+				set_progress (raw_convert<float>(*progress));
+			}
 		}
 	}
 }
diff --git a/src/lib/cross.h b/src/lib/cross.h
index 6d3c2bd2e..7bbb394a1 100644
--- a/src/lib/cross.h
+++ b/src/lib/cross.h
@@ -124,5 +124,9 @@ private:
 };
 
 std::vector<Drive> get_drives ();
+/** Unmount any mounted partitions on a drive.
+ *  @return true on success, false on failure.
+ */
+bool unmount_drive (std::string drive);
 
 #endif
diff --git a/src/lib/cross_common.cc b/src/lib/cross_common.cc
index b3a39402a..10ffb0677 100644
--- a/src/lib/cross_common.cc
+++ b/src/lib/cross_common.cc
@@ -46,6 +46,6 @@ Drive::description () const
 		name = _("Unknown");
 	}
 
-	return String::compose("%1 (%2 GB) [%3]", name, gb, _internal_name);
+	return String::compose(_("%1 (%2 GB) [%3]"), name, gb, _internal_name);
 }
 
diff --git a/src/lib/cross_linux.cc b/src/lib/cross_linux.cc
index 8c81c5008..14a7e5b47 100644
--- a/src/lib/cross_linux.cc
+++ b/src/lib/cross_linux.cc
@@ -24,6 +24,7 @@
 #include "dcpomatic_log.h"
 #include "config.h"
 #include "exceptions.h"
+#include "dcpomatic_log.h"
 #include <dcp/raw_convert.h>
 #include <glib.h>
 extern "C" {
@@ -37,6 +38,7 @@ extern "C" {
 #include <unistd.h>
 #include <mntent.h>
 #include <sys/types.h>
+#include <sys/mount.h>
 #include <ifaddrs.h>
 #include <netinet/in.h>
 #include <arpa/inet.h>
@@ -256,25 +258,34 @@ running_32_on_64 ()
 	return false;
 }
 
-vector<Drive>
-get_drives ()
+vector<pair<string, string> >
+get_mounts (string prefix)
 {
-	vector<Drive> drives;
+	vector<pair<string, string> > mounts;
 
-	using namespace boost::filesystem;
-	list<string> mounted_devices;
 	std::ifstream f("/proc/mounts");
 	string line;
 	while (f.good()) {
 		getline(f, line);
 		vector<string> bits;
 		boost::algorithm::split (bits, line, boost::is_any_of(" "));
-		if (bits.size() > 0 && boost::algorithm::starts_with(bits[0], "/dev/")) {
-			mounted_devices.push_back(bits[0]);
-			LOG_DISK("Mounted device %1", bits[0]);
+		if (bits.size() > 1 && boost::algorithm::starts_with(bits[0], prefix)) {
+			mounts.push_back(make_pair(bits[0], bits[1]));
+			LOG_DISK("Found mounted device %1 from prefix %2", bits[0], prefix);
 		}
 	}
 
+	return mounts;
+}
+
+vector<Drive>
+get_drives ()
+{
+	vector<Drive> drives;
+
+	using namespace boost::filesystem;
+	vector<pair<string, string> > mounted_devices = get_mounts("/dev/");
+
 	for (directory_iterator i = directory_iterator("/sys/block"); i != directory_iterator(); ++i) {
 		string const name = i->path().filename().string();
 		path device_type_file("/sys/block/" + name + "/device/type");
@@ -300,8 +311,8 @@ get_drives ()
 				model = dcp::file_to_string("/sys/block/" + name + "/device/model");
 				boost::trim(*model);
 			} catch (...) {}
-			BOOST_FOREACH (string j, mounted_devices) {
-				if (boost::algorithm::starts_with(j, "/dev/" + name)) {
+			for (vector<pair<string, string> >::const_iterator j = mounted_devices.begin(); j != mounted_devices.end(); ++j) {
+				if (boost::algorithm::starts_with(j->first, "/dev/" + name)) {
 					mounted = true;
 				}
 			}
@@ -342,3 +353,18 @@ config_path ()
 	p /= "dcpomatic2";
 	return p;
 }
+
+bool
+unmount_drive (string drive)
+{
+	vector<pair<string, string> > mounts = get_mounts (drive);
+	for (vector<pair<string, string> >::const_iterator i = mounts.begin(); i != mounts.end(); ++i) {
+		int const r = umount(i->second.c_str());
+		LOG_DISK("Tried to unmount %1 and got %2 and %3", i->second, r, errno);
+		if (r == -1) {
+			return false;
+		}
+	}
+	return true;
+}
+
diff --git a/src/lib/cross_osx.cc b/src/lib/cross_osx.cc
index c52dfc434..a363f0570 100644
--- a/src/lib/cross_osx.cc
+++ b/src/lib/cross_osx.cc
@@ -485,3 +485,12 @@ config_path ()
 	p /= "2";
 	return p;
 }
+
+bool
+unmount_device (string device)
+{
+	int const r = umount(device.c_str());
+	LOG_DISK("Tried to unmount %1 and got %2 and %3", device, r, errno);
+	return r == 0;
+
+}
diff --git a/src/lib/exceptions.cc b/src/lib/exceptions.cc
index d394ad4b2..3991c57b9 100644
--- a/src/lib/exceptions.cc
+++ b/src/lib/exceptions.cc
@@ -1,5 +1,5 @@
 /*
-    Copyright (C) 2012-2014 Carl Hetherington <cth@carlh.net>
+    Copyright (C) 2012-2020 Carl Hetherington <cth@carlh.net>
 
     This file is part of DCP-o-matic.
 
@@ -25,6 +25,7 @@
 
 using std::string;
 using std::runtime_error;
+using boost::optional;
 
 /** @param f File that we were trying to open */
 OpenFileError::OpenFileError (boost::filesystem::path f, int error, Mode mode)
@@ -115,14 +116,20 @@ GLError::GLError (char const * last, int e)
 
 }
 
-CopyError::CopyError (string m, int n)
-	: runtime_error (String::compose("%1 (%2)", m, n))
+CopyError::CopyError (string m, optional<int> n)
+	: runtime_error (String::compose("%1%2", m, n ? String::compose(" (%1)", *n) : ""))
 	, _message (m)
 	, _number (n)
 {
 
 }
 
+CommunicationFailedError::CommunicationFailedError ()
+	: CopyError (_("Lost communication between main and writer processes"))
+{
+
+}
+
 VerifyError::VerifyError (string m, int n)
 	: runtime_error (String::compose("%1 (%2)", m, n))
 	, _message (m)
diff --git a/src/lib/exceptions.h b/src/lib/exceptions.h
index 0f8a2eda2..98534bb32 100644
--- a/src/lib/exceptions.h
+++ b/src/lib/exceptions.h
@@ -30,6 +30,7 @@ extern "C" {
 #include <libavutil/pixfmt.h>
 }
 #include <boost/filesystem.hpp>
+#include <boost/optional.hpp>
 #include <stdexcept>
 #include <cstring>
 
@@ -320,22 +321,33 @@ public:
 class CopyError : public std::runtime_error
 {
 public:
-	CopyError (std::string s, int n);
+	CopyError (std::string s, boost::optional<int> n = boost::optional<int>());
 	virtual ~CopyError () throw () {}
 
 	std::string message () const {
 		return _message;
 	}
 
-	int number () const {
+	boost::optional<int> number () const {
 		return _number;
 	}
 
 private:
 	std::string _message;
-	int _number;
+	boost::optional<int> _number;
 };
 
+
+/** @class CommunicationFailedError
+ *  @brief Communcation between dcpomatic2_disk and _disk_writer failed somehow.
+ */
+class CommunicationFailedError : public CopyError
+{
+public:
+	CommunicationFailedError ();
+};
+
+
 /** @class VerifyError
  *  @brief An error which occurs when verifying a DCP that we copied to a distribution drive.
  */
diff --git a/src/lib/nanomsg.cc b/src/lib/nanomsg.cc
index 57220cd54..0fc0dd357 100644
--- a/src/lib/nanomsg.cc
+++ b/src/lib/nanomsg.cc
@@ -20,6 +20,7 @@
 
 #include "nanomsg.h"
 #include "dcpomatic_log.h"
+#include "exceptions.h"
 #include <nanomsg/nn.h>
 #include <nanomsg/pair.h>
 #include <stdexcept>
@@ -48,23 +49,16 @@ Nanomsg::Nanomsg (bool server)
 	}
 }
 
-void
-Nanomsg::blocking_send (string s)
+bool
+Nanomsg::send (string s, int timeout)
 {
-	int const r = nn_send (_socket, s.c_str(), s.length(), 0);
-	if (r < 0) {
-		throw runtime_error(String::compose("Could not send to nanomsg socket (%1)", errno));
-	} else if (r != int(s.length())) {
-		throw runtime_error("Could not send to nanomsg socket (message too big)");
+	if (timeout != 0) {
+		nn_setsockopt (_socket, NN_SOL_SOCKET, NN_SNDTIMEO, &timeout, sizeof(int));
 	}
-}
 
-bool
-Nanomsg::nonblocking_send (string s)
-{
-	int const r = nn_send (_socket, s.c_str(), s.length(), NN_DONTWAIT);
+	int const r = nn_send (_socket, s.c_str(), s.length(), timeout ? 0 : NN_DONTWAIT);
 	if (r < 0) {
-		if (errno == EAGAIN) {
+		if (errno == ETIMEDOUT || errno == EAGAIN) {
 			return false;
 		}
 		throw runtime_error(String::compose("Could not send to nanomsg socket (%1)", errno));
@@ -88,17 +82,17 @@ Nanomsg::get_from_pending ()
 }
 
 void
-Nanomsg::recv_and_parse (bool blocking)
+Nanomsg::recv_and_parse (int flags)
 {
 	char* buf = 0;
-	int const received = nn_recv (_socket, &buf, NN_MSG, blocking ? 0 : NN_DONTWAIT);
+	int const received = nn_recv (_socket, &buf, NN_MSG, flags);
 	if (received < 0)
 	{
-		if (!blocking && errno == EAGAIN) {
+		if (errno == ETIMEDOUT || errno == EAGAIN) {
 			return;
 		}
 
-		throw runtime_error ("Could not communicate with subprocess");
+		throw CommunicationFailedError ();
 	}
 
 	char* p = buf;
@@ -114,32 +108,24 @@ Nanomsg::recv_and_parse (bool blocking)
 	nn_freemsg (buf);
 }
 
-string
-Nanomsg::blocking_get ()
+optional<string>
+Nanomsg::receive (int timeout)
 {
+	if (timeout != 0) {
+		nn_setsockopt (_socket, NN_SOL_SOCKET, NN_RCVTIMEO, &timeout, sizeof(int));
+	}
+
 	optional<string> l = get_from_pending ();
 	if (l) {
 		return *l;
 	}
 
-	recv_and_parse (true);
+	recv_and_parse (timeout ? 0 : NN_DONTWAIT);
 
-	l = get_from_pending ();
+	return get_from_pending ();
 	if (!l) {
-		throw runtime_error ("Could not communicate with subprocess");
+		throw CommunicationFailedError ();
 	}
 
 	return *l;
 }
-
-optional<string>
-Nanomsg::nonblocking_get ()
-{
-	optional<string> l = get_from_pending ();
-	if (l) {
-		return *l;
-	}
-
-	recv_and_parse (false);
-	return get_from_pending ();
-}
diff --git a/src/lib/nanomsg.h b/src/lib/nanomsg.h
index dc84a6ce7..9bd87098b 100644
--- a/src/lib/nanomsg.h
+++ b/src/lib/nanomsg.h
@@ -28,18 +28,21 @@ class Nanomsg : public boost::noncopyable
 public:
 	explicit Nanomsg (bool server);
 
-	void blocking_send (std::string s);
-	/** Try to send a message, returning true if successful, false
-	 *  if we should try again (EAGAIN) or throwing an exception on any other
-	 *  error.
+	/** Try to send a message, waiting for some timeout before giving up.
+	 *  @param timeout Timeout in milliseconds, or -1 for infinite timeout.
+	 *  @return true if the send happened, false if there was a timeout.
 	 */
-	bool nonblocking_send (std::string s);
-	std::string blocking_get ();
-	boost::optional<std::string> nonblocking_get ();
+	bool send (std::string s, int timeout);
+	
+	/** Try to receive a message, waiting for some timeout before giving up.
+	 *  @param timeout Timeout in milliseconds, or -1 for infinite timeout.
+	 *  @return Empty if the timeout was reached, otherwise the received string.
+	 */
+	boost::optional<std::string> receive (int timeout);
 
 private:
 	boost::optional<std::string> get_from_pending ();
-	void recv_and_parse (bool blocking);
+	void recv_and_parse (int flags);
 
 	int _socket;
 	std::list<std::string> _pending;