+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 $@;
+ }
+}
+