qspawn: workaround Perl 5.16.3 leak, re-enable Deflater

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

diff --git a/lib/PublicInbox/Qspawn.pm b/lib/PublicInbox/Qspawn.pm
index 54976b008d839b302a0d4c262157691f32818a07..cb3dc516304d74e70467c8ba0aac43e8d0004c66 100644 (file)
--- a/lib/PublicInbox/Qspawn.pm
+++ b/lib/PublicInbox/Qspawn.pm
@@ -44,11 +44,11 @@ sub new ($$$;) {
 }
 
 sub _do_spawn {
-       my ($self, $cb) = @_;
+       my ($self, $start_cb, $limiter) = @_;
        my $err;
        my ($cmd, $env, $opts) = @{$self->{args}};
        my %opts = %{$opts || {}};
-       my $limiter = $self->{limiter};
+       $self->{limiter} = $limiter;
        foreach my $k (PublicInbox::Spawn::RLIMITS()) {
                if (defined(my $rlimit = $limiter->{$k})) {
                        $opts{$k} = $rlimit;
@@ -66,7 +66,7 @@ sub _do_spawn {
        } else {
                $self->{err} = $!;
        }
-       $cb->($self->{rpipe});
+       $start_cb->($self->{rpipe});
 }
 
 sub child_err ($) {
@@ -94,21 +94,21 @@ sub waitpid_err ($$) {
                $err = "W: waitpid($xpid, 0) => $pid: $!";
        } # else should not be called with pid == 0
 
+       my $env = delete $self->{env};
+
        # done, spawn whatever's in the queue
        my $limiter = $self->{limiter};
        my $running = --$limiter->{running};
 
-       # limiter->{max} may change dynamically
-       if (($running || $limiter->{running}) < $limiter->{max}) {
-               if (my $next = shift @{$limiter->{run_queue}}) {
-                       _do_spawn(@$next);
+       if ($running < $limiter->{max}) {
+               if (my $next = shift(@{$limiter->{run_queue}})) {
+                       _do_spawn(@$next, $limiter);
                }
        }
 
        return unless $err;
        $self->{err} = $err;
-       my $env = $self->{env} or return;
-       if (!$env->{'qspawn.quiet'}) {
+       if ($env && !$env->{'qspawn.quiet'}) {
                log_err($env, join(' ', @{$self->{args}}) . ": $err");
        }
 }
@@ -116,10 +116,9 @@ sub waitpid_err ($$) {
 sub do_waitpid ($;$) {
        my ($self, $env) = @_;
        my $pid = $self->{pid};
-       eval { # PublicInbox::DS may not be loaded
-               PublicInbox::DS::dwaitpid($pid, \&waitpid_err, $self);
-               $self->{env} = $env;
-       };
+       $self->{env} = $env;
+       # PublicInbox::DS may not be loaded
+       eval { PublicInbox::DS::dwaitpid($pid, \&waitpid_err, $self) };
        # done if we're running in PublicInbox::DS::EventLoop
        if ($@) {
                # non public-inbox-{httpd,nntpd} callers may block:
@@ -133,25 +132,14 @@ sub finish ($;$) {
        if (delete $self->{rpipe}) {
                do_waitpid($self, $env);
        }
-
-       # limiter->{max} may change dynamically
-       my $limiter = $self->{limiter};
-       if ($limiter->{running} < $limiter->{max}) {
-               if (my $next = shift @{$limiter->{run_queue}}) {
-                       _do_spawn(@$next);
-               }
-       }
-       $self->{err}; # may be meaningless if non-blocking
 }
 
 sub start {
-       my ($self, $limiter, $cb) = @_;
-       $self->{limiter} = $limiter;
-
+       my ($self, $limiter, $start_cb) = @_;
        if ($limiter->{running} < $limiter->{max}) {
-               _do_spawn($self, $cb);
+               _do_spawn($self, $start_cb, $limiter);
        } else {
-               push @{$limiter->{run_queue}}, [ $self, $cb ];
+               push @{$limiter->{run_queue}}, [ $self, $start_cb ];
        }
 }
 
@@ -187,11 +175,12 @@ reread:
                }
        };
        $limiter ||= $def_limiter ||= PublicInbox::Qspawn::Limiter->new(32);
-       $self->start($limiter, sub { # may run later, much later...
+       $self->start($limiter, sub { # start_cb, may run later, much later...
                ($rpipe) = @_; # popen_rd result
                if ($async) {
                # PublicInbox::HTTPD::Async->new($rpipe, $cb, $end)
                        $async = $async->($rpipe, $cb, $end);
+                       # $cb will call ->async_pass or ->close
                } else { # generic PSGI
                        $cb->() while $qx;
                }
@@ -266,7 +255,7 @@ sub psgi_return {
                $ret;
        };
 
-       my $wcb = delete $env->{'qspawn.wcb'};
+       my $wcb = delete $env->{'qspawn.wcb'}; # or PSGI server supplies it
        my $async = $env->{'pi-httpd.async'};
 
        my $cb = sub {
@@ -292,6 +281,10 @@ sub psgi_return {
                                                                $buf, $filter);
                        $wcb->($r);
                }
+
+               # Workaround a leak under Perl 5.16.3 when combined with
+               # Plack::Middleware::Deflater:
+               $wcb = undef;
        };
        $limiter ||= $def_limiter ||= PublicInbox::Qspawn::Limiter->new(32);
        my $start_cb = sub { # may run later, much later...
@@ -299,6 +292,7 @@ sub psgi_return {
                if ($async) {
                        # PublicInbox::HTTPD::Async->new($rpipe, $cb, $end)
                        $async = $async->($rpipe, $cb, $end);
+                       # $cb will call ->async_pass or ->close
                } else { # generic PSGI
                        $cb->() while $rd_hdr;
                }