Revert "ipc: add support for asynchronous callbacks"

[public-inbox.git] / lib / PublicInbox / IPC.pm

diff --git a/lib/PublicInbox/IPC.pm b/lib/PublicInbox/IPC.pm
index 3713b56bc659e2e0645f14d1b2bb8779ef5af050..7e5a0b16b7caef020a39645177ba3a94f11397d3 100644 (file)
--- 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 {