Add decode and write circular buffers

Makefile

@@ -10,7 +10,7 @@ FUNCS_LIB = $(FUNCS_DIR)/libfuncs.a
 TS_DIR = libts
 TS_LIB = $(TS_DIR)/libts.a
 
-tsdecrypt_OBJS = data.o udp.o util.o camd.o tables.o tsdecrypt.o $(FUNCS_LIB) $(TS_LIB)
+tsdecrypt_OBJS = cbuf.o data.o udp.o util.o camd.o tables.o tsdecrypt.o $(FUNCS_LIB) $(TS_LIB)
 tsdecrypt_LIBS = -lcrypto -ldvbcsa -lpthread
 
 CLEAN_OBJS = tsdecrypt $(tsdecrypt_OBJS) *~

cbuf.c

@@ -0,0 +1,290 @@
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <netdb.h>
+#include <pthread.h>
+#include <assert.h>
+
+#include "libfuncs/libfuncs.h"
+
+#include "cbuf.h"
+
+static void cbuf_lock(CBUF *b) {
+	pthread_mutex_lock(b->lock);
+}
+
+static void cbuf_unlock(CBUF *b) {
+	pthread_mutex_unlock(b->lock);
+}
+
+/* Returns how much data is filled in the buffer */
+int cbuf_free_data_size(CBUF *b) {
+	int ret = b->size - (b->input - b->output);
+	assert(ret >= 0);
+	return ret;
+}
+
+void cbuf_dump(CBUF *b) {
+	LOGf("CBUF  [%10s]: size:%d pos:%d writepos:%d input:%llu output:%llu free_data:%d buffered:%lld\n",
+		b->name,
+		b->size,
+		b->pos,
+		b->writepos,
+		b->input,
+		b->output,
+		cbuf_free_data_size(b),
+		b->input - b->output
+	);
+/*
+	char *z = b->buffer;
+	printf("cbuf(%s), dump:", b->name);
+	int i;
+	for (i=0;i<b->size;i++) {
+		printf("%c", z[i]);
+	}
+	printf("\n\n");
+*/
+}
+
+CBUF *cbuf_init(int buffer_size, char *name) {
+	CBUF *b = calloc(1, sizeof(CBUF));
+	if (!b)
+		return NULL;
+	if (!buffer_size)
+		return 0;
+	pthread_mutex_t *mutex = malloc(sizeof(pthread_mutex_t));
+	if (pthread_mutex_init(mutex, NULL) != 0) {
+		perror("cbuf_new: mutex_init");
+		return NULL;
+	}
+	b->lock     = mutex;
+	b->name     = strdup(name);
+	b->size     = buffer_size;
+	b->pos      = 0;
+	b->writepos = 0;
+	b->buffer   = calloc(1, buffer_size);
+	if (!b->buffer) {
+		free(b);
+		LOGf("CBUF  [%10s]: Can't allocate buffer size: %d\n", name, buffer_size);
+		return NULL;
+	}
+	return b;
+}
+
+void cbuf_free(CBUF **pb) {
+	CBUF *b = *pb;
+	if (!b)
+		return;
+	pthread_mutex_destroy(b->lock);
+	FREE(b->lock);
+	FREE(b->buffer);
+	FREE(b->name);
+	FREE(*pb);
+}
+
+// Returns -1 on buffer wrap around
+int cbuf_fill(CBUF *b, uint8_t *data, int datasize) {
+	int ret = 0;
+	cbuf_lock(b);
+//	LOGf("  cbuf_fill(%s, '%s', %d)\n", b->name, data, datasize);
+//	cbuf_dump(b);
+	assert(datasize <= b->size);
+	int to_copy = min(datasize, (b->size - b->pos));
+	if (!to_copy || !data) {
+		LOGf("CBUF [%10s]: Nothing to fill.\n", b->name);
+		ret = -2;
+		goto OUT;
+	}
+	if (cbuf_free_data_size(b)-to_copy <= 0) {
+//		LOGf("CBUF [%10s]: Buffer will wrap by (%d bytes). Data not filled, consume more!\n", b->name, -(cbuf_free_data_size(b)-to_copy));
+//		cbuf_dump(b);
+//		b->debug_get = 1;
+		ret = -1;
+		goto OUT;
+	}
+	memcpy(b->buffer + b->pos, data, to_copy);
+	int copied = to_copy;
+	b->pos   += copied; // Move current buffer position
+	b->input += copied;
+	assert(b->pos <= b->size);
+	if (b->pos == b->size) { // Buffer wrap around
+		b->pos = 0;
+	}
+	if (copied < datasize) { // Move the rest
+//		Logs when wrapping
+//		LOGf("cbuf(%10s) copied < datasize, copied:%d datasize:%d datasize-copied: %d pos:%d\n",
+//			b->name, copied, datasize, datasize - copied, b->pos);
+//		cbuf_dump(b);
+		cbuf_unlock(b);
+		ret = cbuf_fill(b, data + copied, datasize - copied);
+		goto OUT;
+	}
+
+OUT:
+	cbuf_unlock(b);
+	return ret;
+}
+
+
+
+
+
+/* Returns how much space is left to the end of the buffer */
+static int cbuf_size_to_end(CBUF *b) {
+	int ret = b->input - b->output;
+	if (b->writepos + ret > b->size) {
+		ret = b->size - b->writepos;
+	}
+	return ret;
+}
+
+int cbuf_data_size(CBUF *b) {
+	return cbuf_size_to_end(b);
+}
+
+void *cbuf_get(CBUF *b, int size, int *ret_datasize) {
+	cbuf_lock(b);
+	void *ret = NULL;
+	int new_size = min(size, cbuf_size_to_end(b));
+	if (b->debug_get) {
+		LOGf("1 cbuf_get(%s, %d) new_size: %d size_to_end: %d\n",
+				b->name, size, new_size, cbuf_size_to_end(b));
+		cbuf_dump(b);
+	}
+	if (new_size <= 0) { // No data
+		*ret_datasize = 0;
+		ret = NULL;
+		goto OUT;
+	}
+	*ret_datasize = new_size;
+	ret = b->buffer + b->writepos;
+	b->writepos += new_size; // Move writepos
+	b->output   += new_size;
+	if (b->writepos > b->size) {
+		LOGf("!!! b->writepos > b->size !!! size:%d new_size:%d\n", size, new_size);
+		cbuf_dump(b);
+		assert(b->writepos <= b->size);
+	}
+	if (b->writepos == b->size) // Buffer wraparound
+		b->writepos = 0;
+
+OUT:
+	if (b->debug_get) {
+		LOGf("2 cbuf_get(%s, %d) new_size: %d size_to_end: %d ret_sz:%d\n",
+				b->name, size, new_size, cbuf_size_to_end(b), *ret_datasize);
+		cbuf_dump(b);
+		b->debug_get = 0;
+	}
+	cbuf_unlock(b);
+	return ret;
+}
+
+void *cbuf_peek(CBUF *b, int size, int *ret_datasize) {
+	cbuf_lock(b);
+	void *ret = NULL;
+	int new_size = min(size, cbuf_size_to_end(b));
+
+	if (new_size <= 0) { // No data
+		*ret_datasize = 0;
+		ret = NULL;
+		goto OUT;
+	}
+	*ret_datasize = new_size;
+	ret = b->buffer + b->writepos;
+
+OUT:
+	cbuf_unlock(b);
+	return ret;
+}
+
+void cbuf_copy(CBUF *from, CBUF *to) {
+//	LOGf("cbuf_copy(%s, %s)\n", from->name, to->name);
+	int data_size;
+	void *data;
+	do {
+		data = cbuf_get(from, from->input - from->output, &data_size);
+		if (from->debug_get)
+			LOGf("copied from %s to %s size=%d\n", from->name, to->name, data_size);
+		if (!data || data_size <= 0)
+			break;
+		cbuf_fill(to, data, data_size);
+	} while (1);
+}
+
+void cbuf_poison(CBUF *b, char poison_byte) {
+	memset(b->buffer, poison_byte, b->size);
+}
+
+
+/*
+void consume(CBUF *b, int size) {
+	int data_size, i;
+	char *data = cbuf_get(b, size, &data_size);
+	if (data && data_size > 0) {
+		printf("Consumed %d Data: \"", data_size);
+		for (i=0;i<data_size;i++) {
+			printf("%c", data[i]);
+		}
+		printf("\"\n");
+	} else {
+		printf("%s", "There is nothing to consume!\n");
+	}
+}
+
+void cbuf_test() {
+	CBUF *in;
+
+	CBUF *out;
+	out = cbuf_init(64, "out");
+	cbuf_poison(out, 'O');
+	cbuf_dump(out);
+
+	in = cbuf_init(4, "in");
+	cbuf_poison(in, '*');
+
+	cbuf_fill(in, "12", 2);
+	cbuf_fill(in, "34", 2);
+	cbuf_fill(in, "z" , 1);
+	cbuf_dump(in);
+
+	cbuf_copy(in, out);
+	cbuf_dump(out);
+	consume(in, 16);
+	cbuf_dump(in);
+
+	cbuf_fill(in, "a", 1);
+	cbuf_fill(in, "b", 1);
+	cbuf_fill(in, "c", 1);
+	cbuf_fill(in, "d", 1);
+	cbuf_dump(in);
+
+	cbuf_copy(in, out);
+	cbuf_dump(out);
+
+	consume(in, 4);
+	cbuf_dump(in);
+
+	cbuf_fill(in, "gfgf", 4);
+	cbuf_dump(in);
+
+	consume(in, 4);
+	cbuf_dump(in);
+
+	cbuf_fill(in, "r", 1);
+	cbuf_fill(in, "r", 1);
+	cbuf_fill(in, "r", 1);
+	cbuf_fill(in, "r", 1);
+	cbuf_dump(in);
+
+	consume(out, 6);
+	cbuf_copy(in, out);
+	consume(out, 6);
+	consume(out, 6);
+	consume(out, 6);
+	consume(out, 6);
+
+	cbuf_free(in);
+	cbuf_free(out);
+
+}
+*/

cbuf.h

@@ -0,0 +1,34 @@
+#ifndef CBUF_H
+#define CBUF_H
+
+#include <netdb.h>
+
+// Circular buffer
+typedef struct {
+	pthread_mutex_t *lock;
+	char *name;
+	int size;			/* Buffer size, must be (bufsize % 1316) == 0 */
+	int pos;			/* Up to where the buffer is filled */
+	int writepos;		/* Up to where the buffer is get */
+	void *buffer;		/* The buffer data */
+	uint64_t input;
+	uint64_t output;
+	int pos_wrapped;
+	int debug_get;
+} CBUF;
+
+CBUF *cbuf_init(int buffer_size, char *name);
+void cbuf_free(CBUF **buffer);
+
+int  cbuf_fill(CBUF *b, uint8_t *data, int datasize);
+void *cbuf_get(CBUF *b, int size, int *ret_datasize);
+void *cbuf_peek(CBUF *b, int size, int *ret_datasize);
+void cbuf_copy(CBUF *src, CBUF *dest);
+
+int cbuf_data_size(CBUF *b);
+
+void cbuf_poison(CBUF *b, char poison_byte);
+
+void cbuf_dump(CBUF *b);
+
+#endif

data.c

@@ -52,6 +52,9 @@ void data_init(struct ts *ts) {
 	ts->output.fd   = 1; // STDOUT
 	ts->output.type = FILE_IO;
 	ts->output.ttl  = 1;
+
+	ts->decode_buf  = cbuf_init((7 * dvbcsa_bs_batch_size() * 188) * 2, "decode");
+	ts->write_buf   = cbuf_init((7 * dvbcsa_bs_batch_size() * 188) * 2, "write");
 }
 
 void data_free(struct ts *ts) {
@@ -72,6 +75,9 @@ void data_free(struct ts *ts) {
 	dvbcsa_bs_key_free(ts->key.bs_csakey[0]);
 	dvbcsa_bs_key_free(ts->key.bs_csakey[1]);
 
+	cbuf_free(&ts->decode_buf);
+	cbuf_free(&ts->write_buf);
+
 	FREE(ts->input.fname);
 	FREE(ts->output.fname);
 }

data.h

@@ -11,6 +11,8 @@
 #include "libfuncs/libfuncs.h"
 #include "libts/tsfuncs.h"
 
+#include "cbuf.h"
+
 struct key {
 	uint8_t				cw[16];
 	int					is_valid_cw;
@@ -90,6 +92,16 @@ struct ts {
 
 	int					camd_stop;
 	int					is_cw_error;
+
+	int					threaded;
+
+	int					decode_stop;
+	pthread_t			decode_thread;
+	CBUF				*decode_buf;
+
+	int					write_stop;
+	pthread_t			write_thread;
+	CBUF				*write_buf;
 };
 
 enum msg_type { EMM_MSG, ECM_MSG };