get rid of unnecessary bytes::length usage

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

diff --git a/lib/PublicInbox/IPC.pm b/lib/PublicInbox/IPC.pm
index 24237773903373ca63f3b0783da537b678a841b8..9efe551bd2398574f4d82bb4d44e81466f2bd06f 100644 (file)
--- a/lib/PublicInbox/IPC.pm
+++ b/lib/PublicInbox/IPC.pm
@@ -17,9 +17,9 @@ use PublicInbox::Spawn;
 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
@@ -213,7 +213,7 @@ sub ipc_sibling_atfork_child {
 
 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;
@@ -235,7 +235,7 @@ sub recv_and_run {
        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;
 }
@@ -268,27 +268,37 @@ sub wq_broadcast {
        }
 }
 
+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
        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]);
-               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;
@@ -343,7 +353,6 @@ sub wq_workers_start {
                die "socketpair: $!";
        $self->ipc_atfork_prepare;
        $nr_workers //= $self->{-wq_nr_workers};
-       $nr_workers = $WQ_MAX_WORKERS if $nr_workers > $WQ_MAX_WORKERS;
        my $sigset = $oldset // PublicInbox::DS::block_signals();
        $self->{-wq_workers} = {};
        $self->{-wq_ident} = $ident;
@@ -356,7 +365,6 @@ sub wq_worker_incr { # SIGTTIN handler
        my ($self, $oldset, $fields) = @_;
        $self->{-wq_s2} or return;
        die "-wq_nr_workers locked" if defined $self->{-wq_nr_workers};
-       return if wq_workers($self) >= $WQ_MAX_WORKERS;
        $self->ipc_atfork_prepare;
        my $sigset = $oldset // PublicInbox::DS::block_signals();
        _wq_worker_start($self, $sigset, $fields);
@@ -432,8 +440,6 @@ sub wq_kill {
        kill($sig // 'TERM', keys %$workers);
 }
 
-sub WQ_MAX_WORKERS { $WQ_MAX_WORKERS }
-
 sub DESTROY {
        my ($self) = @_;
        my $ppid = $self->{-wq_ppid};