]> Sergey Matveev's repositories - public-inbox.git/blob - lib/PublicInbox/Qspawn.pm
script/*: set executable bit on -learn and -imapd
[public-inbox.git] / lib / PublicInbox / Qspawn.pm
1 # Copyright (C) 2016-2020 all contributors <meta@public-inbox.org>
2 # License: AGPL-3.0+ <https://www.gnu.org/licenses/agpl-3.0.txt>
3
4 # Like most Perl modules in public-inbox, this is internal and
5 # NOT subject to any stability guarantees!  It is only documented
6 # for other hackers.
7 #
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.
14 #
15 # This does not depend on PublicInbox::DS or any other external
16 # scheduling mechanism, you just need to call start() and finish()
17 # appropriately. However, public-inbox-httpd (which uses PublicInbox::DS)
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.
21 #
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).
25
26 package PublicInbox::Qspawn;
27 use strict;
28 use PublicInbox::Spawn qw(popen_rd);
29 use PublicInbox::GzipFilter;
30
31 # n.b.: we get EAGAIN with public-inbox-httpd, and EINTR on other PSGI servers
32 use Errno qw(EAGAIN EINTR);
33
34 my $def_limiter;
35
36 # declares a command to spawn (but does not spawn it).
37 # $cmd is the command to spawn
38 # $cmd_env is the environ for the child process (not PSGI env)
39 # $opt can include redirects and perhaps other process spawning options
40 sub new ($$$;) {
41         my ($class, $cmd, $cmd_env, $opt) = @_;
42         bless { args => [ $cmd, $cmd_env, $opt ] }, $class;
43 }
44
45 sub _do_spawn {
46         my ($self, $start_cb, $limiter) = @_;
47         my $err;
48         my ($cmd, $cmd_env, $opt) = @{delete $self->{args}};
49         my %o = %{$opt || {}};
50         $self->{limiter} = $limiter;
51         foreach my $k (@PublicInbox::Spawn::RLIMITS) {
52                 if (defined(my $rlimit = $limiter->{$k})) {
53                         $o{$k} = $rlimit;
54                 }
55         }
56         $self->{cmd} = $o{quiet} ? undef : $cmd;
57         eval {
58                 # popen_rd may die on EMFILE, ENFILE
59                 ($self->{rpipe}, $self->{pid}) = popen_rd($cmd, $cmd_env, \%o);
60
61                 die "E: $!" unless defined($self->{pid});
62
63                 $limiter->{running}++;
64                 $start_cb->($self); # EPOLL_CTL_ADD may ENOSPC/ENOMEM
65         };
66         finish($self, $@) if $@;
67 }
68
69 sub child_err ($) {
70         my ($child_error) = @_; # typically $?
71         my $exitstatus = ($child_error >> 8) or return;
72         my $sig = $child_error & 127;
73         my $msg = "exit status=$exitstatus";
74         $msg .= " signal=$sig" if $sig;
75         $msg;
76 }
77
78 sub log_err ($$) {
79         my ($env, $msg) = @_;
80         $env->{'psgi.errors'}->print($msg, "\n");
81 }
82
83 sub finalize ($$) {
84         my ($self, $err) = @_;
85
86         my ($env, $qx_cb, $qx_arg, $qx_buf) =
87                 delete @$self{qw(psgi_env qx_cb qx_arg qx_buf)};
88
89         # done, spawn whatever's in the queue
90         my $limiter = $self->{limiter};
91         my $running = --$limiter->{running};
92
93         if ($running < $limiter->{max}) {
94                 if (my $next = shift(@{$limiter->{run_queue}})) {
95                         _do_spawn(@$next, $limiter);
96                 }
97         }
98
99         if ($err) {
100                 if (defined $self->{err}) {
101                         $self->{err} .= "; $err";
102                 } else {
103                         $self->{err} = $err;
104                 }
105                 if ($env && $self->{cmd}) {
106                         log_err($env, join(' ', @{$self->{cmd}}) . ": $err");
107                 }
108         }
109         if ($qx_cb) {
110                 eval { $qx_cb->($qx_buf, $qx_arg) };
111         } elsif (my $wcb = delete $env->{'qspawn.wcb'}) {
112                 # have we started writing, yet?
113                 require PublicInbox::WwwStatic;
114                 $wcb->(PublicInbox::WwwStatic::r(500));
115         }
116 }
117
118 # callback for dwaitpid
119 sub waitpid_err ($$) {
120         my ($self, $pid) = @_;
121         my $xpid = delete $self->{pid};
122         my $err;
123         if (defined $pid) {
124                 if ($pid > 0) { # success!
125                         $err = child_err($?);
126                 } elsif ($pid < 0) { # ??? does this happen in our case?
127                         $err = "W: waitpid($xpid, 0) => $pid: $!";
128                 } # else should not be called with pid == 0
129         }
130         finalize($self, $err);
131 }
132
133 sub do_waitpid ($) {
134         my ($self) = @_;
135         my $pid = $self->{pid};
136         # PublicInbox::DS may not be loaded
137         eval { PublicInbox::DS::dwaitpid($pid, \&waitpid_err, $self) };
138         # done if we're running in PublicInbox::DS::EventLoop
139         if ($@) {
140                 # non public-inbox-{httpd,nntpd} callers may block:
141                 my $ret = waitpid($pid, 0);
142                 waitpid_err($self, $ret);
143         }
144 }
145
146 sub finish ($;$) {
147         my ($self, $err) = @_;
148         if (delete $self->{rpipe}) {
149                 do_waitpid($self);
150         } else {
151                 finalize($self, $err);
152         }
153 }
154
155 sub start ($$$) {
156         my ($self, $limiter, $start_cb) = @_;
157         if ($limiter->{running} < $limiter->{max}) {
158                 _do_spawn($self, $start_cb, $limiter);
159         } else {
160                 push @{$limiter->{run_queue}}, [ $self, $start_cb ];
161         }
162 }
163
164 sub psgi_qx_init_cb {
165         my ($self) = @_;
166         my $async = delete $self->{async};
167         my ($r, $buf);
168         my $qx_fh = $self->{qx_fh};
169 reread:
170         $r = sysread($self->{rpipe}, $buf, 65536);
171         if ($async) {
172                 $async->async_pass($self->{psgi_env}->{'psgix.io'},
173                                         $qx_fh, \$buf);
174         } elsif (defined $r) {
175                 $r ? (print $qx_fh $buf) : event_step($self, undef);
176         } else {
177                 return if $! == EAGAIN; # try again when notified
178                 goto reread if $! == EINTR;
179                 event_step($self, $!);
180         }
181 }
182
183 sub psgi_qx_start {
184         my ($self) = @_;
185         if (my $async = $self->{psgi_env}->{'pi-httpd.async'}) {
186                 # PublicInbox::HTTPD::Async->new(rpipe, $cb, cb_arg, $end_obj)
187                 $self->{async} = $async->($self->{rpipe},
188                                         \&psgi_qx_init_cb, $self, $self);
189                 # init_cb will call ->async_pass or ->close
190         } else { # generic PSGI
191                 psgi_qx_init_cb($self) while $self->{qx_fh};
192         }
193 }
194
195 # Similar to `backtick` or "qx" ("perldoc -f qx"), it calls $qx_cb with
196 # the stdout of the given command when done; but respects the given limiter
197 # $env is the PSGI env.  As with ``/qx; only use this when output is small
198 # and safe to slurp.
199 sub psgi_qx {
200         my ($self, $env, $limiter, $qx_cb, $qx_arg) = @_;
201         $self->{psgi_env} = $env;
202         my $qx_buf = '';
203         open(my $qx_fh, '+>', \$qx_buf) or die; # PerlIO::scalar
204         $self->{qx_cb} = $qx_cb;
205         $self->{qx_arg} = $qx_arg;
206         $self->{qx_fh} = $qx_fh;
207         $self->{qx_buf} = \$qx_buf;
208         $limiter ||= $def_limiter ||= PublicInbox::Qspawn::Limiter->new(32);
209         start($self, $limiter, \&psgi_qx_start);
210 }
211
212 # this is called on pipe EOF to reap the process, may be called
213 # via PublicInbox::DS event loop OR via GetlineBody for generic
214 # PSGI servers.
215 sub event_step {
216         my ($self, $err) = @_; # $err: $!
217         log_err($self->{psgi_env}, "psgi_{return,qx} $err") if defined($err);
218         finish($self);
219         my ($fh, $qx_fh) = delete(@$self{qw(fh qx_fh)});
220         $fh->close if $fh; # async-only (psgi_return)
221 }
222
223 sub rd_hdr ($) {
224         my ($self) = @_;
225         # typically used for reading CGI headers
226         # we must loop until EAGAIN for EPOLLET in HTTPD/Async.pm
227         # We also need to check EINTR for generic PSGI servers.
228         my $ret;
229         my $total_rd = 0;
230         my $hdr_buf = $self->{hdr_buf};
231         my ($ph_cb, $ph_arg) = @{$self->{parse_hdr}};
232         do {
233                 my $r = sysread($self->{rpipe}, $$hdr_buf, 4096,
234                                 length($$hdr_buf));
235                 if (defined($r)) {
236                         $total_rd += $r;
237                         eval { $ret = $ph_cb->($total_rd, $hdr_buf, $ph_arg) };
238                         if ($@) {
239                                 log_err($self->{psgi_env}, "parse_hdr: $@");
240                                 $ret = [ 500, [], [ "Internal error\n" ] ];
241                         }
242                 } else {
243                         # caller should notify us when it's ready:
244                         return if $! == EAGAIN;
245                         next if $! == EINTR; # immediate retry
246                         log_err($self->{psgi_env}, "error reading header: $!");
247                         $ret = [ 500, [], [ "Internal error\n" ] ];
248                 }
249         } until (defined $ret);
250         delete $self->{parse_hdr}; # done parsing headers
251         $ret;
252 }
253
254 sub psgi_return_init_cb {
255         my ($self) = @_;
256         my $r = rd_hdr($self) or return;
257         my $env = $self->{psgi_env};
258         my $filter = delete $env->{'qspawn.filter'} //
259                 PublicInbox::GzipFilter::qsp_maybe($r->[1], $env);
260
261         my $wcb = delete $env->{'qspawn.wcb'};
262         my $async = delete $self->{async};
263         if (scalar(@$r) == 3) { # error
264                 if ($async) {
265                         # calls rpipe->close && ->event_step
266                         $async->close;
267                 } else {
268                         $self->{rpipe}->close;
269                         event_step($self);
270                 }
271                 $wcb->($r);
272         } elsif ($async) {
273                 # done reading headers, handoff to read body
274                 my $fh = $wcb->($r); # scalar @$r == 2
275                 $fh = $filter->attach($fh) if $filter;
276                 $self->{fh} = $fh;
277                 $async->async_pass($env->{'psgix.io'}, $fh,
278                                         delete($self->{hdr_buf}));
279         } else { # for synchronous PSGI servers
280                 require PublicInbox::GetlineBody;
281                 $r->[2] = PublicInbox::GetlineBody->new($self->{rpipe},
282                                         \&event_step, $self,
283                                         ${$self->{hdr_buf}}, $filter);
284                 $wcb->($r);
285         }
286 }
287
288 sub psgi_return_start { # may run later, much later...
289         my ($self) = @_;
290         if (my $async = $self->{psgi_env}->{'pi-httpd.async'}) {
291                 # PublicInbox::HTTPD::Async->new(rpipe, $cb, $cb_arg, $end_obj)
292                 $self->{async} = $async->($self->{rpipe},
293                                         \&psgi_return_init_cb, $self, $self);
294         } else { # generic PSGI
295                 psgi_return_init_cb($self) while $self->{parse_hdr};
296         }
297 }
298
299 # Used for streaming the stdout of one process as a PSGI response.
300 #
301 # $env is the PSGI env.
302 # optional keys in $env:
303 #   $env->{'qspawn.wcb'} - the write callback from the PSGI server
304 #                          optional, use this if you've already
305 #                          captured it elsewhere.  If not given,
306 #                          psgi_return will return an anonymous
307 #                          sub for the PSGI server to call
308 #
309 #   $env->{'qspawn.filter'} - filter object, responds to ->attach for
310 #                             pi-httpd.async and ->translate for generic
311 #                             PSGI servers
312 #
313 # $limiter - the Limiter object to use (uses the def_limiter if not given)
314 #
315 # $parse_hdr - Initial read function; often for parsing CGI header output.
316 #              It will be given the return value of sysread from the pipe
317 #              and a string ref of the current buffer.  Returns an arrayref
318 #              for PSGI responses.  2-element arrays in PSGI mean the
319 #              body will be streamed, later, via writes (push-based) to
320 #              psgix.io.  3-element arrays means the body is available
321 #              immediately (or streamed via ->getline (pull-based)).
322 sub psgi_return {
323         my ($self, $env, $limiter, $parse_hdr, $hdr_arg) = @_;
324         $self->{psgi_env} = $env;
325         $self->{hdr_buf} = \(my $hdr_buf = '');
326         $self->{parse_hdr} = [ $parse_hdr, $hdr_arg ];
327         $limiter ||= $def_limiter ||= PublicInbox::Qspawn::Limiter->new(32);
328
329         # the caller already captured the PSGI write callback from
330         # the PSGI server, so we can call ->start, here:
331         $env->{'qspawn.wcb'} and
332                 return start($self, $limiter, \&psgi_return_start);
333
334         # the caller will return this sub to the PSGI server, so
335         # it can set the response callback (that is, for
336         # PublicInbox::HTTP, the chunked_wcb or identity_wcb callback),
337         # but other HTTP servers are supported:
338         sub {
339                 $env->{'qspawn.wcb'} = $_[0];
340                 start($self, $limiter, \&psgi_return_start);
341         }
342 }
343
344 package PublicInbox::Qspawn::Limiter;
345 use strict;
346 use warnings;
347
348 sub new {
349         my ($class, $max) = @_;
350         bless {
351                 # 32 is same as the git-daemon connection limit
352                 max => $max || 32,
353                 running => 0,
354                 run_queue => [],
355                 # RLIMIT_CPU => undef,
356                 # RLIMIT_DATA => undef,
357                 # RLIMIT_CORE => undef,
358         }, $class;
359 }
360
361 sub setup_rlimit {
362         my ($self, $name, $config) = @_;
363         foreach my $rlim (@PublicInbox::Spawn::RLIMITS) {
364                 my $k = lc($rlim);
365                 $k =~ tr/_//d;
366                 $k = "publicinboxlimiter.$name.$k";
367                 defined(my $v = $config->{$k}) or next;
368                 my @rlimit = split(/\s*,\s*/, $v);
369                 if (scalar(@rlimit) == 1) {
370                         push @rlimit, $rlimit[0];
371                 } elsif (scalar(@rlimit) != 2) {
372                         warn "could not parse $k: $v\n";
373                 }
374                 eval { require BSD::Resource };
375                 if ($@) {
376                         warn "BSD::Resource missing for $rlim";
377                         next;
378                 }
379                 foreach my $i (0..$#rlimit) {
380                         next if $rlimit[$i] ne 'INFINITY';
381                         $rlimit[$i] = BSD::Resource::RLIM_INFINITY();
382                 }
383                 $self->{$rlim} = \@rlimit;
384         }
385 }
386
387 1;