use PublicInbox::OnDestroy;
use PublicInbox::WQWorker;
use Socket qw(AF_UNIX MSG_EOR SOCK_STREAM);
+my $MY_MAX_ARG_STRLEN = 4096 * 33; # extra 4K for serialization
my $SEQPACKET = eval { Socket::SOCK_SEQPACKET() }; # portable enough?
our @EXPORT_OK = qw(ipc_freeze ipc_thaw);
-my $WQ_MAX_WORKERS = 4096;
my ($enc, $dec);
# ->imports at BEGIN turns sereal_*_with_object into custom ops on 5.14+
# and eliminate method call overhead
# and writing to the same store
sub ipc_lock_init {
my ($self, $f) = @_;
+ $f // die 'BUG: no filename given';
require PublicInbox::Lock;
$self->{-ipc_lock} //= bless { lock_path => $f }, 'PublicInbox::Lock'
}
sub recv_and_run {
my ($self, $s2, $len, $full_stream) = @_;
- my @fds = $recv_cmd->($s2, my $buf, $len);
+ my @fds = $recv_cmd->($s2, my $buf, $len // $MY_MAX_ARG_STRLEN);
return if scalar(@fds) && !defined($fds[0]);
my $n = length($buf) or return 0;
my $nfd = 0;
undef $buf;
my $sub = shift @$args;
eval { $self->$sub(@$args) };
- warn "$$ wq_worker: $@" if $@;
+ warn "$$ $0 wq_worker: $@" if $@;
delete @$self{0..($nfd-1)};
$n;
}
sub wq_worker_loop ($) {
- my ($self) = @_;
+ my ($self, $bcast_a) = @_;
my $wqw = PublicInbox::WQWorker->new($self);
+ PublicInbox::WQWorker->new($self, '-wq_bcast2');
PublicInbox::DS->SetPostLoopCallback(sub { $wqw->{sock} });
PublicInbox::DS->EventLoop;
PublicInbox::DS->Reset;
sub do_sock_stream { # via wq_io_do, for big requests
my ($self, $len) = @_;
- recv_and_run($self, delete $self->{0}, $len, 1);
+ recv_and_run($self, my $s2 = delete $self->{0}, $len, 1);
+}
+
+sub wq_broadcast {
+ my ($self, $sub, @args) = @_;
+ if (my $wkr = $self->{-wq_workers}) {
+ for my $bcast1 (values %$wkr) {
+ my $buf = ipc_freeze([$sub, @args]);
+ send($bcast1, $buf, MSG_EOR) // croak "send: $!";
+ # XXX shouldn't have to deal with EMSGSIZE here...
+ }
+ } else {
+ eval { $self->$sub(@args) };
+ warn "wq_broadcast: $@" if $@;
+ }
+}
+
+sub stream_in_full ($$$) {
+ my ($s1, $fds, $buf) = @_;
+ socketpair(my $r, my $w, AF_UNIX, SOCK_STREAM, 0) or
+ croak "socketpair: $!";
+ my $n = $send_cmd->($s1, [ fileno($r) ],
+ ipc_freeze(['do_sock_stream', length($buf)]),
+ MSG_EOR) // croak "sendmsg: $!";
+ undef $r;
+ $n = $send_cmd->($w, $fds, $buf, 0) // croak "sendmsg: $!";
+ while ($n < length($buf)) {
+ my $x = syswrite($w, $buf, length($buf) - $n, $n) //
+ croak "syswrite: $!";
+ croak "syswrite wrote 0 bytes" if $x == 0;
+ $n += $x;
+ }
}
sub wq_io_do { # always async
if (my $s1 = $self->{-wq_s1}) { # run in worker
my $fds = [ map { fileno($_) } @$ios ];
my $buf = ipc_freeze([$sub, @args]);
- my $n = $send_cmd->($s1, $fds, $buf, MSG_EOR);
- return if defined($n); # likely
- croak "sendmsg: $! (check RLIMIT_NOFILE)" if $!{ETOOMANYREFS};
- croak "sendmsg: $!" if !$!{EMSGSIZE};
- socketpair(my $r, my $w, AF_UNIX, SOCK_STREAM, 0) or
- croak "socketpair: $!";
- $n = $send_cmd->($s1, [ fileno($r) ],
- ipc_freeze(['do_sock_stream', length($buf)]),
- MSG_EOR) // croak "sendmsg: $!";
- undef $r;
- $n = $send_cmd->($w, $fds, $buf, 0) // croak "sendmsg: $!";
- while ($n < length($buf)) {
- my $x = syswrite($w, $buf, length($buf) - $n, $n) //
- croak "syswrite: $!";
- croak "syswrite wrote 0 bytes" if $x == 0;
- $n += $x;
+ if (length($buf) > $MY_MAX_ARG_STRLEN) {
+ stream_in_full($s1, $fds, $buf);
+ } else {
+ my $n = $send_cmd->($s1, $fds, $buf, MSG_EOR);
+ return if defined($n); # likely
+ $!{ETOOMANYREFS} and
+ croak "sendmsg: $! (check RLIMIT_NOFILE)";
+ $!{EMSGSIZE} ? stream_in_full($s1, $fds, $buf) :
+ croak("sendmsg: $!");
}
} else {
@$self{0..$#$ios} = @$ios;
sub _wq_worker_start ($$$) {
my ($self, $oldset, $fields) = @_;
+ my ($bcast1, $bcast2);
+ socketpair($bcast1, $bcast2, AF_UNIX, $SEQPACKET, 0) or
+ die "socketpair: $!";
my $seed = rand(0xffffffff);
my $pid = fork // die "fork: $!";
if ($pid == 0) {
srand($seed);
+ undef $bcast1;
eval { PublicInbox::DS->Reset };
- delete @$self{qw(-wq_s1 -wq_workers -wq_ppid)};
+ delete @$self{qw(-wq_s1 -wq_ppid)};
+ $self->{-wq_worker_nr} =
+ keys %{delete($self->{-wq_workers}) // {}};
$SIG{$_} = 'IGNORE' for (qw(PIPE));
$SIG{$_} = 'DEFAULT' for (qw(TTOU TTIN TERM QUIT INT CHLD));
- local $0 = $self->{-wq_ident};
+ local $0 = "$self->{-wq_ident} $self->{-wq_worker_nr}";
# ensure we properly exit even if warn() dies:
my $end = PublicInbox::OnDestroy->new($$, sub { exit(!!$@) });
eval {
my $on_destroy = $self->ipc_atfork_child;
local %SIG = %SIG;
PublicInbox::DS::sig_setmask($oldset);
+ $self->{-wq_bcast2} = $bcast2;
wq_worker_loop($self);
};
warn "worker $self->{-wq_ident} PID:$$ died: $@" if $@;
undef $end; # trigger exit
} else {
- $self->{-wq_workers}->{$pid} = \undef;
+ $self->{-wq_workers}->{$pid} = $bcast1;
}
}
socketpair($self->{-wq_s1}, $self->{-wq_s2}, AF_UNIX, $SEQPACKET, 0) or
die "socketpair: $!";
$self->ipc_atfork_prepare;
- $nr_workers //= 4;
- $nr_workers = $WQ_MAX_WORKERS if $nr_workers > $WQ_MAX_WORKERS;
+ $nr_workers //= $self->{-wq_nr_workers};
my $sigset = $oldset // PublicInbox::DS::block_signals();
$self->{-wq_workers} = {};
$self->{-wq_ident} = $ident;
sub wq_worker_incr { # SIGTTIN handler
my ($self, $oldset, $fields) = @_;
$self->{-wq_s2} or return;
- return if wq_workers($self) >= $WQ_MAX_WORKERS;
+ 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);
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
kill($sig // 'TERM', keys %$workers);
}
-sub WQ_MAX_WORKERS { $WQ_MAX_WORKERS }
-
sub DESTROY {
my ($self) = @_;
my $ppid = $self->{-wq_ppid};