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

This reverts commit c6fe034f4c. Origin commit data ------------------ Branch: master Commit: 677776e03a Author: max_10 <max_10@gmx.de> Date: 2014-07-07 (Mon, 07 Jul 2014) ------------------ This commit was generated by Migit
2025-08-26 23:12:44 +02:00 · 2014-07-07 21:14:22 +02:00
parent 1da50312ec
commit 19e7fe3cac
2 changed files with 104 additions and 88 deletions
--- a/libspark/record.cpp
+++ b/libspark/record.cpp
@@ -8,12 +8,14 @@
 #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 functions to call the cpp thread loops */
+/* helper function to call the cpp thread loop */
 void *execute_record_thread(void *c)
 {
 	cRecord *obj = (cRecord *)c;
@@ -21,13 +23,6 @@ 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);
@@ -179,32 +174,6 @@ 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__);
@@ -212,12 +181,15 @@ void cRecord::RecordThread()
 	strncpy(threadname, "RecordThread", sizeof(threadname));
 	threadname[16] = 0;
 	prctl (PR_SET_NAME, (unsigned long)&threadname);
-	int readsize = (bufsize/(RECORD_WRITER_CHUNKS*188)) * 188;
-	uint8_t *buf = (uint8_t *)malloc(readsize * RECORD_WRITER_CHUNKS);
-fprintf(stderr, "%s %d\n", __FILE__, __LINE__);
+	int readsize = bufsize/16;
+	int buf_pos = 0;
+	int queued = 0;
+	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)
@@ -230,62 +202,113 @@ fprintf(stderr, "%s %d\n", __FILE__, __LINE__);
 	if (fcntl(file_fd, F_SETFL, val|O_APPEND))
 		lt_info("%s: O_APPEND? (%m)\n", __func__);

-	for (unsigned int chunk = 0; chunk < RECORD_WRITER_CHUNKS; chunk++) {
-		io_buf[chunk] = buf + chunk * readsize;
-		io_len[chunk] = 0;
-	}
+	memset(&a, 0, sizeof(a));
+	a.aio_fildes = file_fd;
+	a.aio_sigevent.sigev_notify = SIGEV_NONE;

-	sem_init(&sem, 0, 0);
-	pthread_t writer_thread;
-fprintf(stderr, "%s %d\n", __FILE__, __LINE__);
-	if (pthread_create(&writer_thread, 0, execute_writer_thread, this))
-		exit_flag = RECORD_FAILED_FILE;
-	else {
-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)
+	dmx->Start();
+	int overflow_count = 0;
+	bool overflow = false;
+	int r = 0;
+	while (exit_flag == RECORD_RUNNING)
+	{
+		if (buf_pos < bufsize)
 		{
-			ssize_t s = dmx->Read(io_buf[chunk], readsize, 50);
-			lt_debug("%s: Read chunk=%d size=%d\n", __func__, chunk, s);
+			if (overflow_count) {
+				lt_info("%s: Overflow cleared after %d iterations\n", __func__, overflow_count);
+				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_count > 63 /* arbitrary */))
+				if (errno != EAGAIN && (errno != EOVERFLOW || !overflow))
 				{
 					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;
 			}
-			if (overflow_count) {
-				lt_info("%s: Overflow cleared after %d iterations\n", __func__, overflow_count);
-				overflow_count = 0;
+			else
+			{
+				overflow = false;
+				buf_pos += s;
 			}
-			if (!s)
-				continue;
-
-			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);
+		else
+		{
+			if (!overflow)
+				overflow_count = 0;
+			overflow = true;
+			if (!(overflow_count % 10))
+				lt_info("%s: buffer full! Overflow? (%d)\n", __func__, ++overflow_count);
+		}
+		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);
+
 #if 0
 	// TODO: do we need to notify neutrino about failing recording?
 	CEventServer eventServer;
--- a/libspark/record_lib.h
+++ b/libspark/record_lib.h
@@ -2,7 +2,6 @@
 #define __RECORD_TD_H

 #include <pthread.h>
-#include <semaphore.h>
 #include "dmx_lib.h"

 #define REC_STATUS_OK 0
@@ -33,11 +32,6 @@ 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; }
@@ -52,6 +46,5 @@ class cRecord
 		bool ChangePids(unsigned short vpid, unsigned short *apids, int numapids);

 		void RecordThread();
-		void WriterThread();
 };
 #endif