use HTTP::Date qw(time2str);
use IO::Handle;
require PublicInbox::EvCleanup;
+PublicInbox::DS->import(qw(msg_more write_in_full));
+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;
$pipet = undef;
$pipelineq = [];
foreach (@$q) {
- next if $_->{closed};
+ next unless $_->{sock};
rbuf_process($_);
}
}
sub new ($$$) {
my ($class, $sock, $addr, $httpd) = @_;
my $self = fields::new($class);
- $self->SUPER::new($sock);
+ $self->SUPER::new($sock, EPOLLIN | EPOLLONESHOT);
$self->{httpd} = $httpd;
$self->{rbuf} = '';
($self->{remote_addr}, $self->{remote_port}) =
PublicInbox::Daemon::host_with_port($addr);
- $self->watch_read(1);
$self;
}
-sub event_read { # called by PublicInbox::DS
+sub event_step { # called by PublicInbox::DS
my ($self) = @_;
- return event_read_input($self) if defined $self->{env};
+ return unless $self->flush_write && $self->{sock};
+
+ # 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 defined $self->{env};
my $off = length($self->{rbuf});
my $r = sysread($self->{sock}, $self->{rbuf}, 8192, $off);
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;
+ $! == EAGAIN ? $self->watch_in1 : $self->close;
}
sub rbuf_process {
($r == -2 && length($self->{rbuf}) > 0x4000)) {
return quit($self, 400);
}
- return $self->watch_read(1) if $r < 0; # incomplete
+ return $self->watch_in1 if $r < 0; # incomplete
$self->{rbuf} = substr($self->{rbuf}, $r);
my $len = input_prepare($self, \%env);
defined $len or return write_err($self, undef); # EMFILE/ENFILE
- $len ? event_read_input($self) : app_dispatch($self);
+ $len ? read_input($self) : app_dispatch($self);
}
-sub event_read_input ($) {
+sub read_input ($) {
my ($self) = @_;
my $env = $self->{env};
- return event_read_input_chunked($self) if env_chunked($env);
+ return if $env->{REMOTE_ADDR}; # in app dispatch
+ return read_input_chunked($self) if env_chunked($env);
# env->{CONTENT_LENGTH} (identity)
my $sock = $self->{sock};
- my $len = $self->{input_left};
- $self->{input_left} = undef;
+ my $len = delete $self->{input_left};
my $rbuf = \($self->{rbuf});
my $input = $env->{'psgi.input'};
while ($len > 0) {
if ($$rbuf ne '') {
- my $w = write_in_full($input, $rbuf, $len);
+ my $w = write_in_full($input, $rbuf, $len, 0);
return write_err($self, $len) unless $w;
$len -= $w;
die "BUG: $len < 0 (w=$w)" if $len < 0;
sub app_dispatch {
my ($self, $input) = @_;
- $self->watch_read(0);
my $env = $self->{env};
$env->{REMOTE_ADDR} = $self->{remote_addr};
$env->{REMOTE_PORT} = $self->{remote_port};
$h .= 'Date: ' . http_date() . "\r\n\r\n";
if (($len || $chunked) && $env->{REQUEST_METHOD} ne 'HEAD') {
- more($self, $h);
+ msg_more($self, $h);
} else {
- $self->write($h);
+ $self->write(\$h);
}
$alive;
}
my ($self) = @_;
sub {
return if $_[0] eq '';
- more($self, sprintf("%x\r\n", bytes::length($_[0])));
- more($self, $_[0]);
+ msg_more($self, sprintf("%x\r\n", bytes::length($_[0])));
+ msg_more($self, $_[0]);
- # use $self->write("\n\n") if you care about real-time
+ # use $self->write(\"\n\n") if you care about real-time
# streaming responses, public-inbox WWW does not.
- more($self, "\r\n");
+ msg_more($self, "\r\n");
}
}
sub next_request ($) {
my ($self) = @_;
- $self->watch_write(0);
if ($self->{rbuf} eq '') { # wait for next request
- $self->watch_read(1);
+ $self->watch_in1;
} else { # avoid recursion for pipelined requests
push @$pipelineq, $self;
$pipet ||= PublicInbox::EvCleanup::asap(*process_pipelineq);
sub response_done_cb ($$) {
my ($self, $alive) = @_;
sub {
- my $env = $self->{env};
- $self->{env} = undef;
- $self->write("0\r\n\r\n") if $alive == 2;
- $self->write(sub{$alive ? next_request($self) : $self->close});
+ my $env = delete $self->{env};
+ $self->write(\"0\r\n\r\n") if $alive == 2;
+ $self->write($alive ? \&next_request : \&close);
}
}
my $buf = eval { $forward->getline };
if (defined $buf) {
$write->($buf); # may close in PublicInbox::DS::write
- unless ($self->{closed}) {
+ if ($self->{sock}) {
my $next = $self->{pull};
- if ($self->{write_buf_size}) {
+ if ($self->{wbuf}) {
$self->write($next);
} else {
PublicInbox::EvCleanup::asap($next);
}
}
- $self->{forward} = $self->{pull} = undef;
+ delete @$self{qw(forward pull)};
# avoid recursion
if ($forward) {
eval { $forward->close };
}
}
-use constant MSG_MORE => ($^O eq 'linux') ? 0x8000 : 0;
-sub more ($$) {
- my $self = $_[0];
- return if $self->{closed};
- if (MSG_MORE && !$self->{write_buf_size}) {
- my $n = send($self->{sock}, $_[1], MSG_MORE);
- if (defined $n) {
- my $nlen = length($_[1]) - $n;
- return 1 if $nlen == 0; # all done!
-
- # PublicInbox::DS::write queues the unwritten substring:
- return $self->write(substr($_[1], $n, $nlen));
- }
- }
- $self->write($_[1]);
-}
-
sub input_prepare {
my ($self, $env) = @_;
my $input;
sub recv_err {
my ($self, $r, $len) = @_;
return $self->close if (defined $r && $r == 0);
- if ($!{EAGAIN}) {
+ if ($! == EAGAIN) {
$self->{input_left} = $len;
- return;
+ return $self->watch_in1;
}
err($self, "error reading for input: $! ($len bytes remaining)");
quit($self, 500);
}
-sub write_in_full {
- my ($fh, $rbuf, $len) = @_;
- my $rv = 0;
- my $off = 0;
- while ($len > 0) {
- my $w = syswrite($fh, $$rbuf, $len, $off);
- return ($rv ? $rv : $w) unless $w; # undef or 0
- $rv += $w;
- $off += $w;
- $len -= $w;
- }
- $rv
-}
-
-sub event_read_input_chunked { # unlikely...
+sub read_input_chunked { # unlikely...
my ($self) = @_;
my $input = $self->{env}->{'psgi.input'};
my $sock = $self->{sock};
- my $len = $self->{input_left};
- $self->{input_left} = undef;
+ my $len = delete $self->{input_left};
my $rbuf = \($self->{rbuf});
while (1) { # chunk start
# drain the current chunk
until ($len <= 0) {
if ($$rbuf ne '') {
- my $w = write_in_full($input, $rbuf, $len);
+ my $w = write_in_full($input, $rbuf, $len, 0);
return write_err($self, "$len chunk") if !$w;
$len -= $w;
if ($len == 0) {
sub quit {
my ($self, $status) = @_;
my $h = "HTTP/1.1 $status " . status_message($status) . "\r\n\r\n";
- $self->write($h);
+ $self->write(\$h);
$self->close;
}
-# callbacks for PublicInbox::DS
-
-sub event_hup { $_[0]->close }
-sub event_err { $_[0]->close }
-
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
+ }
+ delete $self->{pull};
+ 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->{write_buf_size});
+ ($self->{rbuf} ne '' || $self->{env} || $self->{wbuf});
}
+# fires after pending writes are complete:
+sub restart_pass ($) {
+ $_[0]->{forward}->restart_read; # see PublicInbox::HTTPD::Async
+}
+
+sub enqueue_restart_pass ($) { $_[0]->write(\&restart_pass) }
+
1;