lib/PublicInbox/IPC.pm | 23 +++++++++++++++++++++-- t/ipc.t | 10 +++++++--- diff --git a/lib/PublicInbox/IPC.pm b/lib/PublicInbox/IPC.pm index 4d29532cc188ef53855ce4379ee3dc1b0d4d97fa..8a3120c9d47e762efd452c93fa7bb44659e893f4 100644 --- a/lib/PublicInbox/IPC.pm +++ b/lib/PublicInbox/IPC.pm @@ -276,7 +276,7 @@ my $pid = fork // die "fork: $!"; if ($pid == 0) { eval { PublicInbox::DS->Reset }; close(delete $self->{-wq_s1}); - delete $self->{qw(-wq_workers -wq_quit)}; + delete $self->{qw(-wq_workers -wq_quit -wq_ppid)}; my $quit = sub { $self->{-wq_quit} = 1 }; $SIG{$_} = $quit for (qw(TERM INT QUIT)); $SIG{$_} = 'IGNORE' for (qw(TTOU TTIN)); @@ -347,15 +347,34 @@ $self->{-wq_exit_pending}--; dwaitpid($pid, \&ipc_worker_reap, $self); } +# set or retrieve number of workers +sub wq_workers { + my ($self, $nr) = @_; + my $cur = $self->{-wq_workers} or return; + if (defined $nr) { + while (scalar(keys(%$cur)) > $nr) { + $self->wq_worker_decr; + $self->wq_worker_decr_wait; + } + $self->wq_worker_incr while scalar(keys(%$cur)) < $nr; + } + scalar(keys(%$cur)); +} + sub wq_close { my ($self) = @_; delete @$self{qw(-wq_s1 -wq_s2)} or return; - my $ppid = delete $self->{-wq_ppid} // die 'BUG: no wq_ppid'; + my $ppid = delete $self->{-wq_ppid} or return; my $workers = delete $self->{-wq_workers} // die 'BUG: no wq_workers'; return if $ppid != $$; # can't reap siblings or parents for my $pid (keys %$workers) { dwaitpid($pid, \&ipc_worker_reap, $self); } +} + +sub DESTROY { + wq_close($_[0]); + ipc_worker_stop($_[0]); } 1; diff --git a/t/ipc.t b/t/ipc.t index 51e347c68860fa5e1819358bac499e574ac36d5a..903294c5b61336890c18768077a530ddc7736981 100644 --- a/t/ipc.t +++ b/t/ipc.t @@ -170,11 +170,15 @@ is($warn[2], $warn[1], 'worker did not die'); $SIG{__WARN__} = 'DEFAULT'; is($ipc->wq_workers_start('wq', 1), $$, 'workers started again'); -is(scalar(keys %{$ipc->{-wq_workers}}), 1, '1 worker started'); +is($ipc->wq_workers, 1, '1 worker started'); $ipc->wq_worker_incr; -is(scalar(keys %{$ipc->{-wq_workers}}), 2, 'worker count bumped'); +is($ipc->wq_workers, 2, 'worker count bumped'); $ipc->wq_worker_decr; $ipc->wq_worker_decr_wait(10); -is(scalar(keys %{$ipc->{-wq_workers}}), 1, 'worker count lowered'); +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'); +$ipc->wq_close; +is($ipc->wq_workers, undef, 'workers undef after close'); done_testing;