X-Git-Url: http://www.git.stargrave.org/?a=blobdiff_plain;f=lib%2FPublicInbox%2FLEI.pm;h=51b0e95e17281f901ddb720d2f91bef023d9aa1b;hb=64f6a4c97b05a709de60aea9c3a5f51d7a37f226;hp=148a5b1e530d00e0ab73d02aa0f214532d28ac0b;hpb=9d0f17840479508de4aaf76fe6c150e94a9f79c3;p=public-inbox.git

diff --git a/lib/PublicInbox/LEI.pm b/lib/PublicInbox/LEI.pm
index 148a5b1e..51b0e95e 100644
--- a/lib/PublicInbox/LEI.pm
+++ b/lib/PublicInbox/LEI.pm
@@ -18,8 +18,7 @@ use POSIX qw(strftime);
 use IO::Handle ();
 use Fcntl qw(SEEK_SET);
 use PublicInbox::Config;
-use PublicInbox::Syscall qw(SFD_NONBLOCK EPOLLIN EPOLLET);
-use PublicInbox::Sigfd;
+use PublicInbox::Syscall qw(EPOLLIN EPOLLET);
 use PublicInbox::DS qw(now dwaitpid);
 use PublicInbox::Spawn qw(spawn popen_rd);
 use PublicInbox::Lock;
@@ -201,12 +200,17 @@ our %CMD = ( # sorted in order of importance/use:
 'rediff' => [ '--stdin|LOCATION...',
 		'regenerate a diff with different options',
 	'stdin|', # /|\z/ must be first for lone dash
-	qw(git-dir=s@ cwd! verbose|v+ color:s no-color),
+	qw(git-dir=s@ cwd! verbose|v+ color:s no-color drq:1 dequote-only:1),
 	@diff_opt, @lxs_opt, @net_opt, @c_opt ],
 
+'mail-diff' => [ '--stdin|LOCATION...', 'diff the contents of emails',
+	'stdin|', # /|\z/ must be first for lone dash
+	qw(verbose|v+ color:s no-color raw-header),
+	@diff_opt, @net_opt, @c_opt ],
+
 '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 +424,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' ],
 
@@ -517,8 +523,7 @@ sub fail ($$;$) {
 	my ($self, $buf, $exit_code) = @_;
 	$self->{failed}++;
 	err($self, $buf) if defined $buf;
-	# calls fail_handler
-	$self->{pkt_op_p}->pkt_do('!') if $self->{pkt_op_p};
+	$self->{pkt_op_p}->pkt_do('fail_handler') if $self->{pkt_op_p};
 	x_it($self, ($exit_code // 1) << 8);
 	undef;
 }
@@ -550,7 +555,7 @@ sub child_error { # passes non-fatal curl exit codes to user
 sub note_sigpipe { # triggers sigpipe_handler
 	my ($self, $fd) = @_;
 	close(delete($self->{$fd})); # explicit close silences Perl warning
-	$self->{pkt_op_p}->pkt_do('|') if $self->{pkt_op_p};
+	$self->{pkt_op_p}->pkt_do('sigpipe_handler') if $self->{pkt_op_p};
 	x_it($self, 13);
 }
 
@@ -577,6 +582,7 @@ sub _lei_atfork_child {
 	close $listener if $listener;
 	undef $listener;
 	$dir_idle->force_close if $dir_idle;
+	undef $dir_idle;
 	%PATH2CFG = ();
 	$MDIR2CFGPATH = {};
 	eval 'no warnings; undef $PublicInbox::LeiNoteEvent::to_flush';
@@ -612,11 +618,11 @@ sub incr {
 
 sub pkt_ops {
 	my ($lei, $ops) = @_;
-	$ops->{'!'} = [ \&fail_handler, $lei ];
-	$ops->{'|'} = [ \&sigpipe_handler, $lei ];
-	$ops->{x_it} = [ \&x_it, $lei ];
-	$ops->{child_error} = [ \&child_error, $lei ];
-	$ops->{incr} = [ \&incr, $lei ];
+	$ops->{fail_handler} = [ $lei ];
+	$ops->{sigpipe_handler} = [ $lei ];
+	$ops->{x_it} = [ $lei ];
+	$ops->{child_error} = [ $lei ];
+	$ops->{incr} = [ $lei ];
 	$ops;
 }
 
@@ -1110,7 +1116,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 +1123,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 +1156,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 +1205,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 +1230,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 +1242,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;
@@ -1284,23 +1295,11 @@ sub lazy_start {
 		USR1 => \&noop,
 		USR2 => \&noop,
 	};
-	my $sigfd = PublicInbox::Sigfd->new($sig, SFD_NONBLOCK);
-	local @SIG{keys %$sig} = values(%$sig) unless $sigfd;
-	undef $sig;
-	local $SIG{PIPE} = 'IGNORE';
 	require PublicInbox::DirIdle;
 	local $dir_idle = PublicInbox::DirIdle->new([$sock_dir], sub {
 		# just rely on wakeup to hit PostLoopCallback set below
 		dir_idle_handler($_[0]) if $_[0]->fullname ne $path;
 	}, 1);
-	if ($sigfd) {
-		undef $sigfd; # unref, already in DS::DescriptorMap
-	} else {
-		# wake up every second to accept signals if we don't
-		# have signalfd or IO::KQueue:
-		PublicInbox::DS::sig_setmask($oldset);
-		PublicInbox::DS->SetLoopTimeout(1000);
-	}
 	PublicInbox::DS->SetPostLoopCallback(sub {
 		my ($dmap, undef) = @_;
 		if (@st = defined($path) ? stat($path) : ()) {
@@ -1337,8 +1336,10 @@ sub lazy_start {
 	open STDERR, '>&STDIN' or die "redirect stderr failed: $!";
 	open STDOUT, '>&STDIN' or die "redirect stdout failed: $!";
 	# $daemon pipe to `lei' closed, main loop begins:
-	eval { PublicInbox::DS->EventLoop };
+	eval { PublicInbox::DS::event_loop($sig, $oldset) };
 	warn "event loop error: $@\n" if $@;
+	# exit() may trigger waitpid via various DESTROY, ensure interruptible
+	PublicInbox::DS::sig_setmask($oldset);
 	dump_and_clear_log();
 	exit($exit_code // 0);
 }
@@ -1379,7 +1380,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 ($) {
@@ -1405,6 +1406,7 @@ sub add_maildir_watch ($$) {
 
 sub refresh_watches {
 	my ($lei) = @_;
+	$dir_idle or return;
 	my $cfg = _lei_cfg($lei) or return;
 	my $old = $cfg->{-watches};
 	my $watches = $cfg->{-watches} //= {};
@@ -1415,7 +1417,7 @@ sub refresh_watches {
 		require PublicInbox::LeiWatch;
 		$watches->{$url} //= PublicInbox::LeiWatch->new($url);
 		$seen{$url} = undef;
-		my $state = $cfg->get_1("watch.$url", 'state');
+		my $state = $cfg->get_1("watch.$url.state");
 		if (!watch_state_ok($state)) {
 			$lei->err("watch.$url.state=$state not supported");
 			next;