From b2c5c71a014b6acc897b12c75ae9fc13476a5b3f Mon Sep 17 00:00:00 2001
From: Eric Wong <e@80x24.org>
Date: Fri, 8 Oct 2021 10:20:03 +0000
Subject: [PATCH] git: async_abort includes --batch-check requests
We need to abort both check-only and cat requests when
aborting, since we'll be aborting more aggressively
in in read-write paths.
---
lib/PublicInbox/Gcf2Client.pm | 4 ++--
lib/PublicInbox/Git.pm | 40 +++++++++++++++++++---------------
lib/PublicInbox/GitAsyncCat.pm | 4 ++--
3 files changed, 26 insertions(+), 22 deletions(-)
diff --git a/lib/PublicInbox/Gcf2Client.pm b/lib/PublicInbox/Gcf2Client.pm
index c5695db1..09c3aa06 100644
--- a/lib/PublicInbox/Gcf2Client.pm
+++ b/lib/PublicInbox/Gcf2Client.pm
@@ -18,7 +18,7 @@ use PublicInbox::Syscall qw(EPOLLIN EPOLLET);
# pid.owner => process which spawned {pid}
# in => same as {sock}, for compatibility with PublicInbox::Git
# inflight => array (see PublicInbox::Git)
-# cat_rbuf => scalarref, may be non-existent or empty
+# rbuf => scalarref, may be non-existent or empty
sub new {
my ($rdr) = @_;
my $self = bless {}, __PACKAGE__;
@@ -68,7 +68,7 @@ sub event_step {
return $self->close unless $self->{in}; # process died
# ok, more to do, requeue for fairness
- $self->requeue if @$inflight || exists($self->{cat_rbuf});
+ $self->requeue if @$inflight || exists($self->{rbuf});
}
}
diff --git a/lib/PublicInbox/Git.pm b/lib/PublicInbox/Git.pm
index 77783000..f15ace1a 100644
--- a/lib/PublicInbox/Git.pm
+++ b/lib/PublicInbox/Git.pm
@@ -202,7 +202,7 @@ sub cat_async_step ($$) {
my ($self, $inflight) = @_;
die 'BUG: inflight empty or odd' if scalar(@$inflight) < 3;
my ($req, $cb, $arg) = splice(@$inflight, 0, 3);
- my $rbuf = delete($self->{cat_rbuf}) // \(my $new = '');
+ my $rbuf = delete($self->{rbuf}) // \(my $new = '');
my ($bref, $oid, $type, $size);
my $head = my_readline($self->{in}, $rbuf);
# ->fail may be called via Gcf2Client.pm
@@ -225,7 +225,7 @@ sub cat_async_step ($$) {
my $err = $! ? " ($!)" : '';
$self->fail("bad result from async cat-file: $head$err");
}
- $self->{cat_rbuf} = $rbuf if $$rbuf ne '';
+ $self->{rbuf} = $rbuf if $$rbuf ne '';
eval { $cb->($bref, $oid, $type, $size, $arg) };
warn "E: $oid: $@\n" if $@;
}
@@ -259,7 +259,7 @@ sub check_async_step ($$) {
my ($self, $inflight_c) = @_;
die 'BUG: inflight empty or odd' if scalar(@$inflight_c) < 3;
my ($req, $cb, $arg) = splice(@$inflight_c, 0, 3);
- my $rbuf = delete($self->{chk_rbuf}) // \(my $new = '');
+ my $rbuf = delete($self->{rbuf_c}) // \(my $new = '');
chomp(my $line = my_readline($self->{in_c}, $rbuf));
my ($hex, $type, $size) = split(/ /, $line);
@@ -271,7 +271,7 @@ sub check_async_step ($$) {
my $ret = my_read($self->{in_c}, $rbuf, $type + 1);
$self->fail(defined($ret) ? 'read EOF' : "read: $!") if !$ret;
}
- $self->{chk_rbuf} = $rbuf if $$rbuf ne '';
+ $self->{rbuf_c} = $rbuf if $$rbuf ne '';
eval { $cb->($hex, $type, $size, $arg, $self) };
warn "E: check($req) $@\n" if $@;
}
@@ -333,24 +333,28 @@ sub _destroy {
dwaitpid($p) if $$ == $self->{"$pid.owner"};
}
-sub cat_async_abort ($) {
+sub async_abort ($) {
my ($self) = @_;
- if (my $inflight = $self->{inflight}) {
- while (@$inflight) {
- my ($req, $cb, $arg) = splice(@$inflight, 0, 3);
- $req =~ s/ .*//; # drop git_dir for Gcf2Client
- eval { $cb->(undef, $req, undef, undef, $arg) };
- warn "E: $req: $@ (in abort)\n" if $@;
+ while (scalar(@{$self->{inflight_c} // []}) ||
+ scalar(@{$self->{inflight} // []})) {
+ for my $c ('', '_c') {
+ my $q = $self->{"inflight$c"};
+ while (@$q) {
+ my ($req, $cb, $arg) = splice(@$q, 0, 3);
+ $req =~ s/ .*//; # drop git_dir for Gcf2Client
+ eval { $cb->(undef, $req, undef, undef, $arg) };
+ warn "E: $req: $@ (in abort)\n" if $@;
+ }
+ delete $self->{"inflight$c"};
+ delete $self->{"rbuf$c"};
}
- delete $self->{cat_rbuf};
- delete $self->{inflight};
}
cleanup($self);
}
sub fail { # may be augmented in subclasses
my ($self, $msg) = @_;
- cat_async_abort($self);
+ async_abort($self);
croak(ref($self) . ' ' . ($self->{git_dir} // '') . ": $msg");
}
@@ -391,8 +395,8 @@ sub async_wait_all ($) {
my ($self) = @_;
while (scalar(@{$self->{inflight_c} // []}) ||
scalar(@{$self->{inflight} // []})) {
- $self->check_async_wait;
- $self->cat_async_wait;
+ check_async_wait($self);
+ cat_async_wait($self);
}
}
@@ -406,8 +410,8 @@ sub cleanup {
async_wait_all($self);
delete $self->{inflight};
delete $self->{inflight_c};
- _destroy($self, qw(cat_rbuf in out pid));
- _destroy($self, qw(chk_rbuf in_c out_c pid_c err_c));
+ _destroy($self, qw(rbuf in out pid));
+ _destroy($self, qw(rbuf_c in_c out_c pid_c err_c));
undef;
}
diff --git a/lib/PublicInbox/GitAsyncCat.pm b/lib/PublicInbox/GitAsyncCat.pm
index 57c194d9..cea3f539 100644
--- a/lib/PublicInbox/GitAsyncCat.pm
+++ b/lib/PublicInbox/GitAsyncCat.pm
@@ -16,7 +16,7 @@ our $GCF2C; # singleton PublicInbox::Gcf2Client
sub close {
my ($self) = @_;
if (my $git = delete $self->{git}) {
- $git->cat_async_abort;
+ $git->async_abort;
}
$self->SUPER::close; # PublicInbox::DS::close
}
@@ -32,7 +32,7 @@ sub event_step {
# child death?
if (($git->{in} // 0) != ($self->{sock} // 1)) {
$self->close;
- } elsif (@$inflight || exists $git->{cat_rbuf}) {
+ } elsif (@$inflight || exists $git->{rbuf}) {
# ok, more to do, requeue for fairness
$self->requeue;
}
--
2.51.0