lei q: support --jobs [SEARCHERS],[WRITERS]

[public-inbox.git] / lib / PublicInbox / LEI.pm

diff --git a/lib/PublicInbox/LEI.pm b/lib/PublicInbox/LEI.pm
index d6fa814cf197411ab67abb33415fe79cc5327ce4..9afc90cfaa1067d7a0b852debf8a0bf6575da9ed 100644 (file)
--- a/lib/PublicInbox/LEI.pm
+++ b/lib/PublicInbox/LEI.pm
@@ -104,7 +104,7 @@ our %CMD = ( # sorted in order of importance/use:
 'q' => [ 'SEARCH_TERMS...', 'search for messages matching terms', qw(
        save-as=s output|mfolder|o=s format|f=s dedupe|d=s thread|t augment|a
        sort|s=s reverse|r offset=i remote! local! external! pretty
-       include|I=s@ exclude=s@ only=s@
+       include|I=s@ exclude=s@ only=s@ jobs|j=s
        mua-cmd|mua=s no-torsocks torsocks=s verbose|v quiet|q
        received-after=s received-before=s sent-after=s sent-since=s),
        PublicInbox::LeiQuery::curl_opt(), opt_dash('limit|n=i', '[0-9]+') ],
@@ -236,6 +236,9 @@ my %OPTDESC = (
 'q     only=s@' => [ 'URL_OR_PATHNAME',
                'only use specified external(s) for search' ],
 
+'q     jobs=s' => [ '[SEARCH_JOBS][,WRITER_JOBS]',
+               'control number of search and writer jobs' ],
+
 'ls-query      format|f=s' => $ls_format,
 'ls-external   format|f=s' => $ls_format,
 
@@ -284,20 +287,22 @@ sub x_it ($$) {
        dump_and_clear_log();
        if (my $sock = $self->{sock}) {
                send($sock, "x_it $code", MSG_EOR);
-       } elsif (!$self->{oneshot}) {
-               return; # client disconnected, noop
-       } elsif (my $signum = ($code & 127)) { # usually SIGPIPE (13)
-               $SIG{PIPE} = 'DEFAULT'; # $SIG{$signum} doesn't work
-               kill $signum, $$;
-               sleep; # wait for signal
-       } else {
+       } elsif ($self->{oneshot}) {
                # don't want to end up using $? from child processes
                for my $f (qw(lxs l2m)) {
                        my $wq = delete $self->{$f} or next;
                        $wq->DESTROY;
                }
-               $quit->($code >> 8);
-       }
+               # cleanup anything that has tempfiles
+               delete @$self{qw(ovv dedupe)};
+               if (my $signum = ($code & 127)) { # usually SIGPIPE (13)
+                       $SIG{PIPE} = 'DEFAULT'; # $SIG{$signum} doesn't work
+                       kill $signum, $$;
+                       sleep; # wait for signal
+               } else {
+                       $quit->($code >> 8);
+               }
+       } # else ignore if client disconnected
 }
 
 sub err ($;@) {
@@ -869,6 +874,7 @@ sub accept_dispatch { # Listener {post_accept} callback
 
 sub dclose {
        my ($self) = @_;
+       delete $self->{-progress};
        for my $f (qw(lxs l2m)) {
                my $wq = delete $self->{$f} or next;
                if ($wq->wq_kill) {