Adding upstream version 0.3.65.upstream/0.3.65upstream

Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>

1 files changed, 1095 insertions, 0 deletions

diff --git a/src/modules/module-rt.c b/src/modules/module-rt.c
new file mode 100644
index 0000000..f6b4ade
--- /dev/null
+++ b/src/modules/module-rt.c
@@ -0,0 +1,1095 @@
+/* PipeWire
+ *
+ * Copyright © 2022 Wim Taymans
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a
+ * copy of this software and associated documentation files (the "Software"),
+ * to deal in the Software without restriction, including without limitation
+ * the rights to use, copy, modify, merge, publish, distribute, sublicense,
+ * and/or sell copies of the Software, and to permit persons to whom the
+ * Software is furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice (including the next
+ * paragraph) shall be included in all copies or substantial portions of the
+ * Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.  IN NO EVENT SHALL
+ * THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
+ * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
+ * DEALINGS IN THE SOFTWARE.
+ */
+/***
+  Copyright 2009 Lennart Poettering
+  Copyright 2010 David Henningsson <diwic@ubuntu.com>
+
+  Permission is hereby granted, free of charge, to any person
+  obtaining a copy of this software and associated documentation files
+  (the "Software"), to deal in the Software without restriction,
+  including without limitation the rights to use, copy, modify, merge,
+  publish, distribute, sublicense, and/or sell copies of the Software,
+  and to permit persons to whom the Software is furnished to do so,
+  subject to the following conditions:
+
+  The above copyright notice and this permission notice shall be
+  included in all copies or substantial portions of the Software.
+
+  THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
+  EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
+  MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
+  NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS
+  BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN
+  ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
+  CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+  SOFTWARE.
+***/
+
+#include <stdlib.h>
+#include <stdbool.h>
+#include <string.h>
+#include <stdio.h>
+#include <errno.h>
+#include <sys/stat.h>
+#if defined(__FreeBSD__) || defined(__MidnightBSD__)
+#include <sys/thr.h>
+#endif
+#include <fcntl.h>
+#include <unistd.h>
+#include <pthread.h>
+#include <sys/resource.h>
+#include <sys/syscall.h>
+
+#include "config.h"
+
+#include <spa/utils/result.h>
+#include <spa/utils/string.h>
+
+#include <pipewire/impl.h>
+#include <pipewire/thread.h>
+
+#ifdef HAVE_DBUS
+#include <spa/support/dbus.h>
+#include <dbus/dbus.h>
+#endif
+
+/** \page page_module_rt PipeWire Module: RT
+ *
+ * The `rt` modules can give real-time priorities to processing threads.
+ *
+ * It uses the operating system's scheduler to enable realtime scheduling
+ * for certain threads to assist with low latency audio processing.
+ * This requires `RLIMIT_RTPRIO` to be set to a value that's equal to this
+ * module's `rt.prio` parameter or higher. Most distros will come with some
+ * package that configures this for certain groups or users. If this is not set
+ * up and DBus is available, then this module will fall back to using RTKit.
+ *
+ * ## Module Options
+ *
+ * - `nice.level`: The nice value set for the application thread. It improves
+ *                 performance of the communication with the pipewire daemon.
+ * - `rt.prio`: The realtime priority of the data thread. Higher values are
+ *              higher priority.
+ * - `rt.time.soft`, `rt.time.hard`: The amount of CPU time an RT thread can
+ *              consume without doing any blocking calls before the kernel kills
+ *              the thread. This is a safety measure to avoid lockups of the complete
+ *              system when some thread consumes 100%.
+
+ * The nice level is by default set to an invalid value so that clients don't
+ * automatically have the nice level raised.
+ *
+ * The PipeWire server processes are explicitly configured with a valid nice level.
+ *
+ * ## Example configuration
+ *
+ *\code{.unparsed}
+ * context.modules = [
+ * {   name = libpipewire-module-rt
+ *     args = {
+ *         #nice.level   = 20
+ *         #rt.prio      = 88
+ *         #rt.time.soft = -1
+ *         #rt.time.hard = -1
+ *     }
+ *     flags = [ ifexists nofail ]
+ * }
+ * ]
+ *\endcode
+ */
+
+#define NAME "rt"
+
+PW_LOG_TOPIC_STATIC(mod_topic, "mod." NAME);
+#define PW_LOG_TOPIC_DEFAULT mod_topic
+
+#define REALTIME_POLICY         SCHED_FIFO
+#ifdef SCHED_RESET_ON_FORK
+#define PW_SCHED_RESET_ON_FORK  SCHED_RESET_ON_FORK
+#else
+/* FreeBSD compat */
+#define PW_SCHED_RESET_ON_FORK  0
+#endif
+
+#define IS_VALID_NICE_LEVEL(l)	((l)>=-20 && (l)<=19)
+
+#define DEFAULT_NICE_LEVEL	20
+#define DEFAULT_RT_PRIO_MIN	11
+#define DEFAULT_RT_PRIO		88
+#define DEFAULT_RT_TIME_SOFT	-1
+#define DEFAULT_RT_TIME_HARD	-1
+
+#define MODULE_USAGE	"[nice.level=<priority: default "SPA_STRINGIFY(DEFAULT_NICE_LEVEL)"(don't change)>] "	\
+			"[rt.prio=<priority: default "SPA_STRINGIFY(DEFAULT_RT_PRIO)">] "		\
+			"[rt.time.soft=<in usec: default "SPA_STRINGIFY(DEFAULT_RT_TIME_SOFT)"] "	\
+			"[rt.time.hard=<in usec: default "SPA_STRINGIFY(DEFAULT_RT_TIME_HARD)"] "
+
+static const struct spa_dict_item module_props[] = {
+	{ PW_KEY_MODULE_AUTHOR, "Wim Taymans <wim.taymans@gmail.com>" },
+	{ PW_KEY_MODULE_DESCRIPTION, "Use realtime thread scheduling, falling back to RTKit" },
+	{ PW_KEY_MODULE_USAGE, MODULE_USAGE },
+	{ PW_KEY_MODULE_VERSION, PACKAGE_VERSION },
+};
+
+#ifdef HAVE_DBUS
+#define RTKIT_SERVICE_NAME "org.freedesktop.RealtimeKit1"
+#define RTKIT_OBJECT_PATH "/org/freedesktop/RealtimeKit1"
+#define RTKIT_INTERFACE "org.freedesktop.RealtimeKit1"
+
+#define XDG_PORTAL_SERVICE_NAME "org.freedesktop.portal.Desktop"
+#define XDG_PORTAL_OBJECT_PATH "/org/freedesktop/portal/desktop"
+#define XDG_PORTAL_INTERFACE "org.freedesktop.portal.Realtime"
+
+/** \cond */
+struct pw_rtkit_bus {
+	DBusConnection *bus;
+};
+/** \endcond */
+
+struct thread {
+	struct impl *impl;
+	struct spa_list link;
+	pthread_t thread;
+	pid_t pid;
+	void *(*start)(void*);
+	void *arg;
+};
+#endif /* HAVE_DBUS */
+
+struct impl {
+	struct pw_context *context;
+
+	struct spa_thread_utils thread_utils;
+
+	int nice_level;
+	int rt_prio;
+	rlim_t rt_time_soft;
+	rlim_t rt_time_hard;
+
+	struct spa_hook module_listener;
+
+#ifdef HAVE_DBUS
+	bool use_rtkit;
+	/* For D-Bus. These are const static. */
+	const char* service_name;
+	const char* object_path;
+	const char* interface;
+	struct pw_rtkit_bus *rtkit_bus;
+
+	/* These are only for the RTKit implementation to fill in the `thread`
+	 * struct. Since there's barely any overhead here we'll do this
+	 * regardless of which backend is used. */
+	pthread_mutex_t lock;
+	pthread_cond_t cond;
+	struct spa_list threads_list;
+#endif
+};
+
+#ifndef RLIMIT_RTTIME
+#define RLIMIT_RTTIME 15
+#endif
+
+static pid_t _gettid(void)
+{
+#if defined(HAVE_GETTID)
+	return (pid_t) gettid();
+#elif defined(__linux__)
+	return syscall(SYS_gettid);
+#elif defined(__FreeBSD__) || defined(__MidnightBSD__)
+	long pid;
+	thr_self(&pid);
+	return (pid_t)pid;
+#else
+#error "No gettid impl"
+#endif
+}
+
+#ifdef HAVE_DBUS
+struct pw_rtkit_bus *pw_rtkit_bus_get(DBusBusType bus_type)
+{
+	struct pw_rtkit_bus *bus;
+	DBusError error;
+
+	if (getenv("DISABLE_RTKIT")) {
+		errno = ENOTSUP;
+		return NULL;
+	}
+
+	dbus_error_init(&error);
+
+	bus = calloc(1, sizeof(struct pw_rtkit_bus));
+	if (bus == NULL)
+		return NULL;
+
+	bus->bus = dbus_bus_get_private(bus_type, &error);
+	if (bus->bus == NULL)
+		goto error;
+
+	dbus_connection_set_exit_on_disconnect(bus->bus, false);
+
+	return bus;
+
+error:
+	free(bus);
+	pw_log_error("Failed to connect to %s bus: %s",
+		     bus_type == DBUS_BUS_SYSTEM ? "system" : "session", error.message);
+	dbus_error_free(&error);
+	errno = ECONNREFUSED;
+	return NULL;
+}
+
+struct pw_rtkit_bus *pw_rtkit_bus_get_system(void)
+{
+	return pw_rtkit_bus_get(DBUS_BUS_SYSTEM);
+}
+
+struct pw_rtkit_bus *pw_rtkit_bus_get_session(void)
+{
+	return pw_rtkit_bus_get(DBUS_BUS_SESSION);
+}
+
+bool pw_rtkit_check_xdg_portal(struct pw_rtkit_bus *system_bus)
+{
+	if (!dbus_bus_name_has_owner(system_bus->bus, XDG_PORTAL_SERVICE_NAME, NULL)) {
+		pw_log_warn("Can't find %s. Is xdg-desktop-portal running?", XDG_PORTAL_SERVICE_NAME);
+		return false;
+	}
+
+	return true;
+}
+
+void pw_rtkit_bus_free(struct pw_rtkit_bus *system_bus)
+{
+	dbus_connection_close(system_bus->bus);
+	dbus_connection_unref(system_bus->bus);
+	free(system_bus);
+}
+
+static int translate_error(const char *name)
+{
+	pw_log_warn("RTKit error: %s", name);
+
+	if (spa_streq(name, DBUS_ERROR_NO_MEMORY))
+		return -ENOMEM;
+	if (spa_streq(name, DBUS_ERROR_SERVICE_UNKNOWN) ||
+	    spa_streq(name, DBUS_ERROR_NAME_HAS_NO_OWNER))
+		return -ENOENT;
+	if (spa_streq(name, DBUS_ERROR_ACCESS_DENIED) ||
+	    spa_streq(name, DBUS_ERROR_AUTH_FAILED))
+		return -EACCES;
+
+	return -EIO;
+}
+
+static long long rtkit_get_int_property(struct impl *impl, const char *propname,
+					long long *propval)
+{
+	DBusMessage *m = NULL, *r = NULL;
+	DBusMessageIter iter, subiter;
+	dbus_int64_t i64;
+	dbus_int32_t i32;
+	DBusError error;
+	int current_type;
+	long long ret;
+	struct pw_rtkit_bus *connection = impl->rtkit_bus;
+
+	dbus_error_init(&error);
+
+	if (!(m = dbus_message_new_method_call(impl->service_name,
+					       impl->object_path,
+					       "org.freedesktop.DBus.Properties", "Get"))) {
+		ret = -ENOMEM;
+		goto finish;
+	}
+
+	if (!dbus_message_append_args(m,
+				      DBUS_TYPE_STRING, &impl->interface,
+				      DBUS_TYPE_STRING, &propname, DBUS_TYPE_INVALID)) {
+		ret = -ENOMEM;
+		goto finish;
+	}
+
+	if (!(r = dbus_connection_send_with_reply_and_block(connection->bus, m, -1, &error))) {
+		ret = translate_error(error.name);
+		goto finish;
+	}
+
+	if (dbus_set_error_from_message(&error, r)) {
+		ret = translate_error(error.name);
+		goto finish;
+	}
+
+	ret = -EBADMSG;
+	dbus_message_iter_init(r, &iter);
+	while ((current_type = dbus_message_iter_get_arg_type(&iter)) != DBUS_TYPE_INVALID) {
+
+		if (current_type == DBUS_TYPE_VARIANT) {
+			dbus_message_iter_recurse(&iter, &subiter);
+
+			while ((current_type =
+				dbus_message_iter_get_arg_type(&subiter)) != DBUS_TYPE_INVALID) {
+
+				if (current_type == DBUS_TYPE_INT32) {
+					dbus_message_iter_get_basic(&subiter, &i32);
+					*propval = i32;
+					ret = 0;
+				}
+
+				if (current_type == DBUS_TYPE_INT64) {
+					dbus_message_iter_get_basic(&subiter, &i64);
+					*propval = i64;
+					ret = 0;
+				}
+
+				dbus_message_iter_next(&subiter);
+			}
+		}
+		dbus_message_iter_next(&iter);
+	}
+
+finish:
+
+	if (m)
+		dbus_message_unref(m);
+
+	if (r)
+		dbus_message_unref(r);
+
+	dbus_error_free(&error);
+
+	return ret;
+}
+
+int pw_rtkit_get_max_realtime_priority(struct impl *impl)
+{
+	long long retval;
+	int err;
+
+	err = rtkit_get_int_property(impl, "MaxRealtimePriority", &retval);
+	return err < 0 ? err : retval;
+}
+
+int pw_rtkit_get_min_nice_level(struct impl *impl, int *min_nice_level)
+{
+	long long retval;
+	int err;
+
+	err = rtkit_get_int_property(impl, "MinNiceLevel", &retval);
+	if (err >= 0)
+		*min_nice_level = retval;
+	return err;
+}
+
+long long pw_rtkit_get_rttime_usec_max(struct impl *impl)
+{
+	long long retval;
+	int err;
+
+	err = rtkit_get_int_property(impl, "RTTimeUSecMax", &retval);
+	return err < 0 ? err : retval;
+}
+
+int pw_rtkit_make_realtime(struct impl *impl, pid_t thread, int priority)
+{
+	DBusMessage *m = NULL, *r = NULL;
+	dbus_uint64_t pid;
+	dbus_uint64_t u64;
+	dbus_uint32_t u32;
+	DBusError error;
+	int ret;
+	struct pw_rtkit_bus *connection = impl->rtkit_bus;
+
+	dbus_error_init(&error);
+
+	if (thread == 0)
+		thread = _gettid();
+
+	if (!(m = dbus_message_new_method_call(impl->service_name,
+					       impl->object_path, impl->interface,
+					       "MakeThreadRealtimeWithPID"))) {
+		ret = -ENOMEM;
+		goto finish;
+	}
+
+	pid = (dbus_uint64_t) getpid();
+	u64 = (dbus_uint64_t) thread;
+	u32 = (dbus_uint32_t) priority;
+
+	if (!dbus_message_append_args(m,
+				      DBUS_TYPE_UINT64, &pid,
+				      DBUS_TYPE_UINT64, &u64,
+				      DBUS_TYPE_UINT32, &u32, DBUS_TYPE_INVALID)) {
+		ret = -ENOMEM;
+		goto finish;
+	}
+
+	if (!(r = dbus_connection_send_with_reply_and_block(connection->bus, m, -1, &error))) {
+		ret = translate_error(error.name);
+		goto finish;
+	}
+
+
+	if (dbus_set_error_from_message(&error, r)) {
+		ret = translate_error(error.name);
+		goto finish;
+	}
+
+	ret = 0;
+
+finish:
+
+	if (m)
+		dbus_message_unref(m);
+
+	if (r)
+		dbus_message_unref(r);
+
+	dbus_error_free(&error);
+
+	return ret;
+}
+
+int pw_rtkit_make_high_priority(struct impl *impl, pid_t thread, int nice_level)
+{
+	DBusMessage *m = NULL, *r = NULL;
+	dbus_uint64_t pid;
+	dbus_uint64_t u64;
+	dbus_int32_t s32;
+	DBusError error;
+	int ret;
+	struct pw_rtkit_bus *connection = impl->rtkit_bus;
+
+	dbus_error_init(&error);
+
+	if (thread == 0)
+		thread = _gettid();
+
+	if (!(m = dbus_message_new_method_call(impl->service_name,
+					       impl->object_path, impl->interface,
+					       "MakeThreadHighPriorityWithPID"))) {
+		ret = -ENOMEM;
+		goto finish;
+	}
+
+	pid = (dbus_uint64_t) getpid();
+	u64 = (dbus_uint64_t) thread;
+	s32 = (dbus_int32_t) nice_level;
+
+	if (!dbus_message_append_args(m,
+				      DBUS_TYPE_UINT64, &pid,
+				      DBUS_TYPE_UINT64, &u64,
+				      DBUS_TYPE_INT32, &s32, DBUS_TYPE_INVALID)) {
+		ret = -ENOMEM;
+		goto finish;
+	}
+
+
+
+	if (!(r = dbus_connection_send_with_reply_and_block(connection->bus, m, -1, &error))) {
+		ret = translate_error(error.name);
+		goto finish;
+	}
+
+
+	if (dbus_set_error_from_message(&error, r)) {
+		ret = translate_error(error.name);
+		goto finish;
+	}
+
+	ret = 0;
+
+finish:
+
+	if (m)
+		dbus_message_unref(m);
+
+	if (r)
+		dbus_message_unref(r);
+
+	dbus_error_free(&error);
+
+	return ret;
+}
+#endif /* HAVE_DBUS */
+
+static void module_destroy(void *data)
+{
+	struct impl *impl = data;
+	pw_context_set_object(impl->context, SPA_TYPE_INTERFACE_ThreadUtils, NULL);
+	spa_hook_remove(&impl->module_listener);
+
+#ifdef HAVE_DBUS
+	if (impl->rtkit_bus)
+		pw_rtkit_bus_free(impl->rtkit_bus);
+#endif
+
+	free(impl);
+}
+
+static const struct pw_impl_module_events module_events = {
+	PW_VERSION_IMPL_MODULE_EVENTS,
+	.destroy = module_destroy,
+};
+
+static int get_rt_priority_range(int *out_min, int *out_max)
+{
+	int min, max;
+
+	if ((min = sched_get_priority_min(REALTIME_POLICY)) < 0)
+		return -errno;
+	if ((max = sched_get_priority_max(REALTIME_POLICY)) < 0)
+		return -errno;
+
+	if (out_min)
+		*out_min = min;
+	if (out_max)
+		*out_max = max;
+
+	return 0;
+}
+/**
+ * Check if the current user has permissions to use realtime scheduling at the
+ * specified priority.
+ */
+static bool check_realtime_privileges(struct impl *impl)
+{
+	rlim_t priority = impl->rt_prio;
+	int err, old_policy, new_policy, min, max;
+	struct sched_param old_sched_params;
+	struct sched_param new_sched_params;
+	int try = 0;
+
+	while (try++ < 2) {
+		/* We could check `RLIMIT_RTPRIO`, but the BSDs generally don't have
+		 * that available, and there are also other ways to use realtime
+		 * scheduling without that rlimit being set such as `CAP_SYS_NICE` or
+		 * running as root. Instead of checking a bunch of preconditions, we
+		 * just try if setting realtime scheduling works or not. */
+		if ((err = pthread_getschedparam(pthread_self(), &old_policy, &old_sched_params)) != 0) {
+			pw_log_warn("Failed to check RLIMIT_RTPRIO: %s", strerror(err));
+			return false;
+		}
+		if ((err = get_rt_priority_range(&min, &max)) < 0) {
+			pw_log_warn("Failed to get priority range: %s", strerror(err));
+			return false;
+		}
+		if (try == 2) {
+			struct rlimit rlim;
+			/* second try, try to clamp to RLIMIT_RTPRIO */
+			if (getrlimit(RLIMIT_RTPRIO, &rlim) == 0 && max > (int)rlim.rlim_max) {
+				pw_log_info("Clamp rtprio %d to %d", (int)priority, (int)rlim.rlim_max);
+				max = (int)rlim.rlim_max;
+			}
+			else
+				break;
+		}
+		if (max < DEFAULT_RT_PRIO_MIN) {
+			pw_log_info("Priority max (%d) must be at least %d", max, DEFAULT_RT_PRIO_MIN);
+			return false;
+		}
+
+		/* If the current scheduling policy has `SCHED_RESET_ON_FORK` set, then
+		 * this also needs to be set here or `pthread_setschedparam()` will return
+		 * an error code. Similarly, if it is not set, then we don't want to set
+		 * it here as it would irreversible change the current thread's
+		 * scheduling policy. */
+		spa_zero(new_sched_params);
+		new_sched_params.sched_priority = SPA_CLAMP((int)priority, min, max);
+		new_policy = REALTIME_POLICY;
+		if ((old_policy & PW_SCHED_RESET_ON_FORK) != 0)
+			new_policy |= PW_SCHED_RESET_ON_FORK;
+
+		if (pthread_setschedparam(pthread_self(), new_policy, &new_sched_params) == 0) {
+			impl->rt_prio = new_sched_params.sched_priority;
+			pthread_setschedparam(pthread_self(), old_policy, &old_sched_params);
+			return true;
+		}
+	}
+	pw_log_info("Can't set rt prio to %d: %m (try increasing rlimits)", (int)priority);
+	return false;
+}
+
+static int sched_set_nice(int nice_level)
+{
+	if (setpriority(PRIO_PROCESS, _gettid(), nice_level) == 0)
+		return 0;
+	else
+		return -errno;
+}
+
+static int set_nice(struct impl *impl, int nice_level, bool warn)
+{
+	int res = 0;
+
+#ifdef HAVE_DBUS
+	if (impl->use_rtkit)
+		res = pw_rtkit_make_high_priority(impl, 0, nice_level);
+	else
+		res = sched_set_nice(nice_level);
+#else
+	res = sched_set_nice(nice_level);
+#endif
+
+	if (res < 0) {
+		if (warn)
+			pw_log_warn("could not set nice-level to %d: %s",
+					nice_level, spa_strerror(res));
+	} else {
+		pw_log_info("main thread nice level set to %d",
+				nice_level);
+	}
+	return res;
+}
+
+static int set_rlimit(struct impl *impl)
+{
+	struct rlimit rl;
+	int res = 0;
+
+	spa_zero(rl);
+	rl.rlim_cur = impl->rt_time_soft;
+	rl.rlim_max = impl->rt_time_hard;
+
+#ifdef HAVE_DBUS
+	if (impl->use_rtkit) {
+		long long rttime;
+		rttime = pw_rtkit_get_rttime_usec_max(impl);
+		if (rttime >= 0) {
+			if ((rlim_t)rttime < rl.rlim_cur) {
+				pw_log_debug("clamping rt.time.soft from %llu to %lld because of RTKit",
+					     (long long)rl.rlim_cur, rttime);
+			}
+
+			rl.rlim_cur = SPA_MIN(rl.rlim_cur, (rlim_t)rttime);
+			rl.rlim_max = SPA_MIN(rl.rlim_max, (rlim_t)rttime);
+		}
+	}
+#endif
+
+	if (setrlimit(RLIMIT_RTTIME, &rl) < 0)
+		res = -errno;
+
+	if (res < 0)
+		pw_log_debug("setrlimit() failed: %s", spa_strerror(res));
+	else
+		pw_log_debug("rt.time.soft:%"PRIi64" rt.time.hard:%"PRIi64,
+				(int64_t)rl.rlim_cur, (int64_t)rl.rlim_max);
+
+	return res;
+}
+
+static int acquire_rt_sched(struct spa_thread *thread, int priority)
+{
+	int err, min, max;
+	struct sched_param sp;
+	pthread_t pt = (pthread_t)thread;
+
+	if ((err = get_rt_priority_range(&min, &max)) < 0)
+		return err;
+
+	if (priority < min || priority > max) {
+		pw_log_info("clamping priority %d to range %d - %d for policy %d",
+				priority, min, max, REALTIME_POLICY);
+		priority = SPA_CLAMP(priority, min, max);
+	}
+
+	spa_zero(sp);
+	sp.sched_priority = priority;
+	if ((err = pthread_setschedparam(pt, REALTIME_POLICY | PW_SCHED_RESET_ON_FORK, &sp)) != 0) {
+		pw_log_warn("could not make thread %p realtime: %s", thread, strerror(err));
+		return -err;
+	}
+
+	pw_log_info("acquired realtime priority %d for thread %p", priority, thread);
+	return 0;
+}
+
+static int impl_drop_rt_generic(void *object, struct spa_thread *thread)
+{
+	struct sched_param sp;
+	pthread_t pt = (pthread_t)thread;
+	int err;
+
+	spa_zero(sp);
+	if ((err = pthread_setschedparam(pt, SCHED_OTHER | PW_SCHED_RESET_ON_FORK, &sp)) != 0) {
+		pw_log_debug("thread %p: SCHED_OTHER|SCHED_RESET_ON_FORK failed: %s",
+				thread, strerror(err));
+		return -err;
+	}
+	pw_log_info("thread %p dropped realtime priority", thread);
+	return 0;
+}
+
+#ifdef HAVE_DBUS
+static struct thread *find_thread_by_pt(struct impl *impl, pthread_t pt)
+{
+	struct thread *t;
+
+	spa_list_for_each(t, &impl->threads_list, link) {
+		if (pthread_equal(t->thread, pt))
+			return t;
+	}
+	return NULL;
+}
+
+static void *custom_start(void *data)
+{
+	struct thread *this = data;
+	struct impl *impl = this->impl;
+
+	pthread_mutex_lock(&impl->lock);
+	this->pid = _gettid();
+	pthread_cond_broadcast(&impl->cond);
+	pthread_mutex_unlock(&impl->lock);
+
+	return this->start(this->arg);
+}
+
+static struct spa_thread *impl_create(void *object, const struct spa_dict *props,
+		void *(*start_routine)(void*), void *arg)
+{
+	struct impl *impl = object;
+	struct thread *this;
+	struct spa_thread *thread;
+
+	this = calloc(1, sizeof(*this));
+	this->impl = impl;
+	this->start = start_routine;
+	this->arg = arg;
+
+	/* This thread list is only used for the RTKit implementation */
+	pthread_mutex_lock(&impl->lock);
+	thread = pw_thread_utils_create(props, custom_start, this);
+	if (thread == NULL)
+		goto exit;
+
+	this->thread = (pthread_t)thread;
+	pthread_cond_wait(&impl->cond, &impl->lock);
+
+	spa_list_append(&impl->threads_list, &this->link);
+exit:
+	pthread_mutex_unlock(&impl->lock);
+
+	if (thread == NULL) {
+		free(this);
+		return NULL;
+	}
+	return thread;
+}
+
+static int impl_join(void *object, struct spa_thread *thread, void **retval)
+{
+	struct impl *impl = object;
+	pthread_t pt = (pthread_t)thread;
+	struct thread *thr;
+
+	pthread_mutex_lock(&impl->lock);
+	if ((thr = find_thread_by_pt(impl, pt)) != NULL) {
+		spa_list_remove(&thr->link);
+		free(thr);
+	}
+	pthread_mutex_unlock(&impl->lock);
+
+	return pthread_join(pt, retval);
+}
+
+
+static int get_rtkit_priority_range(struct impl *impl, int *min, int *max)
+{
+	if (min)
+		*min = 1;
+	if (max) {
+		if ((*max = pw_rtkit_get_max_realtime_priority(impl)) < 0)
+			return *max;
+		if (*max < 1)
+			*max = 1;
+	}
+	return 0;
+}
+
+static int impl_get_rt_range(void *object, const struct spa_dict *props,
+		int *min, int *max)
+{
+	struct impl *impl = object;
+	int res;
+	if (impl->use_rtkit)
+		res = get_rtkit_priority_range(impl, min, max);
+	else
+		res = get_rt_priority_range(min, max);
+	return res;
+}
+
+static pid_t impl_gettid(struct impl *impl, pthread_t pt)
+{
+	struct thread *thr;
+	pid_t pid;
+
+	pthread_mutex_lock(&impl->lock);
+	if ((thr = find_thread_by_pt(impl, pt)) != NULL)
+		pid = thr->pid;
+	else
+		pid = _gettid();
+	pthread_mutex_unlock(&impl->lock);
+
+	return pid;
+}
+
+static int impl_acquire_rt(void *object, struct spa_thread *thread, int priority)
+{
+	struct impl *impl = object;
+	struct sched_param sp;
+	int err;
+	pthread_t pt = (pthread_t)thread;
+	pid_t pid;
+
+	/* See the docstring on `spa_thread_utils_methods::acquire_rt` */
+	if (priority == -1) {
+		priority = impl->rt_prio;
+	}
+
+	if (impl->use_rtkit) {
+		int min, max;
+
+		if ((err = get_rtkit_priority_range(impl, &min, &max)) < 0)
+			return err;
+
+		pid = impl_gettid(impl, pt);
+
+		if (priority < min || priority > max) {
+			pw_log_info("clamping requested priority %d for thread %d "
+					"between %d  and %d", priority, pid, min, max);
+			priority = SPA_CLAMP(priority, min, max);
+		}
+
+		spa_zero(sp);
+		sp.sched_priority = priority;
+
+		if (pthread_setschedparam(pt, SCHED_OTHER | PW_SCHED_RESET_ON_FORK, &sp) == 0) {
+			pw_log_debug("SCHED_OTHER|SCHED_RESET_ON_FORK worked.");
+		}
+
+		if ((err = pw_rtkit_make_realtime(impl, pid, priority)) < 0) {
+			pw_log_warn("could not make thread %d realtime using RTKit: %s", pid, spa_strerror(err));
+			return err;
+		}
+
+		pw_log_info("acquired realtime priority %d for thread %d using RTKit", priority, pid);
+		return 0;
+	} else {
+		return acquire_rt_sched(thread, priority);
+	}
+}
+
+static const struct spa_thread_utils_methods impl_thread_utils = {
+	SPA_VERSION_THREAD_UTILS_METHODS,
+	.create = impl_create,
+	.join = impl_join,
+	.get_rt_range = impl_get_rt_range,
+	.acquire_rt = impl_acquire_rt,
+	.drop_rt = impl_drop_rt_generic,
+};
+
+#else /* HAVE_DBUS */
+
+static struct spa_thread *impl_create(void *object, const struct spa_dict *props,
+		void *(*start_routine)(void*), void *arg)
+{
+	return pw_thread_utils_create(props, start_routine, arg);
+}
+
+static int impl_join(void *object, struct spa_thread *thread, void **retval)
+{
+	return pw_thread_utils_join(thread, retval);
+}
+
+static int impl_get_rt_range(void *object, const struct spa_dict *props,
+		int *min, int *max)
+{
+	return get_rt_priority_range(min, max);
+}
+
+static int impl_acquire_rt(void *object, struct spa_thread *thread, int priority)
+{
+	struct impl *impl = object;
+
+	/* See the docstring on `spa_thread_utils_methods::acquire_rt` */
+	if (priority == -1)
+		priority = impl->rt_prio;
+
+	return acquire_rt_sched(thread, priority);
+}
+
+static const struct spa_thread_utils_methods impl_thread_utils = {
+	SPA_VERSION_THREAD_UTILS_METHODS,
+	.create = impl_create,
+	.join = impl_join,
+	.get_rt_range = impl_get_rt_range,
+	.acquire_rt = impl_acquire_rt,
+	.drop_rt = impl_drop_rt_generic,
+};
+#endif /* HAVE_DBUS */
+
+
+#ifdef HAVE_DBUS
+static int check_rtkit(struct impl *impl, struct pw_context *context, bool *can_use_rtkit)
+{
+	const struct pw_properties *context_props;
+	const char *str;
+
+	*can_use_rtkit = true;
+
+	if ((context_props = pw_context_get_properties(context)) != NULL &&
+	    (str = pw_properties_get(context_props, "support.dbus")) != NULL &&
+	    !pw_properties_parse_bool(str))
+		*can_use_rtkit = false;
+
+	return 0;
+}
+#endif /* HAVE_DBUS */
+
+SPA_EXPORT
+int pipewire__module_init(struct pw_impl_module *module, const char *args)
+{
+	struct pw_context *context = pw_impl_module_get_context(module);
+	struct impl *impl;
+	struct pw_properties *props;
+	int res = 0;
+
+	PW_LOG_TOPIC_INIT(mod_topic);
+
+	impl = calloc(1, sizeof(struct impl));
+	if (impl == NULL)
+		return -ENOMEM;
+
+	pw_log_debug("module %p: new", impl);
+
+	props = args ? pw_properties_new_string(args) : pw_properties_new(NULL, NULL);
+	if (!props) {
+		res = -errno;
+		goto error;
+	}
+
+	impl->context = context;
+	impl->nice_level = pw_properties_get_int32(props, "nice.level", DEFAULT_NICE_LEVEL);
+	impl->rt_prio = pw_properties_get_int32(props, "rt.prio", DEFAULT_RT_PRIO);
+	impl->rt_time_soft = pw_properties_get_int32(props, "rt.time.soft", DEFAULT_RT_TIME_SOFT);
+	impl->rt_time_hard = pw_properties_get_int32(props, "rt.time.hard", DEFAULT_RT_TIME_HARD);
+
+	bool can_use_rtkit = false, use_rtkit = false;
+
+#ifdef HAVE_DBUS
+	spa_list_init(&impl->threads_list);
+	pthread_mutex_init(&impl->lock, NULL);
+	pthread_cond_init(&impl->cond, NULL);
+
+	if ((res = check_rtkit(impl, context, &can_use_rtkit)) < 0)
+		goto error;
+#endif
+	/* If the user has permissions to use regular realtime scheduling, as well as
+	 * the nice level we want, then we'll use that instead of RTKit */
+	if (!check_realtime_privileges(impl)) {
+		if (!can_use_rtkit) {
+			res = -ENOTSUP;
+			pw_log_warn("regular realtime scheduling not available (RTKit fallback disabled)");
+			goto error;
+		}
+		use_rtkit = true;
+	}
+
+	if (IS_VALID_NICE_LEVEL(impl->nice_level)) {
+		if (set_nice(impl, impl->nice_level, !can_use_rtkit) < 0)
+			use_rtkit = can_use_rtkit;
+	}
+
+#ifdef HAVE_DBUS
+	impl->use_rtkit = use_rtkit;
+	if (impl->use_rtkit) {
+		/* Checking xdg-desktop-portal. It works fine in all situations. */
+		impl->rtkit_bus = pw_rtkit_bus_get_session();
+		if (impl->rtkit_bus != NULL) {
+			if (pw_rtkit_check_xdg_portal(impl->rtkit_bus)) {
+				impl->service_name = XDG_PORTAL_SERVICE_NAME;
+				impl->object_path = XDG_PORTAL_OBJECT_PATH;
+				impl->interface = XDG_PORTAL_INTERFACE;
+			} else {
+				pw_log_warn("found session bus but no portal");
+				pw_rtkit_bus_free(impl->rtkit_bus);
+				impl->rtkit_bus = NULL;
+			}
+		}
+		/* Failed to get xdg-desktop-portal, try to use rtkit. */
+		if (impl->rtkit_bus == NULL) {
+			impl->rtkit_bus = pw_rtkit_bus_get_system();
+			if (impl->rtkit_bus != NULL) {
+				impl->service_name = RTKIT_SERVICE_NAME;
+				impl->object_path = RTKIT_OBJECT_PATH;
+				impl->interface = RTKIT_INTERFACE;
+			} else {
+				res = -errno;
+				pw_log_warn("could not get system bus: %m");
+				goto error;
+			}
+		}
+		/* Retry set_nice with rtkit */
+		if (IS_VALID_NICE_LEVEL(impl->nice_level))
+			set_nice(impl, impl->nice_level, true);
+	}
+#endif
+	set_rlimit(impl);
+
+	impl->thread_utils.iface = SPA_INTERFACE_INIT(
+			SPA_TYPE_INTERFACE_ThreadUtils,
+			SPA_VERSION_THREAD_UTILS,
+			&impl_thread_utils, impl);
+
+	pw_context_set_object(context, SPA_TYPE_INTERFACE_ThreadUtils,
+			&impl->thread_utils);
+
+	pw_impl_module_add_listener(module, &impl->module_listener, &module_events, impl);
+
+	pw_impl_module_update_properties(module, &SPA_DICT_INIT_ARRAY(module_props));
+	pw_impl_module_update_properties(module, &props->dict);
+
+#ifdef HAVE_DBUS
+	if (impl->use_rtkit) {
+		pw_log_debug("initialized using RTKit");
+	} else {
+		pw_log_debug("initialized using regular realtime scheduling");
+	}
+#else
+	pw_log_debug("initialized using regular realtime scheduling");
+#endif
+
+	goto done;
+
+error:
+#ifdef HAVE_DBUS
+	if (impl->rtkit_bus)
+		pw_rtkit_bus_free(impl->rtkit_bus);
+#endif
+	free(impl);
+done:
+	pw_properties_free(props);
+
+	return res;
+}