OSDN Git Service

tunerec: large change to record multi-threaded
authorLong.inus <long.inus@hotmail.co.jp>
Wed, 10 Jul 2013 11:33:29 +0000 (20:33 +0900)
committerLong.inus <long.inus@hotmail.co.jp>
Wed, 10 Jul 2013 11:33:29 +0000 (20:33 +0900)
tunerec/Makefile
tunerec/tunerec.c

index 4d5310d..0c80392 100755 (executable)
@@ -6,8 +6,8 @@ TARGETS = tunerec
 LANG            = C
 CC              = gcc
 
-#CFLAGS         = -std=c99 -O2 -Wall -g
-CFLAGS          = -std=c99 -O2 -Wall -ggdb
+#CFLAGS         = -std=c99 -O2 -Wall -pthread -g
+CFLAGS          = -std=c99 -O2 -Wall -pthread -ggdb
 
 LIBS            =
 
index f426778..3224b35 100755 (executable)
 #include <stdio.h>
 #include <stdlib.h>
 #include <time.h>
+#include <unistd.h>
+#include <errno.h>
+#include <pthread.h>
+#include <stdbool.h>
+#include <string.h>
 
 #define ARRAY_SIZE(arr) (sizeof(arr) / sizeof((arr)[0]))
 #ifndef DTV_STREAM_ID
 #define DTV_STREAM_ID DTV_ISDBS_TS_ID
 #endif
+#define DVB_READ_BUFFER_SIZE 188 * 128
+#define DVB_WRITE_BUFFER_SIZE 1024 * 1024
+#define DMX_BUFFER_SIZE 1024 * 1024
+#define RING_BUFFER_SIZE 1024 * 1024 * 16
+
+typedef struct {
+       void* rb_ptr;
+       int rb_wt;
+       int rb_rt;
+       int rb_ab;
+       pthread_mutex_t rb_mtx;
+} mrb;
+
+typedef struct {
+       int fin;
+       int fout;
+       int rectime;
+       mrb rb;
+       bool done;
+} my_thread_arg;
+
+void * record_read(void * priv);
+void * record_write(void * priv);
+
+void mrb_init(mrb* rb, size_t size) {
+       rb->rb_ptr = malloc(size);
+       rb->rb_wt = rb->rb_rt = rb->rb_ab = 0;
+       pthread_mutex_init(&rb->rb_mtx, NULL);
+}
+
+void mrb_destroy(mrb* rb) {
+       free(rb->rb_ptr);
+       pthread_mutex_destroy(&rb->rb_mtx);
+}
+
+size_t mrb_data_len(mrb* rb) {
+       return rb->rb_ab;
+}
+
+bool mrb_put(mrb* rb, char *source, size_t size) {
+       int partial;
+
+       if ( rb->rb_ab + size > RING_BUFFER_SIZE ) {
+               // RingBuffer overflow.
+               return false;
+       }
+       if ( rb->rb_wt + size <= RING_BUFFER_SIZE ) {
+               // written bytes + newly added bytes <= buffer max size
+               memcpy(rb->rb_ptr + rb->rb_wt, source, size);
+               rb->rb_wt = rb->rb_wt + size;
+       }
+       else {
+               // written bytes + newly added bytes > buffer max size
+               // wrap around.
+               partial = RING_BUFFER_SIZE - rb->rb_wt;
+               memcpy(rb->rb_ptr + rb->rb_wt, source, partial);
+               memcpy(rb->rb_ptr, source + partial, size - partial);
+               rb->rb_wt = rb->rb_wt + size - RING_BUFFER_SIZE;
+       }
+       pthread_mutex_lock(&rb->rb_mtx);
+       rb->rb_ab += size;
+       pthread_mutex_unlock(&rb->rb_mtx);
+
+       return true;
+}
+
+bool mrb_get(mrb* rb, char *target, size_t size) {
+       int partial;
+
+       if ( rb->rb_ab < size ) {
+               // RingBuffer underflow.
+               return false;
+       }
+       if ( rb->rb_rt + size <= rb->rb_wt ) {
+               // read bytes + newly taken bytes <= written max size
+               memcpy(target, rb->rb_ptr + rb->rb_rt, size);
+               rb->rb_rt = rb->rb_rt + size;
+       }
+       else {
+               // Ring Buffer wrapped.
+               partial = RING_BUFFER_SIZE - rb->rb_rt;
+               memcpy(target, rb->rb_ptr + rb->rb_rt, partial);
+               memcpy(target + partial, rb->rb_ptr, size - partial);
+               rb->rb_rt = rb->rb_rt + size - RING_BUFFER_SIZE;
+       }
+       pthread_mutex_lock(&rb->rb_mtx);
+       rb->rb_ab -= size;
+       pthread_mutex_unlock(&rb->rb_mtx);
+
+       return true;
+}
 
 static int search(int adapter_nr, unsigned int frequency, unsigned int ts_id)
 {
@@ -104,9 +200,8 @@ static int track(int adapter_nr)
 void record(int adapter_nr, char* output, int rectime) {
        int fin, fout;
        char input[256];
-       time_t start_time, current_time;
-       char buf[1316];
-       ssize_t rt, wt, shift;
+       pthread_t read_thread, write_thread;
+       my_thread_arg arg;
 
        fout = open(output, (O_WRONLY | O_CREAT | O_TRUNC), 0666);
        if ( fout < 0 ) {
@@ -114,11 +209,88 @@ void record(int adapter_nr, char* output, int rectime) {
                return;
        }
        sprintf(input, "/dev/dvb/adapter%d/dvr0", adapter_nr);
-       start_time = time(NULL);
        fin = open(input, (O_RDONLY));
-       while ( ( rt = read(fin, buf, 1316) ) > 0 ) {
+       ioctl(fin, DMX_SET_BUFFER_SIZE, DMX_BUFFER_SIZE);
+
+       mrb_init(&arg.rb, RING_BUFFER_SIZE);
+       arg.done = false;
+       arg.rectime = rectime;
+       arg.fin  = fin;
+       arg.fout = fout;
+       pthread_create(&read_thread , NULL, record_read , &arg);
+       pthread_create(&write_thread, NULL, record_write, &arg);
+       pthread_join(write_thread, NULL);
+       pthread_join(read_thread, NULL);
+
+       mrb_destroy(&arg.rb);
+       close(fin);
+       close(fout);
+}
+
+void * record_read(void * priv) {
+       ssize_t rt;
+       time_t start_time, current_time;
+       char buf[DVB_READ_BUFFER_SIZE];
+       my_thread_arg* p = (my_thread_arg*)priv;
+
+       start_time = time(NULL);
+       while ( 1 ) {
+               rt = read(p->fin, buf, DVB_READ_BUFFER_SIZE);
+               //printf("read %d %d\n", p->fin, rt);
+               if ( rt == 0 ) {
+                       printf("dvr0 file read failed\n");
+                       goto error;
+               }
+               if ( rt < 0 ) {
+                       switch (errno) {
+                       case EOVERFLOW:
+                               printf("read overflow, continue\n");
+                               continue;
+                       default:
+                               printf("read finished. err=%d\n", errno);
+                               goto error;
+                       }
+               }
+
+               mrb_put(&p->rb, buf, rt);
+
+               // TODO: time() at each DVB_READ_BUFFER_SIZE bytes read is not efficient, reduce frequency
+               current_time = time(NULL);
+               if ( current_time - start_time > p->rectime || p->done ) {
+                       break;
+               }
+       }
+       error:
+       p->done = true;
+       return NULL;
+}
+
+void * record_write(void * priv) {
+       ssize_t rt = 0, wt, shift;
+       char buf[DVB_WRITE_BUFFER_SIZE];
+       my_thread_arg* p = (my_thread_arg*)priv;
+
+       while ( 1 ) {
+               if ( mrb_get(&p->rb, buf, DVB_WRITE_BUFFER_SIZE) ) {
+                       // sufficient buffer
+                       rt = DVB_WRITE_BUFFER_SIZE;
+               }
+               else if ( p->done && mrb_data_len(&p->rb) > 0) {
+                       // insufficient buffer, but no more buffer fill
+                       rt = mrb_data_len(&p->rb);
+                       mrb_get(&p->rb, buf, mrb_data_len(&p->rb));
+               }
+               else if ( p->done ) {
+                       // normal exit
+                       break;
+               }
+               else {
+                       usleep(100);
+               }
+
                shift = 0;
-               while ( ( wt = write(fout, buf + shift, rt) ) > 0) {
+               while ( ( wt = write(p->fout, buf + shift, rt) ) > 0) {
+                       //printf("write %d %d\n", p->fout, rt);
                        rt -= wt;
                        if ( rt == 0 ) break;
                        shift += wt;
@@ -128,17 +300,10 @@ void record(int adapter_nr, char* output, int rectime) {
                        printf("output file write failed\n");
                        goto error;
                }
-
-               // TODO: time() at each 1316 bytes read is not efficient, reduce frequency
-               current_time = time(NULL);
-               if ( current_time - start_time > rectime ) {
-                       break;
-               }
        }
-
        error:
-       close(fin);
-       close(fout);
+       p->done = true;
+       return NULL;
 }
 
 int main(int argc, char *argv[]) {
@@ -149,7 +314,7 @@ int main(int argc, char *argv[]) {
        int ret;
        int rectime;
 
-       if (argc <= 2) {
+       if (argc != 6) {
                fprintf(stderr, "Usage: %s adapter_nr freq tsid rectime output\n", argv[0]);
                return 1;
        }