X-Git-Url: http://www.git.stargrave.org/?a=blobdiff_plain;f=lib%2FPublicInbox%2FHTTPD%2FAsync.pm;h=9e592f47412bf4f7aa1c32555196ee92b13a06de;hb=9ea503ef65426070303fe1929f456b3591d74d93;hp=35d171506cd52b3253456aca2728d4a68685b30b;hpb=7c83d3e706811095cedab0bf62ac530d7b0f3a5a;p=public-inbox.git diff --git a/lib/PublicInbox/HTTPD/Async.pm b/lib/PublicInbox/HTTPD/Async.pm index 35d17150..9e592f47 100644 --- a/lib/PublicInbox/HTTPD/Async.pm +++ b/lib/PublicInbox/HTTPD/Async.pm @@ -1,93 +1,108 @@ -# Copyright (C) 2016-2018 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(PublicInbox::DS); -use fields qw(cb cleanup); -require PublicInbox::EvCleanup; +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 ($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); - die 'cleanup unsupported w/o $io' if $cleanup; + PublicInbox::DS::requeue($cb ? $cb : $arg); + die '$end_obj unsupported w/o $io' if $end_obj; return; } - - my $self = fields::new($class); - IO::Handle::blocking($io, 0); - $self->SUPER::new($io, PublicInbox::DS::EPOLLIN()); - $self->{cb} = $cb; - $self->{cleanup} = $cleanup; - $self; + 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; + $pp->{fh}->blocking(0) // die "$io->blocking(0): $!"; + $self->SUPER::new($io, EPOLLIN); } -sub restart_read ($) { $_[0]->watch(PublicInbox::DS::EPOLLIN()) } - -sub main_cb ($$$) { - my ($http, $fh, $bref) = @_; - sub { - my ($self) = @_; - my $r = sysread($self->{sock}, $$bref, 8192); +sub event_step { + my ($self) = @_; + if (my $cb = delete $self->{cb}) { + # this may call async_pass when headers are done + $cb->(my $refcnt_guard = delete $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); # may call $http->close + $self->{fh}->write($buf); # may call $http->close + # let other clients get some work done, too + return if $http->{sock}; # !closed - if ($http->{sock}) { # !closed - if ($http->{wbuf}) { - # HTTP client could not keep up, so - # stop reading and buffering. - $self->watch(0); - - # Tell the HTTP socket to restart us - # when HTTP client is done draining - # $http->{wbuf}: - $http->enqueue_restart_pass; - } - # stay in EPOLLIN, but let other clients - # get some work done, too. - return; - } - # fall through to close below... - } elsif (!defined $r) { - return restart_read($self) if $! == EAGAIN; + # 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 + # Done! Error handling will happen in $self->{fh}->close + # called by end_obj->event_step handler delete $http->{forward}; - $self->close; - } + $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 $fh->write sub async_pass { my ($self, $http, $fh, $bref) = @_; # In case the client HTTP connection ($http) dies, it # will automatically close this ($self) object. $http->{forward} = $self; - $fh->write($$bref); # PublicInbox:HTTP::{chunked,identity}_wcb - $self->{cb} = main_cb($http, $fh, $bref); -} -sub event_step { $_[0]->{cb}->(@_) } + # 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 + $fh->write($$bref); + + # we're done with this, free this memory up ASAP since the + # calls after this may use much memory: + $$bref = undef; + + $self->{http} = $http; + $self->{fh} = $fh; +} +# may be called as $forward->close in PublicInbox::HTTP or EOF (event_step) sub close { my $self = $_[0]; - delete $self->{cb}; - $self->SUPER::close; + $self->SUPER::close; # DS::close # we defer this to the next timer loop since close is deferred - if (my $cleanup = delete $self->{cleanup}) { - PublicInbox::DS::requeue($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); } }