Merge branch 'videoconvert' into 'master'

Draft: videoconvert: Add vulkan converter See merge request pipewire/pipewire!1704
2026-05-30 21:37:53 -04:00 · 2024-01-22 13:39:40 +00:00 · 2024-01-22 13:39:40 +00:00 · 4caff02d86
commit 4caff02d86
parent 846f994ac8 092d01240b
37 changed files with 6002 additions and 424 deletions
--- a/spa/plugins/videoconvert/videoadapter.c
+++ b/spa/plugins/videoconvert/videoadapter.c
@ -28,9 +28,14 @@
 #define SPA_LOG_TOPIC_DEFAULT &log_topic
 SPA_LOG_TOPIC_DEFINE_STATIC(log_topic, "spa.videoadapter");

+#if 1
+#define VIDEOCONVERTER 1
+#define VIDEOCONVERTER_FACTORY spa_videoconvert_vulkan_factory
+#endif
+
 #define DEFAULT_ALIGN	16

-#define MAX_PORTS	1
+#define MAX_PORTS	(1+1)

 /** \cond */

@ -48,14 +53,15 @@ struct impl {

 	struct spa_node *follower;
 	struct spa_hook follower_listener;
-	uint32_t follower_flags;
+	uint64_t follower_flags;
 	struct spa_video_info follower_current_format;
 	struct spa_video_info default_format;
+	int in_set_param;

 	struct spa_handle *hnd_convert;
 	struct spa_node *convert;
 	struct spa_hook convert_listener;
-	uint32_t convert_flags;
+	uint64_t convert_flags;

 	uint32_t n_buffers;
 	struct spa_buffer **buffers;
@ -78,6 +84,7 @@ struct impl {
 	struct spa_param_info params[N_NODE_PARAMS];
 	uint32_t convert_params_flags[N_NODE_PARAMS];
 	uint32_t follower_params_flags[N_NODE_PARAMS];
+	uint64_t follower_port_flags;

 	struct spa_hook_list hooks;
 	struct spa_callbacks callbacks;
@ -85,6 +92,8 @@ 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;
@ -120,6 +129,27 @@ static int follower_enum_params(struct impl *this,
 	return 0;
 }

+static int convert_enum_port_config(struct impl *this,
+		int seq, uint32_t id, uint32_t start, uint32_t num,
+		const struct spa_pod *filter, struct spa_pod_builder *builder)
+{
+	struct spa_pod *f1, *f2 = NULL;
+	int res;
+
+	f1 = spa_pod_builder_add_object(builder,
+		SPA_TYPE_OBJECT_ParamPortConfig, id,
+			SPA_PARAM_PORT_CONFIG_direction, SPA_POD_Id(this->direction));
+
+	if (filter) {
+		if ((res = spa_pod_filter(builder, &f2, f1, filter)) < 0)
+			return res;
+	}
+	else {
+		f2 = f1;
+	}
+	return spa_node_enum_params(this->convert, seq, id, start, num, f2);
+}
+
 static int impl_node_enum_params(void *object, int seq,
 				 uint32_t id, uint32_t start, uint32_t num,
 				 const struct spa_pod *filter)
@ -145,11 +175,30 @@ next:

 	switch (id) {
 	case SPA_PARAM_EnumPortConfig:
-	case SPA_PARAM_PortConfig:
 		if (this->convert == NULL)
 			return 0;
-		res = spa_node_enum_params(this->convert, seq, id, start, num, filter);
-		return res;
+		return convert_enum_port_config(this, seq, id, start, num, filter, &b.b);
+	case SPA_PARAM_PortConfig:
+		if (this->passthrough) {
+			switch (result.index) {
+			case 0:
+				result.param = spa_pod_builder_add_object(&b.b,
+					SPA_TYPE_OBJECT_ParamPortConfig, id,
+					SPA_PARAM_PORT_CONFIG_direction, SPA_POD_Id(this->direction),
+					SPA_PARAM_PORT_CONFIG_mode,      SPA_POD_Id(
+						SPA_PARAM_PORT_CONFIG_MODE_passthrough));
+				result.next++;
+				res = 1;
+				break;
+			default:
+				return 0;
+			}
+		} else {
+			if (this->convert == NULL)
+				return 0;
+			return convert_enum_port_config(this, seq, id, start, num, filter, &b.b);
+		}
+		break;
 	case SPA_PARAM_PropInfo:
 		res = follower_enum_params(this,
 				id, IDX_PropInfo, &result, filter, &b.b);
@ -200,21 +249,6 @@ static int link_io(struct impl *this)
 	spa_zero(this->io_rate_match);
 	this->io_rate_match.rate = 1.0;

-	if ((res = spa_node_port_set_io(this->follower,
-			this->direction, 0,
-			SPA_IO_RateMatch,
-			&this->io_rate_match, sizeof(this->io_rate_match))) < 0) {
-		spa_log_debug(this->log, "%p: set RateMatch on follower disabled %d %s", this,
-			res, spa_strerror(res));
-	}
-	else if ((res = spa_node_port_set_io(this->convert,
-			SPA_DIRECTION_REVERSE(this->direction), 0,
-			SPA_IO_RateMatch,
-			&this->io_rate_match, sizeof(this->io_rate_match))) < 0) {
-		spa_log_warn(this->log, "%p: set RateMatch on convert failed %d %s", this,
-			res, spa_strerror(res));
-	}
-
 	this->io_buffers = SPA_IO_BUFFERS_INIT;

 	if ((res = spa_node_port_set_io(this->follower,
@ -280,7 +314,7 @@ static int debug_params(struct impl *this, struct spa_node *node,

 	if (filter) {
 		spa_log_error(this->log, "with this filter:");
-		spa_debug_log_pod(this->log, SPA_LOG_LEVEL_DEBUG, 2, NULL, filter);
+		spa_debug_log_pod(this->log, SPA_LOG_LEVEL_ERROR, 2, NULL, filter);
 	} else {
 		spa_log_error(this->log, "there was no filter");
 	}
@ -298,7 +332,7 @@ static int debug_params(struct impl *this, struct spa_node *node,
 			break;
 		}
 		spa_log_error(this->log, "unmatched %s %d:", debug, count);
-		spa_debug_log_pod(this->log, SPA_LOG_LEVEL_DEBUG, 2, NULL, param);
+		spa_debug_log_pod(this->log, SPA_LOG_LEVEL_ERROR, 2, NULL, param);
 		count++;
 	}
 	if (count == 0)
@ -318,13 +352,15 @@ static int negotiate_buffers(struct impl *this)
 	uint32_t i, size, buffers, blocks, align, flags, stride = 0;
 	uint32_t *aligns;
 	struct spa_data *datas;
-	uint32_t follower_flags, conv_flags;
+	uint64_t follower_flags, conv_flags;

 	spa_log_debug(this->log, "%p: %d", this, this->n_buffers);

 	if (this->target == this->follower)
 		return 0;

+	spa_log_debug(this->log, "%p: n_buffers:%d", this, this->n_buffers);
+
 	if (this->n_buffers > 0)
 		return 0;

@ -478,7 +514,7 @@ static int configure_convert(struct impl *this, uint32_t mode)

 	spa_pod_builder_init(&b, buffer, sizeof(buffer));

-	spa_log_debug(this->log, "%p: configure convert %p", this, this->target);
+	spa_log_debug(this->log, "%p: configure convert %p %d", this, this->target, mode);

 	param = spa_pod_builder_add_object(&b,
 		SPA_TYPE_OBJECT_ParamPortConfig, SPA_PARAM_PortConfig,
@ -488,7 +524,7 @@ static int configure_convert(struct impl *this, uint32_t mode)
 	return spa_node_set_param(this->convert, SPA_PARAM_PortConfig, 0, param);
 }

-extern const struct spa_handle_factory spa_videoconvert_factory;
+extern const struct spa_handle_factory VIDEOCONVERTER_FACTORY;

 static const struct spa_node_events follower_node_events;

@ -530,12 +566,13 @@ static int reconfigure_mode(struct impl *this, bool passthrough,
 		} else {
 			/* add converter ports */
 			configure_convert(this, SPA_PARAM_PORT_CONFIG_MODE_dsp);
-			link_io(this);
 		}
+		link_io(this);
 	}
-
 	this->info.change_mask |= SPA_NODE_CHANGE_MASK_FLAGS | SPA_NODE_CHANGE_MASK_PARAMS;
-	this->info.flags &= ~SPA_NODE_FLAG_NEED_CONFIGURE;
+	SPA_FLAG_CLEAR(this->info.flags, SPA_NODE_FLAG_NEED_CONFIGURE);
+	SPA_FLAG_UPDATE(this->info.flags, SPA_NODE_FLAG_ASYNC,
+			this->async && this->follower == this->target);
 	this->params[IDX_Props].user++;

 	emit_node_info(this, false);
@ -562,9 +599,8 @@ static int impl_node_set_param(void *object, uint32_t id, uint32_t flags,
 		if ((res = spa_format_parse(param, &info.media_type, &info.media_subtype)) < 0)
 			return res;
 		if (info.media_type != SPA_MEDIA_TYPE_video ||
-		    info.media_subtype != SPA_MEDIA_SUBTYPE_raw)
-			return -EINVAL;
-
+			info.media_subtype != SPA_MEDIA_SUBTYPE_raw)
+				return -EINVAL;
 		if (spa_format_video_raw_parse(param, &info.info.raw) < 0)
 			return -EINVAL;

@ -599,8 +635,9 @@ static int impl_node_set_param(void *object, uint32_t id, uint32_t flags,
 			    info.media_subtype != SPA_MEDIA_SUBTYPE_raw)
 				return -ENOTSUP;

-			if (spa_format_video_raw_parse(format, &info.info.raw) >= 0)
+			if (spa_format_video_raw_parse(format, &info.info.raw) >= 0) {
 				this->default_format = info;
+			}
 		}

 		switch (mode) {
@ -629,13 +666,16 @@ static int impl_node_set_param(void *object, uint32_t id, uint32_t flags,
 	}

 	case SPA_PARAM_Props:
-		if (this->target != this->follower)
-			res = spa_node_set_param(this->target, id, flags, param);
-		res2 = spa_node_set_param(this->follower, id, flags, param);
+	{
+		int in_set_param = ++this->in_set_param;
+		res = spa_node_set_param(this->follower, id, flags, param);
+		if (this->target != this->follower && this->in_set_param == in_set_param)
+			res2 = spa_node_set_param(this->target, id, flags, param);
 		if (res < 0 && res2 < 0)
 			return res;
 		res = 0;
 		break;
+	}
 	case SPA_PARAM_ProcessLatency:
 		res = spa_node_set_param(this->follower, id, flags, param);
 		break;
@ -712,10 +752,12 @@ static int negotiate_format(struct impl *this)
 	struct spa_pod_builder b = { 0 };
 	int res;

-	if (this->have_format)
+	if (this->target == this->follower)
 		return 0;

-	if (this->target == this->follower)
+	spa_log_debug(this->log, "%p: have_format:%d", this, this->have_format);
+
+	if (this->have_format)
 		return 0;

 	spa_pod_builder_init(&b, buffer, sizeof(buffer));
@ -787,18 +829,30 @@ static int impl_node_send_command(void *object, const struct spa_command *comman

 	switch (SPA_NODE_COMMAND_ID(command)) {
 	case SPA_NODE_COMMAND_Start:
+		spa_log_debug(this->log, "%p: starting %d", this, this->started);
+		if (this->started)
+			return 0;
 		if ((res = negotiate_format(this)) < 0)
 			return res;
 		if ((res = negotiate_buffers(this)) < 0)
 			return res;
+		this->ready = true;
+		this->warned = false;
 		break;
 	case SPA_NODE_COMMAND_Suspend:
-		configure_format(this, 0, NULL);
-		SPA_FALLTHROUGH
+		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:
+		spa_log_debug(this->log, "%p: flushing", this);
 		this->io_buffers.status = SPA_STATUS_OK;
 		break;
 	default:
@ -809,20 +863,35 @@ static int impl_node_send_command(void *object, const struct spa_command *comman
 		spa_log_error(this->log, "%p: can't send command %d: %s",
 				this, SPA_NODE_COMMAND_ID(command),
 				spa_strerror(res));
-		return res;
 	}

-	if (this->target != this->follower) {
+	if (res >= 0 && this->target != this->follower) {
 		if ((res = spa_node_send_command(this->follower, command)) < 0) {
 			spa_log_error(this->log, "%p: can't send command %d: %s",
 					this, SPA_NODE_COMMAND_ID(command),
 					spa_strerror(res));
-			return res;
 		}
 	}
 	switch (SPA_NODE_COMMAND_ID(command)) {
 	case SPA_NODE_COMMAND_Start:
-		this->started = true;
+		if (res < 0) {
+			spa_log_debug(this->log, "%p: start failed", this);
+			this->ready = false;
+			configure_format(this, 0, NULL);
+		} else {
+			this->started = true;
+			spa_log_debug(this->log, "%p: started", this);
+		}
+		break;
+	case SPA_NODE_COMMAND_Suspend:
+		configure_format(this, 0, NULL);
+		spa_log_debug(this->log, "%p: suspended", this);
+		break;
+	case SPA_NODE_COMMAND_Pause:
+		spa_log_debug(this->log, "%p: paused", this);
+		break;
+	case SPA_NODE_COMMAND_Flush:
+		spa_log_debug(this->log, "%p: flushed", this);
 		break;
 	}
 	return res;
@ -866,10 +935,11 @@ static void convert_node_info(void *data, const struct spa_node_info *info)
 				(this->params[idx].flags & SPA_PARAM_INFO_SERIAL) |
 				(info->params[i].flags & SPA_PARAM_INFO_READWRITE);

-			if (!this->add_listener) {
-				this->params[idx].user++;
-				spa_log_debug(this->log, "param %d changed", info->params[i].id);
-			}
+			if (this->add_listener)
+				continue;
+
+			this->params[idx].user++;
+			spa_log_debug(this->log, "param %d changed", info->params[i].id);
 		}
 	}
 	emit_node_info(this, false);
@ -880,12 +950,19 @@ static void convert_port_info(void *data,
 		const struct spa_port_info *info)
 {
 	struct impl *this = data;
+	struct spa_port_info pi;

 	if (direction != this->direction) {
+		/* skip the converter output port into the follower */
 		if (port_id == 0)
 			return;
 		else
+			/* the monitor ports are exposed */
 			port_id--;
+	} else if (info) {
+		pi = *info;
+		pi.flags = this->follower_port_flags;
+		info = &pi;
 	}

 	spa_log_debug(this->log, "%p: port info %d:%d", this,
@ -938,6 +1015,8 @@ static void follower_info(void *data, const struct spa_node_info *info)
 		this->info.flags |= SPA_NODE_FLAG_OUT_PORT_CONFIG;
 		this->info.max_output_ports = MAX_PORTS;
 	}
+	SPA_FLAG_UPDATE(this->info.flags, SPA_NODE_FLAG_ASYNC,
+			this->async && this->follower == this->target);

 	spa_log_debug(this->log, "%p: follower info %s", this,
 			this->direction == SPA_DIRECTION_INPUT ?
@ -974,10 +1053,11 @@ static void follower_info(void *data, const struct spa_node_info *info)
 				(this->params[idx].flags & SPA_PARAM_INFO_SERIAL) |
 				(info->params[i].flags & SPA_PARAM_INFO_READWRITE);

-			if (!this->add_listener) {
-				this->params[idx].user++;
-				spa_log_debug(this->log, "param %d changed", info->params[i].id);
-			}
+			if (this->add_listener)
+				continue;
+
+			this->params[idx].user++;
+			spa_log_debug(this->log, "param %d changed", info->params[i].id);
 		}
 	}
 	emit_node_info(this, false);
@ -987,6 +1067,7 @@ static void follower_info(void *data, const struct spa_node_info *info)

 }

+/*
 static int recalc_latency(struct impl *this, enum spa_direction direction, uint32_t port_id)
 {
 	struct spa_pod_builder b = { 0 };
@ -1005,8 +1086,10 @@ static int recalc_latency(struct impl *this, enum spa_direction direction, uint3
 		spa_pod_builder_init(&b, buffer, sizeof(buffer));
 		if ((res = spa_node_port_enum_params_sync(this->follower,
 						direction, port_id, SPA_PARAM_Latency,
-						&index, NULL, &param, &b)) != 1)
-			return res;
+						&index, NULL, &param, &b)) != 1) {
+			param = NULL;
+			break;
+		}
 		if ((res = spa_latency_parse(param, &latency)) < 0)
 			return res;
 		if (latency.direction == direction)
@ -1019,6 +1102,7 @@ static int recalc_latency(struct impl *this, enum spa_direction direction, uint3

 	return 0;
 }
+*/

 static void follower_port_info(void *data,
 		enum spa_direction direction, uint32_t port_id,
@ -1033,6 +1117,11 @@ static void follower_port_info(void *data,
 	      return;
 	}

+	this->follower_port_flags = info->flags &
+		(SPA_PORT_FLAG_LIVE |
+		 SPA_PORT_FLAG_PHYSICAL |
+		 SPA_PORT_FLAG_TERMINAL);
+
 	spa_log_debug(this->log, "%p: follower port info %s %p %08"PRIx64, this,
 			this->direction == SPA_DIRECTION_INPUT ?
 				"Input" : "Output", info, info->change_mask);
@ -1054,26 +1143,33 @@ static void follower_port_info(void *data,
 			default:
 				continue;
 			}
+
 			if (!this->add_listener &&
 			    this->follower_params_flags[idx] == info->params[i].flags)
 				continue;

+			this->info.change_mask |= SPA_NODE_CHANGE_MASK_PARAMS;
 			this->follower_params_flags[idx] = info->params[i].flags;
 			this->params[idx].flags =
 				(this->params[idx].flags & SPA_PARAM_INFO_SERIAL) |
 				(info->params[i].flags & SPA_PARAM_INFO_READWRITE);

+			if (this->add_listener)
+				continue;
+
 			if (idx == IDX_Latency) {
-				res = recalc_latency(this, direction, port_id);
+				// res = recalc_latency(this, direction, port_id);
+				res = 0;
 				spa_log_debug(this->log, "latency: %d (%s)", res,
 						spa_strerror(res));
 			}
-
-			this->info.change_mask |= SPA_NODE_CHANGE_MASK_PARAMS;
-			if (!this->add_listener) {
-				this->params[idx].user++;
-				spa_log_debug(this->log, "param %d changed", info->params[i].id);
+			if (idx == IDX_EnumFormat) {
+				spa_log_debug(this->log, "new formats");
+				configure_format(this, 0, NULL);
 			}
+
+			this->params[idx].user++;
+			spa_log_debug(this->log, "param %d changed", info->params[i].id);
 		}
 	}
 	emit_node_info(this, false);
@ -1124,6 +1220,11 @@ static int follower_ready(void *data, int status)

 	spa_log_trace_fp(this->log, "%p: ready %d", this, status);

+	if (!this->ready) {
+		spa_log_info(this->log, "%p: ready stopped node", this);
+		return -EIO;
+	}
+
 	if (this->target != this->follower) {
 		this->driver = true;

@ -1369,6 +1470,13 @@ static int impl_node_process(void *object)
 	struct impl *this = object;
 	int status = 0, fstatus, retry = 8;

+	if (!this->started) {
+		if (!this->warned)
+			spa_log_warn(this->log, "%p: scheduling stopped node", this);
+		this->warned = true;
+		return -EIO;
+	}
+
 	spa_log_trace_fp(this->log, "%p: process convert:%p driver:%d",
 			this, this->convert, this->driver);

@ -1524,8 +1632,8 @@ impl_get_size(const struct spa_handle_factory *factory,
 {
 	size_t size = 0;

-#if 0
-	size += spa_handle_factory_get_size(&spa_videoconvert_factory, params);
+#if VIDEOCONVERTER
+	size += spa_handle_factory_get_size(&VIDEOCONVERTER_FACTORY, params);
 #endif
 	size += sizeof(struct impl);

@ -1540,7 +1648,7 @@ impl_init(const struct spa_handle_factory *factory,
 	  uint32_t n_support)
 {
 	struct impl *this;
-#if 0
+#if VIDEOCONVERTER
 	void *iface;
 #endif
 	const char *str;
@ -1576,13 +1684,16 @@ impl_init(const struct spa_handle_factory *factory,
 			SPA_VERSION_NODE,
 			&impl_node, this);

-#if 0
+#if VIDEOCONVERTER
 	this->hnd_convert = SPA_PTROFF(this, sizeof(struct impl), struct spa_handle);
-	spa_handle_factory_init(&spa_videoconvert_factory,
+	spa_handle_factory_init(&VIDEOCONVERTER_FACTORY,
 				this->hnd_convert,
 				info, support, n_support);

 	spa_handle_get_interface(this->hnd_convert, SPA_TYPE_INTERFACE_Node, &iface);
+	if (iface == NULL)
+		return -EINVAL;
+
 	this->convert = iface;
 #endif
 	this->target = this->convert;
@ -1611,7 +1722,14 @@ impl_init(const struct spa_handle_factory *factory,
 		spa_node_add_listener(this->convert,
 				&this->convert_listener, &convert_node_events, this);

-	reconfigure_mode(this, true, this->direction, NULL);
+	if (this->direction == SPA_DIRECTION_INPUT) {
+		// MODE: CONVERT
+		configure_convert(this, SPA_PARAM_PORT_CONFIG_MODE_convert);
+		// reconfigure_mode(this, true, this->direction, NULL);
+	} else {
+		// MODE: PASSTHROUGH
+		reconfigure_mode(this, true, this->direction, NULL);
+	}

 	link_io(this);