pulse-server: fill in active port for monitors

The monitor sources also list the port of the sink and so the active
port needs to be collected as well so it doesn't fall back to the first
port (which might not be available).

pipewire-jack/src/pipewire-jack.c

@@ -1210,6 +1210,11 @@ static void on_error(void *data, uint32_t id, int seq, int res, const char *mess
 			id, seq, res, spa_strerror(res), message);
 
 	if (id == PW_ID_CORE) {
+		/* This happens when we did something on a proxy that
+		 * was destroyed on the server already */
+		if (res == -ENOENT)
+			return;
+
 		client->last_res = res;
 		if (res == -EPIPE && !client->destroyed) {
 			queue_notify(client, NOTIFY_TYPE_SHUTDOWN,
@@ -4064,6 +4069,7 @@ server_failed:
 exit_unlock:
 	pw_thread_loop_unlock(client->context.loop);
 exit:
+	pw_log_info("%p: error %d", client, *status);
 	jack_client_close((jack_client_t *) client);
 	return NULL;
 disabled:

spa/plugins/alsa/alsa-pcm.c

@@ -2181,8 +2181,6 @@ static int set_swparams(struct state *state)
 	/* start the transfer */
 	CHECK(snd_pcm_sw_params_set_start_threshold(hndl, params, LONG_MAX), "set_start_threshold");
 
-	CHECK(snd_pcm_sw_params_set_period_event(hndl, params, state->disable_tsched), "set_period_event");
-
 	if (state->disable_tsched) {
 		snd_pcm_uframes_t avail_min;
 
@@ -2532,15 +2530,6 @@ static int get_status(struct state *state, uint64_t current_time, snd_pcm_uframe
 	return 0;
 }
 
-
-static uint64_t get_time_ns(struct state *state)
-{
-	struct timespec now;
-	if (spa_system_clock_gettime(state->data_system, CLOCK_MONOTONIC, &now) < 0)
-		return 0;
-	return SPA_TIMESPEC_TO_NSEC(&now);
-}
-
 static int update_time(struct state *state, uint64_t current_time, snd_pcm_sframes_t delay,
 		snd_pcm_sframes_t target, bool follower)
 {
@@ -2548,15 +2537,8 @@ static int update_time(struct state *state, uint64_t current_time, snd_pcm_sfram
 	int32_t diff;
 
 	if (state->disable_tsched && !follower) {
-		uint64_t now = get_time_ns(state);
-
-		if (SPA_UNLIKELY(state->dll.bw == 0.0)) {
-			current_time = now;
-			err = 0.0;
-		} else {
-			err = (int64_t)(now - current_time);
-			err = err / 1e9 * state->rate;
-		}
+		err = (int64_t)(current_time - state->next_time);
+		err = err / 1e9 * state->rate;
 	} else {
 		if (state->stream == SND_PCM_STREAM_PLAYBACK)
 			err = delay - target;
@@ -3183,6 +3165,14 @@ static int capture_ready(struct state *state)
 	return 0;
 }
 
+static uint64_t get_time_ns(struct state *state)
+{
+	struct timespec now;
+	if (spa_system_clock_gettime(state->data_system, CLOCK_MONOTONIC, &now) < 0)
+		return 0;
+	return SPA_TIMESPEC_TO_NSEC(&now);
+}
+
 static void alsa_wakeup_event(struct spa_source *source)
 {
 	struct state *state = source->data, *follower;
@@ -3194,6 +3184,8 @@ static void alsa_wakeup_event(struct spa_source *source)
 		int err;
 		unsigned short revents;
 
+		current_time = get_time_ns(state);
+
 		for (int i = 0; i < state->n_fds; i++) {
 			state->pfds[i].revents = state->source[i].rmask;
 			/* Reset so that we only handle all our sources' events once */
@@ -3229,8 +3221,8 @@ static void alsa_wakeup_event(struct spa_source *source)
 				return;
 			}
 		}
+		current_time = state->next_time;
 	}
-	current_time = state->next_time;
 
 	/* first do all the sync */
 	if (state->stream == SND_PCM_STREAM_CAPTURE)

spa/plugins/audioconvert/audioadapter.c

@@ -89,13 +89,14 @@ struct impl {
 	unsigned int add_listener:1;
 	unsigned int have_format:1;
 	unsigned int started:1;
-	unsigned int warned:1;
 	unsigned int ready:1;
-	unsigned int driver:1;
 	unsigned int async:1;
 	unsigned int passthrough:1;
 	unsigned int follower_removing:1;
 	unsigned int in_recalc;
+
+	unsigned int warned:1;
+	unsigned int driver:1;
 };
 
 /** \endcond */
@@ -926,15 +927,9 @@ static int impl_node_send_command(void *object, const struct spa_command *comman
 		this->warned = false;
 		break;
 	case SPA_NODE_COMMAND_Suspend:
-		this->started = false;
-		this->ready = false;
-		this->warned = false;
 		spa_log_debug(this->log, "%p: suspending", this);
 		break;
 	case SPA_NODE_COMMAND_Pause:
-		this->started = false;
-		this->ready = false;
-		this->warned = false;
 		spa_log_debug(this->log, "%p: pausing", this);
 		break;
 	case SPA_NODE_COMMAND_Flush:
@@ -971,9 +966,15 @@ static int impl_node_send_command(void *object, const struct spa_command *comman
 		break;
 	case SPA_NODE_COMMAND_Suspend:
 		configure_format(this, 0, NULL);
+		this->started = false;
+		this->warned = false;
+		this->ready = false;
 		spa_log_debug(this->log, "%p: suspended", this);
 		break;
 	case SPA_NODE_COMMAND_Pause:
+		this->started = false;
+		this->warned = false;
+		this->ready = false;
 		spa_log_debug(this->log, "%p: paused", this);
 		break;
 	case SPA_NODE_COMMAND_Flush:
@@ -1571,7 +1572,7 @@ static int impl_node_process(void *object)
 	struct impl *this = object;
 	int status = 0, fstatus, retry = 8;
 
-	if (!this->started) {
+	if (!this->ready) {
 		if (!this->warned)
 			spa_log_warn(this->log, "%p: scheduling stopped node", this);
 		this->warned = true;

src/gst/gstpipewiredeviceprovider.c

@@ -180,9 +180,11 @@ struct node_data {
   struct pw_node_info *info;
   GstCaps *caps;
   GstDevice *dev;
+  struct spa_list ports;
 };
 
 struct port_data {
+  struct spa_list link;
   struct node_data *node_data;
   struct pw_port *proxy;
   struct spa_hook proxy_listener;
@@ -353,6 +355,9 @@ static void port_event_info(void *data, const struct pw_port_info *info)
 
   pw_log_debug("%p", port_data);
 
+  if (node_data == NULL)
+    return;
+
   if (info->change_mask & PW_PORT_CHANGE_MASK_PARAMS) {
     for (i = 0; i < info->n_params; i++) {
       uint32_t id = info->params[i].id;
@@ -375,6 +380,9 @@ static void port_event_param(void *data, int seq, uint32_t id,
   struct node_data *node_data = port_data->node_data;
   GstCaps *c1;
 
+  if (node_data == NULL)
+    return;
+
   c1 = gst_caps_from_format (param);
   if (c1 && node_data->caps)
       gst_caps_append (node_data->caps, c1);
@@ -438,11 +446,17 @@ static void
 destroy_node (void *data)
 {
   struct node_data *nd = data;
+  struct port_data *pd;
   GstPipeWireDeviceProvider *self = nd->self;
   GstDeviceProvider *provider = GST_DEVICE_PROVIDER (self);
 
   pw_log_debug("destroy %p", nd);
 
+  spa_list_consume(pd, &nd->ports, link) {
+	  spa_list_remove(&pd->link);
+	  pd->node_data = NULL;
+  }
+
   if (nd->dev != NULL) {
     gst_device_provider_device_remove (provider, GST_DEVICE (nd->dev));
   }
@@ -472,6 +486,7 @@ destroy_port (void *data)
 {
   struct port_data *pd = data;
   pw_log_debug("destroy %p", pd);
+  spa_list_remove(&pd->link);
 }
 
 static const struct pw_proxy_events proxy_port_events = {
@@ -515,6 +530,7 @@ static void registry_event_global(void *data, uint32_t id, uint32_t permissions,
     nd->id = id;
     if (!props || !spa_atou64(spa_dict_lookup(props, PW_KEY_OBJECT_SERIAL), &nd->serial, 0))
       nd->serial = SPA_ID_INVALID;
+    spa_list_init(&nd->ports);
     spa_list_append(&self->nodes, &nd->link);
     pw_node_add_listener(node, &nd->node_listener, &node_events, nd);
     pw_proxy_add_listener((struct pw_proxy*)node, &nd->proxy_listener, &proxy_node_events, nd);
@@ -541,6 +557,7 @@ static void registry_event_global(void *data, uint32_t id, uint32_t permissions,
     pd->id = id;
     if (!props || !spa_atou64(spa_dict_lookup(props, PW_KEY_OBJECT_SERIAL), &pd->serial, 0))
       pd->serial = SPA_ID_INVALID;
+    spa_list_append(&nd->ports, &pd->link);
     pw_port_add_listener(port, &pd->port_listener, &port_events, pd);
     pw_proxy_add_listener((struct pw_proxy*)port, &pd->proxy_listener, &proxy_port_events, pd);
     resync(self);
@@ -581,7 +598,6 @@ gst_pipewire_device_provider_probe (GstDeviceProvider * provider)
   pw_thread_loop_lock (self->core->loop);
 
   spa_list_init(&self->nodes);
-  spa_list_init(&self->pending);
   self->end = FALSE;
   self->error = 0;
   self->list_only = TRUE;
@@ -631,7 +647,6 @@ gst_pipewire_device_provider_start (GstDeviceProvider * provider)
   pw_thread_loop_lock (self->core->loop);
 
   spa_list_init(&self->nodes);
-  spa_list_init(&self->pending);
   self->end = FALSE;
   self->error = 0;
   self->list_only = FALSE;

src/gst/gstpipewiredeviceprovider.h

@@ -70,7 +70,6 @@ struct _GstPipeWireDeviceProvider {
   struct pw_registry *registry;
   struct spa_hook registry_listener;
   struct spa_list nodes;
-  struct spa_list pending;
   int seq;
 
   int error;

src/modules/module-pipe-tunnel.c

@@ -187,7 +187,6 @@ struct impl {
 
 	unsigned int do_disconnect:1;
 	unsigned int driving:1;
-	unsigned int have_sync:1;
 	unsigned int may_pause:1;
 	unsigned int paused:1;
 
@@ -203,6 +202,7 @@ struct impl {
 	float corr;
 
 	uint64_t next_time;
+	unsigned int have_sync:1;
 };
 
 static uint64_t get_time_ns(struct impl *impl)

src/modules/module-protocol-pulse/collect.c

@@ -234,7 +234,7 @@ static void collect_device_info(struct pw_manager_object *device, struct pw_mana
 {
 	struct pw_manager_param *p;
 
-	if (card && !monitor) {
+	if (card) {
 		spa_list_for_each(p, &card->param_list, link) {
 			uint32_t index, dev;
 			struct spa_pod *props;
@@ -251,7 +251,7 @@ static void collect_device_info(struct pw_manager_object *device, struct pw_mana
 			if (dev != dev_info->device)
 				continue;
 			dev_info->active_port = index;
-			if (props) {
+			if (props && !monitor) {
 				volume_parse_param(props, &dev_info->volume_info, monitor);
 				dev_info->have_volume = true;
 			}

src/pipewire/data-loop.c

@@ -188,7 +188,12 @@ int pw_data_loop_start(struct pw_data_loop *loop)
 
 		if ((utils = loop->thread_utils) == NULL)
 			utils = pw_thread_utils_get();
-		thr = spa_thread_utils_create(utils, NULL, do_loop, loop);
+
+		static const struct spa_dict_item items[] = {
+			{ SPA_KEY_THREAD_NAME, "pw-data-loop" },
+		};
+
+		thr = spa_thread_utils_create(utils, &SPA_DICT_INIT_ARRAY(items), do_loop, loop);
 		loop->thread = (pthread_t)thr;
 		if (thr == NULL) {
 			pw_log_error("%p: can't create thread: %m", loop);