<commit_before>/*
 * Copyright (C) 2009-2010 <NAME>
 *
 * Author: <NAME> <<EMAIL>>
 *
 * This file may be used under the terms of the GNU Lesser General Public
 * License version 2.1, a copy of which is found in LICENSE included in the
 * packaging of this file.
 */

#include "gstdspbase.h"
#include "gstdspbuffer.h"
#include "plugin.h"

#include "dsp_bridge.h"

#include <string.h> /* for memcpy */

#include "util.h"
#include "log.h"

#define GST_CAT_DEFAULT gstdsp_debug

static inline bool send_buffer(GstDspBase *self, struct td_buffer *tb);

static inline void
map_buffer(GstDspBase *self,
	   GstBuffer *g_buf,
	   struct td_buffer *tb);

struct td_codec td_fake_codec = {
	.uuid = NULL,
};

static inline long
get_elapsed_eos(GstDspBase *self)
{
	struct timespec cur;
	long start, elapsed;

	if (!self->eos_start.tv_sec)
		return 0;

	clock_gettime(CLOCK_MONOTONIC, &cur);

	start = self->eos_start.tv_sec * 1000 + self->eos_start.tv_nsec / 1000000;
	elapsed = cur.tv_sec * 1000 + cur.tv_nsec / 1000000 - start;

	return elapsed;
}

du_port_t *
du_port_new(int id,
	    int dir)
{
	du_port_t *p;
	p = calloc(1, sizeof(*p));
	if (!p)
		return NULL;

	p->id = id;
	p->queue = async_queue_new();
	p->dir = dir;

	return p;
}

void
du_port_free(du_port_t *p)
{
	if (!p)
		return;

	free(p->buffers);
	async_queue_free(p->queue);

	free(p);
}

void
du_port_alloc_buffers(du_port_t *p, guint num_buffers)
{
	p->num_buffers = num_buffers;
	free(p->buffers);
	p->buffers = calloc(num_buffers, sizeof(*p->buffers));
	for (unsigned i = 0; i < p->num_buffers; i++)
		p->buffers[i].port = p;
}

static inline void
du_port_flush(du_port_t *p)
{
	guint i;
	struct td_buffer *tb = p->buffers;

	for (i = 0; i < p->num_buffers; i++, tb++) {
		dmm_buffer_t *b = tb->data;
		if (!b)
			continue;
		if (tb->user_data)
			gst_buffer_unref(tb->user_data);
		dmm_buffer_free(b);
		tb->data = NULL;
	}
	async_queue_flush(p->queue);
}

static inline void
g_sem_down_status(GSem *sem,
		  const GstFlowReturn *status)
{
	GstFlowReturn ret = GST_FLOW_OK;
	g_mutex_lock(sem->mutex);

	while (sem->count == 0 &&
	       (ret = g_atomic_int_get(status)) == GST_FLOW_OK)
		g_cond_wait(sem->condition, sem->mutex);

	if (ret == GST_FLOW_OK)
		sem->count--;

	g_mutex_unlock(sem->mutex);
}

static inline void
g_sem_signal(GSem *sem)
{
	g_mutex_lock(sem->mutex);
	g_cond_signal(sem->condition);
	g_mutex_unlock(sem->mutex);
}

static inline void
g_sem_reset(GSem *sem,
	    guint count)
{
	g_mutex_lock(sem->mutex);
	sem->count = count;
	g_mutex_unlock(sem->mutex);
}

typedef struct {
	uint32_t buffer_data;
	uint32_t buffer_size;
	uint32_t param_data;
	uint32_t param_size;
	uint32_t buffer_len;
	uint32_t silly_eos;
	uint32_t silly_buf_state;
	uint32_t silly_buf_active;
	uint32_t silly_buf_id;
#if SN_API >= 2
	uint32_t nb_available_buf;
	uint32_t donot_flush_buf;
	uint32_t donot_invalidate_buf;
#endif
	uint32_t reserved;
	uint32_t msg_virt;
	uint32_t buffer_virt;
	uint32_t param_virt;
	uint32_t silly_out_buffer_index;
	uint32_t silly_in_buffer_index;
	uint32_t user_data;
	uint32_t stream_id;
} dsp_comm_t;

static GstElementClass *parent_class;

static inline void
got_message(GstDspBase *self,
	    struct dsp_msg *msg)
{
	int32_t id;
	uint32_t command_id;

	id = msg->cmd & 0x000000ff;
	command_id = msg->cmd & 0xffffff00;

	switch (command_id) {
	case 0x0600: {
		dmm_buffer_t *b;
		du_port_t *p;
		dsp_comm_t *msg_data;
		dmm_buffer_t *param;
		unsigned i;
		struct td_buffer *tb = NULL;

		for (i = 0; i < ARRAY_SIZE(self->ports); i++)
			if (self->ports[i]->id == id) {
				p = self->ports[i];
				break;
			}

		if (i >= ARRAY_SIZE(self->ports))
			g_error("bad port index: %i", id);

		pr_debug(self, "got %s buffer", id == 0 ? "input" : "output");

		for (i = 0; i < p->num_buffers; i++) {
			if (msg->arg_1 == (uint32_t) p->buffers[i].comm->map) {
				tb = &p->buffers[i];
				break;
			}
		}

		if (!tb)
			g_error("buffer mismatch");

		dmm_buffer_end(tb->comm, tb->comm->size);

		msg_data = tb->comm->data;
		b = (void *) msg_data->user_data;
		b->len = msg_data->buffer_len;

		if (G_UNLIKELY(b->len > b->size))
			g_error("wrong buffer size");

		if (tb->pinned)
			dmm_buffer_end(b, b->len);
		else
			dmm_buffer_unmap(b);

		param = (void *) msg_data->param_virt;
		if (param)
			dmm_buffer_end(param, param->size);

		if (p->recv_cb)
			p->recv_cb(self, tb);

		if (id == 0) {
			if (tb->user_data) {
				gst_buffer_unref(tb->user_data);
				tb->user_data = NULL;
			}
		}

		async_queue_push(p->queue, tb);
		break;
	}
	case 0x0500:
		pr_debug(self, "got flush");
		break;
	case 0x0200:
		pr_debug(self, "got stop");
		g_sem_up(self->flush);
		break;
	case 0x0400:
		pr_debug(self, "got alg ctrl");
		dmm_buffer_free(self->alg_ctrl);
		self->alg_ctrl = NULL;
		break;
	case 0x0e00:
		if (msg->arg_1 == 1 && msg->arg_2 == 0x0500) {
			pr_debug(self, "playback completed");
			break;
		}

		if (msg->arg_1 == 1 && (msg->arg_2 & 0x0600) == 0x0600) {
			struct td_codec *codec = self->codec;
			if (codec->update_params)
				codec->update_params(self, self->node, msg->arg_2);
			break;
		}

		pr_warning(self, "DSP event: cmd=0x%04X, arg1=%u, arg2=0x%04X",
			   msg->cmd, msg->arg_1, msg->arg_2);
		if ((msg->arg_2 & 0x0F00) == 0x0F00)
			gstdsp_got_error(self, 0, "algo error");
		break;
	default:
		pr_warning(self, "unhandled command: %u", command_id);
	}
}

static inline void
setup_buffers(GstDspBase *self)
{
	GstBuffer *buf = NULL;
	dmm_buffer_t *b;
	du_port_t *p;
	guint i;

	p = self->ports[0];
	for (i = 0; i < p->num_buffers; i++) {
		p->buffers[i].data = b = dmm_buffer_new(self->dsp_handle, self->proc, p->dir);
		async_queue_push(p->queue, &p->buffers[i]);
	}

	p = self->ports[1];
	for (i = 0; i < p->num_buffers; i++) {
		struct td_buffer *tb = &p->buffers[i];
		tb->data = b = dmm_buffer_new(self->dsp_handle, self->proc, p->dir);

		if (self->use_pad_alloc) {
			GstFlowReturn ret;
			ret = gst_pad_alloc_buffer_and_set_caps(self->srcpad,
								GST_BUFFER_OFFSET_NONE,
								self->output_buffer_size,
								GST_PAD_CAPS(self->srcpad),
								&buf);
			/* might fail if not (yet) linked */
			if (G_UNLIKELY(ret != GST_FLOW_OK)) {
				pr_err(self, "couldn't allocate buffer: %s", gst_flow_get_name(ret));
				dmm_buffer_allocate(b, self->output_buffer_size);
				b->need_copy = true;
			} else {
				map_buffer(self, buf, tb);
				gst_buffer_unref(buf);
			}
		}
		else {
			dmm_buffer_allocate(b, self->output_buffer_size);
			if (self->use_pinned) {
				dmm_buffer_map(b);
				tb->pinned = tb->clean = true;
			}
		}

		self->send_buffer(self, tb);
	}
}

static inline void
pause_task(GstDspBase *self, GstFlowReturn status)
{
	bool deferred_eos;

	/* synchronize to ensure we are not dropping the EOS event */
	g_mutex_lock(self->ts_mutex);
	(void) g_atomic_int_compare_and_exchange(&self->status, GST_FLOW_OK, status);
	deferred_eos = g_atomic_int_compare_and_exchange(&self->deferred_eos, true, false);
	g_mutex_unlock(self->ts_mutex);

	pr_info(self, "pausing task; reason %s", gst_flow_get_name(status));
	gst_pad_pause_task(self->srcpad);

	/* avoid waiting for buffers that will never come */
	async_queue_disable(self->ports[0]->queue);
	async_queue_disable(self->ports[1]->queue);

	/* there's a pending deferred EOS, it's now or never */
	if (deferred_eos) {
		pr_info(self, "send elapsed eos");
		self->eos_start.tv_sec = self->eos_start.tv_nsec = 0;
		gst_pad_push_event(self->srcpad, gst_event_new_eos());
		g_atomic_int_set(&self->eos, true);
	}
}

static inline GstFlowReturn
check_status(GstDspBase *self)
{
	GstFlowReturn ret;
	ret = g_atomic_int_get(&self->status);
	if (G_UNLIKELY(ret != GST_FLOW_OK))
		pause_task(self, ret);
	return ret;
}

static void
output_loop(gpointer data)
{
	GstPad *pad;
	GstDspBase *self;
	GstFlowReturn ret = GST_FLOW_OK;
	GstBuffer *out_buf = NULL;
	dmm_buffer_t *b;
	gboolean flush_buffer;
	gboolean got_eos = FALSE;
	gboolean keyframe = FALSE;
	GstEvent *event;
	du_port_t *p;
	struct td_buffer *tb;
	bool handled;
	GstClockTime timestamp, duration;

	pad = data;
	self = GST_DSP_BASE(GST_OBJECT_PARENT(pad));
	p = self->ports[1];

	pr_debug(self, "begin");
	tb = async_queue_pop(p->queue);

	/*
	 * queue might have been disabled above, so perhaps tb == NULL,
	 * but then right here in between self->status may have been set to
	 * OK by e.g. FLUSH_STOP
	 */
	if (G_UNLIKELY(!tb)) {
		pr_info(self, "no buffer");
		ret = check_status(self);
		goto nok;
	}

	b = tb->data;

	ret = check_status(self);
	if (G_UNLIKELY(ret != GST_FLOW_OK)) {
		async_queue_push(p->queue, tb);
		goto end;
	}

	if (G_UNLIKELY(self->skip_hack_2 > 0)) {
		self->skip_hack_2--;
		goto leave;
	}

	/* check for too many buffers returned */
	g_mutex_lock(self->ts_mutex);
	if (G_UNLIKELY(b->len && !self->ts_count)) {
		pr_warning(self, "no timestamp; unexpected buffer");
		g_mutex_unlock(self->ts_mutex);
		goto leave;
	}
	g_mutex_unlock(self->ts_mutex);

	/* first clear pending events */
	g_mutex_lock(self->ts_mutex);
	while ((event = self->ts_array[self->ts_out_pos].event)) {
		self->ts_array[self->ts_out_pos].event = NULL;
		flush_buffer = (self->ts_out_pos != self->ts_push_pos);
		self->ts_out_pos = (self->ts_out_pos + 1) % ARRAY_SIZE(self->ts_array);
		if (G_LIKELY(!flush_buffer)) {
			self->ts_push_pos = self->ts_out_pos;
			pr_debug(self, "pushing event: %s", GST_EVENT_TYPE_NAME(event));
			gst_pad_push_event(self->srcpad, event);
		} else {
			pr_debug(self, "ignored flushed event: %s", GST_EVENT_TYPE_NAME(event));
			gst_event_unref(event);
		}
	}
	g_mutex_unlock(self->ts_mutex);

	/* a pending reallocation from the previous run */
	if (G_UNLIKELY(!b->data)) {
		dmm_buffer_allocate(b, self->output_buffer_size);
		send_buffer(self, tb);
		goto end;
	}

	if (G_UNLIKELY(!b->len)) {
		/* no need to process this buffer */
		/* no real frame data, so no need to consume a real frame's ts */
		goto leave;
	}

	g_mutex_lock(self->ts_mutex);
	flush_buffer = (self->ts_out_pos != self->ts_push_pos);
	g_mutex_unlock(self->ts_mutex);

	if (G_UNLIKELY(flush_buffer)) {
		g_mutex_lock(self->ts_mutex);
		pr_debug(self, "ignored flushed output buffer for %" GST_TIME_FORMAT,
			 GST_TIME_ARGS((self->ts_array[self->ts_out_pos].time)));
		self->ts_count--;
		self->ts_out_pos = (self->ts_out_pos + 1) % ARRAY_SIZE(self->ts_array);
		g_mutex_unlock(self->ts_mutex);
		goto leave;
	}

	/* now go after the data, but let's first see if it is keyframe */
	keyframe = tb->keyframe;

	if (self->use_pad_alloc) {
		GstBuffer *new_buf;

		ret = gst_pad_alloc_buffer_and_set_caps(self->srcpad,
							GST_BUFFER_OFFSET_NONE,
							self->output_buffer_size,
							GST_PAD_CAPS(self->srcpad),
							&new_buf);

		if (G_UNLIKELY(ret != GST_FLOW_OK)) {
			pr_info(self, "couldn't allocate buffer: %s", gst_flow_get_name(ret));
			async_queue_push(p->queue, tb);
			goto nok;
		}

		if (tb->user_data) {
			out_buf = tb->user_data;
			tb->user_data = NULL;
			map_buffer(self, new_buf, tb);
			gst_buffer_unref(new_buf);
		}
		else
			out_buf = new_buf;

		if (b->need_copy) {
			pr_info(self, "copy");
			memcpy(GST_BUFFER_DATA(out_buf), b->data, b->len);
		}

		GST_BUFFER_SIZE(out_buf) = b->len;
	}
	else {
		out_buf = gst_dsp_buffer_new(self, tb);

		if (!self->use_pinned)
			/* invalidate data to force reallocation */
			b->data = b->allocated_data = NULL;
	}

	if (G_UNLIKELY(self->skip_hack > 0)) {
		self->skip_hack--;
		gst_buffer_unref(out_buf);
		goto leave;
	}

	if (!keyframe)
		GST_BUFFER_FLAGS(out_buf) |= GST_BUFFER_FLAG_DELTA_UNIT;

	g_mutex_lock(self->ts_mutex);
	timestamp = self->ts_array[self->ts_out_pos].time;
	duration = self->ts_array[self->ts_out_pos].duration;
	self->ts_out_pos = (self->ts_out_pos + 1) % ARRAY_SIZE(self->ts_array);
	self->ts_push_pos = self->ts_out_pos;
	self->ts_count--;

	if (G_UNLIKELY(g_atomic_int_get(&self->deferred_eos)) && self->ts_count == 0)
		got_eos = TRUE;
#ifdef TS_COUNT
	if (self->ts_count > 2 || self->ts_count < 1)
		pr_info(self, "tsc=%lu", self->ts_count);
#endif
	g_mutex_unlock(self->ts_mutex);

	if (!GST_CLOCK_TIME_IS_VALID(duration) && self->default_duration) {
		duration = self->default_duration;
		pr_debug(self, "using default duration %" GST_TIME_FORMAT, GST_TIME_ARGS(duration));
	}
	else if (GST_CLOCK_TIME_IS_VALID(duration) && !self->default_duration)
		self->default_duration = duration;

	GST_BUFFER_TIMESTAMP(out_buf) = timestamp;
	GST_BUFFER_DURATION(out_buf) = duration;

	pr_debug(self, "pushing buffer %" GST_TIME_FORMAT,
		 GST_TIME_ARGS(GST_BUFFER_TIMESTAMP(out_buf)));
	ret = gst_pad_push(self->srcpad, out_buf);
	if (G_UNLIKELY(ret != GST_FLOW_OK)) {
		pr_info(self, "pad push failed: %s", gst_flow_get_name(ret));
		goto leave;
	}

leave:
	handled = tb->pinned && out_buf;
	if (G_UNLIKELY(got_eos)) {
		pr_info(self, "got eos");
		self->eos_start.tv_sec = self->eos_start.tv_nsec = 0;
		gst_pad_push_event(self->srcpad, gst_event_new_eos());
		g_atomic_int_set(&self->eos, true);
		g_atomic_int_set(&self->deferred_eos, false);
		ret = GST_FLOW_UNEXPECTED;
		if (self->use_pinned) {
			if (!handled)
				self->send_buffer(self, tb);
			goto nok;
		}
		/*<commit_msg>base: add debug messages for eos time<commit_after>/*
 * Copyright (C) 2009-2010 <NAME>
 *
 * Author: <NAME> <<EMAIL>>
 *
 * This file may be used under the terms of the GNU Lesser General Public
 * License version 2.1, a copy of which is found in LICENSE included in the
 * packaging of this file.
 */

#include "gstdspbase.h"
#include "gstdspbuffer.h"
#include "plugin.h"

#include "dsp_bridge.h"

#include <string.h> /* for memcpy */

#include "util.h"
#include "log.h"

#define GST_CAT_DEFAULT gstdsp_debug

static inline bool send_buffer(GstDspBase *self, struct td_buffer *tb);

static inline void
map_buffer(GstDspBase *self,
	   GstBuffer *g_buf,
	   struct td_buffer *tb);

struct td_codec td_fake_codec = {
	.uuid = NULL,
};

static inline long
get_elapsed_eos(GstDspBase *self)
{
	struct timespec cur;
	long start, elapsed;

	if (!self->eos_start.tv_sec)
		return 0;

	clock_gettime(CLOCK_MONOTONIC, &cur);

	start = self->eos_start.tv_sec * 1000 + self->eos_start.tv_nsec / 1000000;
	elapsed = cur.tv_sec * 1000 + cur.tv_nsec / 1000000 - start;

	return elapsed;
}

static inline void
check_elapsed(GstDspBase *self)
{
	long elapsed = get_elapsed_eos(self);
	if (elapsed > 500)
		pr_warning(self, "eos took %lu ms", elapsed);
}

du_port_t *
du_port_new(int id,
	    int dir)
{
	du_port_t *p;
	p = calloc(1, sizeof(*p));
	if (!p)
		return NULL;

	p->id = id;
	p->queue = async_queue_new();
	p->dir = dir;

	return p;
}

void
du_port_free(du_port_t *p)
{
	if (!p)
		return;

	free(p->buffers);
	async_queue_free(p->queue);

	free(p);
}

void
du_port_alloc_buffers(du_port_t *p, guint num_buffers)
{
	p->num_buffers = num_buffers;
	free(p->buffers);
	p->buffers = calloc(num_buffers, sizeof(*p->buffers));
	for (unsigned i = 0; i < p->num_buffers; i++)
		p->buffers[i].port = p;
}

static inline void
du_port_flush(du_port_t *p)
{
	guint i;
	struct td_buffer *tb = p->buffers;

	for (i = 0; i < p->num_buffers; i++, tb++) {
		dmm_buffer_t *b = tb->data;
		if (!b)
			continue;
		if (tb->user_data)
			gst_buffer_unref(tb->user_data);
		dmm_buffer_free(b);
		tb->data = NULL;
	}
	async_queue_flush(p->queue);
}

static inline void
g_sem_down_status(GSem *sem,
		  const GstFlowReturn *status)
{
	GstFlowReturn ret = GST_FLOW_OK;
	g_mutex_lock(sem->mutex);

	while (sem->count == 0 &&
	       (ret = g_atomic_int_get(status)) == GST_FLOW_OK)
		g_cond_wait(sem->condition, sem->mutex);

	if (ret == GST_FLOW_OK)
		sem->count--;

	g_mutex_unlock(sem->mutex);
}

static inline void
g_sem_signal(GSem *sem)
{
	g_mutex_lock(sem->mutex);
	g_cond_signal(sem->condition);
	g_mutex_unlock(sem->mutex);
}

static inline void
g_sem_reset(GSem *sem,
	    guint count)
{
	g_mutex_lock(sem->mutex);
	sem->count = count;
	g_mutex_unlock(sem->mutex);
}

typedef struct {
	uint32_t buffer_data;
	uint32_t buffer_size;
	uint32_t param_data;
	uint32_t param_size;
	uint32_t buffer_len;
	uint32_t silly_eos;
	uint32_t silly_buf_state;
	uint32_t silly_buf_active;
	uint32_t silly_buf_id;
#if SN_API >= 2
	uint32_t nb_available_buf;
	uint32_t donot_flush_buf;
	uint32_t donot_invalidate_buf;
#endif
	uint32_t reserved;
	uint32_t msg_virt;
	uint32_t buffer_virt;
	uint32_t param_virt;
	uint32_t silly_out_buffer_index;
	uint32_t silly_in_buffer_index;
	uint32_t user_data;
	uint32_t stream_id;
} dsp_comm_t;

static GstElementClass *parent_class;

static inline void
got_message(GstDspBase *self,
	    struct dsp_msg *msg)
{
	int32_t id;
	uint32_t command_id;

	id = msg->cmd & 0x000000ff;
	command_id = msg->cmd & 0xffffff00;

	switch (command_id) {
	case 0x0600: {
		dmm_buffer_t *b;
		du_port_t *p;
		dsp_comm_t *msg_data;
		dmm_buffer_t *param;
		unsigned i;
		struct td_buffer *tb = NULL;

		for (i = 0; i < ARRAY_SIZE(self->ports); i++)
			if (self->ports[i]->id == id) {
				p = self->ports[i];
				break;
			}

		if (i >= ARRAY_SIZE(self->ports))
			g_error("bad port index: %i", id);

		pr_debug(self, "got %s buffer", id == 0 ? "input" : "output");

		for (i = 0; i < p->num_buffers; i++) {
			if (msg->arg_1 == (uint32_t) p->buffers[i].comm->map) {
				tb = &p->buffers[i];
				break;
			}
		}

		if (!tb)
			g_error("buffer mismatch");

		dmm_buffer_end(tb->comm, tb->comm->size);

		msg_data = tb->comm->data;
		b = (void *) msg_data->user_data;
		b->len = msg_data->buffer_len;

		if (G_UNLIKELY(b->len > b->size))
			g_error("wrong buffer size");

		if (tb->pinned)
			dmm_buffer_end(b, b->len);
		else
			dmm_buffer_unmap(b);

		param = (void *) msg_data->param_virt;
		if (param)
			dmm_buffer_end(param, param->size);

		if (p->recv_cb)
			p->recv_cb(self, tb);

		if (id == 0) {
			if (tb->user_data) {
				gst_buffer_unref(tb->user_data);
				tb->user_data = NULL;
			}
		}

		async_queue_push(p->queue, tb);
		break;
	}
	case 0x0500:
		pr_debug(self, "got flush");
		break;
	case 0x0200:
		pr_debug(self, "got stop");
		g_sem_up(self->flush);
		break;
	case 0x0400:
		pr_debug(self, "got alg ctrl");
		dmm_buffer_free(self->alg_ctrl);
		self->alg_ctrl = NULL;
		break;
	case 0x0e00:
		if (msg->arg_1 == 1 && msg->arg_2 == 0x0500) {
			pr_debug(self, "playback completed");
			break;
		}

		if (msg->arg_1 == 1 && (msg->arg_2 & 0x0600) == 0x0600) {
			struct td_codec *codec = self->codec;
			if (codec->update_params)
				codec->update_params(self, self->node, msg->arg_2);
			break;
		}

		pr_warning(self, "DSP event: cmd=0x%04X, arg1=%u, arg2=0x%04X",
			   msg->cmd, msg->arg_1, msg->arg_2);
		if ((msg->arg_2 & 0x0F00) == 0x0F00)
			gstdsp_got_error(self, 0, "algo error");
		break;
	default:
		pr_warning(self, "unhandled command: %u", command_id);
	}
}

static inline void
setup_buffers(GstDspBase *self)
{
	GstBuffer *buf = NULL;
	dmm_buffer_t *b;
	du_port_t *p;
	guint i;

	p = self->ports[0];
	for (i = 0; i < p->num_buffers; i++) {
		p->buffers[i].data = b = dmm_buffer_new(self->dsp_handle, self->proc, p->dir);
		async_queue_push(p->queue, &p->buffers[i]);
	}

	p = self->ports[1];
	for (i = 0; i < p->num_buffers; i++) {
		struct td_buffer *tb = &p->buffers[i];
		tb->data = b = dmm_buffer_new(self->dsp_handle, self->proc, p->dir);

		if (self->use_pad_alloc) {
			GstFlowReturn ret;
			ret = gst_pad_alloc_buffer_and_set_caps(self->srcpad,
								GST_BUFFER_OFFSET_NONE,
								self->output_buffer_size,
								GST_PAD_CAPS(self->srcpad),
								&buf);
			/* might fail if not (yet) linked */
			if (G_UNLIKELY(ret != GST_FLOW_OK)) {
				pr_err(self, "couldn't allocate buffer: %s", gst_flow_get_name(ret));
				dmm_buffer_allocate(b, self->output_buffer_size);
				b->need_copy = true;
			} else {
				map_buffer(self, buf, tb);
				gst_buffer_unref(buf);
			}
		}
		else {
			dmm_buffer_allocate(b, self->output_buffer_size);
			if (self->use_pinned) {
				dmm_buffer_map(b);
				tb->pinned = tb->clean = true;
			}
		}

		self->send_buffer(self, tb);
	}
}

static inline void
pause_task(GstDspBase *self, GstFlowReturn status)
{
	bool deferred_eos;

	/* synchronize to ensure we are not dropping the EOS event */
	g_mutex_lock(self->ts_mutex);
	(void) g_atomic_int_compare_and_exchange(&self->status, GST_FLOW_OK, status);
	deferred_eos = g_atomic_int_compare_and_exchange(&self->deferred_eos, true, false);
	g_mutex_unlock(self->ts_mutex);

	pr_info(self, "pausing task; reason %s", gst_flow_get_name(status));
	gst_pad_pause_task(self->srcpad);

	/* avoid waiting for buffers that will never come */
	async_queue_disable(self->ports[0]->queue);
	async_queue_disable(self->ports[1]->queue);

	/* there's a pending deferred EOS, it's now or never */
	if (deferred_eos) {
		pr_info(self, "send elapsed eos");
		check_elapsed(self);
		self->eos_start.tv_sec = self->eos_start.tv_nsec = 0;
		gst_pad_push_event(self->srcpad, gst_event_new_eos());
		g_atomic_int_set(&self->eos, true);
	}
}

static inline GstFlowReturn
check_status(GstDspBase *self)
{
	GstFlowReturn ret;
	ret = g_atomic_int_get(&self->status);
	if (G_UNLIKELY(ret != GST_FLOW_OK))
		pause_task(self, ret);
	return ret;
}

static void
output_loop(gpointer data)
{
	GstPad *pad;
	GstDspBase *self;
	GstFlowReturn ret = GST_FLOW_OK;
	GstBuffer *out_buf = NULL;
	dmm_buffer_t *b;
	gboolean flush_buffer;
	gboolean got_eos = FALSE;
	gboolean keyframe = FALSE;
	GstEvent *event;
	du_port_t *p;
	struct td_buffer *tb;
	bool handled;
	GstClockTime timestamp, duration;

	pad = data;
	self = GST_DSP_BASE(GST_OBJECT_PARENT(pad));
	p = self->ports[1];

	pr_debug(self, "begin");
	tb = async_queue_pop(p->queue);

	/*
	 * queue might have been disabled above, so perhaps tb == NULL,
	 * but then right here in between self->status may have been set to
	 * OK by e.g. FLUSH_STOP
	 */
	if (G_UNLIKELY(!tb)) {
		pr_info(self, "no buffer");
		ret = check_status(self);
		goto nok;
	}

	b = tb->data;

	ret = check_status(self);
	if (G_UNLIKELY(ret != GST_FLOW_OK)) {
		async_queue_push(p->queue, tb);
		goto end;
	}

	if (G_UNLIKELY(self->skip_hack_2 > 0)) {
		self->skip_hack_2--;
		goto leave;
	}

	/* check for too many buffers returned */
	g_mutex_lock(self->ts_mutex);
	if (G_UNLIKELY(b->len && !self->ts_count)) {
		pr_warning(self, "no timestamp; unexpected buffer");
		g_mutex_unlock(self->ts_mutex);
		goto leave;
	}
	g_mutex_unlock(self->ts_mutex);

	/* first clear pending events */
	g_mutex_lock(self->ts_mutex);
	while ((event = self->ts_array[self->ts_out_pos].event)) {
		self->ts_array[self->ts_out_pos].event = NULL;
		flush_buffer = (self->ts_out_pos != self->ts_push_pos);
		self->ts_out_pos = (self->ts_out_pos + 1) % ARRAY_SIZE(self->ts_array);
		if (G_LIKELY(!flush_buffer)) {
			self->ts_push_pos = self->ts_out_pos;
			pr_debug(self, "pushing event: %s", GST_EVENT_TYPE_NAME(event));
			gst_pad_push_event(self->srcpad, event);
		} else {
			pr_debug(self, "ignored flushed event: %s", GST_EVENT_TYPE_NAME(event));
			gst_event_unref(event);
		}
	}
	g_mutex_unlock(self->ts_mutex);

	/* a pending reallocation from the previous run */
	if (G_UNLIKELY(!b->data)) {
		dmm_buffer_allocate(b, self->output_buffer_size);
		send_buffer(self, tb);
		goto end;
	}

	if (G_UNLIKELY(!b->len)) {
		/* no need to process this buffer */
		/* no real frame data, so no need to consume a real frame's ts */
		goto leave;
	}

	g_mutex_lock(self->ts_mutex);
	flush_buffer = (self->ts_out_pos != self->ts_push_pos);
	g_mutex_unlock(self->ts_mutex);

	if (G_UNLIKELY(flush_buffer)) {
		g_mutex_lock(self->ts_mutex);
		pr_debug(self, "ignored flushed output buffer for %" GST_TIME_FORMAT,
			 GST_TIME_ARGS((self->ts_array[self->ts_out_pos].time)));
		self->ts_count--;
		self->ts_out_pos = (self->ts_out_pos + 1) % ARRAY_SIZE(self->ts_array);
		g_mutex_unlock(self->ts_mutex);
		goto leave;
	}

	/* now go after the data, but let's first see if it is keyframe */
	keyframe = tb->keyframe;

	if (self->use_pad_alloc) {
		GstBuffer *new_buf;

		ret = gst_pad_alloc_buffer_and_set_caps(self->srcpad,
							GST_BUFFER_OFFSET_NONE,
							self->output_buffer_size,
							GST_PAD_CAPS(self->srcpad),
							&new_buf);

		if (G_UNLIKELY(ret != GST_FLOW_OK)) {
			pr_info(self, "couldn't allocate buffer: %s", gst_flow_get_name(ret));
			async_queue_push(p->queue, tb);
			goto nok;
		}

		if (tb->user_data) {
			out_buf = tb->user_data;
			tb->user_data = NULL;
			map_buffer(self, new_buf, tb);
			gst_buffer_unref(new_buf);
		}
		else
			out_buf = new_buf;

		if (b->need_copy) {
			pr_info(self, "copy");
			memcpy(GST_BUFFER_DATA(out_buf), b->data, b->len);
		}

		GST_BUFFER_SIZE(out_buf) = b->len;
	}
	else {
		out_buf = gst_dsp_buffer_new(self, tb);

		if (!self->use_pinned)
			/* invalidate data to force reallocation */
			b->data = b->allocated_data = NULL;
	}

	if (G_UNLIKELY(self->skip_hack > 0)) {
		self->skip_hack--;
		gst_buffer_unref(out_buf);
		goto leave;
	}

	if (!keyframe)
		GST_BUFFER_FLAGS(out_buf) |= GST_BUFFER_FLAG_DELTA_UNIT;

	g_mutex_lock(self->ts_mutex);
	timestamp = self->ts_array[self->ts_out_pos].time;
	duration = self->ts_array[self->ts_out_pos].duration;
	self->ts_out_pos = (self->ts_out_pos + 1) % ARRAY_SIZE(self->ts_array);
	self->ts_push_pos = self->ts_out_pos;
	self->ts_count--;

	if (G_UNLIKELY(g_atomic_int_get(&self->deferred_eos)) && self->ts_count == 0)
		got_eos = TRUE;
#ifdef TS_COUNT
	if (self->ts_count > 2 || self->ts_count < 1)
		pr_info(self, "tsc=%lu", self->ts_count);
#endif
	g_mutex_unlock(self->ts_mutex);

	if (!GST_CLOCK_TIME_IS_VALID(duration) && self->default_duration) {
		duration = self->default_duration;
		pr_debug(self, "using default duration %" GST_TIME_FORMAT, GST_TIME_ARGS(duration));
	}
	else if (GST_CLOCK_TIME_IS_VALID(duration) && !self->default_duration)
		self->default_duration = duration;

	GST_BUFFER_TIMESTAMP(out_buf) = timestamp;
	GST_BUFFER_DURATION(out_buf) = duration;

	pr_debug(self, "pushing buffer %" GST_TIME_FORMAT,
		 GST_TIME_ARGS(GST_BUFFER_TIMESTAMP(out_buf)));
	ret = gst_pad_push(self->srcpad, out_buf);
	if (G_UNLIKELY(ret != GST_FLOW_OK)) {
		pr_info(self, "pad push failed: %s", gst_flow_get_name(ret));
		goto leave;
	}

leave:
	handled = tb->pinned && out_buf;
	if (G_UNLIKELY(got_eos)) {
		pr_info(self, "got eos");
		check_elapsed(self);
		self->eos_start.tv_sec = self->eos_start.tv_nsec = 0;
		gst_pad_push_event(self->srcpad, gst_event_new_eos());
		g_atomic_int_set(&self->eos, true);
		g_atomic_int_set(&self->deferred_eos, false);
		ret = GST_FLOW_UNEXPECTED;
		if (self->use_pinned) {
			if (!handled)
				self->send_buffer(self, tb);
			goto nok;
		}
		/*
{'fraction_non_alphanumeric': 0.1483062049171041, 'fraction_numerical': 0.01401110313833604, 'mean_word_length': 4.265460330085504, 'pattern_counts': {'":': 0, '<': 27, '<?xml version=': 0, '>': 446, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '169672', 'n_tokens_mistral': 12222, 'n_tokens_neox': 11025, 'n_words': 2145}