thread/Cond: add wait() overload which takes a unique_lock<>

Just like std::condition_variable, which however has no way to specify
the std::mutex directly.

src/command/FingerprintCommands.cxx

@@ -270,7 +270,7 @@ GetChromaprintCommand::OpenUri(const char *uri2)
 	auto is = InputStream::Open(uri2, mutex);
 	is->SetHandler(this);
 
-	const std::lock_guard<Mutex> lock(mutex);
+	std::unique_lock<Mutex> lock(mutex);
 	while (true) {
 		if (cancel)
 			throw StopDecoder();
@@ -281,7 +281,7 @@ GetChromaprintCommand::OpenUri(const char *uri2)
 			return is;
 		}
 
-		cond.wait(mutex);
+		cond.wait(lock);
 	}
 }
 
@@ -294,7 +294,7 @@ GetChromaprintCommand::Read(InputStream &is, void *buffer, size_t length)
 	if (length == 0)
 		return 0;
 
-	std::lock_guard<Mutex> lock(mutex);
+	std::unique_lock<Mutex> lock(mutex);
 
 	while (true) {
 		if (cancel)
@@ -303,7 +303,7 @@ GetChromaprintCommand::Read(InputStream &is, void *buffer, size_t length)
 		if (is.IsAvailable())
 			break;
 
-		cond.wait(mutex);
+		cond.wait(lock);
 	}
 
 	return is.Read(buffer, length);

src/decoder/Bridge.cxx

@@ -366,7 +366,7 @@ DecoderBridge::OpenUri(const char *uri)
 	auto is = InputStream::Open(uri, mutex);
 	is->SetHandler(&dc);
 
-	const std::lock_guard<Mutex> lock(mutex);
+	std::unique_lock<Mutex> lock(mutex);
 	while (true) {
 		if (dc.command == DecoderCommand::STOP)
 			throw StopDecoder();
@@ -377,7 +377,7 @@ DecoderBridge::OpenUri(const char *uri)
 			return is;
 		}
 
-		cond.wait(mutex);
+		cond.wait(lock);
 	}
 }
 
@@ -391,7 +391,7 @@ try {
 	if (length == 0)
 		return 0;
 
-	std::lock_guard<Mutex> lock(is.mutex);
+	std::unique_lock<Mutex> lock(is.mutex);
 
 	while (true) {
 		if (CheckCancelRead())
@@ -400,7 +400,7 @@ try {
 		if (is.IsAvailable())
 			break;
 
-		dc.cond.wait(is.mutex);
+		dc.cond.wait(lock);
 	}
 
 	size_t nbytes = is.Read(buffer, length);

src/event/Call.cxx

@@ -52,9 +52,9 @@ public:
 		defer_event.Schedule();
 
 		{
-			const std::lock_guard<Mutex> lock(mutex);
+			std::unique_lock<Mutex> lock(mutex);
 			while (!done)
-				cond.wait(mutex);
+				cond.wait(lock);
 		}
 
 		if (exception)

src/input/BufferedInputStream.cxx

@@ -149,7 +149,7 @@ BufferedInputStream::RunThread() noexcept
 {
 	SetThreadName("input_buffered");
 
-	const std::lock_guard<Mutex> lock(mutex);
+	std::unique_lock<Mutex> lock(mutex);
 
 	while (!stop) {
 		assert(size == buffer.size());
@@ -205,6 +205,6 @@ BufferedInputStream::RunThread() noexcept
 			client_cond.notify_one();
 			InvokeOnAvailable();
 		} else
-			wake_cond.wait(mutex);
+			wake_cond.wait(lock);
 	}
 }

src/input/Open.cxx

@@ -57,14 +57,14 @@ InputStream::OpenReady(const char *uri, Mutex &mutex)
 	is->SetHandler(&handler);
 
 	{
-		const std::lock_guard<Mutex> protect(mutex);
+		std::unique_lock<Mutex> lock(mutex);
 
 		while (true) {
 			is->Update();
 			if (is->IsReady())
 				break;
 
-			handler.cond.wait(mutex);
+			handler.cond.wait(lock);
 		}
 
 		is->Check();

src/input/ThreadInputStream.cxx

@@ -67,7 +67,7 @@ ThreadInputStream::ThreadFunc() noexcept
 {
 	FormatThreadName("input:%s", plugin);
 
-	const std::lock_guard<Mutex> lock(mutex);
+	std::unique_lock<Mutex> lock(mutex);
 
 	try {
 		Open();
@@ -85,7 +85,7 @@ ThreadInputStream::ThreadFunc() noexcept
 
 		auto w = buffer.Write();
 		if (w.empty()) {
-			wake_cond.wait(mutex);
+			wake_cond.wait(lock);
 		} else {
 			size_t nbytes;
 

src/lib/nfs/Blocking.hxx

@@ -59,9 +59,9 @@ public:
 
 private:
 	bool LockWaitFinished() noexcept {
-		const std::lock_guard<Mutex> protect(mutex);
+		std::unique_lock<Mutex> lock(mutex);
 		while (!finished)
-			if (!cond.wait_for(mutex, timeout))
+			if (!cond.wait_for(lock, timeout))
 				return false;
 
 		return true;

src/neighbor/plugins/SmbclientNeighborPlugin.cxx

@@ -238,7 +238,7 @@ SmbclientNeighborExplorer::ThreadFunc() noexcept
 {
 	SetThreadName("smbclient");
 
-	const std::lock_guard<Mutex> lock(mutex);
+	std::unique_lock<Mutex> lock(mutex);
 
 	while (!quit) {
 		Run();
@@ -247,7 +247,7 @@ SmbclientNeighborExplorer::ThreadFunc() noexcept
 			break;
 
 		// TODO: sleep for how long?
-		cond.wait_for(mutex, std::chrono::seconds(10));
+		cond.wait_for(lock, std::chrono::seconds(10));
 	}
 }
 

src/output/Thread.cxx

@@ -412,7 +412,7 @@ AudioOutputControl::Task() noexcept
 
 	SetThreadTimerSlackUS(100);
 
-	const std::lock_guard<Mutex> lock(mutex);
+	std::unique_lock<Mutex> lock(mutex);
 
 	while (true) {
 		switch (command) {
@@ -516,7 +516,7 @@ AudioOutputControl::Task() noexcept
 
 		if (command == Command::NONE) {
 			woken_for_play = false;
-			wake_cond.wait(mutex);
+			wake_cond.wait(lock);
 		}
 	}
 }

src/output/plugins/AlsaOutputPlugin.cxx

@@ -796,7 +796,7 @@ AlsaOutput::DrainInternal()
 void
 AlsaOutput::Drain()
 {
-	const std::lock_guard<Mutex> lock(mutex);
+	std::unique_lock<Mutex> lock(mutex);
 
 	if (error)
 		std::rethrow_exception(error);
@@ -806,7 +806,7 @@ AlsaOutput::Drain()
 	Activate();
 
 	while (drain && active)
-		cond.wait(mutex);
+		cond.wait(lock);
 
 	if (error)
 		std::rethrow_exception(error);
@@ -882,7 +882,7 @@ AlsaOutput::Play(const void *chunk, size_t size)
 		   been played */
 		return size;
 
-	const std::lock_guard<Mutex> lock(mutex);
+	std::unique_lock<Mutex> lock(mutex);
 
 	while (true) {
 		if (error)
@@ -905,7 +905,7 @@ AlsaOutput::Play(const void *chunk, size_t size)
 
 		/* wait for the DispatchSockets() to make room in the
 		   ring_buffer */
-		cond.wait(mutex);
+		cond.wait(lock);
 	}
 }
 

src/output/plugins/httpd/HttpdOutputPlugin.cxx

@@ -277,9 +277,9 @@ HttpdOutput::BroadcastFromEncoder()
 {
 	/* synchronize with the IOThread */
 	{
-		const std::lock_guard<Mutex> lock(mutex);
+		std::unique_lock<Mutex> lock(mutex);
 		while (!pages.empty())
-			cond.wait(mutex);
+			cond.wait(lock);
 	}
 
 	bool empty = true;

src/output/plugins/sles/SlesOutputPlugin.cxx

@@ -317,13 +317,13 @@ SlesOutput::Play(const void *chunk, size_t size)
 		pause = false;
 	}
 
-	const std::lock_guard<Mutex> protect(mutex);
+	std::unique_lock<Mutex> lock(mutex);
 
 	assert(filled < BUFFER_SIZE);
 
 	while (n_queued == N_BUFFERS) {
 		assert(filled == 0);
-		cond.wait(mutex);
+		cond.wait(lock);
 	}
 
 	size_t nbytes = std::min(BUFFER_SIZE - filled, size);
@@ -346,12 +346,12 @@ SlesOutput::Play(const void *chunk, size_t size)
 void
 SlesOutput::Drain()
 {
-	const std::lock_guard<Mutex> protect(mutex);
+	std::unique_lock<Mutex> lock(mutex);
 
 	assert(filled < BUFFER_SIZE);
 
 	while (n_queued > 0)
-		cond.wait(mutex);
+		cond.wait(lock);
 }
 
 void

src/storage/plugins/CurlStorage.cxx

@@ -124,9 +124,9 @@ public:
 	}
 
 	void Wait() {
-		const std::lock_guard<Mutex> lock(mutex);
+		std::unique_lock<Mutex> lock(mutex);
 		while (!done)
-			cond.wait(mutex);
+			cond.wait(lock);
 
 		if (postponed_error)
 			std::rethrow_exception(postponed_error);

src/storage/plugins/NfsStorage.cxx

@@ -167,7 +167,7 @@ private:
 	}
 
 	void WaitConnected() {
-		const std::lock_guard<Mutex> protect(mutex);
+		std::unique_lock<Mutex> lock(mutex);
 
 		while (true) {
 			switch (state) {
@@ -179,7 +179,7 @@ private:
 				}
 
 				if (state == State::INITIAL)
-					cond.wait(mutex);
+					cond.wait(lock);
 				break;
 
 			case State::CONNECTING:

src/storage/plugins/UdisksStorage.cxx

@@ -198,7 +198,7 @@ UdisksStorage::OnListReply(ODBus::Message reply) noexcept
 void
 UdisksStorage::MountWait()
 {
-	const std::lock_guard<Mutex> lock(mutex);
+	std::unique_lock<Mutex> lock(mutex);
 
 	if (mounted_storage)
 		/* already mounted */
@@ -210,7 +210,7 @@ UdisksStorage::MountWait()
 	}
 
 	while (want_mount)
-		cond.wait(mutex);
+		cond.wait(lock);
 
 	if (mount_error)
 		std::rethrow_exception(mount_error);
@@ -272,7 +272,7 @@ try {
 void
 UdisksStorage::UnmountWait()
 {
-	const std::lock_guard<Mutex> lock(mutex);
+	std::unique_lock<Mutex> lock(mutex);
 
 	if (!mounted_storage)
 		/* not mounted */
@@ -281,7 +281,7 @@ UdisksStorage::UnmountWait()
 	defer_unmount.Schedule();
 
 	while (mounted_storage)
-		cond.wait(mutex);
+		cond.wait(lock);
 
 	if (mount_error)
 		std::rethrow_exception(mount_error);

src/thread/PosixCond.hxx

@@ -33,6 +33,7 @@
 #include "PosixMutex.hxx"
 
 #include <chrono>
+#include <mutex>
 
 #include <sys/time.h>
 
@@ -74,6 +75,11 @@ public:
 		pthread_cond_wait(&cond, &mutex.mutex);
 	}
 
+	template<typename M>
+	void wait(std::unique_lock<M> &lock) noexcept {
+		wait(*lock.mutex());
+	}
+
 private:
 	bool wait_for(PosixMutex &mutex, uint_least32_t timeout_us) noexcept {
 		struct timeval now;
@@ -102,6 +108,12 @@ public:
 
 		return wait_for(mutex, timeout_us);
 	}
+
+	template<typename M>
+	bool wait_for(std::unique_lock<M> &lock,
+		      std::chrono::steady_clock::duration timeout) noexcept {
+		return wait_for(*lock.mutex(), timeout);
+	}
 };
 
 #endif

src/thread/WindowsCond.hxx

@@ -33,6 +33,7 @@
 #include "CriticalSection.hxx"
 
 #include <chrono>
+#include <mutex>
 
 /**
  * Wrapper for a CONDITION_VARIABLE, backend for the Cond class.
@@ -69,9 +70,20 @@ public:
 		return wait_for(mutex, timeout_ms);
 	}
 
+	template<typename M>
+	bool wait_for(std::unique_lock<M> &lock,
+		      std::chrono::steady_clock::duration timeout) noexcept {
+		return wait_for(*lock.mutex(), timeout);
+	}
+
 	void wait(CriticalSection &mutex) noexcept {
 		wait_for(mutex, INFINITE);
 	}
+
+	template<typename M>
+	void wait(std::unique_lock<M> &lock) noexcept {
+		wait(*lock.mutex());
+	}
 };
 
 #endif

test/run_input.cxx

@@ -175,9 +175,9 @@ class DumpRemoteTagHandler final : public RemoteTagHandler {
 
 public:
 	Tag Wait() {
-		const std::lock_guard<Mutex> lock(mutex);
+		std::unique_lock<Mutex> lock(mutex);
 		while (!done)
-			cond.wait(mutex);
+			cond.wait(lock);
 
 		if (error)
 			std::rethrow_exception(error);