libspark/record: Implement writer thread. May or may not improve anything.

Origin commit data
------------------
Branch: master
Commit: ada3f5e24a
Author: martii <m4rtii@gmx.de>
Date: 2014-06-08 (Sun, 08 Jun 2014)


------------------
No further description and justification available within origin commit message!

------------------
This commit was generated by Migit

libspark/record.cpp

@@ -8,14 +8,12 @@
 #include <cstdio>
 #include <cstring>
 
-#include <aio.h>
-
 #include "record_lib.h"
 #include "lt_debug.h"
 #define lt_debug(args...) _lt_debug(TRIPLE_DEBUG_RECORD, this, args)
 #define lt_info(args...) _lt_info(TRIPLE_DEBUG_RECORD, this, args)
 
-/* helper function to call the cpp thread loop */
+/* helper functions to call the cpp thread loops */
 void *execute_record_thread(void *c)
 {
 	cRecord *obj = (cRecord *)c;
@@ -23,6 +21,13 @@ void *execute_record_thread(void *c)
 	return NULL;
 }
 
+void *execute_writer_thread(void *c)
+{
+	cRecord *obj = (cRecord *)c;
+	obj->WriterThread();
+	return NULL;
+}
+
 cRecord::cRecord(int num, int bs_dmx, int bs)
 {
 	lt_info("%s %d\n", __func__, num);
@@ -174,6 +179,32 @@ bool cRecord::AddPid(unsigned short pid)
 	return dmx->addPid(pid);
 }
 
+void cRecord::WriterThread()
+{
+	char threadname[17];
+	strncpy(threadname, "WriterThread", sizeof(threadname));
+	threadname[16] = 0;
+	prctl (PR_SET_NAME, (unsigned long)&threadname);
+	unsigned int chunk = 0;
+	while (!sem_wait(&sem)) {
+		if (!io_len[chunk]) // empty, assume end of recording
+			return;
+		unsigned char *p_buf = io_buf[chunk];
+		size_t p_len = io_len[chunk];
+		while (p_len) {
+			ssize_t written = write(file_fd, p_buf, p_len);
+			if (written < 0)
+				break;
+			p_len -= written;
+			p_buf += written;
+		}
+		if (posix_fadvise(file_fd, 0, 0, POSIX_FADV_DONTNEED))
+			perror("posix_fadvise");
+		chunk++;
+		chunk %= RECORD_WRITER_CHUNKS;
+	}
+}
+
 void cRecord::RecordThread()
 {
 	lt_info("%s: begin\n", __func__);
@@ -181,15 +212,12 @@ void cRecord::RecordThread()
 	strncpy(threadname, "RecordThread", sizeof(threadname));
 	threadname[16] = 0;
 	prctl (PR_SET_NAME, (unsigned long)&threadname);
-	int readsize = bufsize/16;
+	int readsize = (bufsize/(RECORD_WRITER_CHUNKS*188)) * 188;
-	int buf_pos = 0;
+	uint8_t *buf = (uint8_t *)malloc(readsize * RECORD_WRITER_CHUNKS);
-	int queued = 0;
+fprintf(stderr, "%s %d\n", __FILE__, __LINE__);
-	uint8_t *buf;
-	struct aiocb a;
-
-	buf = (uint8_t *)malloc(bufsize);
 	if (!buf)
 	{
+fprintf(stderr, "%s %d\n", __FILE__, __LINE__);
 		exit_flag = RECORD_FAILED_MEMORY;
 		lt_info("%s: unable to allocate buffer! (out of memory)\n", __func__);
 		if (failureCallback)
@@ -202,113 +230,62 @@ void cRecord::RecordThread()
 	if (fcntl(file_fd, F_SETFL, val|O_APPEND))
 		lt_info("%s: O_APPEND? (%m)\n", __func__);
 
-	memset(&a, 0, sizeof(a));
+	for (unsigned int chunk = 0; chunk < RECORD_WRITER_CHUNKS; chunk++) {
-	a.aio_fildes = file_fd;
+		io_buf[chunk] = buf + chunk * readsize;
-	a.aio_sigevent.sigev_notify = SIGEV_NONE;
+		io_len[chunk] = 0;
+	}
 
-	dmx->Start();
+	sem_init(&sem, 0, 0);
-	int overflow_count = 0;
+	pthread_t writer_thread;
-	bool overflow = false;
+fprintf(stderr, "%s %d\n", __FILE__, __LINE__);
-	int r = 0;
+	if (pthread_create(&writer_thread, 0, execute_writer_thread, this))
-	while (exit_flag == RECORD_RUNNING)
+		exit_flag = RECORD_FAILED_FILE;
-	{
+	else {
-		if (buf_pos < bufsize)
+fprintf(stderr, "%s %d\n", __FILE__, __LINE__);
+		dmx->Start();
+fprintf(stderr, "%s %d\n", __FILE__, __LINE__);
+		int overflow_count = 0;
+		unsigned int chunk = 0;
+
+		while (exit_flag == RECORD_RUNNING)
 		{
-			if (overflow_count) {
+			ssize_t s = dmx->Read(io_buf[chunk], readsize, 50);
-				lt_info("%s: Overflow cleared after %d iterations\n", __func__, overflow_count);
+			lt_debug("%s: Read chunk=%d size=%d\n", __func__, chunk, s);
-				overflow_count = 0;
-			}
-			int toread = bufsize - buf_pos;
-			if (toread > readsize)
-				toread = readsize;
-			ssize_t s = dmx->Read(buf + buf_pos, toread, 50);
-			lt_debug("%s: buf_pos %6d s %6d / %6d\n", __func__,
-				buf_pos, (int)s, bufsize - buf_pos);
 			if (s < 0)
 			{
-				if (errno != EAGAIN && (errno != EOVERFLOW || !overflow))
+				if (errno != EAGAIN && (errno != EOVERFLOW || overflow_count > 63 /* arbitrary */))
 				{
 					lt_info("%s: read failed: %m\n", __func__);
 					exit_flag = RECORD_FAILED_READ;
 					break;
 				}
+				if (!overflow_count)
+					lt_info("%s: dmx->Read(): %m\n", __func__);
+				overflow_count++;
+				continue;
 			}
-			else
+			if (overflow_count) {
-			{
+				lt_info("%s: Overflow cleared after %d iterations\n", __func__, overflow_count);
-				overflow = false;
-				buf_pos += s;
-			}
-		}
-		else
-		{
-			if (!overflow)
 				overflow_count = 0;
-			overflow = true;
+			}
-			if (!(overflow_count % 10))
+			if (!s)
-				lt_info("%s: buffer full! Overflow? (%d)\n", __func__, ++overflow_count);
+				continue;
-		}
-		r = aio_error(&a);
-		if (r == EINPROGRESS)
-		{
-			lt_debug("%s: aio in progress, free: %d\n", __func__, bufsize - buf_pos);
-			continue;
-		}
-		// not calling aio_return causes a memory leak  --martii
-		r = aio_return(&a);
-		if (r < 0)
-		{
-			exit_flag = RECORD_FAILED_FILE;
-			lt_debug("%s: aio_return = %d (%m)\n", __func__, r);
-			break;
-		}
-		else
-			lt_debug("%s: aio_return = %d, free: %d\n", __func__, r, bufsize - buf_pos);
-		if (posix_fadvise(file_fd, 0, 0, POSIX_FADV_DONTNEED))
-			perror("posix_fadvise");
-		if (queued)
-		{
-			memmove(buf, buf + queued, buf_pos - queued);
-			buf_pos -= queued;
-		}
-		queued = buf_pos;
-		a.aio_buf = buf;
-		a.aio_nbytes = queued;
-		r = aio_write(&a);
-		if (r)
-		{
-			lt_info("%s: aio_write %d (%m)\n", __func__, r);
-			exit_flag = RECORD_FAILED_FILE;
-			break;
-		}
-	}
-	dmx->Stop();
-	while (true) /* write out the unwritten buffer content */
-	{
-		lt_debug("%s: run-out write, buf_pos %d\n", __func__, buf_pos);
-		r = aio_error(&a);
-		if (r == EINPROGRESS)
-		{
-			usleep(50000);
-			continue;
-		}
-		r = aio_return(&a);
-		if (r < 0)
-		{
-			exit_flag = RECORD_FAILED_FILE;
-			lt_info("%s: aio_result: %d (%m)\n", __func__, r);
-			break;
-		}
-		if (!queued)
-			break;
-		memmove(buf, buf + queued, buf_pos - queued);
-		buf_pos -= queued;
-		queued = buf_pos;
-		a.aio_buf = buf;
-		a.aio_nbytes = queued;
-		r = aio_write(&a);
-	}
-	free(buf);
 
+			io_len[chunk] = s;
+			sem_post(&sem);
+			chunk++;
+			chunk %= RECORD_WRITER_CHUNKS;
+		}
+fprintf(stderr, "%s %d\n", __FILE__, __LINE__);
+		dmx->Stop();
+fprintf(stderr, "%s %d\n", __FILE__, __LINE__);
+		io_len[chunk] = 0;
+		sem_post(&sem);
+fprintf(stderr, "trying to join writer\n");
+		pthread_join(writer_thread, NULL);
+fprintf(stderr, "%s %d\n", __FILE__, __LINE__);
+		free(buf);
+	}
 #if 0
 	// TODO: do we need to notify neutrino about failing recording?
 	CEventServer eventServer;

libspark/record_lib.h

@@ -2,6 +2,7 @@
 #define __RECORD_TD_H
 
 #include <pthread.h>
+#include <semaphore.h>
 #include "dmx_lib.h"
 
 #define REC_STATUS_OK 0
@@ -32,6 +33,11 @@ class cRecord
 		int bufsize_dmx;
 		void (*failureCallback)(void *);
 		void *failureData;
+
+		sem_t sem;
+#define RECORD_WRITER_CHUNKS 16
+		unsigned char *io_buf[RECORD_WRITER_CHUNKS];
+		size_t io_len[RECORD_WRITER_CHUNKS];
 	public:
 		cRecord(int num = 0, int bs_dmx = 100 * 188 * 1024, int bs = 100 * 188 * 1024);
 		void setFailureCallback(void (*f)(void *), void *d) { failureCallback = f; failureData = d; }
@@ -46,5 +52,6 @@ class cRecord
 		bool ChangePids(unsigned short vpid, unsigned short *apids, int numapids);
 
 		void RecordThread();
+		void WriterThread();
 };
 #endif