X-Git-Url: http://www.git.stargrave.org/?a=blobdiff_plain;f=lib%2FPublicInbox%2FHTTP.pm;h=19b57d59518d87ec79e0b564bf63a81ed5935c71;hb=9bd675d33ad1e49bd2ebe12a1d216216e61380de;hp=afa71ea53f3cb3ddda35d96a11117b104ead6dc8;hpb=f41dc46f6213661ba51443d6cb0d6a9ba4d41472;p=public-inbox.git diff --git a/lib/PublicInbox/HTTP.pm b/lib/PublicInbox/HTTP.pm index afa71ea5..19b57d59 100644 --- a/lib/PublicInbox/HTTP.pm +++ b/lib/PublicInbox/HTTP.pm @@ -1,4 +1,4 @@ -# Copyright (C) 2016-2018 all contributors +# Copyright (C) 2016-2019 all contributors # License: AGPL-3.0+ # # Generic PSGI server for convenience. It aims to provide @@ -11,7 +11,7 @@ package PublicInbox::HTTP; use strict; use warnings; use base qw(PublicInbox::DS); -use fields qw(httpd env rbuf input_left remote_addr remote_port forward pull); +use fields qw(httpd env input_left remote_addr remote_port forward); use bytes (); # only for bytes::length use Fcntl qw(:seek); use Plack::HTTPParser qw(parse_http_request); # XS or pure Perl @@ -19,19 +19,19 @@ use HTTP::Status qw(status_message); use HTTP::Date qw(time2str); use IO::Handle; require PublicInbox::EvCleanup; -PublicInbox::DS->import(qw(msg_more write_in_full)); +use PublicInbox::DS qw(msg_more); +use PublicInbox::Syscall qw(EPOLLIN EPOLLONESHOT); use constant { CHUNK_START => -1, # [a-f0-9]+\r\n CHUNK_END => -2, # \r\n CHUNK_ZEND => -3, # \r\n CHUNK_MAX_HDR => 256, }; +use Errno qw(EAGAIN); my $pipelineq = []; -my $pipet; sub process_pipelineq () { my $q = $pipelineq; - $pipet = undef; $pipelineq = []; foreach (@$q) { next unless $_->{sock}; @@ -56,9 +56,16 @@ sub http_date () { sub new ($$$) { my ($class, $sock, $addr, $httpd) = @_; my $self = fields::new($class); - $self->SUPER::new($sock, PublicInbox::DS::EPOLLIN()); + my $ev = EPOLLIN; + my $wbuf; + if (ref($sock) eq 'IO::Socket::SSL' && !$sock->accept_SSL) { + return CORE::close($sock) if $! != EAGAIN; + $ev = PublicInbox::TLS::epollbit(); + $wbuf = [ \&PublicInbox::DS::accept_tls_step ]; + } + $self->SUPER::new($sock, $ev | EPOLLONESHOT); $self->{httpd} = $httpd; - $self->{rbuf} = ''; + $self->{wbuf} = $wbuf if $wbuf; ($self->{remote_addr}, $self->{remote_port}) = PublicInbox::Daemon::host_with_port($addr); $self; @@ -73,58 +80,59 @@ sub event_step { # called by PublicInbox::DS # otherwise we can be buffering infinitely w/o backpressure return read_input($self) if defined $self->{env}; - - my $off = length($self->{rbuf}); - my $r = sysread($self->{sock}, $self->{rbuf}, 8192, $off); - if (defined $r) { - return $self->close if $r == 0; - return rbuf_process($self); - } - return if $!{EAGAIN}; # no need to call watch_read(1) again - - # common for clients to break connections without warning, - # would be too noisy to log here: - return $self->close; + my $rbuf = $self->{rbuf} // (\(my $x = '')); + $self->do_read($rbuf, 8192, bytes::length($$rbuf)) or return; + rbuf_process($self, $rbuf); } sub rbuf_process { - my ($self) = @_; + my ($self, $rbuf) = @_; + $rbuf //= $self->{rbuf} // (\(my $x = '')); my %env = %{$self->{httpd}->{env}}; # full hash copy - my $r = parse_http_request($self->{rbuf}, \%env); + 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($self->{rbuf}) > 0x4000)) { + ($r == -2 && bytes::length($$rbuf) > 0x4000)) { return quit($self, 400); } - return $self->watch_read(1) if $r < 0; # incomplete - $self->{rbuf} = substr($self->{rbuf}, $r); - + if ($r < 0) { # incomplete + $self->rbuf_idle($rbuf); + return $self->requeue; + } + $$rbuf = substr($$rbuf, $r); my $len = input_prepare($self, \%env); defined $len or return write_err($self, undef); # EMFILE/ENFILE - $len ? read_input($self) : app_dispatch($self); + $len ? read_input($self, $rbuf) : app_dispatch($self, undef, $rbuf); } -sub read_input ($) { - my ($self) = @_; +# IO::Handle::write returns boolean, this returns bytes written: +sub xwrite ($$$) { + my ($fh, $rbuf, $max) = @_; + my $w = bytes::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 = '')); my $env = $self->{env}; return if $env->{REMOTE_ADDR}; # in app dispatch - return read_input_chunked($self) if env_chunked($env); + return read_input_chunked($self, $rbuf) if env_chunked($env); # env->{CONTENT_LENGTH} (identity) - my $sock = $self->{sock}; - my $len = $self->{input_left}; - $self->{input_left} = undef; - my $rbuf = \($self->{rbuf}); + my $len = delete $self->{input_left}; my $input = $env->{'psgi.input'}; while ($len > 0) { if ($$rbuf ne '') { - my $w = write_in_full($input, $rbuf, $len, 0); + my $w = xwrite($input, $rbuf, $len); return write_err($self, $len) unless $w; $len -= $w; die "BUG: $len < 0 (w=$w)" if $len < 0; @@ -134,16 +142,15 @@ sub read_input ($) { } $$rbuf = ''; } - my $r = sysread($sock, $$rbuf, 8192); - return recv_err($self, $r, $len) unless $r; + $self->do_read($rbuf, 8192) or return recv_err($self, $len); # continue looping if $r > 0; } - app_dispatch($self, $input); + app_dispatch($self, $input, $rbuf); } sub app_dispatch { - my ($self, $input) = @_; - $self->watch_read(0); + my ($self, $input, $rbuf) = @_; + $self->rbuf_idle($rbuf); my $env = $self->{env}; $env->{REMOTE_ADDR} = $self->{remote_addr}; $env->{REMOTE_PORT} = $self->{remote_port}; @@ -235,66 +242,67 @@ sub identity_wcb ($) { sub next_request ($) { my ($self) = @_; - if ($self->{rbuf} eq '') { # wait for next request - $self->watch_read(1); - } else { # avoid recursion for pipelined requests + if ($self->{rbuf}) { + # avoid recursion for pipelined requests + PublicInbox::DS::requeue(\&process_pipelineq) if !@$pipelineq; push @$pipelineq, $self; - $pipet ||= PublicInbox::EvCleanup::asap(*process_pipelineq); + } else { # wait for next request + $self->requeue; } } sub response_done_cb ($$) { my ($self, $alive) = @_; sub { - my $env = $self->{env}; - $self->{env} = undef; + my $env = delete $self->{env}; $self->write(\"0\r\n\r\n") if $alive == 2; - $self->write(sub{$alive ? next_request($self) : $self->close}); + $self->write($alive ? \&next_request : \&close); } } -sub getline_cb ($$$) { +sub getline_response ($$$) { my ($self, $write, $close) = @_; - local $/ = \8192; - my $forward = $self->{forward}; - # limit our own running time for fairness with other - # clients and to avoid buffering too much: - if ($forward) { - my $buf = eval { $forward->getline }; + my $pull; # DANGER: self-referential + $pull = sub { + my $forward = $self->{forward}; + # limit our own running time for fairness with other + # clients and to avoid buffering too much: + my $buf = eval { + local $/ = \8192; + $forward->getline; + } if $forward; + if (defined $buf) { $write->($buf); # may close in PublicInbox::DS::write + if ($self->{sock}) { - my $next = $self->{pull}; - if ($self->{wbuf}) { - $self->write($next); - } else { - PublicInbox::EvCleanup::asap($next); - } - return; + my $wbuf = $self->{wbuf} ||= []; + push @$wbuf, $pull; + + # wbuf may be populated by $write->($buf), + # no need to rearm if so: + $self->requeue if scalar(@$wbuf) == 1; + return; # likely } } elsif ($@) { err($self, "response ->getline error: $@"); - $forward = undef; $self->close; } - } - $self->{forward} = $self->{pull} = undef; - # avoid recursion - if ($forward) { - eval { $forward->close }; - if ($@) { - err($self, "response ->close error: $@"); - $self->close; # idempotent + $pull = undef; # all done! + # avoid recursion + if (delete $self->{forward}) { + eval { $forward->close }; + if ($@) { + err($self, "response ->close error: $@"); + $self->close; # idempotent + } } - } - $close->(); -} + $forward = undef; + $close->(); # call response_done_cb + }; -sub getline_response ($$$) { - my ($self, $write, $close) = @_; - my $pull = $self->{pull} = sub { getline_cb($self, $write, $close) }; - $pull->(); + $pull->(); # kick-off! } sub response_write { @@ -316,6 +324,11 @@ sub response_write { } } +sub input_tmpfile ($) { + open($_[0], '+>', undef); + $_[0]->autoflush(1); +} + sub input_prepare { my ($self, $env) = @_; my $input; @@ -325,10 +338,10 @@ sub input_prepare { quit($self, 413); return; } - open($input, '+>', undef); + input_tmpfile($input); } elsif (env_chunked($env)) { $len = CHUNK_START; - open($input, '+>', undef); + input_tmpfile($input); } else { $input = $null_io; } @@ -356,34 +369,31 @@ sub write_err { } sub recv_err { - my ($self, $r, $len) = @_; - return $self->close if (defined $r && $r == 0); - if ($!{EAGAIN}) { + my ($self, $len) = @_; + if ($! == EAGAIN) { # epoll/kevent watch already set by do_read $self->{input_left} = $len; - return; + } else { + err($self, "error reading input: $! ($len bytes remaining)"); } - err($self, "error reading for input: $! ($len bytes remaining)"); - quit($self, 500); } sub read_input_chunked { # unlikely... - my ($self) = @_; + my ($self, $rbuf) = @_; + $rbuf //= $self->{rbuf} // (\(my $x = '')); my $input = $self->{env}->{'psgi.input'}; - my $sock = $self->{sock}; - my $len = $self->{input_left}; - $self->{input_left} = undef; - my $rbuf = \($self->{rbuf}); + my $len = delete $self->{input_left}; while (1) { # chunk start if ($len == CHUNK_ZEND) { $$rbuf =~ s/\A\r\n//s and - return app_dispatch($self, $input); - return quit($self, 400) if length($$rbuf) > 2; + return app_dispatch($self, $input, $rbuf); + + return quit($self, 400) if bytes::length($$rbuf) > 2; } if ($len == CHUNK_END) { if ($$rbuf =~ s/\A\r\n//s) { $len = CHUNK_START; - } elsif (length($$rbuf) > 2) { + } elsif (bytes::length($$rbuf) > 2) { return quit($self, 400); } } @@ -393,16 +403,15 @@ sub read_input_chunked { # unlikely... if (($len + -s $input) > $MAX_REQUEST_BUFFER) { return quit($self, 413); } - } elsif (length($$rbuf) > CHUNK_MAX_HDR) { + } elsif (bytes::length($$rbuf) > CHUNK_MAX_HDR) { return quit($self, 400); } # will break from loop since $len >= 0 } if ($len < 0) { # chunk header is trickled, read more - my $off = length($$rbuf); - my $r = sysread($sock, $$rbuf, 8192, $off); - return recv_err($self, $r, $len) unless $r; + $self->do_read($rbuf, 8192, bytes::length($$rbuf)) or + return recv_err($self, $len); # (implicit) goto chunk_start if $r > 0; } $len = CHUNK_ZEND if $len == 0; @@ -410,7 +419,7 @@ sub read_input_chunked { # unlikely... # drain the current chunk until ($len <= 0) { if ($$rbuf ne '') { - my $w = write_in_full($input, $rbuf, $len, 0); + my $w = xwrite($input, $rbuf, $len); return write_err($self, "$len chunk") if !$w; $len -= $w; if ($len == 0) { @@ -426,8 +435,8 @@ sub read_input_chunked { # unlikely... } if ($$rbuf eq '') { # read more of current chunk - my $r = sysread($sock, $$rbuf, 8192); - return recv_err($self, $r, $len) unless $r; + $self->do_read($rbuf, 8192) or + return recv_err($self, $len); } } } @@ -441,22 +450,21 @@ sub quit { } sub close { - my $self = shift; - my $forward = $self->{forward}; - my $env = $self->{env}; - delete $env->{'psgix.io'} if $env; # prevent circular references - $self->{pull} = $self->{forward} = $self->{env} = undef; - if ($forward) { + my $self = $_[0]; + if (my $env = delete $self->{env}) { + delete $env->{'psgix.io'}; # prevent circular references + } + if (my $forward = delete $self->{forward}) { eval { $forward->close }; err($self, "forward ->close error: $@") if $@; } - $self->SUPER::close(@_); + $self->SUPER::close; # PublicInbox::DS::close } # for graceful shutdown in PublicInbox::Daemon: sub busy () { my ($self) = @_; - ($self->{rbuf} ne '' || $self->{env} || $self->{wbuf}); + ($self->{rbuf} || $self->{env} || $self->{wbuf}); } 1;