From mboxrd@z Thu Jan 1 00:00:00 1970 Return-Path: X-Spam-Checker-Version: SpamAssassin 3.4.2 (2018-09-13) on dcvr.yhbt.net X-Spam-Level: X-Spam-ASN: X-Spam-Status: No, score=-4.0 required=3.0 tests=ALL_TRUSTED,BAYES_00 shortcircuit=no autolearn=ham autolearn_force=no version=3.4.2 Received: from localhost (dcvr.yhbt.net [127.0.0.1]) by dcvr.yhbt.net (Postfix) with ESMTP id 3F2E71FB0F for ; Sun, 19 Sep 2021 12:50:36 +0000 (UTC) From: Eric Wong To: meta@public-inbox.org Subject: [PATCH 06/16] ipc: drop dynamic WQ process counts Date: Sun, 19 Sep 2021 12:50:25 +0000 Message-Id: <20210919125035.6331-7-e@80x24.org> In-Reply-To: <20210919125035.6331-1-e@80x24.org> References: <20210919125035.6331-1-e@80x24.org> MIME-Version: 1.0 Content-Transfer-Encoding: 8bit List-Id: In retrospect, I don't think it's needed; and trying to wire up a user interface for lei to manage process counts doesn't seem worthwhile. It could be resurrected for public-facing daemon use in the future, but that's what version control systems are for. This also lets us automatically avoid setting up broadcast sockets Followup-to: 7b7939d47b336fb7 ("lei: lock worker counts") --- lib/PublicInbox/IPC.pm | 69 ++++++------------------------------- lib/PublicInbox/LEI.pm | 1 - lib/PublicInbox/LeiStore.pm | 1 - t/ipc.t | 13 +------ 4 files changed, 11 insertions(+), 73 deletions(-) diff --git a/lib/PublicInbox/IPC.pm b/lib/PublicInbox/IPC.pm index 92f35189..add5f3df 100644 --- a/lib/PublicInbox/IPC.pm +++ b/lib/PublicInbox/IPC.pm @@ -262,9 +262,10 @@ sub do_sock_stream { # via wq_io_do, for big requests sub wq_broadcast { my ($self, $sub, @args) = @_; if (my $wkr = $self->{-wq_workers}) { + my $buf = ipc_freeze([$sub, @args]); for my $bcast1 (values %$wkr) { - my $buf = ipc_freeze([$sub, @args]); - send($bcast1, $buf, MSG_EOR) // croak "send: $!"; + my $sock = $bcast1 // $self->{-wq_s1} // next; + send($sock, $buf, MSG_EOR) // croak "send: $!"; # XXX shouldn't have to deal with EMSGSIZE here... } } else { @@ -336,11 +337,10 @@ sub wq_do { } } -sub _wq_worker_start ($$$) { - my ($self, $oldset, $fields) = @_; +sub _wq_worker_start ($$$$) { + my ($self, $oldset, $fields, $one) = @_; my ($bcast1, $bcast2); - $self->{-wq_no_bcast} or - socketpair($bcast1, $bcast2, AF_UNIX, $SEQPACKET, 0) or + $one or socketpair($bcast1, $bcast2, AF_UNIX, $SEQPACKET, 0) or die "socketpair: $!"; my $seed = rand(0xffffffff); my $pid = fork // die "fork: $!"; @@ -380,66 +380,17 @@ sub wq_workers_start { socketpair($self->{-wq_s1}, $self->{-wq_s2}, AF_UNIX, $SEQPACKET, 0) or die "socketpair: $!"; $self->ipc_atfork_prepare; - $nr_workers //= $self->{-wq_nr_workers}; + $nr_workers //= $self->{-wq_nr_workers}; # was set earlier my $sigset = $oldset // PublicInbox::DS::block_signals(); $self->{-wq_workers} = {}; $self->{-wq_ident} = $ident; - _wq_worker_start($self, $sigset, $fields) for (1..$nr_workers); + my $one = $nr_workers == 1; + $self->{-wq_nr_workers} = $nr_workers; + _wq_worker_start($self, $sigset, $fields, $one) for (1..$nr_workers); PublicInbox::DS::sig_setmask($sigset) unless $oldset; $self->{-wq_ppid} = $$; } -sub wq_worker_incr { # SIGTTIN handler - my ($self, $oldset, $fields) = @_; - $self->{-wq_s2} or return; - die "-wq_nr_workers locked" if defined $self->{-wq_nr_workers}; - $self->ipc_atfork_prepare; - my $sigset = $oldset // PublicInbox::DS::block_signals(); - _wq_worker_start($self, $sigset, $fields); - PublicInbox::DS::sig_setmask($sigset) unless $oldset; -} - -sub wq_exit { # wakes up wq_worker_decr_wait - send($_[0]->{-wq_s2}, $$, MSG_EOR) // die "$$ send: $!"; - exit; -} - -sub wq_worker_decr { # SIGTTOU handler, kills first idle worker - my ($self) = @_; - return unless wq_workers($self); - die "-wq_nr_workers locked" if defined $self->{-wq_nr_workers}; - $self->wq_io_do('wq_exit'); - # caller must call wq_worker_decr_wait in main loop -} - -sub wq_worker_decr_wait { - my ($self, $timeout, $cb, @args) = @_; - return if $self->{-wq_ppid} != $$; # can't reap siblings or parents - die "-wq_nr_workers locked" if defined $self->{-wq_nr_workers}; - my $s1 = $self->{-wq_s1} // croak 'BUG: no wq_s1'; - vec(my $rin = '', fileno($s1), 1) = 1; - select(my $rout = $rin, undef, undef, $timeout) or - croak 'timed out waiting for wq_exit'; - recv($s1, my $pid, 64, 0) // croak "recv: $!"; - my $workers = $self->{-wq_workers} // croak 'BUG: no wq_workers'; - delete $workers->{$pid} // croak "BUG: PID:$pid invalid"; - dwaitpid($pid, $cb // \&ipc_worker_reap, [ $self, @args ]); -} - -# set or retrieve number of workers -sub wq_workers { - my ($self, $nr, $cb, @args) = @_; - my $cur = $self->{-wq_workers} or return; - if (defined $nr) { - while (scalar(keys(%$cur)) > $nr) { - $self->wq_worker_decr; - $self->wq_worker_decr_wait(undef, $cb, @args); - } - $self->wq_worker_incr while scalar(keys(%$cur)) < $nr; - } - scalar(keys(%$cur)); -} - sub wq_close { my ($self, $nohang, $cb, @args) = @_; delete @$self{qw(-wq_s1 -wq_s2)} or return; diff --git a/lib/PublicInbox/LEI.pm b/lib/PublicInbox/LEI.pm index f62e82dc..def85ef1 100644 --- a/lib/PublicInbox/LEI.pm +++ b/lib/PublicInbox/LEI.pm @@ -627,7 +627,6 @@ sub workers_start { my $end = $lei->pkt_op_pair; my $ident = $wq->{-wq_ident} // "lei-$lei->{cmd} worker"; $flds->{lei} = $lei; - $wq->{-wq_nr_workers} //= $jobs; # lock, no incrementing $wq->wq_workers_start($ident, $jobs, $lei->oldset, $flds); delete $lei->{pkt_op_p}; my $op_c = delete $lei->{pkt_op_c}; diff --git a/lib/PublicInbox/LeiStore.pm b/lib/PublicInbox/LeiStore.pm index 164a9f2d..b4f40912 100644 --- a/lib/PublicInbox/LeiStore.pm +++ b/lib/PublicInbox/LeiStore.pm @@ -565,7 +565,6 @@ sub write_prepare { # Mail we import into lei are private, so headers filtered out # by -mda for public mail are not appropriate local @PublicInbox::MDA::BAD_HEADERS = (); - $self->{-wq_no_bcast} = 1; $self->wq_workers_start("lei/store $dir", 1, $lei->oldset, { lei => $lei, -err_wr => $w, diff --git a/t/ipc.t b/t/ipc.t index 202b1cc6..ce89f94b 100644 --- a/t/ipc.t +++ b/t/ipc.t @@ -180,24 +180,13 @@ SKIP: { is($warn[2], $warn[1], 'worker did not die'); $SIG{__WARN__} = 'DEFAULT'; - is($ipc->wq_workers_start('wq', 1), $$, 'workers started again'); - is($ipc->wq_workers, 1, '1 worker started'); - - $ipc->wq_worker_incr; - is($ipc->wq_workers, 2, 'worker count bumped'); - $ipc->wq_worker_decr; - $ipc->wq_worker_decr_wait(10); - is($ipc->wq_workers, 1, 'worker count lowered'); - is($ipc->wq_workers(2), 2, 'worker count set'); - is($ipc->wq_workers, 2, 'worker count stayed set'); - + is($ipc->wq_workers_start('wq', 2), $$, 'workers started again'); $ipc->wq_broadcast('test_append_pid', "$tmpdir/append_pid"); $ipc->wq_close; open my $fh, '<', "$tmpdir/append_pid" or BAIL_OUT "open: $!"; chomp(my @pids = <$fh>); my %pids = map { $_ => 1 } grep(/\A[0-9]+\z/, @pids); is(scalar keys %pids, 2, 'broadcast hit both PIDs'); - is($ipc->wq_workers, undef, 'workers undef after close'); } done_testing;