about summary refs log tree commit homepage
path: root/lib/PublicInbox/LEI.pm
diff options
context:
space:
mode:
authorEric Wong <e@80x24.org>2021-09-21 09:29:44 +0000
committerEric Wong <e@80x24.org>2021-09-21 19:18:36 +0000
commit5c7d3f916da9a9c93419282e943a7d45c53b9c86 (patch)
tree6a32a05d58629fa19dae5a160144037f6a0500b9 /lib/PublicInbox/LEI.pm
parent1411a6e0ca36d6eda89a2f2f4e937fec3cad2030 (diff)
downloadpublic-inbox-5c7d3f916da9a9c93419282e943a7d45c53b9c86.tar.gz
Sometimes it's useful to pause an expensive query or
refresh-mail-sync to do something else.  While lei-daemon and
lei/store can't be paused since they're shared across clients,
per-invocation WQ workers can be paused safely using the
unblockable SIGSTOP.

While we're at it, drop the ETOOMANYREFS hint since it
hasn't been a problem since we drastically reduced FD passing
early in development.
Diffstat (limited to 'lib/PublicInbox/LEI.pm')
-rw-r--r--lib/PublicInbox/LEI.pm18
1 files changed, 12 insertions, 6 deletions
diff --git a/lib/PublicInbox/LEI.pm b/lib/PublicInbox/LEI.pm
index f94bfa45..2df1f326 100644
--- a/lib/PublicInbox/LEI.pm
+++ b/lib/PublicInbox/LEI.pm
@@ -1118,23 +1118,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 +1151,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;
         };