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 # create a filter for "push"-based streaming PSGI writes used by HTTPD::Async
  65 sub filter_fh ($$) {
  66         my ($fh, $filter) = @_;
  67         Plack::Util::inline_object(
  68                 close => sub {
  69                         $fh->write($filter->(undef));
  70                         $fh->close;
  71                 },
  72                 write => sub {
  73                         $fh->write($filter->($_[0]));
  74                 });
  75 }
  76
  77 sub psgi_return {
  78         my ($self, $env, $limiter, $parse_hdr) = @_;
  79         my ($fh, $rpipe);
  80         my $end = sub {
  81                 my $err = $self->finish;
  82                 if ($err && !$env->{'qspawn.quiet'}) {
  83                         $err = join(' ', @{$self->{args}->[0]}).": $err\n";
  84                         $env->{'psgi.errors'}->print($err);
  85                 }
  86                 $fh->close if $fh; # async-only
  87         };
  88
  89         # Danga::Socket users, we queue up the read_enable callback to
  90         # fire after pending writes are complete:
  91         my $buf = '';
  92         my $rd_hdr = sub {
  93                 my $r = sysread($rpipe, $buf, 1024, length($buf));
  94                 return if !defined($r) && ($!{EINTR} || $!{EAGAIN});
  95                 $parse_hdr->($r, \$buf);
  96         };
  97         my $res;
  98         my $async = $env->{'pi-httpd.async'};
  99         my $cb = sub {
 100                 my $r = $rd_hdr->() or return;
 101                 $rd_hdr = undef;
 102                 my $filter = delete $env->{'qspawn.filter'};
 103                 if (scalar(@$r) == 3) { # error
 104                         if ($async) {
 105                                 $async->close; # calls rpipe->close
 106                         } else {
 107                                 $rpipe->close;
 108                                 $end->();
 109                         }
 110                         $res->($r);
 111                 } elsif ($async) {
 112                         $fh = $res->($r); # scalar @$r == 2
 113                         $fh = filter_fh($fh, $filter) if $filter;
 114                         $async->async_pass($env->{'psgix.io'}, $fh, \$buf);
 115                 } else { # for synchronous PSGI servers
 116                         require PublicInbox::GetlineBody;
 117                         $r->[2] = PublicInbox::GetlineBody->new($rpipe, $end,
 118                                                                 $buf, $filter);
 119                         $res->($r);
 120                 }
 121         };
 122         $limiter ||= $def_limiter ||= PublicInbox::Qspawn::Limiter->new(32);
 123         sub {
 124                 ($res) = @_;
 125                 $self->start($limiter, sub { # may run later, much later...
 126                         ($rpipe) = @_;
 127                         if ($async) {
 128                         # PublicInbox::HTTPD::Async->new($rpipe, $cb, $end)
 129                                 $async = $async->($rpipe, $cb, $end);
 130                         } else { # generic PSGI
 131                                 $cb->() while $rd_hdr;
 132                         }
 133                 });
 134         };
 135 }
 136
 137 package PublicInbox::Qspawn::Limiter;
 138 use strict;
 139 use warnings;
 140
 141 sub new {
 142         my ($class, $max) = @_;
 143         bless {
 144                 # 32 is same as the git-daemon connection limit
 145                 max => $max || 32,
 146                 running => 0,
 147                 run_queue => [],
 148         }, $class;
 149 }
 150
 151 1;