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 finish ($) {
  34         my ($self) = @_;
  35         my $limiter = $self->{limiter};
  36         my $running;
  37         if (delete $self->{rpipe}) {
  38                 my $pid = delete $self->{pid};
  39                 $self->{err} = $pid == waitpid($pid, 0) ? $? :
  40                                 "PID:$pid still running?";
  41                 $running = --$limiter->{running};
  42         }
  43
  44         # limiter->{max} may change dynamically
  45         if (($running || $limiter->{running}) < $limiter->{max}) {
  46                 if (my $next = shift @{$limiter->{run_queue}}) {
  47                         _do_spawn(@$next);
  48                 }
  49         }
  50         $self->{err};
  51 }
  52
  53 sub start {
  54         my ($self, $limiter, $cb) = @_;
  55         $self->{limiter} = $limiter;
  56
  57         if ($limiter->{running} < $limiter->{max}) {
  58                 _do_spawn($self, $cb);
  59         } else {
  60                 push @{$limiter->{run_queue}}, [ $self, $cb ];
  61         }
  62 }
  63
  64 sub _psgi_finish ($$) {
  65         my ($self, $env) = @_;
  66         my $err = $self->finish;
  67         if ($err && !$env->{'qspawn.quiet'}) {
  68                 $err = join(' ', @{$self->{args}->[0]}).": $err\n";
  69                 $env->{'psgi.errors'}->print($err);
  70         }
  71 }
  72
  73 sub psgi_qx {
  74         my ($self, $env, $limiter, $qx_cb) = @_;
  75         my $qx = PublicInbox::Qspawn::Qx->new;
  76         my $end = sub {
  77                 _psgi_finish($self, $env);
  78                 eval { $qx_cb->($qx) };
  79                 $qx = undef;
  80         };
  81         my $rpipe;
  82         my $async = $env->{'pi-httpd.async'};
  83         my $cb = sub {
  84                 my $r = sysread($rpipe, my $buf, 8192);
  85                 if ($async) {
  86                         $async->async_pass($env->{'psgix.io'}, $qx, \$buf);
  87                 } elsif (defined $r) {
  88                         $r ? $qx->write($buf) : $end->();
  89                 } else {
  90                         return if $!{EAGAIN} || $!{EINTR}; # loop again
  91                         $end->();
  92                 }
  93         };
  94         $limiter ||= $def_limiter ||= PublicInbox::Qspawn::Limiter->new(32);
  95         $self->start($limiter, sub { # may run later, much later...
  96                 ($rpipe) = @_;
  97                 if ($async) {
  98                 # PublicInbox::HTTPD::Async->new($rpipe, $cb, $end)
  99                         $async = $async->($rpipe, $cb, $end);
 100                 } else { # generic PSGI
 101                         $cb->() while $qx;
 102                 }
 103         });
 104 }
 105
 106 # create a filter for "push"-based streaming PSGI writes used by HTTPD::Async
 107 sub filter_fh ($$) {
 108         my ($fh, $filter) = @_;
 109         Plack::Util::inline_object(
 110                 close => sub {
 111                         $fh->write($filter->(undef));
 112                         $fh->close;
 113                 },
 114                 write => sub {
 115                         $fh->write($filter->($_[0]));
 116                 });
 117 }
 118
 119 sub psgi_return {
 120         my ($self, $env, $limiter, $parse_hdr) = @_;
 121         my ($fh, $rpipe);
 122         my $end = sub {
 123                 _psgi_finish($self, $env);
 124                 $fh->close if $fh; # async-only
 125         };
 126
 127         my $buf = '';
 128         my $rd_hdr = sub {
 129                 my $r = sysread($rpipe, $buf, 1024, length($buf));
 130                 return if !defined($r) && ($!{EINTR} || $!{EAGAIN});
 131                 $parse_hdr->($r, \$buf);
 132         };
 133         my $res = delete $env->{'qspawn.response'};
 134         my $async = $env->{'pi-httpd.async'};
 135         my $cb = sub {
 136                 my $r = $rd_hdr->() or return;
 137                 $rd_hdr = undef;
 138                 my $filter = delete $env->{'qspawn.filter'};
 139                 if (scalar(@$r) == 3) { # error
 140                         if ($async) {
 141                                 $async->close; # calls rpipe->close and $end
 142                         } else {
 143                                 $rpipe->close;
 144                                 $end->();
 145                         }
 146                         $res->($r);
 147                 } elsif ($async) {
 148                         $fh = $res->($r); # scalar @$r == 2
 149                         $fh = filter_fh($fh, $filter) if $filter;
 150                         $async->async_pass($env->{'psgix.io'}, $fh, \$buf);
 151                 } else { # for synchronous PSGI servers
 152                         require PublicInbox::GetlineBody;
 153                         $r->[2] = PublicInbox::GetlineBody->new($rpipe, $end,
 154                                                                 $buf, $filter);
 155                         $res->($r);
 156                 }
 157         };
 158         $limiter ||= $def_limiter ||= PublicInbox::Qspawn::Limiter->new(32);
 159         my $start_cb = sub { # may run later, much later...
 160                 ($rpipe) = @_;
 161                 if ($async) {
 162                         # PublicInbox::HTTPD::Async->new($rpipe, $cb, $end)
 163                         $async = $async->($rpipe, $cb, $end);
 164                 } else { # generic PSGI
 165                         $cb->() while $rd_hdr;
 166                 }
 167         };
 168
 169         return $self->start($limiter, $start_cb) if $res;
 170
 171         sub {
 172                 ($res) = @_;
 173                 $self->start($limiter, $start_cb);
 174         };
 175 }
 176
 177 package PublicInbox::Qspawn::Limiter;
 178 use strict;
 179 use warnings;
 180
 181 sub new {
 182         my ($class, $max) = @_;
 183         bless {
 184                 # 32 is same as the git-daemon connection limit
 185                 max => $max || 32,
 186                 running => 0,
 187                 run_queue => [],
 188         }, $class;
 189 }
 190
 191 # captures everything into a buffer and executes a callback when done
 192 package PublicInbox::Qspawn::Qx;
 193 use strict;
 194 use warnings;
 195
 196 sub new {
 197         my ($class) = @_;
 198         my $buf = '';
 199         bless \$buf, $class;
 200 }
 201
 202 # called by PublicInbox::HTTPD::Async ($fh->write)
 203 sub write {
 204         ${$_[0]} .= $_[1];
 205         undef;
 206 }
 207
 208 1;