spawn: implement spawn_kill, improve IPTV pipe (busy loop)

src/config.c

@@ -1147,7 +1147,7 @@ dobackup(const char *oldver)
                                      root, oldver);
   tvhinfo("config", "backup: running, output file %s", outfile);
 
-  spawnv(argv[0], (void *)argv, 1, 1);
+  spawnv(argv[0], (void *)argv, NULL, 1, 1);
 
   while ((code = spawn_reap(errtxt, sizeof(errtxt))) == -EAGAIN)
     usleep(20000);

src/dvr/dvr_rec.c

@@ -679,7 +679,7 @@ dvr_spawn_postproc(dvr_entry_t *de, const char *dvr_postproc)
     args[i] = s;
   }
   
-  spawnv(args[0], (void *)args, 1, 1);
+  spawnv(args[0], (void *)args, NULL, 1, 1);
     
   htsstr_argsplit_free(args);
 }

src/epggrab/module.c

@@ -287,7 +287,7 @@ char *epggrab_module_grab_spawn ( void *m )
   tvhlog(LOG_INFO, mod->id, "grab %s", mod->path);
 
   /* Grab */
-  outlen = spawn_and_give_stdout(mod->path, NULL, &rd, 1);
+  outlen = spawn_and_give_stdout(mod->path, NULL, &rd, NULL, 1);
 
   if (outlen < 0)
     goto error;

src/epggrab/module/xmltv.c

@@ -683,7 +683,7 @@ static void _xmltv_load_grabbers ( void )
   char *tmp, *tmp2 = NULL, *path;
 
   /* Load data */
-  if (spawn_and_give_stdout(XMLTV_FIND, NULL, &rd, 1) >= 0)
+  if (spawn_and_give_stdout(XMLTV_FIND, NULL, &rd, NULL, 1) >= 0)
     outlen = file_readall(rd, &outbuf);
   if (rd >= 0)
     close(rd);
@@ -731,7 +731,7 @@ static void _xmltv_load_grabbers ( void )
           if (!(st.st_mode & S_IEXEC)) continue;
           if (!S_ISREG(st.st_mode)) continue;
           rd = -1;
-          if (spawn_and_give_stdout(bin, argv, &rd, 1) >= 0 &&
+          if (spawn_and_give_stdout(bin, argv, &rd, NULL, 1) >= 0 &&
               (outlen = file_readall(rd, &outbuf)) > 0) {
             close(rd);
             if (outbuf[outlen-1] == '\n') outbuf[outlen-1] = '\0';

src/input/mpegts/iptv/iptv_pipe.c

@@ -25,6 +25,7 @@
 #include <sys/types.h>
 #include <sys/ioctl.h>
 #include <fcntl.h>
+#include <signal.h>
 #include <assert.h>
 
 /*
@@ -35,6 +36,7 @@ iptv_pipe_start ( iptv_mux_t *im, const char *_raw, const url_t *url )
 {
   char *argv[64], *f, *raw, *s, *p, *a;
   int i = 1, rd;
+  pid_t pid;
 
   if (strncmp(_raw, "pipe://", 7))
     return -1;
@@ -73,7 +75,7 @@ iptv_pipe_start ( iptv_mux_t *im, const char *_raw, const url_t *url )
   }
   argv[i] = NULL;
 
-  if (spawn_and_give_stdout(raw, argv, &rd, 1)) {
+  if (spawn_and_give_stdout(raw, argv, &rd, &pid, 1)) {
     tvherror("iptv", "Unable to start pipe '%s' (wrong executable?)", raw);
     return -1;
   }
@@ -82,6 +84,7 @@ iptv_pipe_start ( iptv_mux_t *im, const char *_raw, const url_t *url )
   fcntl(rd, F_SETFL, fcntl(rd, F_GETFL) | O_NONBLOCK);
 
   im->mm_iptv_fd = rd;
+  im->im_data = (void *)(intptr_t)pid;
 
   iptv_input_mux_started(im);
   return 0;
@@ -92,6 +95,8 @@ iptv_pipe_stop
   ( iptv_mux_t *im )
 {
   int rd = im->mm_iptv_fd;
+  pid_t pid = (intptr_t)im->im_data;
+  spawn_kill(pid, SIGKILL);
   if (rd > 0)
     close(rd);
   im->mm_iptv_fd = -1;
@@ -103,17 +108,24 @@ iptv_pipe_read ( iptv_mux_t *im )
   int r, rd = im->mm_iptv_fd;
   ssize_t res = 0;
   char buf[64*1024];
+  pid_t pid;
 
-  while (rd > 0 && res < 1024*1024) {
+  while (rd > 0 && res < sizeof(buf)) {
     r = read(rd, buf, sizeof(buf));
     if (r < 0) {
+      if (errno == EAGAIN)
+        break;
       if (ERRNO_AGAIN(errno))
         continue;
-      break;
     }
-    if (!r) {
+    if (r <= 0) {
+      tvherror("iptv", "stdin pipe unexpectedly closed: %s",
+               r < 0 ? strerror(errno) : "No data");
       close(rd);
+      pid = (intptr_t)im->im_data;
+      spawn_kill(pid, SIGKILL);
       im->mm_iptv_fd = -1;
+      im->im_data = NULL;
       break;
     }
     sbuf_append(&im->mm_iptv_buffer, buf, r);

src/spawn.c

@@ -282,6 +282,31 @@ spawn_reaper(void)
   while (spawn_reap(NULL, 0) != -EAGAIN) ;
 }
 
+/**
+ * Kill the pid (only if waiting)
+ */
+int
+spawn_kill(pid_t pid, int sig)
+{
+  int r = -ESRCH;
+  spawn_t *s;
+
+  if (pid > 0) {
+    spawn_reaper();
+
+    pthread_mutex_lock(&spawn_mutex);
+    LIST_FOREACH(s, &spawns, link)
+      if(s->pid == pid)
+        break;
+    if (s) {
+      r = kill(pid, sig);
+      if (r < 0)
+        r = -errno;
+    }
+    pthread_mutex_unlock(&spawn_mutex);
+  }
+  return r;
+}
 
 /**
  * Enqueue a spawn on the pending spawn list
@@ -298,12 +323,13 @@ spawn_enq(const char *name, int pid)
   return s;
 }
 
+
+
 /**
  * Execute the given program and return its standard output as file-descriptor (pipe).
  */
-
 int
-spawn_and_give_stdout(const char *prog, char *argv[], int *rd, int redir_stderr)
+spawn_and_give_stdout(const char *prog, char *argv[], int *rd, pid_t *pid, int redir_stderr)
 {
   pid_t p;
   int fd[2], f, maxfd;
@@ -374,6 +400,8 @@ spawn_and_give_stdout(const char *prog, char *argv[], int *rd, int redir_stderr)
   close(fd[1]);
 
   *rd = fd[0];
+  if (pid)
+    *pid = p;
   return 0;
 }
 
@@ -384,7 +412,7 @@ spawn_and_give_stdout(const char *prog, char *argv[], int *rd, int redir_stderr)
  * The function will return the size of the buffer
  */
 int
-spawnv(const char *prog, char *argv[], int redir_stdout, int redir_stderr)
+spawnv(const char *prog, char *argv[], pid_t *pid, int redir_stdout, int redir_stderr)
 {
   pid_t p, f, maxfd;
   char bin[256];
@@ -445,6 +473,9 @@ spawnv(const char *prog, char *argv[], int redir_stdout, int redir_stderr)
 
   spawn_enq(prog, p);
 
+  if (pid)
+    *pid = p;
+
   return 0;
 }
 

src/spawn.h

@@ -25,12 +25,14 @@ void spawn_error ( const char *fmt, ... );
 
 int find_exec ( const char *name, char *out, size_t len );
 
-int spawn_and_give_stdout(const char *prog, char *argv[], int *rd, int redir_stderr);
+int spawn_and_give_stdout(const char *prog, char *argv[], int *rd, pid_t *pid, int redir_stderr);
 
-int spawnv(const char *prog, char *argv[], int redir_stdout, int redir_stderr);
+int spawnv(const char *prog, char *argv[], pid_t *pid, int redir_stdout, int redir_stderr);
 
 int spawn_reap(char *stxt, size_t stxtlen);
 
+int spawn_kill(pid_t pid, int sig);
+
 void spawn_init(void);
 
 void spawn_done(void);