audioconvert: set IO_Buffers only when buffers are negotiated

The IO_Buffers is used in the data thread to check if the port should be
scheduled or not. Make sure it is only set after we set buffers on the
port and cleared before the buffers are cleared.

Make sure we sync the port->io with the data thread.

See #4094

spa/plugins/audioconvert/audioadapter.c

@@ -268,23 +268,31 @@ static int link_io(struct impl *this)
 			res, spa_strerror(res));
 	}
 
+	return 0;
+}
+
+static int activate_io(struct impl *this, bool active)
+{
+	int res;
+	struct spa_io_buffers *data = active ? &this->io_buffers : NULL;
+	uint32_t size = active ? sizeof(this->io_buffers) : 0;
+
 	if (this->follower == this->target)
 		return 0;
 
-	this->io_buffers = SPA_IO_BUFFERS_INIT;
+	if (active)
+		this->io_buffers = SPA_IO_BUFFERS_INIT;
 
 	if ((res = spa_node_port_set_io(this->follower,
 			this->direction, 0,
-			SPA_IO_Buffers,
+			SPA_IO_Buffers, data, size)) < 0) {
-			&this->io_buffers, sizeof(this->io_buffers))) < 0) {
 		spa_log_warn(this->log, "%p: set Buffers on follower failed %d %s", this,
 			res, spa_strerror(res));
 		return res;
 	}
 	else if ((res = spa_node_port_set_io(this->convert,
 			SPA_DIRECTION_REVERSE(this->direction), 0,
-			SPA_IO_Buffers,
+			SPA_IO_Buffers, data, size)) < 0) {
-			&this->io_buffers, sizeof(this->io_buffers))) < 0) {
 		spa_log_warn(this->log, "%p: set Buffers on convert failed %d %s", this,
 			res, spa_strerror(res));
 		return res;
@@ -484,6 +492,8 @@ static int negotiate_buffers(struct impl *this)
 		       this->buffers, this->n_buffers)) < 0)
 		return res;
 
+	activate_io(this, true);
+
 	return 0;
 }
 
@@ -499,12 +509,20 @@ static int configure_format(struct impl *this, uint32_t flags, const struct spa_
 	uint8_t buffer[4096];
 	int res;
 
+	spa_log_debug(this->log, "%p: configure format:", this);
+
 	if (format == NULL && !this->have_format)
 		return 0;
 
-	spa_log_debug(this->log, "%p: configure format:", this);
+
-	if (format)
+	if (format == NULL) {
+		if (!this->have_format)
+			return 0;
+		activate_io(this, false);
+	}
+	else {
 		spa_debug_log_format(this->log, SPA_LOG_LEVEL_DEBUG, 0, NULL, format);
+	}
 
 	if ((res = spa_node_port_set_param(this->follower,
 					   this->direction, 0,

spa/plugins/audioconvert/audioconvert.c

@@ -193,6 +193,7 @@ struct impl {
 
 	struct spa_log *log;
 	struct spa_cpu *cpu;
+	struct spa_loop *data_loop;
 
 	uint32_t cpu_flags;
 	uint32_t max_align;
@@ -2448,7 +2449,7 @@ static int port_set_format(void *object,
 
 	port = GET_PORT(this, direction, port_id);
 
-	spa_log_debug(this->log, "%p: set format", this);
+	spa_log_debug(this->log, "%p: %d:%d set format", this, direction, port_id);
 
 	if (format == NULL) {
 		port->have_format = false;
@@ -2677,6 +2678,20 @@ impl_node_port_use_buffers(void *object,
 	return 0;
 }
 
+struct io_data {
+	struct port *port;
+	void *data;
+	size_t size;
+};
+
+static int do_set_port_io(struct spa_loop *loop, bool async, uint32_t seq,
+		const void *data, size_t size, void *user_data)
+{
+	const struct io_data *d = user_data;
+	d->port->io = d->data;
+	return 0;
+}
+
 static int
 impl_node_port_set_io(void *object,
 		      enum spa_direction direction, uint32_t port_id,
@@ -2696,7 +2711,12 @@ impl_node_port_set_io(void *object,
 
 	switch (id) {
 	case SPA_IO_Buffers:
-		port->io = data;
+		if (this->data_loop) {
+			struct io_data d = { .port = port, .data = data, .size = size };
+			spa_loop_invoke(this->data_loop, do_set_port_io, 0, NULL, 0, true, &d);
+		}
+		else
+			port->io = data;
 		break;
 	case SPA_IO_RateMatch:
 		this->io_rate_match = data;
@@ -3432,6 +3452,7 @@ impl_init(const struct spa_handle_factory *factory,
 
 	this = (struct impl *) handle;
 
+	this->data_loop = spa_support_find(support, n_support, SPA_TYPE_INTERFACE_DataLoop);
 	this->log = spa_support_find(support, n_support, SPA_TYPE_INTERFACE_Log);
 	spa_log_topic_init(this->log, &log_topic);