X-Git-Url: http://www.git.stargrave.org/?a=blobdiff_plain;f=lib%2FPublicInbox%2FLEI.pm;h=9d5a5a46b3987fc05d4095fd31d751f37fa4c462;hb=69b5ab18a594fa434d79cacee35c4d9a0df7b55c;hp=148a5b1e530d00e0ab73d02aa0f214532d28ac0b;hpb=9d0f17840479508de4aaf76fe6c150e94a9f79c3;p=public-inbox.git diff --git a/lib/PublicInbox/LEI.pm b/lib/PublicInbox/LEI.pm index 148a5b1e..9d5a5a46 100644 --- a/lib/PublicInbox/LEI.pm +++ b/lib/PublicInbox/LEI.pm @@ -206,7 +206,7 @@ our %CMD = ( # sorted in order of importance/use: 'add-external' => [ 'LOCATION', 'add/set priority of a publicinbox|extindex for extra matches', - qw(boost=i mirror=s inbox-version=i verbose|v+), + qw(boost=i mirror=s inbox-version=i epoch=s verbose|v+), @c_opt, index_opt(), @net_opt ], 'ls-external' => [ '[FILTER]', 'list publicinbox|extindex locations', qw(format|f=s z|0 globoff|g invert-match|v local remote), @c_opt ], @@ -420,7 +420,9 @@ my %OPTDESC = ( 'remote' => 'limit operations to those requiring network access', 'remote!' => 'prevent operations requiring network access', -'all:s up' => ['local|remote', 'update all remote or local saved searches' ], +# up, refresh-mail-sync, export-kw +'all:s' => ['TYPE|local|remote', 'all remote or local folders' ], + 'remote-fudge-time=s' => [ 'INTERVAL', 'look for mail INTERVAL older than the last successful query' ], @@ -1110,7 +1112,6 @@ sub dclose { my ($self) = @_; delete $self->{-progress}; _drop_wq($self) if $self->{failed}; - close(delete $self->{1}) if $self->{1}; # may reap_compress $self->close if $self->{-event_init_done}; # PublicInbox::DS::close } @@ -1118,23 +1119,28 @@ sub dclose { sub event_step { my ($self) = @_; local %ENV = %{$self->{env}}; - my $sock = $self->{sock}; local $current_lei = $self; eval { - while (my @fds = $recv_cmd->($sock, my $buf, 4096)) { + my $buf; + while (my @fds = $recv_cmd->($self->{sock}, $buf, 4096)) { if (scalar(@fds) == 1 && !defined($fds[0])) { return if $! == EAGAIN; next if $! == EINTR; last if $! == ECONNRESET; die "recvmsg: $!"; } - for my $fd (@fds) { - open my $rfh, '+<&=', $fd; + for (@fds) { open my $rfh, '+<&=', $_ } + } + if ($buf eq '') { + _drop_wq($self); # EOF, client disconnected + dclose($self); + } elsif ($buf =~ /\A(STOP|CONT)\z/) { + for my $wq (grep(defined, @$self{@WQ_KEYS})) { + $wq->wq_kill($buf) or $wq->wq_kill_old($buf); } + } else { die "unrecognized client signal: $buf"; } - _drop_wq($self); # EOF, client disconnected - dclose($self); }; if (my $err = $@) { eval { $self->fail($err) }; @@ -1146,6 +1152,7 @@ sub event_step_init { my ($self) = @_; my $sock = $self->{sock} or return; $self->{-event_init_done} //= do { # persist til $ops done + $sock->blocking(0); $self->SUPER::new($sock, EPOLLIN|EPOLLET); $sock; }; @@ -1194,7 +1201,7 @@ sub dir_idle_handler ($) { # PublicInbox::DirIdle callback $lei->dispatch('note-event', "maildir:$mdir", $nc, $bn, $fn); }; - warn "E note-event $f: $@\n" if $@; + warn "E: note-event $f: $@\n" if $@; } } if ($ev->can('cancel') && ($ev->IN_IGNORE || $ev->IN_UNMOUNT)) { @@ -1219,6 +1226,7 @@ sub lazy_start { $errors_log = "$sock_dir/errors.log"; my $addr = pack_sockaddr_un($path); my $lk = bless { lock_path => $errors_log }, 'PublicInbox::Lock'; + umask(077) // die("umask(077): $!"); $lk->lock_acquire; socket($listener, AF_UNIX, SOCK_SEQPACKET, 0) or die "socket: $!"; if ($errno == ECONNREFUSED || $errno == ENOENT) { @@ -1230,7 +1238,6 @@ sub lazy_start { $! = $errno; # allow interpolation to stringify in die die "connect($path): $!"; } - umask(077) // die("umask(077): $!"); bind($listener, $addr) or die "bind($path): $!"; $lk->lock_release; undef $lk; @@ -1379,7 +1386,7 @@ sub fchdir { sub wq_eof { # EOF callback for main daemon my ($lei) = @_; my $wq1 = delete $lei->{wq1} // return $lei->fail; # already failed - $wq1->wq_wait_old(\&wq_done_wait, $lei); + $wq1->wq_wait_old($wq1->can('_wq_done_wait') // \&wq_done_wait, $lei); } sub watch_state_ok ($) {