diff options
author | Eric Wong <e@80x24.org> | 2021-09-19 12:50:25 +0000 |
---|---|---|
committer | Eric Wong <e@80x24.org> | 2021-09-19 19:52:47 +0000 |
commit | ee355020415fbbebbeb8fbe79be4e4f3fa2e657b (patch) | |
tree | 13a6d79cb75a7a2d87e04acef236b2b25aefabd0 /lib/PublicInbox | |
parent | 7cbdc00ece1c77d25e6654446624e6457145e955 (diff) | |
download | public-inbox-ee355020415fbbebbeb8fbe79be4e4f3fa2e657b.tar.gz |
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")
Diffstat (limited to 'lib/PublicInbox')
-rw-r--r-- | lib/PublicInbox/IPC.pm | 69 | ||||
-rw-r--r-- | lib/PublicInbox/LEI.pm | 1 | ||||
-rw-r--r-- | lib/PublicInbox/LeiStore.pm | 1 |
3 files changed, 10 insertions, 61 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, |