epg: add compression support for epgdb.v2 file

src/config.c

@@ -1631,6 +1631,7 @@ config_boot ( const char *path, gid_t gid, uid_t uid )
   config.cookie_expires = 7;
   config.dscp = -1;
   config.descrambler_buffer = 9000;
+  config.epg_compress = 1;
 
   /* Generate default */
   if (!path) {
@@ -2089,6 +2090,19 @@ const idclass_t config_class = {
       .opts   = PO_LORDER,
       .group  = 2
     },
+#if ENABLE_ZLIB
+    {
+      .type   = PT_BOOL,
+      .id     = "epg_compress",
+      .name   = N_("Compress EPG database"),
+      .desc   = N_("Compress the EPG database to reduce disk I/O "
+                   "and space."),
+      .off    = offsetof(config_t, epg_compress),
+      .opts   = PO_EXPERT,
+      .def.i  = 1,
+      .group  = 1
+    },
+#endif
     {
       .type   = PT_STR,
       .islist = 1,

src/config.h

@@ -52,6 +52,7 @@ typedef struct config {
   int dscp;
   uint32_t descrambler_buffer;
   int parser_backlog;
+  int epg_compress;
 } config_t;
 
 extern const idclass_t config_class;

src/epgdb.c

@@ -31,6 +31,7 @@
 #include "channels.h"
 #include "epg.h"
 #include "epggrab.h"
+#include "config.h"
 
 #define EPG_DB_VERSION 2
 #define EPG_DB_ALLOC_STEP (1024*1024)
@@ -163,7 +164,7 @@ void epg_init ( void )
   int fd = -1;
   struct stat st;
   size_t remain;
-  uint8_t *mem, *rp;
+  uint8_t *mem, *rp, *zlib_mem = NULL;
   epggrab_stats_t stats;
   int ver = EPG_DB_VERSION;
   struct sigaction act, oldact;
@@ -214,6 +215,16 @@ void epg_init ( void )
     goto end;
   }
 
+#if ENABLE_ZLIB
+  if (remain > 12 && memcmp(rp, "\xff\xffGZIP00", 8) == 0) {
+    uint32_t orig = (rp[8] << 24) | (rp[9] << 16) | (rp[10] << 8) | rp[11];
+    tvhlog(LOG_INFO, "epgdb", "gzip format detected, inflating (ratio %.1f%%)",
+           (float)((remain * 100.0) / orig));
+    rp = zlib_mem = tvh_gzip_inflate(rp + 12, remain - 12, orig);
+    remain = rp ? orig : 0;
+  }
+#endif
+
   /* Process */
   memset(&stats, 0, sizeof(stats));
   while ( remain > 4 ) {
@@ -273,6 +284,7 @@ void epg_init ( void )
 
   /* Close file */
   munmap(mem, st.st_size);
+  free(zlib_mem);
 end:
   sigaction(SIGBUS, &oldact, NULL);
   close(fd);
@@ -325,17 +337,38 @@ static int _epg_write_sect ( sbuf_t *sb, const char *sect )
 static void epg_save_tsk_callback ( void *p, int dearmed )
 {
   sbuf_t *sb = p;
+  size_t size = sb->sb_ptr;
   int fd, r;
 
   tvhinfo("epgdb", "save start");
   fd = hts_settings_open_file(1, "epgdb.v%d", EPG_DB_VERSION);
   if (fd >= 0) {
-    r = tvh_write(fd, sb->sb_data, sb->sb_ptr);
+#if ENABLE_ZLIB
+    if (config.epg_compress) {
+      r = tvh_write(fd, "\xff\xffGZIP00\x00\x00\x00\x00", 12);
+      if (!r)
+        r = tvh_gzip_deflate_fd(fd, sb->sb_data, sb->sb_ptr, &size, 3) < 0;
+      if (!r && size > UINT_MAX)
+        r = 1;
+      if (!r) {
+        r = lseek(fd, 8, SEEK_SET) != (off_t)8;
+        if (!r) {
+          uint8_t data2[4];
+          data2[0] = (sb->sb_ptr >> 24) & 0xff;
+          data2[1] = (sb->sb_ptr >> 16) & 0xff;
+          data2[2] = (sb->sb_ptr >> 8) & 0xff;
+          data2[3] = (sb->sb_ptr & 0xff);
+          r = tvh_write(fd, data2, 4);
+        }
+      }
+   } else
+#endif
+      r = tvh_write(fd, sb->sb_data, sb->sb_ptr);
     close(fd);
     if (r)
-      tvherror("epgdb", "write error (size %d)", sb->sb_ptr);
+      tvherror("epgdb", "write error (size %zd)", size);
     else
-      tvhinfo("epgdb", "stored (size %d)", sb->sb_ptr);
+      tvhinfo("epgdb", "stored (size %zd)", size);
   } else
     tvherror("epgdb", "unable to open epgdb file");
   sbuf_free(sb);

src/main.c

@@ -409,6 +409,8 @@ tasklet_thread ( void *aux )
 {
   tasklet_t *tsk;
 
+  tvhtread_renice(20);
+
   pthread_mutex_lock(&tasklet_lock);
   while (tvheadend_running) {
     tsk = TAILQ_FIRST(&tasklets);

src/tvheadend.h

@@ -678,6 +678,8 @@ int tvhthread_create
    void *(*start_routine) (void *), void *arg,
    const char *name);
 
+int tvhtread_renice(int value);
+
 int tvh_open(const char *pathname, int flags, mode_t mode);
 
 int tvh_socket(int domain, int type, int protocol);
@@ -784,6 +786,7 @@ char *regexp_escape ( const char *str );
 #if ENABLE_ZLIB
 uint8_t *tvh_gzip_inflate ( const uint8_t *data, size_t size, size_t orig );
 uint8_t *tvh_gzip_deflate ( const uint8_t *data, size_t orig, size_t *size );
+int      tvh_gzip_deflate_fd ( int fd, const uint8_t *data, size_t orig, size_t *size, int speed );
 #endif
 
 /* URL decoding */

src/wrappers.c

@@ -4,6 +4,7 @@
 #include <sys/types.h>          /* See NOTES */
 #include <sys/socket.h>
 #include <sys/stat.h>
+#include <sys/resource.h>
 #include <unistd.h>
 #include <signal.h>
 #include <pthread.h>
@@ -196,6 +197,21 @@ tvhthread_create
   return r;
 }
 
+/* linux style: -19 .. 20 */
+int
+tvhtread_renice(int value)
+{
+  int ret = 0;
+#ifdef SYS_gettid
+  pid_t tid;
+  tid = syscall(SYS_gettid);
+  ret = setpriority(PRIO_PROCESS, tid, value);
+#else
+#warning "Implement renice for your platform!"
+#endif
+  return ret;
+}
+
 #if ! ENABLE_QSORT_R
 /*
  * qsort_r wrapper for pre GLIBC 2.8

src/zlib.c

@@ -98,3 +98,57 @@ uint8_t *tvh_gzip_deflate ( const uint8_t *data, size_t orig, size_t *size )
 
   return bufout;
 }
+
+int tvh_gzip_deflate_fd ( int fd, const uint8_t *data, size_t orig, size_t *size, int speed )
+{
+  int r = 0, err;
+  z_stream zstr;
+  uint8_t *bufout;
+  size_t alloc;
+
+  assert(speed >= Z_BEST_SPEED && speed <= Z_BEST_COMPRESSION);
+
+  /* Setup buffers */
+  *size  = 0;
+  alloc  = MIN(orig, 256*1024);
+  bufout = malloc(alloc);
+
+  /* Setup zlib */
+  memset(&zstr, 0, sizeof(zstr));
+  err = deflateInit2(&zstr, speed, Z_DEFLATED, MAX_WBITS + 16 /* gzip */, MAX_MEM_LEVEL, Z_DEFAULT_STRATEGY);
+  zstr.avail_in  = orig;
+  zstr.next_in   = (z_const uint8_t *)data;
+  zstr.avail_out = alloc;
+  zstr.next_out  = bufout;
+
+  /* Compress */
+  while (1) {
+    err = deflate(&zstr, Z_FINISH);
+
+    /* Need more space */
+    if (err == Z_OK && zstr.avail_out == 0) {
+      r = tvh_write(fd, bufout, alloc);
+      if (r) {
+        r = -1;
+        break;
+      }
+      zstr.avail_out = alloc;
+      zstr.next_out  = bufout;
+      continue;
+    }
+
+    /* Error */
+    if ( (err != Z_STREAM_END && err != Z_OK) || zstr.total_out == 0 ) {
+      r = -1;
+    } else {
+      if (zstr.avail_out != alloc)
+        r = tvh_write(fd, bufout, alloc - zstr.avail_out);
+      *size = zstr.total_out;
+    }
+    break;
+  }
+  deflateEnd(&zstr);
+  free(bufout);
+
+  return r;
+}