diff options
Diffstat (limited to 'lib/PublicInbox/HTTP.pm')
| -rw-r--r-- | lib/PublicInbox/HTTP.pm | 166 |
1 files changed, 62 insertions, 104 deletions
diff --git a/lib/PublicInbox/HTTP.pm b/lib/PublicInbox/HTTP.pm index b2c74cf3..7162732e 100644 --- a/lib/PublicInbox/HTTP.pm +++ b/lib/PublicInbox/HTTP.pm @@ -1,4 +1,4 @@ -# Copyright (C) 2016-2021 all contributors <meta@public-inbox.org> +# Copyright (C) all contributors <meta@public-inbox.org> # License: AGPL-3.0+ <https://www.gnu.org/licenses/agpl-3.0.txt> # # Generic PSGI server for convenience. It aims to provide @@ -26,7 +26,6 @@ use Plack::HTTPParser qw(parse_http_request); # XS or pure Perl use Plack::Util; use HTTP::Status qw(status_message); use HTTP::Date qw(time2str); -use IO::Handle; # ->write use PublicInbox::DS qw(msg_more); use PublicInbox::Syscall qw(EPOLLIN EPOLLONESHOT); use PublicInbox::Tmpfile; @@ -38,23 +37,19 @@ use constant { }; use Errno qw(EAGAIN); -my $pipelineq = []; -sub process_pipelineq () { - my $q = $pipelineq; - $pipelineq = []; - foreach (@$q) { - next unless $_->{sock}; - rbuf_process($_); - } -} - # Use the same configuration parameter as git since this is primarily # a slow-client sponge for git-http-backend # TODO: support per-respository http.maxRequestBuffer somehow... our $MAX_REQUEST_BUFFER = $ENV{GIT_HTTP_MAX_REQUEST_BUFFER} || (10 * 1024 * 1024); -open(my $null_io, '<', '/dev/null') or die "failed to open /dev/null: $!"; +open(my $null_io, '<', '/dev/null') or die "open /dev/null: $!"; +{ + my @n = stat($null_io) or die "stat(/dev/null): $!"; + my @i = stat(STDIN) or die "stat(STDIN): $!"; + $null_io = *STDIN{IO} if "@n[0, 1]" eq "@i[0, 1]"; +} + my $http_date; my $prev = 0; sub http_date () { @@ -63,13 +58,13 @@ sub http_date () { } sub new ($$$) { - my ($class, $sock, $addr, $httpd) = @_; - my $self = bless { httpd => $httpd }, $class; + my ($class, $sock, $addr, $srv_env) = @_; + my $self = bless { srv_env => $srv_env }, $class; my $ev = EPOLLIN; my $wbuf; if ($sock->can('accept_SSL') && !$sock->accept_SSL) { - return CORE::close($sock) if $! != EAGAIN; - $ev = PublicInbox::TLS::epollbit() or return CORE::close($sock); + return $sock->close if $! != EAGAIN; + $ev = PublicInbox::TLS::epollbit() or return $sock->close; $wbuf = [ \&PublicInbox::DS::accept_tls_step ]; } $self->{wbuf} = $wbuf if $wbuf; @@ -80,52 +75,36 @@ sub new ($$$) { sub event_step { # called by PublicInbox::DS my ($self) = @_; - - return unless $self->flush_write && $self->{sock}; + local $SIG{__WARN__} = $self->{srv_env}->{'pi-httpd.warn_cb'}; + return unless $self->flush_write && $self->{sock} && !$self->{forward}; # only read more requests if we've drained the write buffer, # otherwise we can be buffering infinitely w/o backpressure return read_input($self) if ref($self->{env}); - my $rbuf = $self->{rbuf} // (\(my $x = '')); - $self->do_read($rbuf, 8192, length($$rbuf)) or return; - rbuf_process($self, $rbuf); -} -sub rbuf_process { - my ($self, $rbuf) = @_; - $rbuf //= $self->{rbuf} // (\(my $x = '')); - - my %env = %{$self->{httpd}->{env}}; # full hash copy - my $r = parse_http_request($$rbuf, \%env); - - # We do not support Trailers in chunked requests, for now - # (they are rarely-used and git (as of 2.7.2) does not use them) - if ($r == -1 || $env{HTTP_TRAILER} || - # this length-check is necessary for PURE_PERL=1: - ($r == -2 && length($$rbuf) > 0x4000)) { - return quit($self, 400); - } - if ($r < 0) { # incomplete - $self->rbuf_idle($rbuf); - return $self->requeue; + my $rbuf = $self->{rbuf} // (\(my $x = '')); + my %env = %{$self->{srv_env}}; # full hash copy + my $r; + while (($r = parse_http_request($$rbuf, \%env)) < 0) { + # We do not support Trailers in chunked requests, for + # now (they are rarely-used and git (as of 2.7.2) does + # not use them). + # this length-check is necessary for PURE_PERL=1: + if ($r == -1 || $env{HTTP_TRAILER} || + ($r == -2 && length($$rbuf) > 0x4000)) { + return quit($self, 400); + } + $self->do_read($rbuf, 8192, length($$rbuf)) or return; } + return quit($self, 400) if grep(/\s/, keys %env); # stop smugglers $$rbuf = substr($$rbuf, $r); - my $len = input_prepare($self, \%env); - defined $len or return write_err($self, undef); # EMFILE/ENFILE + my $len = input_prepare($self, \%env) // + return write_err($self, undef); # EMFILE/ENFILE $len ? read_input($self, $rbuf) : app_dispatch($self, undef, $rbuf); } -# IO::Handle::write returns boolean, this returns bytes written: -sub xwrite ($$$) { - my ($fh, $rbuf, $max) = @_; - my $w = length($$rbuf); - $w = $max if $w > $max; - $fh->write($$rbuf, $w) or return; - $w; -} - sub read_input ($;$) { my ($self, $rbuf) = @_; $rbuf //= $self->{rbuf} // (\(my $x = '')); @@ -138,7 +117,7 @@ sub read_input ($;$) { while ($len > 0) { if ($$rbuf ne '') { - my $w = xwrite($input, $rbuf, $len); + my $w = syswrite($input, $$rbuf, $len); return write_err($self, $len) unless $w; $len -= $w; die "BUG: $len < 0 (w=$w)" if $len < 0; @@ -162,7 +141,7 @@ sub app_dispatch { $env->{REMOTE_ADDR} = $self->{remote_addr}; $env->{REMOTE_PORT} = $self->{remote_port}; if (defined(my $host = $env->{HTTP_HOST})) { - $host =~ s/:([0-9]+)\z// and $env->{SERVER_PORT} = $1; + $host =~ s/:([0-9]+)\z// and $env->{SERVER_PORT} = $1 + 0; $env->{SERVER_NAME} = $host; } if (defined $input) { @@ -172,7 +151,7 @@ sub app_dispatch { # note: NOT $self->{sock}, we want our close (+ PublicInbox::DS::close), # to do proper cleanup: $env->{'psgix.io'} = $self; # for ->close or async_pass - my $res = Plack::Util::run_app($self->{httpd}->{app}, $env); + my $res = Plack::Util::run_app($env->{'pi-httpd.app'}, $env); eval { if (ref($res) eq 'CODE') { $res->(sub { response_write($self, $env, $_[0]) }); @@ -181,7 +160,7 @@ sub app_dispatch { } }; if ($@) { - err($self, "response_write error: $@"); + warn "response_write error: $@"; $self->close; } } @@ -212,6 +191,7 @@ sub response_header_write { my $alive; if (!$term && $prot_persist) { # auto-chunk $chunked = $alive = 2; + $alive = 3 if $env->{REQUEST_METHOD} eq 'HEAD'; $h .= "Transfer-Encoding: chunked\r\n"; # no need for "Connection: keep-alive" with HTTP/1.1 } elsif ($term && ($prot_persist || ($conn =~ /\bkeep-alive\b/i))) { @@ -248,22 +228,19 @@ sub identity_write ($$) { $self->write(\($_[1])) if $_[1] ne ''; } -sub next_request ($) { - my ($self) = @_; - if ($self->{rbuf}) { - # avoid recursion for pipelined requests - PublicInbox::DS::requeue(\&process_pipelineq) if !@$pipelineq; - push @$pipelineq, $self; - } else { # wait for next request - $self->requeue; - } -} - sub response_done { my ($self, $alive) = @_; + if (my $forward = delete $self->{forward}) { # avoid recursion + eval { $forward->close }; + if ($@) { + warn "response forward->close error: $@"; + return $self->close; # idempotent + } + } delete $self->{env}; # we're no longer busy + # HEAD requests set $alive = 3 so we don't send "0\r\n\r\n"; $self->write(\"0\r\n\r\n") if $alive == 2; - $self->write($alive ? \&next_request : \&close); + $self->write($alive ? $self->can('requeue') : \&close); } sub getline_pull { @@ -273,7 +250,7 @@ sub getline_pull { # limit our own running time for fairness with other # clients and to avoid buffering too much: my $buf = eval { - local $/ = \8192; + local $/ = \65536; $forward->getline; } if $forward; @@ -295,17 +272,9 @@ sub getline_pull { return; # likely } } elsif ($@) { - err($self, "response ->getline error: $@"); + warn "response ->getline error: $@"; $self->close; } - # avoid recursion - if (delete $self->{forward}) { - eval { $forward->close }; - if ($@) { - err($self, "response ->close error: $@"); - $self->close; # idempotent - } - } response_done($self, delete $self->{alive}); } @@ -326,19 +295,13 @@ sub response_write { getline_pull($self); # kick-off! } # these are returned to the calling application: - } elsif ($alive == 2) { + } elsif ($alive >= 2) { bless [ $self, $alive ], 'PublicInbox::HTTP::Chunked'; } else { bless [ $self, $alive ], 'PublicInbox::HTTP::Identity'; } } -sub input_tmpfile ($) { - my $input = tmpfile('http.input', $_[0]->{sock}) or return; - $input->autoflush(1); - $input; -} - sub input_prepare { my ($self, $env) = @_; my ($input, $len); @@ -354,39 +317,33 @@ sub input_prepare { return quit($self, 400) if $hte !~ /\Achunked\z/i; $len = CHUNK_START; - $input = input_tmpfile($self); + $input = tmpfile('http.input', $self->{sock}); } else { $len = $env->{CONTENT_LENGTH}; if (defined $len) { # rfc7230 3.3.3.4 return quit($self, 400) if $len !~ /\A[0-9]+\z/; - return quit($self, 413) if $len > $MAX_REQUEST_BUFFER; - $input = $len ? input_tmpfile($self) : $null_io; + $input = $len ? tmpfile('http.input', $self->{sock}) + : $null_io; } else { $input = $null_io; } } # TODO: expire idle clients on ENFILE / EMFILE - return unless $input; - - $env->{'psgi.input'} = $input; + $env->{'psgi.input'} = $input // return; $self->{env} = $env; $self->{input_left} = $len || 0; } sub env_chunked { ($_[0]->{HTTP_TRANSFER_ENCODING} // '') =~ /\Achunked\z/i } -sub err ($$) { - eval { $_[0]->{httpd}->{env}->{'psgi.errors'}->print($_[1]."\n") }; -} - sub write_err { my ($self, $len) = @_; my $msg = $! || '(zero write)'; $msg .= " ($len bytes remaining)" if defined $len; - err($self, "error buffering to input: $msg"); + warn "error buffering to input: $msg"; quit($self, 500); } @@ -395,7 +352,7 @@ sub recv_err { if ($! == EAGAIN) { # epoll/kevent watch already set by do_read $self->{input_left} = $len; } else { - err($self, "error reading input: $! ($len bytes remaining)"); + warn "error reading input: $! ($len bytes remaining)"; } } @@ -441,7 +398,7 @@ sub read_input_chunked { # unlikely... # drain the current chunk until ($len <= 0) { if ($$rbuf ne '') { - my $w = xwrite($input, $rbuf, $len); + my $w = syswrite($input, $$rbuf, $len); return write_err($self, "$len chunk") if !$w; $len -= $w; if ($len == 0) { @@ -476,15 +433,14 @@ sub close { my $self = $_[0]; if (my $forward = delete $self->{forward}) { eval { $forward->close }; - err($self, "forward ->close error: $@") if $@; + warn "forward ->close error: $@" if $@; } $self->SUPER::close; # PublicInbox::DS::close } -# for graceful shutdown in PublicInbox::Daemon: -sub busy () { +sub busy { # for graceful shutdown in PublicInbox::Daemon: my ($self) = @_; - ($self->{rbuf} || exists($self->{env}) || $self->{wbuf}); + defined($self->{rbuf}) || exists($self->{env}) || defined($self->{wbuf}) } # runs $cb on the next iteration of the event loop at earliest @@ -498,11 +454,12 @@ sub next_step { # They may be exposed to the PSGI application when the PSGI app # returns a CODE ref for "push"-based responses package PublicInbox::HTTP::Chunked; -use strict; +use v5.12; sub write { # ([$http], $buf) = @_; - PublicInbox::HTTP::chunked_write($_[0]->[0], $_[1]) + PublicInbox::HTTP::chunked_write($_[0]->[0], $_[1]); + $_[0]->[0]->{sock} ? length($_[1]) : undef; } sub close { @@ -511,12 +468,13 @@ sub close { } package PublicInbox::HTTP::Identity; -use strict; +use v5.12; our @ISA = qw(PublicInbox::HTTP::Chunked); sub write { # ([$http], $buf) = @_; PublicInbox::HTTP::identity_write($_[0]->[0], $_[1]); + $_[0]->[0]->{sock} ? length($_[1]) : undef; } 1; |