X-Git-Url: http://www.git.stargrave.org/?a=blobdiff_plain;f=lib%2FPublicInbox%2FGitAsyncCat.pm;h=613dbf7e9966ed153c09144cc0a507ad03070eed;hb=6dec9bf8c0e1b859703d7a5dfb87052cf4e87846;hp=65e161219699e45730e637be17bde7b94a7e9166;hpb=1dacadf3328241e7c4a378c4566c497d1e33326b;p=public-inbox.git diff --git a/lib/PublicInbox/GitAsyncCat.pm b/lib/PublicInbox/GitAsyncCat.pm index 65e16121..613dbf7e 100644 --- a/lib/PublicInbox/GitAsyncCat.pm +++ b/lib/PublicInbox/GitAsyncCat.pm @@ -1,49 +1,98 @@ -# Copyright (C) 2020 all contributors +# Copyright (C) 2020-2021 all contributors # License: AGPL-3.0+ # -# internal class used by PublicInbox::Git + Danga::Socket +# internal class used by PublicInbox::Git + PublicInbox::DS # This parses the output pipe of "git cat-file --batch" -# -# Note: this does NOT set the non-blocking flag, we expect `git cat-file' -# to be a local process, and git won't start writing a blob until it's -# fully read. So minimize context switching and read as much as possible -# and avoid holding a buffer in our heap any longer than it has to live. package PublicInbox::GitAsyncCat; use strict; use parent qw(PublicInbox::DS Exporter); -use fields qw(git); +use POSIX qw(WNOHANG); use PublicInbox::Syscall qw(EPOLLIN EPOLLET); -our @EXPORT = qw(git_async_msg); - -sub new { - my ($class, $git) = @_; - my $self = fields::new($class); - $git->batch_prepare; - $self->SUPER::new($git->{in}, EPOLLIN|EPOLLET); - $self->{git} = $git; - $self; +our @EXPORT = qw(ibx_async_cat ibx_async_prefetch); +use PublicInbox::Git (); + +our $GCF2C; # singleton PublicInbox::Gcf2Client + +sub close { + my ($self) = @_; + if (my $git = delete $self->{git}) { + $git->async_abort; + } + $self->SUPER::close; # PublicInbox::DS::close } sub event_step { my ($self) = @_; - my $git = $self->{git} or return; # ->close-ed + my $git = $self->{git} or return; + return $self->close if ($git->{in} // 0) != ($self->{sock} // 1); my $inflight = $git->{inflight}; - if (@$inflight) { + if ($inflight && @$inflight) { $git->cat_async_step($inflight); - $self->requeue if @$inflight || exists $git->{cat_rbuf}; + + # child death? + if (($git->{in} // 0) != ($self->{sock} // 1)) { + $self->close; + } elsif (@$inflight || exists $git->{rbuf}) { + # ok, more to do, requeue for fairness + $self->requeue; + } + } elsif ((my $pid = waitpid($git->{pid}, WNOHANG)) > 0) { + # May happen if the child process is killed by a BOFH + # (or segfaults) + delete $git->{pid}; + warn "E: git $pid exited with \$?=$?\n"; + $self->close; } } -sub close { - my ($self) = @_; - delete $self->{git}; - $self->SUPER::close; # PublicInbox::DS::close +sub watch_cat { + my ($git) = @_; + $git->{async_cat} //= do { + my $self = bless { git => $git }, __PACKAGE__; + $git->{in}->blocking(0); + $self->SUPER::new($git->{in}, EPOLLIN|EPOLLET); + \undef; # this is a true ref() + }; } -sub git_async_msg ($$$$) { - my ($ibx, $smsg, $cb, $arg) = @_; - $ibx->git->cat_async($smsg->{blob}, $cb, $arg); - $ibx->{async_cat} //= new(__PACKAGE__, $ibx->{git}); +sub ibx_async_cat ($$$$) { + my ($ibx, $oid, $cb, $arg) = @_; + my $git = $ibx->{git} // $ibx->git; + # {topdir} means ExtSearch (likely [extindex "all"]) with potentially + # 100K alternates. git(1) has a proposed patch for 100K alternates: + # + if (!defined($ibx->{topdir}) && !defined($git->{-tmp}) && + ($GCF2C //= eval { + require PublicInbox::Gcf2Client; + PublicInbox::Gcf2Client::new(); + } // 0)) { # 0: do not retry if libgit2 or Inline::C are missing + $GCF2C->gcf2_async(\"$oid $git->{git_dir}\n", $cb, $arg); + \undef; + } else { # read-only end of git-cat-file pipe + $git->cat_async($oid, $cb, $arg); + watch_cat($git); + } +} + +# this is safe to call inside $cb, but not guaranteed to enqueue +# returns true if successful, undef if not. For fairness, we only +# prefetch if there's no in-flight requests. +sub ibx_async_prefetch { + my ($ibx, $oid, $cb, $arg) = @_; + my $git = $ibx->git; + if (!defined($ibx->{topdir}) && $GCF2C) { + if (!@{$GCF2C->{inflight} // []}) { + $oid .= " $git->{git_dir}\n"; + return $GCF2C->gcf2_async(\$oid, $cb, $arg); # true + } + } elsif ($git->{async_cat} && (my $inflight = $git->{inflight})) { + if (!@$inflight) { + print { $git->{out} } $oid, "\n" or + $git->fail("write error: $!"); + return push(@$inflight, $oid, $cb, $arg); + } + } + undef; } 1;