Improve ringbuffer support

Fix ringbuffer mixing in audiomixer
Add ringbuffer support in audiotestsrc params
Don't recycle buffers before signaling have_output, the app is supposed
to recycle explicitly or with a process_output call.
Add some trace to graph functions in tests
Add ringbuffer support in export-source

spa/include/spa/graph-scheduler6.h

@@ -26,6 +26,16 @@ extern "C" {
 
 #include <spa/graph.h>
 
+struct spa_graph_data {
+	struct spa_graph *graph;
+};
+
+static inline void spa_graph_data_init(struct spa_graph_data *data,
+				       struct spa_graph *graph)
+{
+	data->graph = graph;
+}
+
 static inline int spa_graph_impl_need_input(void *data, struct spa_graph_node *node)
 {
 	struct spa_graph_port *p;

spa/plugins/alsa/alsa-utils.c

@@ -372,7 +372,6 @@ pull_frames(struct state *state,
 
 			spa_ringbuffer_read_data(ringbuffer, d[0].data, index % ringbuffer->size, dst, n_bytes);
 
-
 			spa_ringbuffer_read_update(ringbuffer, index + n_bytes);
 
 			reuse = avail == n_frames || state->n_buffers == 1;

spa/plugins/audiomixer/audiomixer.c

@@ -679,11 +679,11 @@ impl_node_port_send_command(struct spa_node *node,
 static inline void
 add_port_data(struct impl *this, void *out, size_t outsize, size_t next, struct port *port, int layer)
 {
-	void *in;
 	size_t insize;
 	struct buffer *b;
 	struct spa_data *id;
 	uint32_t index = 0, offset, len1, len2;
+	mix_func_t mix = layer == 0 ? this->copy : this->add;
 
 	b = spa_list_first(&port->queue, struct buffer, link);
 
@@ -704,21 +704,11 @@ add_port_data(struct impl *this, void *out, size_t outsize, size_t next, struct
 		outsize = SPA_MIN(outsize, insize);
 		len1 = outsize;
 	}
-	in = SPA_MEMBER(id[0].data, offset, void);
 	len2 = outsize - len1;
 
-	if (layer == 0) {
-		this->copy(out, in, len1);
+	mix(out, SPA_MEMBER(id[0].data, offset, void), len1);
 	if (len2 > 0)
-			this->copy(out + len1, in + len1, len2);
-	}
-	else {
-		this->add(out, in, len1);
-		if (len2 > 0)
-			this->add(out + len1, in + len1, len2);
-	}
-
-	spa_log_trace(this->log, NAME " %p: %d %d %d %zd", this, index, len1, len2, outsize);
+		mix(out + len1, id[0].data, len2);
 
 	if (b->rb)
 		spa_ringbuffer_read_update(&b->rb->ringbuffer, index + outsize);
@@ -774,8 +764,6 @@ static int mix_output(struct impl *this, size_t n_bytes)
 		else
 			len1 = n_bytes;
 		len2 = n_bytes - len1;
-
-		spa_log_trace(this->log, NAME " %p: %d %d %d %ld %d %d", this, index, offset, avail, n_bytes, len1, len2);
 	} else {
 		n_bytes = SPA_MIN(n_bytes, od[0].maxsize);
 		offset = 0;

spa/plugins/audiotestsrc/audiotestsrc.c

@@ -285,7 +285,7 @@ static int make_buffer(struct impl *this)
 
 	if (b->rb) {
 		int32_t filled, avail;
-		uint32_t index, offset;
+		uint32_t index, offset, l0, l1;
 
 		filled = spa_ringbuffer_get_write_index(&b->rb->ringbuffer, &index);
 		avail = b->rb->ringbuffer.size - filled;
@@ -296,15 +296,18 @@ static int make_buffer(struct impl *this)
 		offset = index & b->rb->ringbuffer.mask;
 
 		if (offset + n_bytes > b->rb->ringbuffer.size) {
-			uint32_t l0 = b->rb->ringbuffer.size - offset;
-			this->render_func(this, SPA_MEMBER(b->outbuf->datas[0].data, offset, void),
-					  l0 / this->bpf);
-			this->render_func(this, b->outbuf->datas[0].data,
-					  (n_bytes - l0) / this->bpf);
-		} else {
-			this->render_func(this, SPA_MEMBER(b->outbuf->datas[0].data, offset, void),
-					  n_samples);
+			l0 = (b->rb->ringbuffer.size - offset) / this->bpf;
+			l1 = n_samples - l0;
 		}
+		else {
+			l0 = n_samples;
+			l1 = 0;
+		}
+
+		this->render_func(this, SPA_MEMBER(b->outbuf->datas[0].data, offset, void), l0);
+		if (l1)
+			this->render_func(this, b->outbuf->datas[0].data, l1);
+
 		spa_ringbuffer_write_update(&b->rb->ringbuffer, index + n_bytes);
 	} else {
 		n_samples = n_bytes / this->bpf;
@@ -674,8 +677,20 @@ impl_node_port_enum_params(struct spa_node *node,
 			":", t->param_alloc_meta_enable.size, "i", sizeof(struct spa_meta_header));
 		break;
 
+	case 2:
+		*param = spa_pod_builder_param(&b,
+			t->param_alloc_meta_enable.MetaEnable,
+			":", t->param_alloc_meta_enable.type,	"I", t->meta.Ringbuffer,
+			":", t->param_alloc_meta_enable.size,	"i", sizeof(struct spa_meta_ringbuffer),
+			":", t->param_alloc_meta_enable.ringbufferSize,   "ir", 5512 * this->bpf,
+								2, 16 * this->bpf, INT32_MAX / this->bpf,
+			":", t->param_alloc_meta_enable.ringbufferStride, "i", 0,
+			":", t->param_alloc_meta_enable.ringbufferBlocks, "i", 1,
+			":", t->param_alloc_meta_enable.ringbufferAlign,  "i", 16);
+		break;
+
 	default:
-		return SPA_RESULT_NOT_IMPLEMENTED;
+		return SPA_RESULT_ENUM_END;
 	}
 
 	return SPA_RESULT_OK;

spa/plugins/v4l2/v4l2-utils.c

@@ -930,9 +930,6 @@ static int mmap_read(struct impl *this)
 	d[0].chunk->size = buf.bytesused;
 	d[0].chunk->stride = port->fmt.fmt.pix.bytesperline;
 
-	if (io->buffer_id != SPA_ID_INVALID)
-		spa_v4l2_buffer_recycle(this, io->buffer_id);
-
 	b->outstanding = true;
 	io->buffer_id = b->outbuf->id;
 	io->status = SPA_RESULT_HAVE_BUFFER;

spa/tests/test-graph.c

@@ -35,14 +35,17 @@
 #include <spa/audio/format-utils.h>
 #include <spa/format-utils.h>
 #include <spa/format-builder.h>
-#include <spa/graph.h>
-#include <spa/graph-scheduler1.h>
-
-#include <lib/debug.h>
 
 static SPA_TYPE_MAP_IMPL(default_map, 4096);
 static SPA_LOG_IMPL(default_log);
 
+#define spa_debug(f,...) spa_log_trace(&default_log.log, f, __VA_ARGS__)
+
+#include <spa/graph.h>
+#include <spa/graph-scheduler6.h>
+
+#include <lib/debug.h>
+
 struct type {
 	uint32_t node;
 	uint32_t props;

spa/tests/test-mixer.c

@@ -30,8 +30,6 @@
 #include <spa/log.h>
 #include <spa/log-impl.h>
 #include <spa/loop.h>
-#include <spa/graph.h>
-#include <spa/graph-scheduler1.h>
 #include <spa/type-map.h>
 #include <spa/type-map-impl.h>
 #include <spa/audio/format-utils.h>
@@ -43,6 +41,11 @@
 static SPA_TYPE_MAP_IMPL(default_map, 4096);
 static SPA_LOG_IMPL(default_log);
 
+#define spa_debug(...)	spa_log_trace(&default_log.log,__VA_ARGS__)
+
+#include <spa/graph.h>
+#include <spa/graph-scheduler1.h>
+
 struct type {
 	uint32_t node;
 	uint32_t props;

src/examples/export-source.c

@@ -61,6 +61,7 @@ struct buffer {
 	void *ptr;
 	size_t size;
 	bool mapped;
+	struct spa_meta_ringbuffer *rb;
 };
 
 struct data {
@@ -89,7 +90,6 @@ struct data {
 	struct spa_audio_info_raw format;
 
 	uint8_t params_buffer[1024];
-	struct spa_param *params[2];
 
 	struct buffer buffers[32];
 	int n_buffers;
@@ -168,8 +168,6 @@ static int impl_port_set_format(struct spa_node *node, enum spa_direction direct
 				uint32_t flags, const struct spa_format *format)
 {
 	struct data *d = SPA_CONTAINER_OF(node, struct data, impl_node);
-	struct pw_type *t = d->t;
-	struct spa_pod_builder b = { NULL };
 
 	if (format == NULL)
 		return SPA_RESULT_OK;
@@ -182,21 +180,6 @@ static int impl_port_set_format(struct spa_node *node, enum spa_direction direct
 	if (d->format.format != d->type.audio_format.S16)
 		return SPA_RESULT_ERROR;
 
-	spa_pod_builder_init(&b, d->params_buffer, sizeof(d->params_buffer));
-	d->params[0] = spa_pod_builder_param(&b,
-		t->param_alloc_buffers.Buffers,
-		":", t->param_alloc_buffers.size,    "iru", 1024,
-									2, 32, 4096,
-		":", t->param_alloc_buffers.stride,  "i",   0,
-		":", t->param_alloc_buffers.buffers, "iru", 2,
-									2, 2, 32,
-		":", t->param_alloc_buffers.align,   "i",  16);
-
-	d->params[1] = spa_pod_builder_param(&b,
-		t->param_alloc_meta_enable.MetaEnable,
-		":", t->param_alloc_meta_enable.type, "I", t->meta.Header,
-		":", t->param_alloc_meta_enable.size, "i", sizeof(struct spa_meta_header));
-
 	return SPA_RESULT_OK;
 }
 
@@ -233,12 +216,42 @@ static int impl_port_enum_params(struct spa_node *node, enum spa_direction direc
 				 uint32_t index, struct spa_param **param)
 {
 	struct data *d = SPA_CONTAINER_OF(node, struct data, impl_node);
+	struct pw_type *t = d->t;
+	struct spa_pod_builder b = { NULL };
 
-	if (index >= 2)
+	spa_pod_builder_init(&b, d->params_buffer, sizeof(d->params_buffer));
+
+	switch (index) {
+	case 0:
+		*param = spa_pod_builder_param(&b,
+			t->param_alloc_buffers.Buffers,
+			":", t->param_alloc_buffers.size,    "iru", 1024,
+										2, 32, 4096,
+			":", t->param_alloc_buffers.stride,  "i",   0,
+			":", t->param_alloc_buffers.buffers, "iru", 2,
+										2, 2, 32,
+			":", t->param_alloc_buffers.align,   "i",  16);
+		break;
+	case 1:
+		*param = spa_pod_builder_param(&b,
+			t->param_alloc_meta_enable.MetaEnable,
+			":", t->param_alloc_meta_enable.type, "I", t->meta.Header,
+			":", t->param_alloc_meta_enable.size, "i", sizeof(struct spa_meta_header));
+		break;
+	case 2:
+		*param = spa_pod_builder_param(&b,
+			t->param_alloc_meta_enable.MetaEnable,
+			":", t->param_alloc_meta_enable.type,	"I", t->meta.Ringbuffer,
+			":", t->param_alloc_meta_enable.size,	"i", sizeof(struct spa_meta_ringbuffer),
+			":", t->param_alloc_meta_enable.ringbufferSize,   "ir", 5512 * 4,
+								2, 16 * 4, INT32_MAX / 4,
+			":", t->param_alloc_meta_enable.ringbufferStride, "i", 0,
+			":", t->param_alloc_meta_enable.ringbufferBlocks, "i", 1,
+			":", t->param_alloc_meta_enable.ringbufferAlign,  "i", 16);
+		break;
+	default:
 		return SPA_RESULT_ENUM_END;
-
-	*param = d->params[index];
-
+	}
 	return SPA_RESULT_OK;
 }
 
@@ -273,6 +286,7 @@ static int impl_port_use_buffers(struct spa_node *node, enum spa_direction direc
 		}
 		b->size = datas[0].maxsize;
 		b->buffer = buffers[i];
+		b->rb = spa_buffer_find_meta(buffers[i], d->type.meta.Ringbuffer);
 		pw_log_info("got buffer %d size %zd", i, b->size);
 		spa_list_append(&d->empty, &b->link);
 	}
@@ -297,9 +311,10 @@ static int impl_node_process_output(struct spa_node *node)
 {
 	struct data *d = SPA_CONTAINER_OF(node, struct data, impl_node);
 	struct buffer *b;
-	int i, c, n_samples;
+	int i, c, n_samples, avail;
 	int16_t *dst;
         struct spa_port_io *io = d->io;
+	uint32_t index = 0;
 
 	if (io->buffer_id < d->n_buffers) {
 		reuse_buffer(d, io->buffer_id);
@@ -312,8 +327,23 @@ static int impl_node_process_output(struct spa_node *node)
         b = spa_list_first(&d->empty, struct buffer, link);
         spa_list_remove(&b->link);
 
+	if (b->rb) {
+		uint32_t filled, offset;
+
+		filled = spa_ringbuffer_get_write_index(&b->rb->ringbuffer, &index);
+		avail = b->rb->ringbuffer.size - filled;
+		offset = index % b->rb->ringbuffer.size;
+
+		if (offset + avail > b->rb->ringbuffer.size)
+			avail = b->rb->ringbuffer.size - offset;
+
+		dst = SPA_MEMBER(b->ptr, offset, void);
+	}
+	else {
 		dst = b->ptr;
-        n_samples = b->size / (sizeof(int16_t) * d->format.channels);
+		avail = b->size;
+	}
+        n_samples = avail / (sizeof(int16_t) * d->format.channels);
 
         for (i = 0; i < n_samples; i++) {
                 int16_t val;
@@ -327,6 +357,10 @@ static int impl_node_process_output(struct spa_node *node)
                 for (c = 0; c < d->format.channels; c++)
                         *dst++ = val;
         }
+
+	if (b->rb)
+		spa_ringbuffer_write_update(&b->rb->ringbuffer, index + avail);
+
 	io->buffer_id = b->buffer->id;
 	io->status = SPA_RESULT_HAVE_BUFFER;