xap_helper: implement alarm(2)-based timeout

alarm(2) delivering SIGALRM seems sufficient for Xapian since
Xapian doesn't block signals (which would necessitate the use of
SIGKILL via RLIMIT_CPU hard limit).  When Xapian gets stuck in
`D' state on slow storage, SIGKILL would not make a difference,
either (at least not on Linux).

Relying on RLIMIT_CPU is also trickier since we must account for
CPU time already consumed by a process for unrelated requests.
Thus we just rely on a simple alarm-based timeout.  This also
avoids requiring the optional BSD::Resource module in the (mostly)
Perl implementation (and avoids potential bugs given my meager
arithmetic skills).

diff --git a/lib/PublicInbox/XapHelper.pm b/lib/PublicInbox/XapHelper.pm
index 746b4d6243b7c25d7c02dbf6608fb00556960e4f..2e20660e803c2878b4f5e2f737d01785b460ddb7 100644 (file)
--- a/lib/PublicInbox/XapHelper.pm
+++ b/lib/PublicInbox/XapHelper.pm
@@ -27,6 +27,8 @@ sub cmd_test_inspect {
                ($req->{srch}->has_threadid ? 1 : 0)
 }
 
+sub cmd_test_sleep { select(undef, undef, undef, 0.01) while 1 }
+
 sub iter_retry_check ($) {
        if (ref($@) =~ /\bDatabaseModifiedError\b/) {
                $_[0]->{srch}->reopen;
@@ -193,7 +195,10 @@ sub dispatch {
                $new->{qp} = $new->qparse_new;
                $new;
        };
+       my $timeo = $req->{K};
+       alarm($timeo) if $timeo;
        $fn->($req, @argv);
+       alarm(0) if $timeo;
 }
 
 sub recv_loop {
@@ -212,7 +217,7 @@ sub recv_loop {
                }
                scalar(@fds) or exit(66); # EX_NOINPUT
                die "recvmsg: $!" if !defined($fds[0]);
-               PublicInbox::DS::block_signals();
+               PublicInbox::DS::block_signals(POSIX::SIGALRM);
                my $req = bless {}, __PACKAGE__;
                my $i = 0;
                open($req->{$i++}, '+<&=', $_) for @fds;

diff --git a/lib/PublicInbox/xap_helper.h b/lib/PublicInbox/xap_helper.h
index 7ecea264aa3898561e6d3392b5d2046adbc18278..3df3ce91d196d89d283143fdd988e02d7e0f3092 100644 (file)
--- a/lib/PublicInbox/xap_helper.h
+++ b/lib/PublicInbox/xap_helper.h
@@ -27,6 +27,7 @@
 #include <sys/types.h>
 #include <sys/uio.h>
 #include <sys/wait.h>
+#include <poll.h>
 
 #include <assert.h>
 #include <err.h> // BSD, glibc, and musl all have this
@@ -413,6 +414,11 @@ static bool cmd_test_inspect(struct req *req)
        return false;
 }
 
+static bool cmd_test_sleep(struct req *req)
+{
+       for (;;) poll(NULL, 0, 10);
+       return false;
+}
 #include "xh_mset.h" // read-only (WWW, IMAP, lei) stuff
 #include "xh_cidx.h" // CodeSearchIdx.pm stuff
 
@@ -427,6 +433,7 @@ static const struct cmd_entry {
        CMD(dump_ibx), // many inboxes
        CMD(dump_roots), // per-cidx shard
        CMD(test_inspect), // least common commands last
+       CMD(test_sleep), // least common commands last
 };
 
 #define MY_ARRAY_SIZE(x)       (sizeof(x)/sizeof((x)[0]))
@@ -680,6 +687,9 @@ static void dispatch(struct req *req)
                free_srch(kbuf.srch);
                goto cmd_err; // srch_init already warned
        }
+       if (req->timeout_sec)
+               alarm(req->timeout_sec > UINT_MAX ?
+                       UINT_MAX : (unsigned)req->timeout_sec);
        try {
                if (!req->fn(req))
                        warnx("`%s' failed", req->argv[0]);
@@ -688,6 +698,8 @@ static void dispatch(struct req *req)
        } catch (...) {
                warn("unhandled exception");
        }
+       if (req->timeout_sec)
+               alarm(0);
 cmd_err:
        return; // just be silent on errors, for now
 }
@@ -1025,6 +1037,7 @@ int main(int argc, char *argv[])
        DELSET(SIGXFSZ);
 #undef DELSET
        CHECK(int, 0, sigdelset(&workerset, SIGUSR1));
+       CHECK(int, 0, sigdelset(&fullset, SIGALRM));
 
        if (nworker == 0) { // no SIGTERM handling w/o workers
                recv_loop();

diff --git a/t/xap_helper.t b/t/xap_helper.t
index effe8bc500cfce8af377032b003e58c897e6308b..78be8539cbbdf6d5b9991ad85bad13f2aae65481 100644 (file)
--- a/t/xap_helper.t
+++ b/t/xap_helper.t
@@ -9,6 +9,7 @@ use Socket qw(AF_UNIX SOCK_SEQPACKET SOCK_STREAM);
 require PublicInbox::AutoReap;
 use PublicInbox::IPC;
 require PublicInbox::XapClient;
+use PublicInbox::DS qw(now);
 use autodie;
 my ($tmp, $for_destroy) = tmpdir();
 
@@ -267,6 +268,20 @@ for my $n (@NO_CXX) {
        my @oids = (join('', @res) =~ /^([a-f0-9]{7}) /gms);
        is $nr_out, scalar(@oids), "output count matches $xhc->{impl}" or
                diag explain(\@res, \@err);
+
+       if ($ENV{TEST_XH_TIMEOUT}) {
+               diag 'testing timeouts...';
+               for my $j (qw(0 1)) {
+                       my $t0 = now;
+                       $r = $xhc->mkreq(undef, qw(test_sleep -K 1 -d),
+                                       $ibx_idx[0]);
+                       is readline($r), undef, 'got EOF';
+                       my $diff = now - $t0;
+                       ok $diff < 3, "timeout didn't take too long -j$j";
+                       ok $diff >= 0.9, "timeout didn't fire prematurely -j$j";
+                       $xhc = PublicInbox::XapClient::start_helper('-j1');
+               }
+       }
 }
 
 done_testing;