When a child process times out or otherwise closes on us, kill it.

diff --git a/httpinput.cpp b/httpinput.cpp
index 1b09abf6ce8ad550b6c371853fdbcd0bc0bb8e6f..3a6b3c8af561afba6ca00a23bcb5d621ed79cb17 100644 (file)
--- a/httpinput.cpp
+++ b/httpinput.cpp
@@ -75,7 +75,8 @@ HTTPInput::HTTPInput(const InputProto &serialized)
          http_header(serialized.http_header()),
          stream_header(serialized.stream_header()),
          has_metacube_header(serialized.has_metacube_header()),
          http_header(serialized.http_header()),
          stream_header(serialized.stream_header()),
          has_metacube_header(serialized.has_metacube_header()),

-         sock(serialized.sock())
+         sock(serialized.sock()),
+         child_pid(serialized.child_pid())

 {
        // Set back the close-on-exec flag for the socket.
        // (This can't leak into a child, since we haven't been started yet.)
 {
        // Set back the close-on-exec flag for the socket.
        // (This can't leak into a child, since we haven't been started yet.)


@@ -112,6 +113,13 @@ void HTTPInput::close_socket()
                safe_close(sock);
                sock = -1;
        }
                safe_close(sock);
                sock = -1;
        }

+       if (child_pid != -1) {
+               // Kill the child process group, forcibly.
+               // TODO: Consider using a pidfd on newer kernels, so that we're guaranteed
+               // never to kill the wrong process.
+               kill(-child_pid, SIGKILL);
+       }
+       child_pid = -1;

 
        lock_guard<mutex> lock(stats_mutex);
        stats.connect_time = -1;
 
        lock_guard<mutex> lock(stats_mutex);
        stats.connect_time = -1;


@@ -134,6 +142,7 @@ InputProto HTTPInput::serialize() const
        serialized.set_pending_data(string(pending_data.begin(), pending_data.end()));
        serialized.set_has_metacube_header(has_metacube_header);
        serialized.set_sock(sock);
        serialized.set_pending_data(string(pending_data.begin(), pending_data.end()));
        serialized.set_has_metacube_header(has_metacube_header);
        serialized.set_sock(sock);

+       serialized.set_child_pid(child_pid);

        serialized.set_bytes_received(stats.bytes_received);
        serialized.set_data_bytes_received(stats.data_bytes_received);
        if (isfinite(stats.latency_sec)) {
        serialized.set_bytes_received(stats.bytes_received);
        serialized.set_data_bytes_received(stats.data_bytes_received);
        if (isfinite(stats.latency_sec)) {


@@ -247,15 +256,23 @@ int HTTPInput::open_child_process(const string &cmdline)
        posix_spawn_file_actions_adddup2(&actions, devnullfd, 0);
        posix_spawn_file_actions_adddup2(&actions, pipefd[1], 1);
 
        posix_spawn_file_actions_adddup2(&actions, devnullfd, 0);
        posix_spawn_file_actions_adddup2(&actions, pipefd[1], 1);
 

-       pid_t child_pid;
+       // Make the process a leader of its own process group, so that we can easily
+       // kill it and any of its child processes (unless it's started new process
+       // groups itself, of course).
+       posix_spawnattr_t attr;
+       posix_spawnattr_init(&attr);
+       posix_spawnattr_setflags(&attr, POSIX_SPAWN_SETPGROUP);
+       posix_spawnattr_setpgroup(&attr, 0);
+

        char * const argv[] = {
                strdup("/bin/sh"),
                strdup("-c"),
                strdup(path.c_str()),
                nullptr
        };
        char * const argv[] = {
                strdup("/bin/sh"),
                strdup("-c"),
                strdup(path.c_str()),
                nullptr
        };

-       int err = posix_spawn(&child_pid, "/bin/sh", &actions, /*attrp=*/nullptr, argv, /*envp=*/nullptr);
+       int err = posix_spawn(&child_pid, "/bin/sh", &actions, &attr, argv, /*envp=*/nullptr);

        posix_spawn_file_actions_destroy(&actions);
        posix_spawn_file_actions_destroy(&actions);

+       posix_spawnattr_destroy(&attr);

        free(argv[0]);
        free(argv[1]);
        free(argv[2]);
        free(argv[0]);
        free(argv[1]);
        free(argv[2]);


@@ -264,6 +281,7 @@ int HTTPInput::open_child_process(const string &cmdline)
        if (err == 0) {
                return pipefd[0];
        } else {
        if (err == 0) {
                return pipefd[0];
        } else {

+               child_pid = -1;

                log_perror(cmdline.c_str());
                close(pipefd[0]);
                return -1;
                log_perror(cmdline.c_str());
                close(pipefd[0]);
                return -1;


@@ -397,6 +415,7 @@ void HTTPInput::do_work()
                                        break;
                                }
                        } while (err != 0);
                                        break;
                                }
                        } while (err != 0);

+                       child_pid = -1;

 
                        request.clear();
                        request_bytes_sent = 0;
 
                        request.clear();
                        request_bytes_sent = 0;

diff --git a/httpinput.h b/httpinput.h
index 890cc9ef91256b92145e58dcc3425831fa0747fc..1ee1f8f5c4d134e3066ec65f6923b74effc2489a 100644 (file)
--- a/httpinput.h
+++ b/httpinput.h
@@ -2,6 +2,7 @@
 #define _HTTPINPUT_H 1
 
 #include <stddef.h>
 #define _HTTPINPUT_H 1
 
 #include <stddef.h>

+#include <sys/types.h>

 #include <mutex>
 #include <string>
 #include <vector>
 #include <mutex>
 #include <string>
 #include <vector>


@@ -99,6 +100,9 @@ private:
        // The socket we are downloading on (or -1).
        int sock = -1;
 
        // The socket we are downloading on (or -1).
        int sock = -1;
 

+       // pid of the cihld process (or -1).
+       pid_t child_pid = -1;
+

        // Mutex protecting <stats>.
        mutable std::mutex stats_mutex;
 
        // Mutex protecting <stats>.
        mutable std::mutex stats_mutex;
 

diff --git a/state.proto b/state.proto
index c1e782fb5b71b6e64411bdc736b3a68d9b725b97..52df32ce027ac2af18bbc272f7f8ca5c678a07cb 100644 (file)
--- a/state.proto
+++ b/state.proto
@@ -62,6 +62,7 @@ message InputProto {
        optional bytes pending_data = 7;
        optional bool has_metacube_header = 8;
        optional int32 sock = 9;
        optional bytes pending_data = 7;
        optional bool has_metacube_header = 8;
        optional int32 sock = 9;

+       optional int64 child_pid = 18 [default=-1];

        optional int64 bytes_received = 11;
        optional int64 data_bytes_received = 12;
        optional int64 metadata_bytes_received = 16;
        optional int64 bytes_received = 11;
        optional int64 data_bytes_received = 12;
        optional int64 metadata_bytes_received = 16;