session-manager: split policy and monitor

Separate the session manager in a monitor and policy part.

The monitor manages the devices and endpoints.

The policy watches the nodes/ports/clients and applies the policy
of linking them.

Because both now have a separate connection, we can remove some
hacks in the protocol. When a remote was both the implementer and
user of an object we could get in a deadlock when the user was
blocked waiting and the implementator was blocked sending a reply.
We used to un-busy a client when it was expecting a reply from a
ping or sync for this reason.

Add and use some more keys for the endpoints and streams.

src/examples/media-session/alsa-endpoint.c

@@ -44,25 +44,49 @@
 #include "pipewire/pipewire.h"
 #include "pipewire/private.h"
 
+struct endpoint {
+	struct spa_list link;
+
+	struct pw_properties *props;
+	struct alsa_object *obj;
+
+	struct pw_client_endpoint_proxy *client_endpoint;
+	struct spa_hook client_endpoint_listener;
+	struct pw_endpoint_info client_endpoint_info;
+
+	unsigned int use_ucm:1;
+	snd_use_case_mgr_t *ucm;
+
+	struct spa_list stream_list;
+};
+
+struct stream {
+	struct spa_list link;
+
+	struct pw_properties *props;
+	struct pw_endpoint_stream_info info;
+};
+
 static int client_endpoint_set_id(void *object, uint32_t id)
 {
-	struct alsa_object *obj = object;
+	struct endpoint *endpoint = object;
+	struct alsa_object *obj = endpoint->obj;
 
-	obj->client_endpoint_info.id = id;
-	obj->client_endpoint_info.name = (char*)pw_properties_get(obj->props, PW_KEY_DEVICE_DESCRIPTION);
-	obj->client_endpoint_info.media_class = (char*)pw_properties_get(obj->props, PW_KEY_MEDIA_CLASS);
+	endpoint->client_endpoint_info.id = id;
+	endpoint->client_endpoint_info.name = (char*)pw_properties_get(endpoint->props, PW_KEY_ENDPOINT_NAME);
+	endpoint->client_endpoint_info.media_class = (char*)pw_properties_get(obj->props, PW_KEY_MEDIA_CLASS);
 
-	pw_client_endpoint_proxy_update(obj->client_endpoint,
+	pw_client_endpoint_proxy_update(endpoint->client_endpoint,
 			PW_CLIENT_ENDPOINT_UPDATE_INFO,
 			0, NULL,
-			&obj->client_endpoint_info);
+			&endpoint->client_endpoint_info);
 	return 0;
 }
 
 static int client_endpoint_set_session_id(void *object, uint32_t id)
 {
-	struct alsa_object *obj = object;
-	obj->client_endpoint_info.session_id = id;
+	struct endpoint *endpoint = object;
+	endpoint->client_endpoint_info.session_id = id;
 	return 0;
 }
 
@@ -89,24 +113,44 @@ static const struct pw_client_endpoint_proxy_events client_endpoint_events = {
 };
 
 /** fallback, one stream for each node */
-static int setup_alsa_fallback_endpoint(struct alsa_object *obj)
+static int setup_alsa_fallback_endpoint(struct endpoint *endpoint)
 {
+	struct alsa_object *obj = endpoint->obj;
 	struct alsa_node *n;
+	const char *str;
 
 	spa_list_for_each(n, &obj->node_list, link) {
-		n->info.version = PW_VERSION_ENDPOINT_STREAM_INFO;
-		n->info.id = n->id;
-		n->info.endpoint_id = obj->client_endpoint_info.id;
-		n->info.name = (char*)pw_properties_get(n->props, PW_KEY_NODE_DESCRIPTION);
-		n->info.change_mask = PW_ENDPOINT_STREAM_CHANGE_MASK_PROPS;
-		n->info.props = &n->props->dict;
+		struct stream *s;
+
+		s = calloc(1, sizeof(*s));
+		if (s == NULL)
+			return -errno;
+
+		spa_list_append(&endpoint->stream_list, &s->link);
+
+		s->props = pw_properties_new(NULL, NULL);
+		if ((str = pw_properties_get(n->props, PW_KEY_MEDIA_CLASS)) != NULL)
+			pw_properties_set(s->props, PW_KEY_MEDIA_CLASS, str);
+		if ((str = pw_properties_get(n->props, PW_KEY_PRIORITY_SESSION)) != NULL)
+			pw_properties_set(s->props, PW_KEY_PRIORITY_SESSION, str);
+		if ((str = pw_properties_get(n->props, PW_KEY_NODE_NAME)) != NULL)
+			pw_properties_set(s->props, PW_KEY_STREAM_NAME, str);
+		if ((str = pw_properties_get(n->props, PW_KEY_NODE_DESCRIPTION)) != NULL)
+			pw_properties_set(s->props, PW_KEY_STREAM_DESCRIPTION, str);
+
+		s->info.version = PW_VERSION_ENDPOINT_STREAM_INFO;
+		s->info.id = n->id;
+		s->info.endpoint_id = endpoint->client_endpoint_info.id;
+		s->info.name = (char*)pw_properties_get(s->props, PW_KEY_STREAM_DESCRIPTION);
+		s->info.change_mask = PW_ENDPOINT_STREAM_CHANGE_MASK_PROPS;
+		s->info.props = &s->props->dict;
 
 		pw_log_debug("stream %d", n->id);
-		pw_client_endpoint_proxy_stream_update(obj->client_endpoint,
+		pw_client_endpoint_proxy_stream_update(endpoint->client_endpoint,
 				n->id,
 				PW_CLIENT_ENDPOINT_STREAM_UPDATE_INFO,
 				0, NULL,
-				&n->info);
+				&s->info);
 	}
 	return 0;
 }
@@ -115,8 +159,9 @@ static int setup_alsa_fallback_endpoint(struct alsa_object *obj)
  *
  * We create 1 stream for each verb + modifier combination
  */
-static int setup_alsa_ucm_endpoint(struct alsa_object *obj)
+static int setup_alsa_ucm_endpoint(struct endpoint *endpoint)
 {
+	struct alsa_object *obj = endpoint->obj;
 	const char *str, *card_name = NULL;
 	char *name_free = NULL;
 	int i, res, num_verbs;
@@ -134,12 +179,12 @@ static int setup_alsa_ucm_endpoint(struct alsa_object *obj)
 		goto exit;
 	}
 
-	if ((res = snd_use_case_mgr_open(&obj->ucm, card_name)) < 0) {
+	if ((res = snd_use_case_mgr_open(&endpoint->ucm, card_name)) < 0) {
 		pw_log_error("can not open UCM for %s: %s", card_name, snd_strerror(res));
 		goto exit;
 	}
 
-	num_verbs = snd_use_case_verb_list(obj->ucm, &verb_list);
+	num_verbs = snd_use_case_verb_list(endpoint->ucm, &verb_list);
 	if (num_verbs < 0) {
 		res = num_verbs;
 		pw_log_error("UCM verb list not found for %s: %s", card_name, snd_strerror(num_verbs));
@@ -150,15 +195,15 @@ static int setup_alsa_ucm_endpoint(struct alsa_object *obj)
 		pw_log_debug("verb: %s", verb_list[i]);
 	}
 
-	obj->use_ucm = true;
+	endpoint->use_ucm = true;
 
 	snd_use_case_free_list(verb_list, num_verbs);
 
 	return 0;
 close_exit:
-	snd_use_case_mgr_close(obj->ucm);
+	snd_use_case_mgr_close(endpoint->ucm);
 exit:
-	obj->ucm = NULL;
+	endpoint->ucm = NULL;
 	free(name_free);
 	return res;
 
@@ -168,24 +213,48 @@ static int setup_alsa_endpoint(struct alsa_object *obj)
 {
 	struct impl *impl = obj->monitor->impl;
 	int res;
+	struct pw_proxy *proxy;
+	struct endpoint *endpoint;
+	struct pw_properties *props;
+	const char *str;
 
-	obj->client_endpoint = pw_core_proxy_create_object(impl->core_proxy,
+	props = pw_properties_new(NULL, NULL);
+	if (props == NULL)
+		return -errno;
+
+	if ((str = pw_properties_get(obj->props, PW_KEY_MEDIA_CLASS)) != NULL)
+		pw_properties_set(props, PW_KEY_MEDIA_CLASS, str);
+	if ((str = pw_properties_get(obj->props, PW_KEY_DEVICE_DESCRIPTION)) != NULL)
+		pw_properties_set(props, PW_KEY_ENDPOINT_NAME, str);
+	if ((str = pw_properties_get(obj->props, PW_KEY_DEVICE_ICON_NAME)) != NULL)
+		pw_properties_set(props, PW_KEY_ENDPOINT_ICON_NAME, str);
+
+	proxy = pw_core_proxy_create_object(impl->core_proxy,
 						"client-endpoint",
 						PW_TYPE_INTERFACE_ClientEndpoint,
 						PW_VERSION_CLIENT_ENDPOINT_PROXY,
-						&obj->props->dict, 0);
+						&props->dict, sizeof(*endpoint));
+	if (proxy == NULL) {
+		pw_properties_free(props);
+		return -errno;
+	}
 
-	obj->client_endpoint_info.version = PW_VERSION_ENDPOINT_INFO;
-	obj->client_endpoint_info.name = "name";
-	obj->client_endpoint_info.media_class = "media-class";
+	endpoint = pw_proxy_get_user_data(proxy);
+	endpoint->obj = obj;
+	endpoint->props = props;
+	endpoint->client_endpoint = (struct pw_client_endpoint_proxy *) proxy;
+	endpoint->client_endpoint_info.version = PW_VERSION_ENDPOINT_INFO;
+	endpoint->client_endpoint_info.name = "name";
+	endpoint->client_endpoint_info.media_class = "media-class";
+	spa_list_init(&endpoint->stream_list);
 
-	pw_client_endpoint_proxy_add_listener(obj->client_endpoint,
-			&obj->client_endpoint_listener,
+	pw_client_endpoint_proxy_add_listener(endpoint->client_endpoint,
+			&endpoint->client_endpoint_listener,
 			&client_endpoint_events,
-			obj);
+			endpoint);
 
-	if ((res = setup_alsa_ucm_endpoint(obj)) < 0)
-		res = setup_alsa_fallback_endpoint(obj);
+	if ((res = setup_alsa_ucm_endpoint(endpoint)) < 0)
+		res = setup_alsa_fallback_endpoint(endpoint);
 
 	return res;
 }

src/examples/media-session/alsa-monitor.c

@@ -60,8 +60,6 @@ struct alsa_node {
 
 	struct pw_proxy *proxy;
 	struct spa_node *node;
-
-	struct pw_endpoint_stream_info info;
 };
 
 struct alsa_object {
@@ -84,13 +82,6 @@ struct alsa_object {
 
 	unsigned int first:1;
 	struct spa_list node_list;
-
-	struct pw_client_endpoint_proxy *client_endpoint;
-	struct spa_hook client_endpoint_listener;
-	struct pw_endpoint_info client_endpoint_info;
-
-	unsigned int use_ucm:1;
-	snd_use_case_mgr_t *ucm;
 };
 
 static struct alsa_node *alsa_find_node(struct alsa_object *obj, uint32_t id)
@@ -447,7 +438,6 @@ static void reserve_acquired(void *data, struct rd_device *d)
 	set_profile(obj, 1);
 
 	setup_alsa_endpoint(obj);
-
 }
 
 static void sync_complete_done(void *data, int seq)

src/examples/meson.build

@@ -46,6 +46,8 @@ executable('export-spa-device',
 
 executable('media-session',
   'media-session/media-session.c',
+  'media-session/monitor.c',
+  'media-session/policy.c',
   c_args : [ '-D_GNU_SOURCE' ],
   install: false,
   dependencies : [dbus_dep, pipewire_dep, alsa_dep, mathlib],

src/extensions/session-manager/introspect.h

@@ -34,9 +34,6 @@
 extern "C" {
 #endif
 
-#define PW_KEY_ENDPOINT_ID	"endpoint.id"
-#define PW_KEY_SESSION_ID	"session.id"
-
 enum pw_endpoint_direction {
 	PW_ENDPOINT_DIRECTION_SINK_INPUT = SPA_DIRECTION_INPUT,
 	PW_ENDPOINT_DIRECTION_SOURCE_OUTPUT = SPA_DIRECTION_OUTPUT,

src/extensions/session-manager/keys.h

@@ -30,8 +30,16 @@
 extern "C" {
 #endif
 
-#define PW_KEY_ENDPOINT_ID	"endpoint.id"
-#define PW_KEY_SESSION_ID	"session.id"
+#define PW_KEY_SESSION_ID		"session.id"		/**< id of a session manager */
+
+#define PW_KEY_ENDPOINT_ID		"endpoint.id"		/**< id of an endpoint */
+#define PW_KEY_ENDPOINT_NAME		"endpoint.name"		/**< the name of an endpoint */
+#define PW_KEY_ENDPOINT_ICON_NAME	"endpoint.icon-name"	/**< an XDG icon name for the device.
+								  *  Ex. "sound-card-speakers-usb" */
+
+#define PW_KEY_STREAM_ID		"stream.id"		/**< id of a stream */
+#define PW_KEY_STREAM_NAME		"stream.name"		/**< unique name of a stream */
+#define PW_KEY_STREAM_DESCRIPTION	"stream.description"	/**< description of a stream */
 
 #ifdef __cplusplus
 }

src/modules/module-client-device/protocol-native.c

@@ -88,7 +88,6 @@ static int device_marshal_sync(void *object, int seq)
 	struct pw_resource *resource = object;
 	struct spa_pod_builder *b;
 
-	pw_client_set_busy(pw_resource_get_client(resource), false);
 	b = pw_protocol_native_begin_resource(resource, SPA_DEVICE_METHOD_SYNC, &msg);
 
 	spa_pod_builder_add_struct(b,

src/modules/module-protocol-native.c

@@ -1007,7 +1007,7 @@ int pipewire__module_init(struct pw_module *module, const char *args)
 	pw_protocol_native_init(this);
 	pw_protocol_native0_init(this);
 
-	pw_log_debug(NAME" %p: new %d", this, debug_messages);
+	pw_log_debug(NAME" %p: new debug:%d", this, debug_messages);
 
 	d = pw_protocol_get_user_data(this);
 	d->protocol = this;

src/modules/module-protocol-native/protocol-native.c

@@ -416,7 +416,6 @@ static void core_event_marshal_ping(void *object, uint32_t id, int seq)
 	struct spa_pod_builder *b;
 	struct pw_protocol_native_message *msg;
 
-	pw_client_set_busy(pw_resource_get_client(resource), false);
 	b = pw_protocol_native_begin_resource(resource, PW_CORE_PROXY_EVENT_PING, &msg);
 
 	spa_pod_builder_add_struct(b,

src/modules/module-session-manager/client-endpoint.c

@@ -85,6 +85,8 @@ static int client_endpoint_stream_update(void *object,
 			PW_KEY_ENDPOINT_ID,
 			PW_KEY_MEDIA_CLASS,
 			PW_KEY_PRIORITY_SESSION,
+			PW_KEY_STREAM_NAME,
+			PW_KEY_STREAM_DESCRIPTION,
 			NULL
 		};
 

src/modules/module-session-manager/endpoint.c

@@ -288,6 +288,8 @@ int endpoint_init(struct endpoint *this,
 		PW_KEY_CLIENT_ID,
 		PW_KEY_DEVICE_ID,
 		PW_KEY_MEDIA_CLASS,
+		PW_KEY_ENDPOINT_NAME,
+		PW_KEY_ENDPOINT_ICON_NAME,
 		NULL
 	};