ipc: allow disabling broadcast for wq_workers

Since some lei worker classes only use a single worker,
there's no sense in having broadcast for those cases.

diff --git a/lib/PublicInbox/IPC.pm b/lib/PublicInbox/IPC.pm
index d5e37719bf48f3cd36cb84eedc5a256a8066f678..92f35189a529c1e3de8534e61f2205f262bff782 100644 (file)
--- a/lib/PublicInbox/IPC.pm
+++ b/lib/PublicInbox/IPC.pm
@@ -245,10 +245,10 @@ sub recv_and_run {
        $n;
 }
 
-sub wq_worker_loop ($) {
-       my ($self, $bcast_a) = @_;
-       my $wqw = PublicInbox::WQWorker->new($self);
-       PublicInbox::WQWorker->new($self, '-wq_bcast2');
+sub wq_worker_loop ($$) {
+       my ($self, $bcast2) = @_;
+       my $wqw = PublicInbox::WQWorker->new($self, $self->{-wq_s2});
+       PublicInbox::WQWorker->new($self, $bcast2) if $bcast2;
        PublicInbox::DS->SetPostLoopCallback(sub { $wqw->{sock} });
        PublicInbox::DS->EventLoop;
        PublicInbox::DS->Reset;
@@ -339,8 +339,9 @@ sub wq_do {
 sub _wq_worker_start ($$$) {
        my ($self, $oldset, $fields) = @_;
        my ($bcast1, $bcast2);
-       socketpair($bcast1, $bcast2, AF_UNIX, $SEQPACKET, 0) or
-                                               die "socketpair: $!";
+       $self->{-wq_no_bcast} or
+               socketpair($bcast1, $bcast2, AF_UNIX, $SEQPACKET, 0) or
+                                                       die "socketpair: $!";
        my $seed = rand(0xffffffff);
        my $pid = fork // die "fork: $!";
        if ($pid == 0) {
@@ -361,8 +362,7 @@ sub _wq_worker_start ($$$) {
                        my $on_destroy = $self->ipc_atfork_child;
                        local %SIG = %SIG;
                        PublicInbox::DS::sig_setmask($oldset);
-                       $self->{-wq_bcast2} = $bcast2;
-                       wq_worker_loop($self);
+                       wq_worker_loop($self, $bcast2);
                };
                warn "worker $self->{-wq_ident} PID:$$ died: $@" if $@;
                undef $end; # trigger exit

diff --git a/lib/PublicInbox/WQWorker.pm b/lib/PublicInbox/WQWorker.pm
index f7aa61c555705d2c0c476f8526844ea28830b134..48b901bb139feec7d4abd9ff6f95ccb149cdc026 100644 (file)
--- a/lib/PublicInbox/WQWorker.pm
+++ b/lib/PublicInbox/WQWorker.pm
@@ -11,11 +11,10 @@ use Errno qw(EAGAIN ECONNRESET);
 use IO::Handle (); # blocking
 
 sub new {
-       my ($cls, $wq, $field) = @_;
-       my $s2 = $wq->{$field // '-wq_s2'} // die "BUG: no {$field}";
-       $s2->blocking(0);
-       my $self = bless { sock => $s2, wq => $wq }, $cls;
-       $self->SUPER::new($s2, EPOLLEXCLUSIVE|EPOLLIN|EPOLLET);
+       my ($cls, $wq, $sock) = @_;
+       $sock->blocking(0);
+       my $self = bless { sock => $sock, wq => $wq }, $cls;
+       $self->SUPER::new($sock, EPOLLEXCLUSIVE|EPOLLIN|EPOLLET);
        $self;
 }