diff options
author | Eric Wong <e@80x24.org> | 2023-10-07 21:24:05 +0000 |
---|---|---|
committer | Eric Wong <e@80x24.org> | 2023-10-08 18:54:44 +0000 |
commit | fd799cf3d3f48f0329b125324db173c9d5e4bddf (patch) | |
tree | bac1d9c30f92417a89e08e2470b0cf3f97ab9f21 /lib/PublicInbox/IPC.pm | |
parent | c8b757d945040426f8c4cd909c2f05e34be6ccb1 (diff) | |
download | public-inbox-fd799cf3d3f48f0329b125324db173c9d5e4bddf.tar.gz |
None of the lei internals works properly without forking and sockets. The fallback code increases the potential to accidentally call subs in the wrong process during the teardown phase. We'll still support ipc_do w/o forking for now since it forking doesn't benefit small indexing runs from -mda and such.
Diffstat (limited to 'lib/PublicInbox/IPC.pm')
-rw-r--r-- | lib/PublicInbox/IPC.pm | 43 |
1 files changed, 16 insertions, 27 deletions
diff --git a/lib/PublicInbox/IPC.pm b/lib/PublicInbox/IPC.pm index 068c5623..ba8b5739 100644 --- a/lib/PublicInbox/IPC.pm +++ b/lib/PublicInbox/IPC.pm @@ -256,16 +256,12 @@ 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 $sock = $bcast1 // $self->{-wq_s1} // next; - send($sock, $buf, 0) // croak "send: $!"; - # XXX shouldn't have to deal with EMSGSIZE here... - } - } else { - eval { $self->$sub(@args) }; - warn "wq_broadcast: $@" if $@; + my $wkr = $self->{-wq_workers} or Carp::confess('no -wq_workers'); + my $buf = ipc_freeze([$sub, @args]); + for my $bcast1 (values %$wkr) { + my $sock = $bcast1 // $self->{-wq_s1} // next; + send($sock, $buf, 0) // croak "send: $!"; + # XXX shouldn't have to deal with EMSGSIZE here... } } @@ -291,24 +287,17 @@ sub stream_in_full ($$$) { sub wq_io_do { # always async my ($self, $sub, $ios, @args) = @_; - if (my $s1 = $self->{-wq_s1}) { # run in worker - my $fds = [ map { fileno($_) } @$ios ]; - my $buf = ipc_freeze([$sub, @args]); - if (length($buf) > $MY_MAX_ARG_STRLEN) { - stream_in_full($s1, $fds, $buf); - } else { - my $n = $send_cmd->($s1, $fds, $buf, 0); - return if defined($n); # likely - $!{ETOOMANYREFS} and - croak "sendmsg: $! (check RLIMIT_NOFILE)"; - $!{EMSGSIZE} ? stream_in_full($s1, $fds, $buf) : - croak("sendmsg: $!"); - } + my $s1 = $self->{-wq_s1} or Carp::confess('no -wq_s1'); + my $fds = [ map { fileno($_) } @$ios ]; + my $buf = ipc_freeze([$sub, @args]); + if (length($buf) > $MY_MAX_ARG_STRLEN) { + stream_in_full($s1, $fds, $buf); } else { - @$self{0..$#$ios} = @$ios; - eval { $self->$sub(@args) }; - warn "wq_io_do: $@" if $@; - delete @$self{0..$#$ios}; # don't close + my $n = $send_cmd->($s1, $fds, $buf, 0); + return if defined($n); # likely + $!{ETOOMANYREFS} and croak "sendmsg: $! (check RLIMIT_NOFILE)"; + $!{EMSGSIZE} ? stream_in_full($s1, $fds, $buf) : + croak("sendmsg: $!"); } } |