lib/PublicInbox/Filter/Vger.pm
lib/PublicInbox/Gcf2.pm
lib/PublicInbox/Gcf2Client.pm
-lib/PublicInbox/GetlineBody.pm
lib/PublicInbox/GetlineResponse.pm
lib/PublicInbox/Git.pm
lib/PublicInbox/GitAsyncCat.pm
lib/PublicInbox/GzipFilter.pm
lib/PublicInbox/HTTP.pm
lib/PublicInbox/HTTPD.pm
-lib/PublicInbox/HTTPD/Async.pm
lib/PublicInbox/HlMod.pm
lib/PublicInbox/Hval.pm
lib/PublicInbox/IMAP.pm
+++ /dev/null
-# Copyright (C) 2016-2021 all contributors <meta@public-inbox.org>
-# License: AGPL-3.0+ <https://www.gnu.org/licenses/agpl-3.0.txt>
-
-# Wrap a pipe or file for PSGI streaming response bodies and calls the
-# end callback when the object goes out-of-scope.
-# This depends on rpipe being _blocking_ on getline.
-#
-# This is only used by generic PSGI servers and not public-inbox-httpd
-package PublicInbox::GetlineBody;
-use strict;
-use warnings;
-
-sub new {
- my ($class, $rpipe, $end, $end_arg, $buf, $filter) = @_;
- bless {
- rpipe => $rpipe,
- end => $end,
- end_arg => $end_arg,
- initial_buf => $buf,
- filter => $filter,
- }, $class;
-}
-
-# close should always be called after getline returns undef,
-# but a client aborting a connection can ruin our day; so lets
-# hope our underlying PSGI server does not leak references, here.
-sub DESTROY { $_[0]->close }
-
-sub getline {
- my ($self) = @_;
- my $rpipe = $self->{rpipe} or return; # EOF was set on previous call
- my $buf = delete($self->{initial_buf}) // $rpipe->getline;
- delete($self->{rpipe}) unless defined $buf; # set EOF for next call
- if (my $filter = $self->{filter}) {
- $buf = $filter->translate($buf);
- }
- $buf;
-}
-
-sub close {
- my ($self) = @_;
- my ($end, $end_arg) = delete @$self{qw(end end_arg)};
- $end->($end_arg) if $end;
-}
-
-1;
}
}
- # fallback to WwwCoderepo if cgit 404s. Duplicating $ctx prevents
- # ->finalize from the current Qspawn from using qspawn.wcb.
- # This makes qspawn skip ->async_pass and causes
- # PublicInbox::HTTPD::Async::event_step to close shortly after
+ # fallback to WwwCoderepo if cgit 404s
if ($code == 404 && $ctx->{www} && !$ctx->{_coderepo_tried}++) {
my $wcb = delete $ctx->{env}->{'qspawn.wcb'};
$ctx->{env}->{'plack.skip-deflater'} = 1; # prevent 2x gzip
$ctx->{env}->{'qspawn.fallback'} = $code;
my $res = $ctx->{www}->coderepo->srv($ctx);
- # for ->psgi_return_init_cb
$ctx->{env}->{'qspawn.wcb'} = $wcb;
$res; # CODE or ARRAY ref
} else {
undef;
}
-# for GetlineBody (via Qspawn) when NOT using $env->{'pi-httpd.async'}
+# for GetlineResponse (via Qspawn) when NOT using $env->{'pi-httpd.async'}
# Also used for ->getline callbacks
sub translate {
my $self = shift; # $_[1] => input
use Plack::Util ();
use Plack::Builder;
use PublicInbox::HTTP;
-use PublicInbox::HTTPD::Async;
-
-sub pi_httpd_async { PublicInbox::HTTPD::Async->new(@_) }
# we have a different env for ever listener socket for
# SERVER_NAME, SERVER_PORT and psgi.url_scheme
# this to limit git-http-backend(1) parallelism.
# We also check for the truthiness of this to
# detect when to use async paths for slow blobs
- 'pi-httpd.async' => \&pi_httpd_async,
+ 'pi-httpd.async' => 1,
'pi-httpd.app' => $self->{app},
'pi-httpd.warn_cb' => $self->{warn_cb},
}
+++ /dev/null
-# Copyright (C) all contributors <meta@public-inbox.org>
-# License: AGPL-3.0+ <https://www.gnu.org/licenses/agpl-3.0.txt>
-#
-# XXX This is a totally unstable API for public-inbox internal use only
-# This is exposed via the 'pi-httpd.async' key in the PSGI env hash.
-# The name of this key is not even stable!
-# Currently intended for use with read-only pipes with expensive
-# processes such as git-http-backend(1), cgit(1)
-#
-# fields:
-# http: PublicInbox::HTTP ref
-# fh: PublicInbox::HTTP::{Identity,Chunked} ref (can ->write + ->close)
-# cb: initial read callback
-# arg: arg for {cb}
-# end_obj: CODE or object which responds to ->event_step when ->close is called
-package PublicInbox::HTTPD::Async;
-use v5.12;
-use parent qw(PublicInbox::DS);
-use Errno qw(EAGAIN);
-use PublicInbox::Syscall qw(EPOLLIN);
-use PublicInbox::ProcessIONBF;
-
-# This is called via: $env->{'pi-httpd.async'}->()
-# $io is a read-only pipe ($rpipe) for now, but may be a
-# bidirectional socket in the future.
-sub new {
- my ($class, $io, $cb, $arg, $end_obj) = @_;
- my $self = bless {
- cb => $cb, # initial read callback
- arg => $arg, # arg for $cb
- end_obj => $end_obj, # like END{}, can ->event_step
- }, $class;
- PublicInbox::ProcessIONBF->replace($io);
- $self->SUPER::new($io, EPOLLIN);
-}
-
-sub event_step {
- my ($self) = @_;
- if (defined $self->{cb}) {
- # this may call async_pass when headers are done
- $self->{cb}->($self->{arg});
- } elsif (my $sock = $self->{sock}) {
- # $http may be undef if discarding body output from cgit on 404
- my $http = $self->{http} or return $self->close;
- # $self->{sock} is a read pipe for git-http-backend or cgit
- # and 65536 is the default Linux pipe size
- my $r = sysread($sock, my $buf, 65536);
- if ($r) {
- $self->{ofh}->write($buf); # may call $http->close
- # let other clients get some work done, too
- return if $http->{sock}; # !closed
-
- # else: fall through to close below...
- } elsif (!defined $r && $! == EAGAIN) {
- return; # EPOLLIN means we'll be notified
- }
-
- # Done! Error handling will happen in $self->{ofh}->close
- # called by end_obj->event_step handler
- delete $http->{forward};
- $self->close; # queues end_obj->event_step to be called
- } # else { # we may've been requeued but closed by $http
-}
-
-# once this is called, all data we read is passed to the
-# to the PublicInbox::HTTP instance ($http) via $ofh->write
-# $ofh is typically PublicInbox::HTTP::{Chunked,Identity}, but
-# may be PublicInbox::GzipFilter or $PublicInbox::Qspawn::qx_fh
-sub async_pass {
- my ($self, $http, $ofh, $bref) = @_;
- delete @$self{qw(cb arg)};
- # In case the client HTTP connection ($http) dies, it
- # will automatically close this ($self) object.
- $http->{forward} = $self;
-
- # write anything we overread when we were reading headers.
- # This is typically PublicInbox:HTTP::{chunked,identity}_wcb,
- # but may be PublicInbox::GzipFilter::write. PSGI requires
- # *_wcb methods respond to ->write (and ->close), not ->print
- $ofh->write($$bref);
-
- $self->{http} = $http;
- $self->{ofh} = $ofh;
-}
-
-# may be called as $forward->close in PublicInbox::HTTP or EOF (event_step)
-sub close {
- my $self = $_[0];
- $self->SUPER::close; # DS::close
- delete @$self{qw(cb arg)};
-
- # we defer this to the next timer loop since close is deferred
- if (my $end_obj = delete $self->{end_obj}) {
- # this calls $end_obj->event_step
- # (likely PublicInbox::Qspawn::event_step,
- # NOT PublicInbox::HTTPD::Async::event_step)
- PublicInbox::DS::requeue($end_obj);
- }
-}
-
-1;
start($self, $limiter, undef);
}
-# this is called on pipe EOF to reap the process, may be called
-# via PublicInbox::DS event loop OR via GetlineBody for generic
-# PSGI servers.
-sub event_step {
- my ($self) = @_;
- finish($self);
- my $fh = delete $self->{qfh};
- $fh->close if $fh; # async-only (psgi_return)
-}
-
-sub rd_hdr ($) {
- my ($self) = @_;
- # typically used for reading CGI headers
- # We also need to check EINTR for generic PSGI servers.
- my ($ret, $total_rd);
- my ($bref, $ph_cb, @ph_arg) = ($self->{hdr_buf}, @{$self->{parse_hdr}});
- until (defined($ret)) {
- my $r = sysread($self->{rpipe}, $$bref, 4096, length($$bref));
- if (defined($r)) {
- $total_rd += $r;
- eval { $ret = $ph_cb->($total_rd, $bref, @ph_arg) };
- if ($@) {
- warn "parse_hdr: $@";
- $ret = [ 500, [], [ "Internal error\n" ] ];
- } elsif (!defined($ret) && !$r) {
- warn <<EOM;
-EOF parsing headers from @{$self->{cmd}} ($self->{psgi_env}->{REQUEST_URI})
-EOM
- $ret = [ 500, [], [ "Internal error\n" ] ];
- }
- } else {
- # caller should notify us when it's ready:
- return if $! == EAGAIN;
- next if $! == EINTR; # immediate retry
- warn "error reading header: $!";
- $ret = [ 500, [], [ "Internal error\n" ] ];
- }
- }
- delete $self->{parse_hdr}; # done parsing headers
- $ret;
-}
-
sub yield_pass {
my ($self, $ipipe, $res) = @_; # $ipipe = InputPipe
my $env = $self->{psgi_env};
$self->{qfh} = $qfh; # keep $ipipe open
}
-sub psgi_return_init_cb { # this may be PublicInbox::HTTPD::Async {cb}
- my ($self) = @_;
- my $r = rd_hdr($self) or return; # incomplete
- my $env = $self->{psgi_env};
- my $filter;
-
- # this is for RepoAtom since that can fire after parse_cgi_headers
- if (ref($r) eq 'ARRAY' && blessed($r->[2]) && $r->[2]->can('attach')) {
- $filter = pop @$r;
- }
- $filter //= delete($env->{'qspawn.filter'}) // (ref($r) eq 'ARRAY' ?
- PublicInbox::GzipFilter::qsp_maybe($r->[1], $env) : undef);
-
- my $wcb = delete $env->{'qspawn.wcb'};
- my $async = delete $self->{async}; # PublicInbox::HTTPD::Async
- if (ref($r) ne 'ARRAY' || scalar(@$r) == 3) { # error
- if ($async) { # calls rpipe->close && ->event_step
- $async->close; # PublicInbox::HTTPD::Async::close
- } else { # generic PSGI, use PublicInbox::ProcessIO::CLOSE
- delete($self->{rpipe})->close;
- event_step($self);
- }
- if (ref($r) eq 'ARRAY') { # error
- $wcb->($r)
- } elsif (ref($r) eq 'CODE') { # chain another command
- $r->($wcb);
- $self->{passed} = 1;
- }
- # else do nothing
- } elsif ($async) {
- # done reading headers, handoff to read body
- my $fh = $wcb->($r); # scalar @$r == 2
- $fh = $filter->attach($fh) if $filter;
- $self->{qfh} = $fh;
- $async->async_pass($env->{'psgix.io'}, $fh,
- delete($self->{hdr_buf}));
- } else { # for synchronous PSGI servers
- require PublicInbox::GetlineBody;
- my $buf = delete $self->{hdr_buf};
- $r->[2] = PublicInbox::GetlineBody->new($self->{rpipe},
- \&event_step, $self, $$buf, $filter);
- $wcb->($r);
- }
-}
-
-sub psgi_return_start { # may run later, much later...
- my ($self) = @_;
- if (my $cb = $self->{psgi_env}->{'pi-httpd.async'}) {
- # PublicInbox::HTTPD::Async->new(rpipe, $cb, $cb_arg, $end_obj)
- $self->{async} = $cb->($self->{rpipe},
- \&psgi_return_init_cb, $self, $self);
- } else { # generic PSGI
- psgi_return_init_cb($self) while $self->{parse_hdr};
- }
-}
-
sub r500 () { [ 500, [], [ "Internal error\n" ] ] }
sub parse_hdr_done ($$) {
# $env->{'qspawn.wcb'} - the write callback from the PSGI server
# optional, use this if you've already
# captured it elsewhere. If not given,
-# psgi_return will return an anonymous
+# psgi_yield will return an anonymous
# sub for the PSGI server to call
#
# $env->{'qspawn.filter'} - filter object, responds to ->attach for
# body will be streamed, later, via writes (push-based) to
# psgix.io. 3-element arrays means the body is available
# immediately (or streamed via ->getline (pull-based)).
-sub psgi_return {
- my ($self, $env, $limiter, @parse_hdr_arg)= @_;
- $self->{psgi_env} = $env;
- $self->{hdr_buf} = \(my $hdr_buf = '');
- $self->{parse_hdr} = \@parse_hdr_arg;
- $limiter ||= $def_limiter ||= PublicInbox::Limiter->new(32);
-
- # the caller already captured the PSGI write callback from
- # the PSGI server, so we can call ->start, here:
- $env->{'qspawn.wcb'} and
- return start($self, $limiter, \&psgi_return_start);
-
- # the caller will return this sub to the PSGI server, so
- # it can set the response callback (that is, for
- # PublicInbox::HTTP, the chunked_wcb or identity_wcb callback),
- # but other HTTP servers are supported:
- sub {
- $env->{'qspawn.wcb'} = $_[0];
- start($self, $limiter, \&psgi_return_start);
- }
-}
sub psgi_yield {
my ($self, $env, $limiter, @parse_hdr_arg)= @_;
# called by GzipFilter->close
sub zflush { $_[0]->SUPER::zflush('</feed>') }
-# called by GzipFilter->write or GetlineBody->getline
+# called by GzipFilter->write or GetlineResponse->getline
sub translate {
my $self = shift;
my $rec = $_[0] // return $self->zflush; # getline
# called by GzipFilter->close after translate
sub zflush { $_[0]->SUPER::zflush('</pre>', $_[0]->_html_end) }
-# called by GzipFilter->write or GetlineBody->getline
+# called by GzipFilter->write or GetlineResponse->getline
sub translate {
my $ctx = shift;
my $rec = $_[0] // return zflush($ctx); # getline
my $rdr = { 2 => fileno($null) };
my $cmd = [qw(dd if=/dev/zero count=30 bs=1024k)];
my $qsp = PublicInbox::Qspawn->new($cmd, undef, $rdr);
- return $qsp->psgi_return($env, undef, sub {
+ return $qsp->psgi_yield($env, undef, sub {
my ($r, $bref) = @_;
# make $rd_hdr retry sysread + $parse_hdr in Qspawn:
return until length($$bref) > 8000;
close $null;
[ 200, [ qw(Content-Type application/octet-stream) ]];
});
- } elsif ($path eq '/psgi-return-gzip') {
+ } elsif ($path eq '/psgi-yield-gzip') {
require PublicInbox::Qspawn;
require PublicInbox::GzipFilter;
my $cmd = [qw(echo hello world)];
my $qsp = PublicInbox::Qspawn->new($cmd);
$env->{'qspawn.filter'} = PublicInbox::GzipFilter->new;
- return $qsp->psgi_return($env, undef, sub {
+ return $qsp->psgi_yield($env, undef, sub {
[ 200, [ qw(Content-Type application/octet-stream)]]
});
- } elsif ($path eq '/psgi-return-compressible') {
+ } elsif ($path eq '/psgi-yield-compressible') {
require PublicInbox::Qspawn;
my $cmd = [qw(echo goodbye world)];
my $qsp = PublicInbox::Qspawn->new($cmd);
- return $qsp->psgi_return($env, undef, sub {
+ return $qsp->psgi_yield($env, undef, sub {
[200, [qw(Content-Type text/plain)]]
});
- } elsif ($path eq '/psgi-return-enoent') {
+ } elsif ($path eq '/psgi-yield-enoent') {
require PublicInbox::Qspawn;
my $cmd = [ 'this-better-not-exist-in-PATH'.rand ];
my $qsp = PublicInbox::Qspawn->new($cmd);
- return $qsp->psgi_return($env, undef, sub {
+ return $qsp->psgi_yield($env, undef, sub {
[ 200, [ qw(Content-Type application/octet-stream)]]
});
} elsif ($path eq '/pid') {
is($non_zero, 0, 'read all zeros');
require_mods(@zmods, 4);
- my $buf = xqx([$curl, '-gsS', "$base/psgi-return-gzip"]);
+ my $buf = xqx([$curl, '-gsS', "$base/psgi-yield-gzip"]);
is($?, 0, 'curl succesful');
IO::Uncompress::Gunzip::gunzip(\$buf => \(my $out));
is($out, "hello world\n");
my $curl_rdr = { 2 => \(my $curl_err = '') };
$buf = xqx([$curl, qw(-gsSv --compressed),
- "$base/psgi-return-compressible"], undef, $curl_rdr);
+ "$base/psgi-yield-compressible"], undef, $curl_rdr);
is($?, 0, 'curl --compressed successful');
is($buf, "goodbye world\n", 'gzipped response as expected');
like($curl_err, qr/\bContent-Encoding: gzip\b/,
}
{
- my $conn = conn_for($sock, 'psgi_return ENOENT');
- print $conn "GET /psgi-return-enoent HTTP/1.1\r\n\r\n" or die;
+ my $conn = conn_for($sock, 'psgi_yield ENOENT');
+ print $conn "GET /psgi-yield-enoent HTTP/1.1\r\n\r\n" or die;
my $buf = '';
sysread($conn, $buf, 16384, length($buf)) until $buf =~ /\r\n\r\n/;
like($buf, qr!HTTP/1\.[01] 500\b!, 'got 500 error on ENOENT');
my $app = require $psgi;
test_psgi($app, sub {
my ($cb) = @_;
- my $req = GET('http://example.com/psgi-return-gzip');
+ my $req = GET('http://example.com/psgi-yield-gzip');
my $res = $cb->($req);
my $buf = $res->content;
IO::Uncompress::Gunzip::gunzip(\$buf => \(my $out));
is($out, "hello world\n", 'got expected output');
- $req = GET('http://example.com/psgi-return-enoent');
+ $req = GET('http://example.com/psgi-yield-enoent');
$res = $cb->($req);
is($res->code, 500, 'got error on ENOENT');
seek($tmperr, 0, SEEK_SET) or die;
projects / thirdparty / public-inbox.git / commitdiff
