io: add support for segment

Move fields from the io_position to io_segment. The segment contains
the mapping between raw clock time and stream time in various
formats. We keep an array of pending segments available in the
io_position field so clients can anticipate changes.

Make looping a flag in the segment instead of a state.

Prepare for segment masters. These will be registered clients that
are responsible for updating parts of the extended segment info.

Add namespace to some defines.

pipewire-jack

@@ -1 +1 @@
-Subproject commit f8c3126b525eb01d062d1744b0c8cd3d36126407
+Subproject commit 3daf9f85d27a8af8f73daf975d273c01be1797bd

spa/include/spa/node/io.h

@@ -109,23 +109,23 @@ struct spa_io_sequence {
 	struct spa_pod_sequence sequence;	/**< sequence of timed events */
 };
 
-/** bar and beat position */
-struct spa_io_position_bar {
+/** bar and beat segment */
+struct spa_io_segment_bar {
 	uint32_t offset;		/**< offset in samples of this beat */
 	float signature_num;		/**< time signature numerator */
 	float signature_denom;		/**< time signature denominator */
 	double bpm;			/**< beats per minute */
-	double beat;			/**< current beat in position */
+	double beat;			/**< current beat in segment */
 	uint32_t padding[16];
 };
 
-/** video frame position */
-struct spa_io_position_video {
-	uint32_t offset;		/**< offset of frame against current position */
+/** video frame segment */
+struct spa_io_segment_video {
+	uint32_t offset;		/**< offset of frame against current segment */
 	struct spa_fraction framerate;
-#define SPA_IO_POSITION_VIDEO_FLAG_DROP_FRAME	(1<<0)
-#define SPA_IO_POSITION_VIDEO_FLAG_PULL_DOWN	(1<<1)
-#define SPA_IO_POSITION_VIDEO_FLAG_INTERLACED	(1<<2)
+#define SPA_IO_SEGMENT_VIDEO_FLAG_DROP_FRAME	(1<<0)
+#define SPA_IO_SEGMENT_VIDEO_FLAG_PULL_DOWN	(1<<1)
+#define SPA_IO_SEGMENT_VIDEO_FLAG_INTERLACED	(1<<2)
 	uint32_t flags;			/**< flags */
 	uint32_t hours;
 	uint32_t minutes;
@@ -135,52 +135,77 @@ struct spa_io_position_video {
 	uint32_t padding[16];
 };
 
+/**
+ * A segment converts a raw clock time to a segment (stream) position.
+ *
+ * The segment position is valid when the current clock position is between
+ * clock_start and clock_start + clock_duration. The position is then
+ * calculated as:
+ *
+ *   (clock_start - clock.position) * rate + position;
+ *
+ * Support for looping is done by specifying a non-zero duration. When the
+ * clock reaches clock_start + clock_duration, clock_duration is added to
+ * clock_start and the loop repeats.
+ *
+ * Care has to be taken when the clock.duration extends past the
+ * clock_start + clock_duration from the segment; the user should correctly
+ * wrap around and partially repeat the loop in the current cycle.
+ *
+ * Extra information can be placed in the segment by setting the valid flags
+ * and filling up the corresponding structures.
+ */
+struct spa_io_segment {
+#define SPA_IO_SEGMENT_FLAG_LOOPING	(1<<0)	/**< after the duration, the segment repeats */
+	uint32_t flags;				/**< extra flags */
+#define SPA_IO_SEGMENT_VALID_POSITION	(1<<0)
+#define SPA_IO_SEGMENT_VALID_BAR	(1<<1)
+#define SPA_IO_SEGMENT_VALID_VIDEO	(1<<2)
+	uint32_t valid;				/**< indicates what fields are valid below */
+	uint64_t clock_start;			/**< position against clock position when this
+						  *  info is active. Can be in the future for
+						  *  pending changes. It does not have to be in
+						  *  exact multiples of the clock duration. */
+	uint64_t clock_duration;		/**< duration when this info becomes invalid. If
+						  *  the duration is 0, this segment extends to the
+						  *  next segment. If the segment becomes invalid and
+						  *  the looping flag is set, the segment is repeats. */
+	uint64_t position;			/**< The position when the clock == clock_start. */
+	double rate;				/**< overal rate of the graph, can be negative for
+						  *  backwards time reporting. */
+
+	struct spa_io_segment_bar bar;		/**< when valid & SPA_IO_SEGMENT_VALID_BAR */
+	struct spa_io_segment_video video;	/**< when valid & SPA_IO_SEGMENT_VALID_VIDEO */
+};
+
 enum spa_io_position_state {
 	SPA_IO_POSITION_STATE_STOPPED,
 	SPA_IO_POSITION_STATE_STARTING,
 	SPA_IO_POSITION_STATE_RUNNING,
-	SPA_IO_POSITION_STATE_LOOPING,
 };
 
+/** the maximum number of segments visible in the future */
+#define SPA_IO_POSITION_MAX_SEGMENTS	8
+
 /**
  * The position information adds extra meaning to the raw clock times.
  *
  * It is set on all nodes and the clock id will contain the clock of the
  * master node in the graph.
  *
- * The position is valid when the current clock position is between
- * start_position and end_position. The position is then calculated as:
- *
- *   (clock_start - clock.position) * rate + position;
- *
- * Support for looping is done by specifying an end_position. When the
- * clock reaches the end_position, end_position is copied to start_position
- * and start_position is incremented with the same duration as the previous
- * loop.
+ * The position information contains 1 or more segments that convert the
+ * raw clock times to a stream time. They are sorted based on their
+ * clock_start times, and thus the order in which they will activate in
+ * the future. This makes it possible to look ahead in the scheduled
+ * segments and anticipate the changes in the timeline.
  */
 struct spa_io_position {
 	struct spa_io_clock clock;		/**< clock position of driver, always valid and
 						  *  read only */
-	uint32_t flags;				/**< extra flags */
 	uint32_t state;				/**< one of enum spa_io_position_state */
-	uint64_t clock_start;			/**< position against clock position when this
-						  *  info is active. Can be in the future for
-						  *  pending changes. It does not have to be in
-						  *  exact multiples of the clock duration. */
-	uint64_t clock_duration;		/**< duration when this info becomes invalid. If
-						  *  RUNNING, the state will transition to STOPPED.
-						  *  If the state was LOOPING, the clock_start will
-						  *  be updated with the duration of this info */
-	uint64_t position;			/**< The position when the clock position == clock_start */
 
-	double rate;				/**< overal rate of the graph, can be negative for
-						  *  backwards time reporting. */
-
-#define SPA_IO_POSITION_VALID_BAR	(1<<0)
-#define SPA_IO_POSITION_VALID_VIDEO	(1<<1)
-	uint32_t valid;				/**< indicates what fields are valid */
-	struct spa_io_position_bar bar;		/**< when mask & SPA_IO_POSITION_VALID_BAR */
-	struct spa_io_position_video video;	/**< when mask & SPA_IO_POSITION_VALID_VIDEO */
+	uint32_t n_segments;			/**< number of segments */
+	struct spa_io_segment segments[SPA_IO_POSITION_MAX_SEGMENTS];	/**< segments */
 };
 
 /** rate matching */

spa/plugins/jack/jack-sink.c

@@ -333,19 +333,25 @@ static void client_process(void *data)
 	if (this->position) {
 		jack_position_t *jp = &this->client->pos;
 		struct spa_io_position *p = this->position;
+		struct spa_io_segment *s;
 
-		p->rate = 1.0;
-		p->valid = 0;
+		p->n_segments = 1;
+		s = &p->segments[0];
+
+		s->flags = SPA_IO_SEGMENT_VALID_POSITION;
+		s->position = jp->frame;
+		s->rate = 1.0;
+		s->valid = 0;
 		if (jp->valid & JackPositionBBT) {
-			p->valid |= SPA_IO_POSITION_VALID_BAR;
+			s->valid |= SPA_IO_SEGMENT_VALID_BAR;
 			if (jp->valid & JackBBTFrameOffset)
-				p->bar.offset = jp->bbt_offset;
+				s->bar.offset = jp->bbt_offset;
 			else
-				p->bar.offset = 0;
-			p->bar.signature_num = jp->beats_per_bar;
-			p->bar.signature_denom = jp->beat_type;
-			p->bar.bpm = jp->beats_per_minute;
-			p->bar.beat = jp->bar * jp->beats_per_bar + jp->beat;
+				s->bar.offset = 0;
+			s->bar.signature_num = jp->beats_per_bar;
+			s->bar.signature_denom = jp->beat_type;
+			s->bar.bpm = jp->beats_per_minute;
+			s->bar.beat = jp->bar * jp->beats_per_bar + jp->beat;
 		}
 	}
 	spa_node_call_ready(&this->callbacks, SPA_STATUS_NEED_BUFFER);

src/modules/module-client-node/client-node.c

@@ -880,7 +880,7 @@ static int impl_node_process(void *object)
 	spa_log_trace_fp(this->log, "%p: send process driver:%p", this, impl->this.node->driver_node);
 
 	spa_system_clock_gettime(this->data_system, CLOCK_MONOTONIC, &ts);
-	n->rt.activation->status = TRIGGERED;
+	n->rt.activation->status = PW_NODE_ACTIVATION_TRIGGERED;
 	n->rt.activation->signal_time = SPA_TIMESPEC_TO_NSEC(&ts);
 
 	if (spa_system_eventfd_write(this->data_system, this->writefd, 1) < 0)

src/modules/module-client-node/remote-node.c

@@ -774,7 +774,7 @@ static int link_signal_func(void *user_data)
 	pw_log_trace("link %p: signal", link);
 
 	clock_gettime(CLOCK_MONOTONIC, &ts);
-	link->target.activation->status = TRIGGERED;
+	link->target.activation->status = PW_NODE_ACTIVATION_TRIGGERED;
 	link->target.activation->signal_time = SPA_TIMESPEC_TO_NSEC(&ts);
 
 	if (write(link->signalfd, &cmd, sizeof(cmd)) != sizeof(cmd))
@@ -1050,7 +1050,7 @@ static int node_ready(void *d, int status)
 	}
 
 	clock_gettime(CLOCK_MONOTONIC, &ts);
-	a->status = TRIGGERED;
+	a->status = PW_NODE_ACTIVATION_TRIGGERED;
 	a->signal_time = SPA_TIMESPEC_TO_NSEC(&ts);
 
 	if (write(data->rtwritefd, &cmd, sizeof(cmd)) != sizeof(cmd))

src/pipewire/node.c

@@ -768,7 +768,7 @@ static inline int resume_node(struct pw_node *this, int status)
 
 	spa_system_clock_gettime(data_system, CLOCK_MONOTONIC, &ts);
 	nsec = SPA_TIMESPEC_TO_NSEC(&ts);
-	activation->status = FINISHED;
+	activation->status = PW_NODE_ACTIVATION_FINISHED;
 	activation->finish_time = nsec;
 
 	pw_log_trace_fp(NAME" %p: trigger peers %"PRIu64, this, nsec);
@@ -782,7 +782,7 @@ static inline int resume_node(struct pw_node *this, int status)
                                 state->pending, state->required);
 
 		if (pw_node_activation_state_dec(state, 1)) {
-			t->activation->status = TRIGGERED;
+			t->activation->status = PW_NODE_ACTIVATION_TRIGGERED;
 			t->activation->signal_time = nsec;
 			t->signal(t->data);
 		}
@@ -812,7 +812,7 @@ static inline int process_node(void *data)
 	int status;
 
 	spa_system_clock_gettime(data_system, CLOCK_MONOTONIC, &ts);
-	a->status = AWAKE;
+	a->status = PW_NODE_ACTIVATION_AWAKE;
 	a->awake_time = SPA_TIMESPEC_TO_NSEC(&ts);
 
 	pw_log_trace_fp(NAME" %p: process %"PRIu64, this, a->awake_time);
@@ -830,7 +830,7 @@ static inline int process_node(void *data)
 
 	if (this == this->driver_node && !this->exported) {
 		spa_system_clock_gettime(data_system, CLOCK_MONOTONIC, &ts);
-		a->status = FINISHED;
+		a->status = PW_NODE_ACTIVATION_FINISHED;
 		a->signal_time = a->finish_time;
 		a->finish_time = SPA_TIMESPEC_TO_NSEC(&ts);
 
@@ -963,7 +963,6 @@ struct pw_node *pw_node_new(struct pw_core *core,
 
 	this->rt.activation->position.clock.rate = SPA_FRACTION(1, 48000);
 	this->rt.activation->position.clock.duration = DEFAULT_QUANTUM;
-	this->rt.activation->position.rate = 1.0;
 
 	check_properties(this);
 
@@ -1164,47 +1163,63 @@ static const struct spa_node_events node_events = {
 static void update_position(struct pw_node *node)
 {
 	struct pw_node_activation *a = node->rt.activation;
-	struct spa_io_position position;
+	struct spa_io_segment segment, *seg;
 	uint32_t seq1, seq2, change_mask;
 	enum spa_io_position_state state;
 
 	seq1 = SEQ_READ(&a->pending.seq);
 	change_mask = a->pending.change_mask;
 	state = a->pending.state;
-	position = a->pending.position;
+	segment = a->pending.segment;
 	seq2 = SEQ_READ(&a->pending.seq);
 
+	/* if we can't read a stable update, just ignore it and process it
+	 * in the next cycle. */
         if (SEQ_READ_SUCCESS(seq1, seq2))
 		a->pending.change_mask = 0;
 	else
 		change_mask = 0;
 
-	if (change_mask & UPDATE_POSITION) {
-		pw_log_debug("update position:%lu", position.position);
-		a->position.position = position.position;
-		a->position.clock_start = position.clock_start;
-		a->position.clock_duration = position.clock_duration;
-		a->position.rate = position.rate;
+	seg = &a->position.segments[0];
+
+	if (change_mask & PW_NODE_ACTIVATION_UPDATE_SEGMENT) {
+		pw_log_debug("update segment");
+		if (segment.valid & SPA_IO_SEGMENT_VALID_BAR) {
+			seg->bar = segment.bar;
+			seg->valid |= SPA_IO_SEGMENT_VALID_BAR;
+		}
+		if (segment.valid & SPA_IO_SEGMENT_VALID_VIDEO) {
+			seg->video = segment.video;
+			seg->valid |= SPA_IO_SEGMENT_VALID_BAR;
+		}
 	}
-	if (change_mask & UPDATE_STATE) {
+	if (change_mask & PW_NODE_ACTIVATION_UPDATE_REPOSITION) {
+		pw_log_debug("update position:%lu", segment.position);
+		seg->flags = segment.flags;
+		seg->clock_start = segment.clock_start;
+		seg->clock_duration = segment.clock_duration;
+		seg->position = segment.position;
+		seg->rate = segment.rate;
+	}
+	if (change_mask & PW_NODE_ACTIVATION_UPDATE_STATE) {
 		switch (state) {
 		case SPA_IO_POSITION_STATE_STOPPED:
+		case SPA_IO_POSITION_STATE_RUNNING:
 			a->position.state = state;
 			break;
 		case SPA_IO_POSITION_STATE_STARTING:
 			a->position.state = SPA_IO_POSITION_STATE_RUNNING;
 			break;
-		case SPA_IO_POSITION_STATE_RUNNING:
-		case SPA_IO_POSITION_STATE_LOOPING:
-			a->position.state = state;
-			break;
 		}
 	}
-	if (a->position.clock_start == 0)
-		a->position.clock_start = a->position.clock.position;
+	if (seg->rate == 0.0)
+		seg->rate = 1.0;
+
+	if (seg->clock_start == 0)
+		seg->clock_start = a->position.clock.position;
 
 	if (a->position.state == SPA_IO_POSITION_STATE_STOPPED)
-		a->position.clock_start += a->position.clock.duration;
+		seg->clock_start += a->position.clock.duration;
 }
 
 static int node_ready(void *data, int status)
@@ -1227,7 +1242,7 @@ static int node_ready(void *data, int status)
 		}
 		spa_list_for_each(t, &driver->rt.target_list, link) {
 			pw_node_activation_state_reset(&t->activation->state[0]);
-			t->activation->status = NOT_TRIGGERED;
+			t->activation->status = PW_NODE_ACTIVATION_NOT_TRIGGERED;
 		}
 		a->prev_signal_time = a->signal_time;
 

src/pipewire/private.h

@@ -349,10 +349,10 @@ struct pw_node_target {
 };
 
 struct pw_node_activation {
-#define NOT_TRIGGERED	0
-#define TRIGGERED	1
-#define AWAKE		2
-#define FINISHED	3
+#define PW_NODE_ACTIVATION_NOT_TRIGGERED	0
+#define PW_NODE_ACTIVATION_TRIGGERED		1
+#define PW_NODE_ACTIVATION_AWAKE		2
+#define PW_NODE_ACTIVATION_FINISHED		3
 	int status;
 
 	uint64_t signal_time;
@@ -361,21 +361,32 @@ struct pw_node_activation {
 	uint64_t prev_signal_time;
 
 	struct spa_io_position position;
+	uint32_t segment_master[32];			/* unique id (client id usually) of client
+							 * that will update extra segment info, There
+							 * can be one master for each segment
+							 * bitfield */
+
+	uint32_t version;
 	struct pw_node_activation_state state[2];	/* one current state and one next state,
-							 * as low bit of version in position */
+							 * as low bit of version */
 	float cpu_load[3];				/* averaged over short, medium, long time */
 	uint32_t xrun_count;				/* number of xruns */
 	uint64_t xrun_time;				/* time of last xrun in microseconds */
 	uint64_t xrun_delay;				/* delay of last xrun in microseconds */
 	uint64_t max_delay;				/* max of all xruns in microseconds */
 
+
 	struct {
 		uint32_t seq;
-#define UPDATE_STATE		(1<<0)
-#define UPDATE_POSITION		(1<<1)
+#define PW_NODE_ACTIVATION_UPDATE_STATE		(1<<0)
+#define PW_NODE_ACTIVATION_UPDATE_SEGMENT	(1<<1)
+#define PW_NODE_ACTIVATION_UPDATE_REPOSITION	(1<<2)
 		uint32_t change_mask;
 		enum spa_io_position_state state;
-		struct spa_io_position position;
+		struct spa_io_segment segment;		/* update for the extra segment info
+							 * fields. When REPOSITION update, the segment
+							 * position field will contain the desired
+							 * new position. */
 	} pending;
 };