Add new signal handling.

7 files changed, 541 insertions, 0 deletions

diff --git a/src/lib/signal.cc b/src/lib/signal.cc
new file mode 100644
index 000000000..2ebae8a00
--- /dev/null
+++ b/src/lib/signal.cc
@@ -0,0 +1,108 @@
+/*
+    Copyright (C) 2025 Carl Hetherington <cth@carlh.net>
+
+    This file is part of DCP-o-matic.
+
+    DCP-o-matic 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.
+
+    DCP-o-matic 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 DCP-o-matic.  If not, see <http://www.gnu.org/licenses/>.
+
+*/
+
+
+#include "signal.h"
+
+SignalManager2* dcpomatic::signal::manager = nullptr;
+
+
+Trackable::~Trackable()
+{
+	dcpomatic::signal::manager->unregister_trackable(this);
+}
+
+
+SignalManager2::SignalManager2()
+	: _ui_thread(boost::this_thread::get_id())
+{
+
+}
+
+
+void
+SignalManager2::add_pending(Trackable* trackable, std::function<void ()> pending)
+{
+	boost::mutex::scoped_lock lm(_pending_mutex);
+	_pending.push_back(make_pair(trackable, pending));
+	wake();
+}
+
+
+void
+SignalManager2::process_pending()
+{
+	boost::mutex::scoped_lock lm(_pending_mutex);
+	boost::mutex::scoped_lock lm2(_trackables_mutex);
+	for (auto const& pending: _pending) {
+		if (!pending.first || _trackables.find(pending.first) != _trackables.end()) {
+			pending.second();
+		}
+	}
+	_pending.clear();
+}
+
+
+void
+SignalManager2::register_signal(SignalBase* signal)
+{
+	boost::mutex::scoped_lock lm(_signals_mutex);
+	_signals.insert(signal);
+}
+
+
+void
+SignalManager2::unregister_signal(SignalBase* signal)
+{
+	boost::mutex::scoped_lock lm(_signals_mutex);
+	_signals.erase(signal);
+}
+
+
+void
+SignalManager2::register_trackable(Trackable* trackable)
+{
+	boost::mutex::scoped_lock lm(_trackables_mutex);
+	_trackables.insert(trackable);
+}
+
+
+void SignalManager2::unregister_trackable(Trackable* trackable)
+{
+	boost::mutex::scoped_lock lm(_trackables_mutex);
+	_trackables.erase(trackable);
+}
+
+
+void
+SignalManager2::disconnect(SignalBase* signal, int id)
+{
+	boost::mutex::scoped_lock lm(_signals_mutex);
+	if (_signals.find(signal) != _signals.end()) {
+		signal->disconnect(id);
+	}
+}
+
+
+void
+SignalManager2::wake()
+{
+	process_pending();
+}
diff --git a/src/lib/signal.h b/src/lib/signal.h
new file mode 100644
index 000000000..98094c82a
--- /dev/null
+++ b/src/lib/signal.h
@@ -0,0 +1,237 @@
+/*
+    Copyright (C) 2025 Carl Hetherington <cth@carlh.net>
+
+    This file is part of DCP-o-matic.
+
+    DCP-o-matic 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.
+
+    DCP-o-matic 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 DCP-o-matic.  If not, see <http://www.gnu.org/licenses/>.
+
+*/
+
+
+#ifndef DCPOMATIC_SIGNAL_H
+#define DCPOMATIC_SIGNAL_H
+
+
+#include "dcpomatic_assert.h"
+#include <boost/optional.hpp>
+#include <boost/thread.hpp>
+#include <set>
+
+
+class SignalBase;
+
+
+class Trackable
+{
+public:
+	~Trackable();
+};
+
+
+class SignalManager2
+{
+public:
+	/** Must be called from the UI thread, as the current thread ID will be
+	 *  taken as the UI one.
+	 */
+	SignalManager2();
+
+	virtual ~SignalManager2() {}
+
+	boost::thread::id ui_thread() const {
+		return _ui_thread;
+	}
+
+	void add_pending(Trackable* trackable, std::function<void ()> pending);
+	void process_pending();
+
+	void register_signal(SignalBase* signal);
+	void unregister_signal(SignalBase* signal);
+
+	void register_trackable(Trackable* trackable);
+	void unregister_trackable(Trackable* trackable);
+
+	void disconnect(SignalBase* signal, int id);
+
+	virtual void wake();
+
+private:
+	boost::thread::id _ui_thread;
+	boost::mutex _pending_mutex;
+	std::list<std::pair<Trackable*, std::function<void ()>>> _pending;
+	boost::mutex _signals_mutex;
+	std::set<SignalBase*> _signals;
+	boost::mutex _trackables_mutex;
+	std::set<Trackable*> _trackables;
+};
+
+
+namespace dcpomatic {
+namespace signal {
+	extern SignalManager2* manager;
+}
+}
+
+
+class SignalBase
+{
+public:
+	SignalBase()
+	{
+		DCPOMATIC_ASSERT(dcpomatic::signal::manager);
+		dcpomatic::signal::manager->register_signal(this);
+	}
+
+	virtual ~SignalBase()
+	{
+		dcpomatic::signal::manager->unregister_signal(this);
+	}
+
+	virtual void disconnect(int id) = 0;
+};
+
+
+class Connection
+{
+public:
+	Connection(SignalBase* signal, int id)
+		: _signal(signal)
+		, _id(id)
+	{}
+
+	Connection& operator=(Connection const& other)
+	{
+		_signal = other._signal;
+		_id = other._id;
+		return *this;
+	}
+
+	void disconnect()
+	{
+		DCPOMATIC_ASSERT(dcpomatic::signal::manager);
+		dcpomatic::signal::manager->disconnect(_signal, _id);
+	}
+
+private:
+	SignalBase* _signal;
+	int _id;
+};
+
+
+/** @class ScopedConnection
+ *  @brief Connection that disconnects itself on destruction.
+ */
+class ScopedConnection
+{
+public:
+	ScopedConnection() {}
+
+	ScopedConnection(Connection connection)
+		: _connection(std::move(connection))
+	{}
+
+	ScopedConnection(ScopedConnection const& other) = delete;
+
+	ScopedConnection(ScopedConnection&& other)
+		: _connection(std::move(other._connection))
+	{}
+
+	ScopedConnection& operator=(ScopedConnection const& other) = delete;
+
+	ScopedConnection& operator=(ScopedConnection&& other)
+	{
+		if (this != &other) {
+			_connection = other._connection;
+			other._connection.reset();
+		}
+		return *this;
+	}
+
+	~ScopedConnection()
+	{
+		if (_connection) {
+			_connection->disconnect();
+		}
+	}
+
+private:
+       boost::optional<Connection> _connection;
+};
+
+
+template <class Signature>
+class Signal : public SignalBase
+{
+private:
+	struct Callback
+	{
+		int id;
+		bool same_thread;
+		Trackable* trackable = nullptr;
+		std::function<Signature> function;
+	};
+
+public:
+	Connection connect_same_thread(std::function<Signature> function)
+	{
+		boost::mutex::scoped_lock lm(_mutex);
+		auto const id = _id++;
+		_callbacks.push_back({id, true, nullptr, function});
+		return Connection(this, id);
+	}
+
+	Connection connect_ui_thread(Trackable* trackable, std::function<Signature> function)
+	{
+		dcpomatic::signal::manager->register_trackable(trackable);
+		boost::mutex::scoped_lock lm(_mutex);
+		auto const id = _id++;
+		_callbacks.push_back({id, false, trackable, function});
+		return Connection(this, id);
+	}
+
+	template <typename... Args>
+	void operator()(Args... args)
+	{
+		boost::mutex::scoped_lock lm(_mutex);
+		DCPOMATIC_ASSERT(dcpomatic::signal::manager);
+		bool in_ui_thread = boost::this_thread::get_id() == dcpomatic::signal::manager->ui_thread();
+		for (auto const& callback: _callbacks) {
+			if (callback.same_thread || in_ui_thread) {
+				callback.function(args...);
+			} else {
+				dcpomatic::signal::manager->add_pending(callback.trackable, boost::bind(callback.function, args...));
+			}
+		}
+	}
+
+private:
+	void disconnect(int id) override
+	{
+		boost::mutex::scoped_lock lm(_mutex);
+		auto iter = std::find_if(_callbacks.begin(), _callbacks.end(), [id](Callback const& callback) { return callback.id == id; });
+		if (iter != _callbacks.end()) {
+			_callbacks.erase(iter);
+		}
+	}
+
+	friend class Connection;
+
+	boost::mutex _mutex;
+	std::list<Callback> _callbacks;
+	int _id = 0;
+};
+
+
+#endif
+
diff --git a/src/lib/wscript b/src/lib/wscript
index dc18f713a..b5e9a1f8b 100644
--- a/src/lib/wscript
+++ b/src/lib/wscript
@@ -193,6 +193,7 @@ sources = """
           send_problem_report_job.cc
           server.cc
           shuffler.cc
+          signal.cc
           spl.cc
           spl_entry.cc
           sqlite_database.cc
diff --git a/src/wx/wx_signal.h b/src/wx/wx_signal.h
new file mode 100644
index 000000000..bbaf316f9
--- /dev/null
+++ b/src/wx/wx_signal.h
@@ -0,0 +1,49 @@
+/*
+    Copyright (C) 2025 Carl Hetherington <cth@carlh.net>
+
+    This file is part of DCP-o-matic.
+
+    DCP-o-matic 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.
+
+    DCP-o-matic 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 DCP-o-matic.  If not, see <http://www.gnu.org/licenses/>.
+
+*/
+
+
+#include "lib/signal.h"
+#include <dcp/warnings.h>
+LIBDCP_DISABLE_WARNINGS
+#include <wx/wx.h>
+LIBDCP_ENABLE_WARNINGS
+
+
+/** @class wxThreadWaker
+ *  @brief Class to wake a thread so that it looks for incoming signals that it should deliver.
+ */
+class wxSignalManager2 : public SignalManager2
+{
+public:
+	wxSignalManager2(wxEvtHandler* handler)
+		: _handler(handler)
+	{}
+
+	void wake() override
+	{
+		/* This will result in an idle handler being called, which should process
+		 * pending events.
+		 */
+		_handler->QueueEvent(new wxCommandEvent(wxEVT_IDLE));
+	}
+
+private:
+	wxEvtHandler* _handler;
+};
diff --git a/test/signal_test.cc b/test/signal_test.cc
new file mode 100644
index 000000000..a03f5e38a
--- /dev/null
+++ b/test/signal_test.cc
@@ -0,0 +1,133 @@
+#include "lib/signal.h"
+#include <boost/thread/condition.hpp>
+#include <boost/thread/mutex.hpp>
+#include <boost/test/unit_test.hpp>
+#include <boost/bind/bind.hpp>
+
+using namespace boost::placeholders;
+
+
+BOOST_AUTO_TEST_CASE(signal_test_basic_same_thread)
+{
+	int signal_received = 0;
+
+	Signal<void (int)> signal;
+	signal.connect_same_thread([&](int x) {
+		signal_received = x;
+	});
+
+	signal(42);
+
+	BOOST_CHECK_EQUAL(signal_received, 42);
+}
+
+
+BOOST_AUTO_TEST_CASE(signal_test_disconnect)
+{
+	int signal_received = 0;
+
+	Signal<void (int)> signal;
+	auto connection = signal.connect_same_thread([&](int x) { signal_received = x;});
+	connection.disconnect();
+	signal(42);
+
+	BOOST_CHECK_EQUAL(signal_received, 0);
+}
+
+
+static
+void
+emit_signal_from_thread(Signal<void (int)>* signal)
+{
+	boost::mutex mutex;
+	boost::condition emitted_condition;
+	bool emitted = false;
+
+	boost::thread thread([&]() {
+		(*signal)(42);
+		boost::mutex::scoped_lock lm(mutex);
+		emitted = true;
+		emitted_condition.notify_all();
+	});
+
+	boost::mutex::scoped_lock lm(mutex);
+	while (!emitted) {
+		emitted_condition.wait(lm);
+	}
+
+	thread.join();
+}
+
+
+BOOST_AUTO_TEST_CASE(signal_test_basic_worker_to_ui_thread)
+{
+	int signal_received = 0;
+
+	Signal<void (int)> signal;
+	signal.connect_ui_thread(nullptr, [&](int x) {
+		BOOST_CHECK(boost::this_thread::get_id() == dcpomatic::signal::manager->ui_thread());
+		signal_received = x;
+	});
+
+	emit_signal_from_thread(&signal);
+
+	dcpomatic::signal::manager->process_pending();
+
+	BOOST_CHECK_EQUAL(signal_received, 42);
+}
+
+
+BOOST_AUTO_TEST_CASE(signal_test_signal_going_away)
+{
+	int signal_received = 0;
+
+	auto signal = new Signal<void (int)>();
+	signal->connect_ui_thread(nullptr, [&](int x) {
+		signal_received = x;
+	});
+
+	emit_signal_from_thread(signal);
+	delete signal;
+
+	dcpomatic::signal::manager->process_pending();
+
+	BOOST_CHECK_EQUAL(signal_received, 42);
+}
+
+
+BOOST_AUTO_TEST_CASE(signal_test_disconnection_when_signal_gone_away)
+{
+	auto signal = new Signal<void (int)>();
+	auto connection = signal->connect_same_thread([&](int) {});
+	delete signal;
+	connection.disconnect();
+}
+
+
+BOOST_AUTO_TEST_CASE(signal_test_handler_going_away)
+{
+	Signal<void (int)> signal;
+
+	class A : public Trackable
+	{
+	public:
+		void handler(int x)
+		{
+			_x += x;
+		}
+
+	private:
+		int _x = 1;
+	};
+
+	auto a = new A;
+
+	signal.connect_ui_thread(a, boost::bind(&A::handler, a, _1));
+
+	emit_signal_from_thread(&signal);
+
+	delete a;
+
+	dcpomatic::signal::manager->process_pending();
+}
+
diff --git a/test/test.cc b/test/test.cc
index 6cdb6d510..6a7b4799c 100644
--- a/test/test.cc
+++ b/test/test.cc
@@ -39,6 +39,7 @@
 #include "lib/log_entry.h"
 #include "lib/make_dcp.h"
 #include "lib/ratio.h"
+#include "lib/signal.h"
 #include "lib/signal_manager.h"
 #include "lib/util.h"
 #include "test.h"
@@ -151,6 +152,16 @@ public:
 	}
 };
 
+class TestSignalManager2 : public SignalManager2
+{
+public:
+	/* No wakes in tests: we call ui_idle ourselves */
+	void wake() override
+	{
+
+	}
+};
+
 struct TestConfig
 {
 	TestConfig ()
@@ -167,6 +178,7 @@ struct TestConfig
 		EncodeServerFinder::drop();
 
 		signal_manager = new TestSignalManager ();
+		dcpomatic::signal::manager = new TestSignalManager2();
 
 		dcpomatic_log.reset (new FileLog("build/test/log"));
 
diff --git a/test/wscript b/test/wscript
index 50c86751f..a6a9733d7 100644
--- a/test/wscript
+++ b/test/wscript
@@ -159,6 +159,7 @@ def build(bld):
                  render_subtitles_test.cc
                  scaling_test.cc
                  scoped_temporary_test.cc
+                 signal_test.cc
                  silence_padding_test.cc
                  shuffler_test.cc
                  skip_frame_test.cc