From ada3f5e24ae0df499b53f69d761b9ddfcd16aef9 Mon Sep 17 00:00:00 2001 From: martii Date: Sun, 8 Jun 2014 19:27:12 +0200 Subject: [PATCH] libspark/record: Implement writer thread. May or may not improve anything. --- libspark/record.cpp | 187 ++++++++++++++++++------------------------ libspark/record_lib.h | 7 ++ 2 files changed, 89 insertions(+), 105 deletions(-) diff --git a/libspark/record.cpp b/libspark/record.cpp index ad8a5cf..186d5ef 100644 --- a/libspark/record.cpp +++ b/libspark/record.cpp @@ -8,14 +8,12 @@ #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 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 buf_pos = 0; - int queued = 0; - uint8_t *buf; - struct aiocb a; - - buf = (uint8_t *)malloc(bufsize); + 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__); 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)); - a.aio_fildes = file_fd; - a.aio_sigevent.sigev_notify = SIGEV_NONE; + for (unsigned int chunk = 0; chunk < RECORD_WRITER_CHUNKS; chunk++) { + io_buf[chunk] = buf + chunk * readsize; + io_len[chunk] = 0; + } - dmx->Start(); - int overflow_count = 0; - bool overflow = false; - int r = 0; - while (exit_flag == RECORD_RUNNING) - { - if (buf_pos < bufsize) + 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) { - 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); + ssize_t s = dmx->Read(io_buf[chunk], readsize, 50); + lt_debug("%s: Read chunk=%d size=%d\n", __func__, chunk, s); 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 - { - overflow = false; - buf_pos += s; - } - } - else - { - if (!overflow) + if (overflow_count) { + lt_info("%s: Overflow cleared after %d iterations\n", __func__, overflow_count); 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 (!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); + } #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 e7c5772..15fc87b 100644 --- a/libspark/record_lib.h +++ b/libspark/record_lib.h @@ -2,6 +2,7 @@ #define __RECORD_TD_H #include +#include #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