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 ffbc25039cd0566e8ab7d60246495069c4af3494..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]+') ],
@@ -229,6 +229,16 @@ my %OPTDESC = (
 'q     format|f=s' => [
        'OUT|maildir|mboxrd|mboxcl2|mboxcl|mboxo|html|json|jsonl|concatjson',
                'specify output format, default depends on --output'],
+'q     exclude=s@' => [ 'URL_OR_PATHNAME',
+               'exclude specified external(s) from search' ],
+'q     include|I=s@' => [ 'URL_OR_PATHNAME',
+               'include specified external(s) in search' ],
+'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,
 
@@ -277,18 +287,22 @@ sub x_it ($$) {
        dump_and_clear_log();
        if (my $sock = $self->{sock}) {
                send($sock, "x_it $code", MSG_EOR);
-       } elsif (my $signum = ($code & 127)) { # oneshot, usually SIGPIPE (13)
-               $SIG{PIPE} = 'DEFAULT'; # $SIG{$signum} doesn't work
-               kill $signum, $$;
-               sleep; # wait for signal
-       } else { # oneshot
+       } 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 ($;@) {
@@ -327,10 +341,9 @@ sub child_error { # passes non-fatal curl exit codes to user
        my ($self, $child_error) = @_; # child_error is $?
        if (my $sock = $self->{sock}) { # send to lei(1) client
                send($sock, "child_error $child_error", MSG_EOR);
-       } else { # oneshot
+       } elsif ($self->{oneshot}) {
                $self->{child_error} = $child_error;
-       }
-       undef;
+       } # else noop if client disconnected
 }
 
 sub atfork_prepare_wq {
@@ -737,8 +750,7 @@ sub lei__complete {
                my $opt = quotemeta $1;
                puts $self, map {
                        my $v = $OPTDESC{$_};
-                       $v = $v->[0] if ref($v);
-                       my @v = split(/\|/, $v);
+                       my @v = ref($v) ? split(/\|/, $v->[0]) : ();
                        # get rid of ALL CAPS placeholder (e.g "OUT")
                        # (TODO: completion for external paths)
                        shift(@v) if uc($v[0]) eq $v[0];
@@ -778,7 +790,7 @@ sub start_mua {
        push @cmd, $mfolder unless defined($replaced);
        if (my $sock = $self->{sock}) { # lei(1) client process runs it
                send($sock, exec_buf(\@cmd, {}), MSG_EOR);
-       } else { # oneshot
+       } elsif ($self->{oneshot}) {
                $self->{"mua.pid.$self.$$"} = spawn(\@cmd);
        }
 }
@@ -796,13 +808,16 @@ sub start_pager {
        $new_env->{MORE} = 'FRX' if $^O eq 'freebsd';
        pipe(my ($r, $wpager)) or return warn "pipe: $!";
        my $rdr = { 0 => $r, 1 => $self->{1}, 2 => $self->{2} };
-       my $pgr = [ undef, @$rdr{1, 2}, $$ ];
+       my $pgr = [ undef, @$rdr{1, 2} ];
        if (my $sock = $self->{sock}) { # lei(1) process runs it
                delete @$new_env{keys %$env}; # only set iff unset
                my $fds = [ map { fileno($_) } @$rdr{0..2} ];
                $send_cmd->($sock, $fds, exec_buf([$pager], $new_env), MSG_EOR);
-       } else {
+       } elsif ($self->{oneshot}) {
                $pgr->[0] = spawn([$pager], $new_env, $rdr);
+               $pgr->[3] = $$; # ew'll reap it
+       } else {
+               die 'BUG: start_pager w/o socket';
        }
        $self->{1} = $wpager;
        $self->{2} = $wpager if -t $self->{2};
@@ -817,7 +832,7 @@ sub stop_pager {
        # do not restore original stdout, just close it so we error out
        close(delete($self->{1})) if $self->{1};
        my $pid = $pgr->[0];
-       dwaitpid($pid, undef, $self->{sock}) if $pid && $pgr->[3] == $$;
+       dwaitpid($pid) if $pid && ($pgr->[3] // 0) == $$;
 }
 
 sub accept_dispatch { # Listener {post_accept} callback
@@ -859,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) {
@@ -1050,6 +1066,7 @@ sub oneshot {
        local %PATH2CFG;
        umask(077) // die("umask(077): $!");
        my $self = bless {
+               oneshot => 1,
                0 => *STDIN{GLOB},
                1 => *STDOUT{GLOB},
                2 => *STDERR{GLOB},