port: implement latency reporting some more

Implement a port recalculate latency method that takes the min
and max latency of all peer ports and sets that as the new port
latency.
When a link is made, let the output and input port recalculate
latencies.
Pass latency param in audioconvert.

spa/plugins/audioconvert/audioadapter.c

@@ -958,6 +958,12 @@ impl_node_port_set_param(void *object,
 			flags, param)) < 0)
 		return res;
 
+	if (id == SPA_PARAM_Latency && direction == this->direction) {
+		if ((res = spa_node_port_set_param(this->follower, direction, 0, id,
+				flags, param)) < 0)
+			return res;
+	}
+
 	return res;
 }
 

spa/plugins/audioconvert/audioconvert.c

@@ -1044,6 +1044,7 @@ impl_node_port_set_param(void *object,
 	switch (id) {
 	case SPA_PARAM_Latency:
 		target = this->fmt[SPA_DIRECTION_REVERSE(direction)];
+		port_id = 0;
 		break;
 	default:
 		is_monitor = IS_MONITOR_PORT(this, direction, port_id);

spa/plugins/audioconvert/fmtconvert.c

@@ -683,6 +683,8 @@ impl_node_port_set_param(void *object,
 				this, id, direction, port_id, param);
 
 	switch (id) {
+	case SPA_PARAM_Latency:
+		return 0;
 	case SPA_PARAM_Format:
 		return port_set_format(object, direction, port_id, flags, param);
 	default:

src/pipewire/impl-link.c

@@ -799,9 +799,8 @@ static void port_param_changed(struct pw_impl_link *this, uint32_t id,
 	case SPA_PARAM_EnumFormat:
 		target = PW_IMPL_PORT_STATE_CONFIGURE;
 		break;
-//	case SPA_PARAM_Buffers:
-//		target = PW_IMPL_PORT_STATE_READY;
-//		break;
+	case SPA_PARAM_Latency:
+		return;
 	default:
 		return;
 	}
@@ -933,6 +932,17 @@ static bool pw_impl_node_can_reach(struct pw_impl_node *output, struct pw_impl_n
 	return false;
 }
 
+static void recalculate_latencies(struct impl *impl)
+{
+	struct pw_impl_link *this = &impl->this;
+	/* from output port we get capture latency and propagate this
+	 * on the input port */
+	pw_impl_port_recalc_latency(this->output);
+	/* from input port we get playback latency and propagate that
+	 * on the output port */
+	pw_impl_port_recalc_latency(this->input);
+}
+
 static void try_link_controls(struct impl *impl, struct pw_impl_port *output, struct pw_impl_port *input)
 {
 	struct pw_control *cin, *cout;
@@ -1146,6 +1156,8 @@ struct pw_impl_link *pw_context_create_link(struct pw_context *context,
 	pw_impl_port_emit_link_added(output, this);
 	pw_impl_port_emit_link_added(input, this);
 
+	recalculate_latencies(impl);
+
 	try_link_controls(impl, output, input);
 
 	pw_impl_node_emit_peer_added(impl->onode, impl->inode);

src/pipewire/impl-port.c

@@ -351,6 +351,31 @@ static void emit_params(struct pw_impl_port *port, uint32_t *changed_ids, uint32
 	}
 }
 
+static int process_latency_param(void *data, int seq,
+		uint32_t id, uint32_t index, uint32_t next, struct spa_pod *param)
+{
+	struct pw_impl_port *this = data;
+	uint32_t direction;
+	struct pw_port_latency latency;
+
+	if (id != SPA_PARAM_Latency)
+		return -EINVAL;
+
+	if (spa_pod_parse_object(param,
+			SPA_TYPE_OBJECT_ParamLatency, NULL,
+			SPA_PARAM_LATENCY_direction,SPA_POD_Id(&direction),
+			SPA_PARAM_LATENCY_quantum,SPA_POD_Float(&latency.quantum),
+			SPA_PARAM_LATENCY_min,  SPA_POD_Int(&latency.min),
+			SPA_PARAM_LATENCY_max,  SPA_POD_Int(&latency.max)) < 0)
+		return 0;
+	if (direction != this->direction)
+		return 0;
+	pw_log_info("got latency %f %d %d", latency.quantum, latency.min, latency.max);
+
+	this->latency[direction] = latency;
+	return 0;
+}
+
 static void update_info(struct pw_impl_port *port, const struct spa_port_info *info)
 {
 	uint32_t changed_ids[MAX_PARAMS], n_changed_ids = 0;
@@ -391,6 +416,17 @@ static void update_info(struct pw_impl_port *port, const struct spa_port_info *i
 
 			if (info->params[i].flags & SPA_PARAM_INFO_READ)
 				changed_ids[n_changed_ids++] = id;
+
+			switch (id) {
+			case SPA_PARAM_Latency:
+				if (port->node != NULL)
+					pw_impl_port_for_each_param(port, 0, id, 0, UINT32_MAX,
+							NULL, process_latency_param, port);
+				break;
+			default:
+				break;
+			}
+
 		}
 	}
 
@@ -911,6 +947,7 @@ int pw_impl_port_add(struct pw_impl_port *port, struct pw_impl_node *node)
 	pw_impl_node_emit_port_init(node, port);
 
 	pw_impl_port_for_each_param(port, 0, SPA_PARAM_IO, 0, 0, NULL, check_param_io, port);
+	pw_impl_port_for_each_param(port, 0, SPA_PARAM_Latency, 0, 0, NULL, process_latency_param, port);
 
 	control = PW_IMPL_PORT_IS_CONTROL(port);
 	if (control) {
@@ -1251,6 +1288,60 @@ int pw_impl_port_for_each_link(struct pw_impl_port *port,
 	return res;
 }
 
+static void port_set_latency(struct pw_impl_port *port, struct pw_port_latency *latency)
+{
+	struct pw_port_latency *current = &port->latency[port->direction];
+	struct spa_pod *param;
+	struct spa_pod_builder b = { 0 };
+	uint8_t buffer[1024];
+
+	if (current->quantum == latency->quantum &&
+	    current->min == latency->min &&
+	    current->max == latency->max)
+		return;
+
+	*current = *latency;
+	pw_log_info("port set latency %d %d %f", latency->min, latency->max, latency->quantum);
+
+	spa_pod_builder_init(&b, buffer, sizeof(buffer));
+	param = spa_pod_builder_add_object(&b,
+			SPA_TYPE_OBJECT_ParamLatency, SPA_PARAM_Latency,
+			SPA_PARAM_LATENCY_direction, SPA_POD_Id(port->direction),
+			SPA_PARAM_LATENCY_quantum, SPA_POD_Float(latency->quantum),
+			SPA_PARAM_LATENCY_min, SPA_POD_Int(latency->min),
+			SPA_PARAM_LATENCY_max, SPA_POD_Int(latency->max));
+
+	pw_impl_port_set_param(port, SPA_PARAM_Latency, 0, param);
+}
+
+int pw_impl_port_recalc_latency(struct pw_impl_port *port)
+{
+	struct pw_impl_link *l;
+	struct pw_port_latency latency = { .quantum = 0.0f, .min = UINT32_MAX, .max = 0 };
+	struct pw_impl_port *other;
+
+	if (port->direction == PW_DIRECTION_OUTPUT) {
+		spa_list_for_each(l, &port->links, output_link) {
+			other = l->input;
+			latency.quantum = SPA_MAX(latency.quantum, other->latency[other->direction].quantum);
+			latency.min = SPA_MIN(latency.min, other->latency[other->direction].min);
+			latency.max = SPA_MAX(latency.max, other->latency[other->direction].max);
+		}
+	} else {
+		spa_list_for_each(l, &port->links, input_link) {
+			other = l->input;
+			latency.quantum = SPA_MAX(latency.quantum, other->latency[other->direction].quantum);
+			latency.min = SPA_MIN(latency.min, other->latency[other->direction].min);
+			latency.max = SPA_MAX(latency.max, other->latency[other->direction].max);
+		}
+	}
+	if (latency.min == UINT32_MAX)
+		latency.min = 0;
+
+	port_set_latency(port, &latency);
+	return 0;
+}
+
 SPA_EXPORT
 int pw_impl_port_is_linked(struct pw_impl_port *port)
 {

src/pipewire/private.h

@@ -686,6 +686,12 @@ struct pw_impl_node {
         void *user_data;                /**< extra user data */
 };
 
+struct pw_port_latency {
+	float quantum;		/** quantum multiplier */
+	uint32_t min;		/** min of all peers */
+	uint32_t max;		/** max of all peers */
+};
+
 struct pw_impl_port_mix {
 	struct spa_list link;
 	struct spa_list rt_link;
@@ -790,6 +796,8 @@ struct pw_impl_port {
 	} rt;					/**< data only accessed from the data thread */
 	unsigned int added:1;
 
+	struct pw_port_latency latency[2];	/**< latencies */
+
 	void *owner_data;		/**< extra owner data */
 	void *user_data;                /**< extra user data */
 };
@@ -1152,6 +1160,8 @@ int pw_impl_port_set_param(struct pw_impl_port *port,
 int pw_impl_port_use_buffers(struct pw_impl_port *port, struct pw_impl_port_mix *mix, uint32_t flags,
 		struct spa_buffer **buffers, uint32_t n_buffers);
 
+int pw_impl_port_recalc_latency(struct pw_impl_port *port);
+
 /** Change the state of the node */
 int pw_impl_node_set_state(struct pw_impl_node *node, enum pw_node_state state);