X-Git-Url: http://www.git.stargrave.org/?a=blobdiff_plain;f=lib%2FPublicInbox%2FHTTPD%2FAsync.pm;h=7bbab1e1cd65c22d2a895f5a99b5924fa172afd5;hb=a12cd80b5821cf5cb2c8672aad0fba94c451f728;hp=b00e63701495a73635874e702e16addb62985d6b;hpb=395406118cb22b4cfa3a83c2b3a3a9aa2627b708;p=public-inbox.git diff --git a/lib/PublicInbox/HTTPD/Async.pm b/lib/PublicInbox/HTTPD/Async.pm index b00e6370..7bbab1e1 100644 --- a/lib/PublicInbox/HTTPD/Async.pm +++ b/lib/PublicInbox/HTTPD/Async.pm @@ -1,73 +1,109 @@ -# Copyright (C) 2016 all contributors +# Copyright (C) all contributors # License: AGPL-3.0+ # # 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 is is intended for use with read-only pipes. +# 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 strict; -use warnings; -use base qw(Danga::Socket); -use fields qw(cb cleanup); -use Scalar::Util qw(weaken); -require PublicInbox::EvCleanup; +use v5.12; +use parent qw(PublicInbox::DS); +use Errno qw(EAGAIN); +use PublicInbox::Syscall qw(EPOLLIN); +# 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, $cleanup) = @_; - my $self = fields::new($class); - IO::Handle::blocking($io, 0); - $self->SUPER::new($io); - $self->{cb} = $cb; - $self->{cleanup} = $cleanup; - $self->watch_read(1); - $self; + my ($class, $io, $cb, $arg, $end_obj) = @_; + + # no $io? call $cb at the top of the next event loop to + # avoid recursion: + unless (defined($io)) { + PublicInbox::DS::requeue($cb ? $cb : $arg); + die '$end_obj unsupported w/o $io' if $end_obj; + return; + } + my $self = bless { + cb => $cb, # initial read callback + arg => $arg, # arg for $cb + end_obj => $end_obj, # like END{}, can ->event_step + }, $class; + my $pp = tied *$io; # ProcessPipe + $pp->{fh}->blocking(0) // die "$io->blocking(0): $!"; + $self->SUPER::new($io, EPOLLIN); } -sub async_pass { - my ($self, $io, $fh, $bref) = @_; - # In case the client HTTP connection ($io) dies, it - # will automatically close this ($self) object. - $io->{forward} = $self; - $fh->write($$bref); - weaken($self); - my $restart_read = sub { $self->watch_read(1) }; - $self->{cb} = sub { - my $r = sysread($self->{sock}, $$bref, 8192); +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) { - $fh->write($$bref); - if ($io->{write_buf_size}) { - $self->watch_read(0); - $io->write($restart_read); # D::S::write - } - return; # stay in watch_read - } elsif (!defined $r) { - return if $!{EAGAIN} || $!{EINTR}; + $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 $fh->close - # called by the {cleanup} handler - $io->{forward} = undef; - $self->close; - } + # 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 } -sub event_read { $_[0]->{cb}->() } -sub event_hup { $_[0]->{cb}->() } -sub event_err { $_[0]->{cb}->() } -sub sysread { shift->{sock}->sysread(@_) } +# 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 = shift; - my $cleanup = $self->{cleanup}; - $self->{cleanup} = $self->{cb} = undef; - $self->SUPER::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 - PublicInbox::EvCleanup::asap($cleanup) if $cleanup; + 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); + } } -# do not let ourselves be closed during graceful termination -sub busy () { $_[0]->{cb} } - 1;