From 19e7fe3cac6795cde532ed88fb53cba62d93e4dd Mon Sep 17 00:00:00 2001 From: max_10 Date: Mon, 7 Jul 2014 21:14:22 +0200 Subject: [PATCH] Revert "libspark/record: Implement writer thread. May or may not improve anything." This reverts commit c6fe034f4cb7b9163ac84624da61576e7b69d536. Origin commit data ------------------ Branch: master Commit: https://github.com/neutrino-images/ni-libstb-hal/commit/677776e03a5efc70727838a0bd7b19c09831d5ec Author: max_10 Date: 2014-07-07 (Mon, 07 Jul 2014) ------------------ This commit was generated by Migit --- libspark/record.cpp | 185 ++++++++++++++++++++++++------------------ libspark/record_lib.h | 7 -- 2 files changed, 104 insertions(+), 88 deletions(-) diff --git a/libspark/record.cpp b/libspark/record.cpp index 186d5ef..ad8a5cf 100644 --- a/libspark/record.cpp +++ b/libspark/record.cpp @@ -8,12 +8,14 @@ #include #include +#include + #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; diff --git a/libspark/record_lib.h b/libspark/record_lib.h index 15fc87b..e7c5772 100644 --- a/libspark/record_lib.h +++ b/libspark/record_lib.h @@ -2,7 +2,6 @@ #define __RECORD_TD_H #include -#include #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