lib/PublicInbox/Qspawn.pm

   1 # Copyright (C) 2016-2018 all contributors <meta@public-inbox.org>
   2 # License: AGPL-3.0+ <https://www.gnu.org/licenses/agpl-3.0.txt>
   3
   4 # Limits the number of processes spawned
   5 # This does not depend on Danga::Socket or any other external
   6 # scheduling mechanism, you just need to call start and finish
   7 # appropriately
   8 package PublicInbox::Qspawn;
   9 use strict;
  10 use warnings;
  11 use PublicInbox::Spawn qw(popen_rd);
  12 require Plack::Util;
  13 my $def_limiter;
  14
  15 sub new ($$$;) {
  16         my ($class, $cmd, $env, $opt) = @_;
  17         bless { args => [ $cmd, $env, $opt ] }, $class;
  18 }
  19
  20 sub _do_spawn {
  21         my ($self, $cb) = @_;
  22         my $err;
  23
  24         ($self->{rpipe}, $self->{pid}) = popen_rd(@{$self->{args}});
  25         if (defined $self->{pid}) {
  26                 $self->{limiter}->{running}++;
  27         } else {
  28                 $self->{err} = $!;
  29         }
  30         $cb->($self->{rpipe});
  31 }
  32
  33 sub child_err ($) {
  34         my ($child_error) = @_; # typically $?
  35         my $exitstatus = ($child_error >> 8) or return;
  36         my $sig = $child_error & 127;
  37         my $msg = "exit status=$exitstatus";
  38         $msg .= " signal=$sig" if $sig;
  39         $msg;
  40 }
  41
  42 sub finish ($) {
  43         my ($self) = @_;
  44         my $limiter = $self->{limiter};
  45         my $running;
  46         if (delete $self->{rpipe}) {
  47                 my $pid = delete $self->{pid};
  48                 $self->{err} = $pid == waitpid($pid, 0) ? child_err($?) :
  49                                 "PID:$pid still running?";
  50                 $running = --$limiter->{running};
  51         }
  52
  53         # limiter->{max} may change dynamically
  54         if (($running || $limiter->{running}) < $limiter->{max}) {
  55                 if (my $next = shift @{$limiter->{run_queue}}) {
  56                         _do_spawn(@$next);
  57                 }
  58         }
  59         $self->{err};
  60 }
  61
  62 sub start {
  63         my ($self, $limiter, $cb) = @_;
  64         $self->{limiter} = $limiter;
  65
  66         if ($limiter->{running} < $limiter->{max}) {
  67                 _do_spawn($self, $cb);
  68         } else {
  69                 push @{$limiter->{run_queue}}, [ $self, $cb ];
  70         }
  71 }
  72
  73 sub _psgi_finish ($$) {
  74         my ($self, $env) = @_;
  75         my $err = $self->finish;
  76         if ($err && !$env->{'qspawn.quiet'}) {
  77                 $err = join(' ', @{$self->{args}->[0]}).": $err\n";
  78                 $env->{'psgi.errors'}->print($err);
  79         }
  80 }
  81
  82 sub psgi_qx {
  83         my ($self, $env, $limiter, $qx_cb) = @_;
  84         my $qx = PublicInbox::Qspawn::Qx->new;
  85         my $end = sub {
  86                 _psgi_finish($self, $env);
  87                 eval { $qx_cb->($qx) };
  88                 $qx = undef;
  89         };
  90         my $rpipe;
  91         my $async = $env->{'pi-httpd.async'};
  92         my $cb = sub {
  93                 my $r = sysread($rpipe, my $buf, 8192);
  94                 if ($async) {
  95                         $async->async_pass($env->{'psgix.io'}, $qx, \$buf);
  96                 } elsif (defined $r) {
  97                         $r ? $qx->write($buf) : $end->();
  98                 } else {
  99                         return if $!{EAGAIN} || $!{EINTR}; # loop again
 100                         $end->();
 101                 }
 102         };
 103         $limiter ||= $def_limiter ||= PublicInbox::Qspawn::Limiter->new(32);
 104         $self->start($limiter, sub { # may run later, much later...
 105                 ($rpipe) = @_;
 106                 if ($async) {
 107                 # PublicInbox::HTTPD::Async->new($rpipe, $cb, $end)
 108                         $async = $async->($rpipe, $cb, $end);
 109                 } else { # generic PSGI
 110                         $cb->() while $qx;
 111                 }
 112         });
 113 }
 114
 115 # create a filter for "push"-based streaming PSGI writes used by HTTPD::Async
 116 sub filter_fh ($$) {
 117         my ($fh, $filter) = @_;
 118         Plack::Util::inline_object(
 119                 close => sub {
 120                         $fh->write($filter->(undef));
 121                         $fh->close;
 122                 },
 123                 write => sub {
 124                         $fh->write($filter->($_[0]));
 125                 });
 126 }
 127
 128 sub psgi_return {
 129         my ($self, $env, $limiter, $parse_hdr) = @_;
 130         my ($fh, $rpipe);
 131         my $end = sub {
 132                 _psgi_finish($self, $env);
 133                 $fh->close if $fh; # async-only
 134         };
 135
 136         my $buf = '';
 137         my $rd_hdr = sub {
 138                 my $r = sysread($rpipe, $buf, 1024, length($buf));
 139                 return if !defined($r) && ($!{EINTR} || $!{EAGAIN});
 140                 $parse_hdr->($r, \$buf);
 141         };
 142         my $res = delete $env->{'qspawn.response'};
 143         my $async = $env->{'pi-httpd.async'};
 144         my $cb = sub {
 145                 my $r = $rd_hdr->() or return;
 146                 $rd_hdr = undef;
 147                 my $filter = delete $env->{'qspawn.filter'};
 148                 if (scalar(@$r) == 3) { # error
 149                         if ($async) {
 150                                 $async->close; # calls rpipe->close and $end
 151                         } else {
 152                                 $rpipe->close;
 153                                 $end->();
 154                         }
 155                         $res->($r);
 156                 } elsif ($async) {
 157                         $fh = $res->($r); # scalar @$r == 2
 158                         $fh = filter_fh($fh, $filter) if $filter;
 159                         $async->async_pass($env->{'psgix.io'}, $fh, \$buf);
 160                 } else { # for synchronous PSGI servers
 161                         require PublicInbox::GetlineBody;
 162                         $r->[2] = PublicInbox::GetlineBody->new($rpipe, $end,
 163                                                                 $buf, $filter);
 164                         $res->($r);
 165                 }
 166         };
 167         $limiter ||= $def_limiter ||= PublicInbox::Qspawn::Limiter->new(32);
 168         my $start_cb = sub { # may run later, much later...
 169                 ($rpipe) = @_;
 170                 if ($async) {
 171                         # PublicInbox::HTTPD::Async->new($rpipe, $cb, $end)
 172                         $async = $async->($rpipe, $cb, $end);
 173                 } else { # generic PSGI
 174                         $cb->() while $rd_hdr;
 175                 }
 176         };
 177
 178         return $self->start($limiter, $start_cb) if $res;
 179
 180         sub {
 181                 ($res) = @_;
 182                 $self->start($limiter, $start_cb);
 183         };
 184 }
 185
 186 package PublicInbox::Qspawn::Limiter;
 187 use strict;
 188 use warnings;
 189
 190 sub new {
 191         my ($class, $max) = @_;
 192         bless {
 193                 # 32 is same as the git-daemon connection limit
 194                 max => $max || 32,
 195                 running => 0,
 196                 run_queue => [],
 197         }, $class;
 198 }
 199
 200 # captures everything into a buffer and executes a callback when done
 201 package PublicInbox::Qspawn::Qx;
 202 use strict;
 203 use warnings;
 204
 205 sub new {
 206         my ($class) = @_;
 207         my $buf = '';
 208         bless \$buf, $class;
 209 }
 210
 211 # called by PublicInbox::HTTPD::Async ($fh->write)
 212 sub write {
 213         ${$_[0]} .= $_[1];
 214         undef;
 215 }
 216
 217 1;