1 # Copyright (C) 2016-2019 all contributors <meta@public-inbox.org>
2 # License: AGPL-3.0+ <https://www.gnu.org/licenses/agpl-3.0.txt>
4 # Like most Perl modules in public-inbox, this is internal and
5 # NOT subject to any stability guarantees! It is only documented
8 # This is used to limit the number of processes spawned by the
9 # PSGI server, so it acts like a semaphore and queues up extra
10 # commands to be run if currently at the limit. Multiple "limiters"
11 # may be configured which give inboxes different channels to
12 # operate in. This can be useful to ensure smaller inboxes can
13 # be cloned while cloning of large inboxes is maxed out.
15 # This does not depend on Danga::Socket or any other external
16 # scheduling mechanism, you just need to call start() and finish()
17 # appropriately. However, public-inbox-httpd (which uses Danga::Socket)
18 # will be able to schedule this based on readability of stdout from
19 # the spawned process. See GitHTTPBackend.pm and SolverGit.pm for
20 # usage examples. It does not depend on any form of threading.
22 # This is useful for scheduling CGI execution of both long-lived
23 # git-http-backend(1) process (for "git clone") as well as short-lived
24 # processes such as git-apply(1).
26 package PublicInbox::Qspawn;
29 use PublicInbox::Spawn qw(popen_rd);
34 # declares a command to spawn (but does not spawn it).
35 # $cmd is the command to spawn
36 # $env is the environ for the child process
37 # $opt can include redirects and perhaps other process spawning options
39 my ($class, $cmd, $env, $opt) = @_;
40 bless { args => [ $cmd, $env, $opt ] }, $class;
46 my ($cmd, $env, $opts) = @{$self->{args}};
47 my %opts = %{$opts || {}};
48 my $limiter = $self->{limiter};
49 foreach my $k (PublicInbox::Spawn::RLIMITS()) {
50 if (defined(my $rlimit = $limiter->{$k})) {
55 ($self->{rpipe}, $self->{pid}) = popen_rd($cmd, $env, \%opts);
56 if (defined $self->{pid}) {
57 $limiter->{running}++;
61 $cb->($self->{rpipe});
65 my ($child_error) = @_; # typically $?
66 my $exitstatus = ($child_error >> 8) or return;
67 my $sig = $child_error & 127;
68 my $msg = "exit status=$exitstatus";
69 $msg .= " signal=$sig" if $sig;
75 my $limiter = $self->{limiter};
77 if (delete $self->{rpipe}) {
78 my $pid = delete $self->{pid};
79 $self->{err} = $pid == waitpid($pid, 0) ? child_err($?) :
80 "PID:$pid still running?";
81 $running = --$limiter->{running};
84 # limiter->{max} may change dynamically
85 if (($running || $limiter->{running}) < $limiter->{max}) {
86 if (my $next = shift @{$limiter->{run_queue}}) {
94 my ($self, $limiter, $cb) = @_;
95 $self->{limiter} = $limiter;
97 if ($limiter->{running} < $limiter->{max}) {
98 _do_spawn($self, $cb);
100 push @{$limiter->{run_queue}}, [ $self, $cb ];
104 sub _psgi_finish ($$) {
105 my ($self, $env) = @_;
106 my $err = $self->finish;
107 if ($err && !$env->{'qspawn.quiet'}) {
108 $err = join(' ', @{$self->{args}->[0]}).": $err\n";
109 $env->{'psgi.errors'}->print($err);
113 # Similar to `backtick` or "qx" ("perldoc -f qx"), it calls $qx_cb with
114 # the stdout of the given command when done; but respects the given limiter
115 # $env is the PSGI env. As with ``/qx; only use this when output is small
118 my ($self, $env, $limiter, $qx_cb) = @_;
119 my $qx = PublicInbox::Qspawn::Qx->new;
121 _psgi_finish($self, $env);
122 eval { $qx_cb->($qx) };
126 my $async = $env->{'pi-httpd.async'};
128 my $r = sysread($rpipe, my $buf, 8192);
130 $async->async_pass($env->{'psgix.io'}, $qx, \$buf);
131 } elsif (defined $r) {
132 $r ? $qx->write($buf) : $end->();
134 return if $!{EAGAIN} || $!{EINTR}; # loop again
138 $limiter ||= $def_limiter ||= PublicInbox::Qspawn::Limiter->new(32);
139 $self->start($limiter, sub { # may run later, much later...
142 # PublicInbox::HTTPD::Async->new($rpipe, $cb, $end)
143 $async = $async->($rpipe, $cb, $end);
144 } else { # generic PSGI
150 # create a filter for "push"-based streaming PSGI writes used by HTTPD::Async
152 my ($fh, $filter) = @_;
153 Plack::Util::inline_object(
155 $fh->write($filter->(undef));
159 $fh->write($filter->($_[0]));
163 # Used for streaming the stdout of one process as a PSGI response.
165 # $env is the PSGI env.
166 # optional keys in $env:
167 # $env->{'qspawn.wcb'} - the write callback from the PSGI server
168 # optional, use this if you've already
169 # captured it elsewhere. If not given,
170 # psgi_return will return an anonymous
171 # sub for the PSGI server to call
173 # $env->{'qspawn.filter'} - filter callback, receives a string as input,
176 # $limiter - the Limiter object to use (uses the def_limiter if not given)
178 # $parse_hdr - Initial read function; often for parsing CGI header output.
179 # It will be given the return value of sysread from the pipe
180 # and a string ref of the current buffer. Returns an arrayref
181 # for PSGI responses. 2-element arrays in PSGI mean the
182 # body will be streamed, later, via writes (push-based) to
183 # psgix.io. 3-element arrays means the body is available
184 # immediately (or streamed via ->getline (pull-based)).
186 my ($self, $env, $limiter, $parse_hdr) = @_;
189 _psgi_finish($self, $env);
190 $fh->close if $fh; # async-only
195 my $r = sysread($rpipe, $buf, 1024, length($buf));
196 return if !defined($r) && ($!{EINTR} || $!{EAGAIN});
197 $parse_hdr->($r, \$buf);
200 my $wcb = delete $env->{'qspawn.wcb'};
201 my $async = $env->{'pi-httpd.async'};
204 my $r = $rd_hdr->() or return;
206 my $filter = delete $env->{'qspawn.filter'};
207 if (scalar(@$r) == 3) { # error
209 $async->close; # calls rpipe->close and $end
216 $fh = $wcb->($r); # scalar @$r == 2
217 $fh = filter_fh($fh, $filter) if $filter;
218 $async->async_pass($env->{'psgix.io'}, $fh, \$buf);
219 } else { # for synchronous PSGI servers
220 require PublicInbox::GetlineBody;
221 $r->[2] = PublicInbox::GetlineBody->new($rpipe, $end,
226 $limiter ||= $def_limiter ||= PublicInbox::Qspawn::Limiter->new(32);
227 my $start_cb = sub { # may run later, much later...
230 # PublicInbox::HTTPD::Async->new($rpipe, $cb, $end)
231 $async = $async->($rpipe, $cb, $end);
232 } else { # generic PSGI
233 $cb->() while $rd_hdr;
237 # the caller already captured the PSGI write callback from
238 # the PSGI server, so we can call ->start, here:
239 return $self->start($limiter, $start_cb) if $wcb;
241 # the caller will return this sub to the PSGI server, so
242 # it can set the response callback (that is, for PublicInbox::HTTP,
243 # the chunked_wcb or identity_wcb callback), but other HTTP servers
247 $self->start($limiter, $start_cb);
251 package PublicInbox::Qspawn::Limiter;
256 my ($class, $max) = @_;
258 # 32 is same as the git-daemon connection limit
262 # RLIMIT_CPU => undef,
263 # RLIMIT_DATA => undef,
264 # RLIMIT_CORE => undef,
269 my ($self, $name, $config) = @_;
270 foreach my $rlim (PublicInbox::Spawn::RLIMITS()) {
273 $k = "publicinboxlimiter.$name.$k";
274 defined(my $v = $config->{$k}) or next;
275 my @rlimit = split(/\s*,\s*/, $v);
276 if (scalar(@rlimit) == 1) {
277 push @rlimit, $rlimit[0];
278 } elsif (scalar(@rlimit) != 2) {
279 warn "could not parse $k: $v\n";
281 eval { require BSD::Resource };
283 warn "BSD::Resource missing for $rlim";
286 foreach my $i (0..$#rlimit) {
287 next if $rlimit[$i] ne 'INFINITY';
288 $rlimit[$i] = BSD::Resource::RLIM_INFINITY();
290 $self->{$rlim} = \@rlimit;
294 # captures everything into a buffer and executes a callback when done
295 package PublicInbox::Qspawn::Qx;
305 # called by PublicInbox::HTTPD::Async ($fh->write)