diff options
| -rw-r--r-- | lib/PublicInbox/IPC.pm | 52 | ||||
| -rw-r--r-- | t/ipc.t | 25 |
2 files changed, 3 insertions, 74 deletions
diff --git a/lib/PublicInbox/IPC.pm b/lib/PublicInbox/IPC.pm index 3713b56b..7e5a0b16 100644 --- a/lib/PublicInbox/IPC.pm +++ b/lib/PublicInbox/IPC.pm @@ -18,7 +18,6 @@ use PublicInbox::OnDestroy; use PublicInbox::WQWorker; use Socket qw(AF_UNIX MSG_EOR SOCK_STREAM); my $SEQPACKET = eval { Socket::SOCK_SEQPACKET() }; # portable enough? -use constant PIPE_BUF => $^O eq 'linux' ? 4096 : POSIX::_POSIX_PIPE_BUF(); our @EXPORT_OK = qw(ipc_freeze ipc_thaw); my $WQ_MAX_WORKERS = 4096; my ($enc, $dec); @@ -59,15 +58,10 @@ sub _get_rec ($) { ipc_thaw($buf); } -sub _pack_rec ($) { - my ($ref) = @_; - my $buf = ipc_freeze($ref); - length($buf) . "\n" . $buf; -} - sub _send_rec ($$) { my ($w, $ref) = @_; - print $w _pack_rec($ref) or croak "print: $!"; + my $buf = ipc_freeze($ref); + print $w length($buf), "\n", $buf or croak "print: $!"; } sub ipc_return ($$$) { @@ -188,21 +182,6 @@ sub ipc_lock_init { $self->{-ipc_lock} //= bless { lock_path => $f }, 'PublicInbox::Lock' } -sub ipc_async_wait ($$) { - my ($self, $max) = @_; # max == -1 to wait for all - my $aif = $self->{-async_inflight} or return; - my $r_res = $self->{-ipc_res} or die 'BUG: no ipc_res'; - while (my ($sub, $bytes, $cb, $cb_arg) = splice(@$aif, 0, 4)) { - my $ret = _get_rec($r_res) // - die "no response on $sub (req.size=$bytes)"; - $self->{-async_inflight_bytes} -= $bytes; - - eval { $cb->($cb_arg, $ret) }; - warn "E: $sub callback error: $@\n" if $@; - return if --$max == 0; - } -} - # call $self->$sub(@args), on a worker if ipc_worker_spawn was used sub ipc_do { my ($self, $sub, @args) = @_; @@ -210,8 +189,7 @@ sub ipc_do { my $ipc_lock = $self->{-ipc_lock}; my $lock = $ipc_lock ? $ipc_lock->lock_for_scope : undef; if (defined(wantarray)) { - my $r_res = $self->{-ipc_res} or die 'BUG: no ipc_res'; - ipc_async_wait($self, -1); + my $r_res = $self->{-ipc_res} or die 'no ipc_res'; _send_rec($w_req, [ wantarray, $sub, @args ]); my $ret = _get_rec($r_res) // die "no response on $sub"; die $$ret if ref($ret) eq 'PublicInbox::IPC::Die'; @@ -224,30 +202,6 @@ sub ipc_do { } } -sub ipc_async { - my ($self, $sub, $sub_args, $cb, $cb_arg) = @_; - if (my $w_req = $self->{-ipc_req}) { # run in worker - my $rec = _pack_rec([ 1, $sub, @$sub_args ]); - my $cur_bytes = \($self->{-async_inflight_bytes} //= 0); - while (($$cur_bytes + length($rec)) > PIPE_BUF) { - ipc_async_wait($self, 1); - } - my $ipc_lock = $self->{-ipc_lock}; - my $lock = $ipc_lock ? $ipc_lock->lock_for_scope : undef; - print $w_req $rec or croak "print: $!"; - $$cur_bytes += length($rec); - push @{$self->{-async_inflight}}, - $sub, length($rec), $cb, $cb_arg; - } else { - my $ret = [ eval { $self->$sub(@$sub_args) } ]; - if (my $exc = $@) { - $ret = ( bless(\$exc, 'PublicInbox::IPC::Die') ); - } - eval { $cb->($cb_arg, $ret) }; - warn "E: $sub callback error: $@\n" if $@; - } -} - # needed when there's multiple IPC workers and the parent forking # causes newer siblings to inherit older siblings sockets sub ipc_sibling_atfork_child { @@ -37,7 +37,6 @@ my $ipc = bless {}, 'PublicInbox::IPC'; my @t = qw(array scalar scalarref undef); my $test = sub { my $x = shift; - my @res; for my $type (@t) { my $m = "test_$type"; my @ret = $ipc->ipc_do($m); @@ -46,34 +45,10 @@ my $test = sub { $ipc->ipc_do($m); - $ipc->ipc_async($m, [], sub { push @res, \@_ }, \$m); - my $ret = $ipc->ipc_do($m); my $exp = $ipc->$m; is_deeply($ret, $exp, "!wantarray $m $x"); - - is_deeply(\@res, [ [ \$m, \@exp ] ], "async $m $x"); - @res = (); } - $ipc->ipc_async_wait(-1); - is_deeply(\@res, [], 'no leftover results'); - $ipc->ipc_async('test_die', ['die test'], - sub { push @res, \@_ }, 'die arg'); - $ipc->ipc_async_wait(1); - is(scalar(@res), 1, 'only one result'); - is(scalar(@{$res[0]}), 2, 'result has 2-element array'); - is($res[0]->[0], 'die arg', 'got async die arg '.$x); - is(ref($res[0]->[1]), 'PublicInbox::IPC::Die', - "exception type $x"); - { - my $nr = PublicInbox::IPC::PIPE_BUF(); - my $count = 0; - my $cb = sub { ++$count }; - $ipc->ipc_async('test_undef', [], $cb) for (1..$nr); - $ipc->ipc_async_wait(-1); - is($count, $nr, "$x async runs w/o deadlock"); - } - my $ret = eval { $ipc->test_die('phail') }; my $exp = $@; $ret = eval { $ipc->ipc_do('test_die', 'phail') }; |