http: use bigger, but shorter-lived buffers for pipes

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

diff --git a/lib/PublicInbox/Qspawn.pm b/lib/PublicInbox/Qspawn.pm
index 509a441246e7407932a4db62d252e8d46caa392f..8f0b9fe240ebc378a81ce86a7207f73d71ec4f63 100644 (file)
--- a/lib/PublicInbox/Qspawn.pm
+++ b/lib/PublicInbox/Qspawn.pm
@@ -12,9 +12,9 @@
 # operate in.  This can be useful to ensure smaller inboxes can
 # be cloned while cloning of large inboxes is maxed out.
 #
-# This does not depend on Danga::Socket or any other external
+# This does not depend on PublicInbox::DS or any other external
 # scheduling mechanism, you just need to call start() and finish()
-# appropriately. However, public-inbox-httpd (which uses Danga::Socket)
+# appropriately. However, public-inbox-httpd (which uses PublicInbox::DS)
 # will be able to schedule this based on readability of stdout from
 # the spawned process.  See GitHTTPBackend.pm and SolverGit.pm for
 # usage examples.  It does not depend on any form of threading.
@@ -29,6 +29,9 @@ use warnings;
 use PublicInbox::Spawn qw(popen_rd);
 require Plack::Util;
 
+# n.b.: we get EAGAIN with public-inbox-httpd, and EINTR on other PSGI servers
+use Errno qw(EAGAIN EINTR);
+
 my $def_limiter;
 
 # declares a command to spawn (but does not spawn it).
@@ -43,10 +46,18 @@ sub new ($$$;) {
 sub _do_spawn {
        my ($self, $cb) = @_;
        my $err;
+       my ($cmd, $env, $opts) = @{$self->{args}};
+       my %opts = %{$opts || {}};
+       my $limiter = $self->{limiter};
+       foreach my $k (PublicInbox::Spawn::RLIMITS()) {
+               if (defined(my $rlimit = $limiter->{$k})) {
+                       $opts{$k} = $rlimit;
+               }
+       }
 
-       ($self->{rpipe}, $self->{pid}) = popen_rd(@{$self->{args}});
+       ($self->{rpipe}, $self->{pid}) = popen_rd($cmd, $env, \%opts);
        if (defined $self->{pid}) {
-               $self->{limiter}->{running}++;
+               $limiter->{running}++;
        } else {
                $self->{err} = $!;
        }
@@ -114,22 +125,22 @@ sub psgi_qx {
                eval { $qx_cb->($qx) };
                $qx = undef;
        };
-       my $rpipe;
+       my $rpipe; # comes from popen_rd
        my $async = $env->{'pi-httpd.async'};
        my $cb = sub {
-               my $r = sysread($rpipe, my $buf, 8192);
+               my $r = sysread($rpipe, my $buf, 65536);
                if ($async) {
                        $async->async_pass($env->{'psgix.io'}, $qx, \$buf);
                } elsif (defined $r) {
                        $r ? $qx->write($buf) : $end->();
                } else {
-                       return if $!{EAGAIN} || $!{EINTR}; # loop again
+                       return if $! == EAGAIN || $! == EINTR; # loop again
                        $end->();
                }
        };
        $limiter ||= $def_limiter ||= PublicInbox::Qspawn::Limiter->new(32);
        $self->start($limiter, sub { # may run later, much later...
-               ($rpipe) = @_;
+               ($rpipe) = @_; # popen_rd result
                if ($async) {
                # PublicInbox::HTTPD::Async->new($rpipe, $cb, $end)
                        $async = $async->($rpipe, $cb, $end);
@@ -185,7 +196,7 @@ sub psgi_return {
        my $buf = '';
        my $rd_hdr = sub {
                my $r = sysread($rpipe, $buf, 1024, length($buf));
-               return if !defined($r) && ($!{EINTR} || $!{EAGAIN});
+               return if !defined($r) && $! == EAGAIN || $! == EINTR;
                $parse_hdr->($r, \$buf);
        };
 
@@ -251,9 +262,38 @@ sub new {
                max => $max || 32,
                running => 0,
                run_queue => [],
+               # RLIMIT_CPU => undef,
+               # RLIMIT_DATA => undef,
+               # RLIMIT_CORE => undef,
        }, $class;
 }
 
+sub setup_rlimit {
+       my ($self, $name, $config) = @_;
+       foreach my $rlim (PublicInbox::Spawn::RLIMITS()) {
+               my $k = lc($rlim);
+               $k =~ tr/_//d;
+               $k = "publicinboxlimiter.$name.$k";
+               defined(my $v = $config->{$k}) or next;
+               my @rlimit = split(/\s*,\s*/, $v);
+               if (scalar(@rlimit) == 1) {
+                       push @rlimit, $rlimit[0];
+               } elsif (scalar(@rlimit) != 2) {
+                       warn "could not parse $k: $v\n";
+               }
+               eval { require BSD::Resource };
+               if ($@) {
+                       warn "BSD::Resource missing for $rlim";
+                       next;
+               }
+               foreach my $i (0..$#rlimit) {
+                       next if $rlimit[$i] ne 'INFINITY';
+                       $rlimit[$i] = BSD::Resource::RLIM_INFINITY();
+               }
+               $self->{$rlim} = \@rlimit;
+       }
+}
+
 # captures everything into a buffer and executes a callback when done
 package PublicInbox::Qspawn::Qx;
 use strict;