git: improve coupling with {sock} and {inflight} fields

While the {inflight} array should be tied to the IO object even
more tightly, that's not an easy task with our current code.  So
take some small steps by introducing a gcf_inflight helper to
validate the ownership of the process and to drain the inflight
array via the awaitpid callback.

This hopefully fix problems with t/lei-q-save.t (still) hanging
occasionally on v2 outputs since git->cleanup/->DESTROY was getting
called in v2 shard workers.

diff --git a/lib/PublicInbox/Gcf2Client.pm b/lib/PublicInbox/Gcf2Client.pm
index 19d77e323b216ad34577e2e7ded5132a903056ee..07ff7dcb60dd4d6dcd21ad49da48b4c3ebb1c9dd 100644 (file)
--- a/lib/PublicInbox/Gcf2Client.pm
+++ b/lib/PublicInbox/Gcf2Client.pm
@@ -31,15 +31,16 @@ sub new  {
        $opt->{0} = $opt->{1} = $s2;
        my $cmd = [$^X, $^W ? ('-w') : (),
                        qw[-MPublicInbox::Gcf2 -e PublicInbox::Gcf2::loop]];
-       PublicInbox::IO::attach_pid($s1, spawn($cmd, $env, $opt));
        $self->{inflight} = [];
+       PublicInbox::IO::attach_pid($s1, spawn($cmd, $env, $opt),
+                       \&PublicInbox::Git::gcf_drain, $self->{inflight});
        $self->{epwatch} = \undef; # for Git->cleanup
        $self->SUPER::new($s1, EPOLLIN);
 }
 
 sub gcf2_async ($$$;$) {
        my ($self, $req, $cb, $arg) = @_;
-       my $inflight = $self->{inflight} or return $self->close;
+       my $inflight = $self->gcf_inflight or return;
        PublicInbox::Git::write_all($self, $req, \&cat_async_step, $inflight);
        push @$inflight, \$req, $cb, $arg; # ref prevents Git.pm retries
 }
@@ -49,6 +50,7 @@ sub alternates_changed {}
 
 no warnings 'once';
 
+*gcf_inflight = \&PublicInbox::Git::gcf_inflight; # for event_step
 *cat_async_step = \&PublicInbox::Git::cat_async_step; # for event_step
 *event_step = \&PublicInbox::Git::event_step;
 *fail = \&PublicInbox::Git::fail;

diff --git a/lib/PublicInbox/Git.pm b/lib/PublicInbox/Git.pm
index 93736cf0ceacde9ec45c5f1ff0b869bf06f5a218..fe8342100059b0082030c3388a642ce28704bdc8 100644 (file)
--- a/lib/PublicInbox/Git.pm
+++ b/lib/PublicInbox/Git.pm
@@ -140,6 +140,18 @@ sub last_check_err {
        $buf;
 }
 
+sub gcf_drain { # awaitpid cb
+       my ($pid, $inflight, $bc) = @_;
+       while (@$inflight) {
+               my ($req, $cb, $arg) = splice(@$inflight, 0, 3);
+               $req = $$req if ref($req);
+               $bc and $req =~ s/\A(?:contents|info) //;
+               $req =~ s/ .*//; # drop git_dir for Gcf2Client
+               eval { $cb->(undef, $req, undef, undef, $arg) };
+               warn "E: (in abort) $req: $@" if $@;
+       }
+}
+
 sub _sock_cmd {
        my ($self, $batch, $err_c) = @_;
        $self->{sock} and Carp::confess('BUG: {sock} exists');
@@ -162,8 +174,11 @@ sub _sock_cmd {
                $self->{err_c} = $opt->{2} = tmpfile($id, undef, 1) or
                                                $self->fail("tmpfile($id): $!");
        }
+       my $inflight = []; # TODO consider moving this into the IO object
        my $pid = spawn(\@cmd, undef, $opt);
-       $self->{sock} = PublicInbox::IO::attach_pid($s1, $pid);
+       $self->{sock} = PublicInbox::IO::attach_pid($s1, $pid,
+                               \&gcf_drain, $inflight, $self->{-bc});
+       $self->{inflight} = $inflight;
 }
 
 sub cat_async_retry ($$) {
@@ -171,8 +186,8 @@ sub cat_async_retry ($$) {
 
        # {inflight} may be non-existent, but if it isn't we delete it
        # here to prevent cleanup() from waiting:
-       delete $self->{inflight};
-       cleanup($self);
+       my ($sock, $epwatch) = delete @$self{qw(sock epwatch inflight)};
+       $self->SUPER::close if $epwatch;
        my $new_inflight = batch_prepare($self);
 
        while (my ($oid, $cb, $arg) = splice(@$old_inflight, 0, 3)) {
@@ -180,13 +195,25 @@ sub cat_async_retry ($$) {
                $oid = \$oid if !@$new_inflight; # to indicate oid retried
                push @$new_inflight, $oid, $cb, $arg;
        }
+       $sock->close if $sock; # only safe once old_inflight is empty
        cat_async_step($self, $new_inflight); # take one step
 }
 
+sub gcf_inflight ($) {
+       my ($self) = @_;
+       if ($self->{sock}) {
+               return $self->{inflight} if $self->{sock}->owner_pid == $$;
+               delete @$self{qw(sock inflight)};
+       } else {
+               $self->close;
+       }
+       undef;
+}
+
 # returns true if prefetch is successful
 sub async_prefetch {
        my ($self, $oid, $cb, $arg) = @_;
-       my $inflight = $self->{inflight} or return;
+       my $inflight = gcf_inflight($self) or return;
        return if @$inflight;
        substr($oid, 0, 0) = 'contents ' if $self->{-bc};
        write_all($self, "$oid\n", \&cat_async_step, $inflight);
@@ -195,7 +222,7 @@ sub async_prefetch {
 
 sub cat_async_step ($$) {
        my ($self, $inflight) = @_;
-       die 'BUG: inflight empty or odd' if scalar(@$inflight) < 3;
+       croak 'BUG: inflight empty or odd' if scalar(@$inflight) < 3;
        my ($req, $cb, $arg) = @$inflight[0, 1, 2];
        my ($bref, $oid, $type, $size);
        my $head = $self->{sock}->my_readline;
@@ -237,11 +264,8 @@ sub cat_async_step ($$) {
 
 sub cat_async_wait ($) {
        my ($self) = @_;
-       return $self->close if !$self->{sock};
-       my $inflight = $self->{inflight} or return;
-       while (scalar(@$inflight)) {
-               cat_async_step($self, $inflight);
-       }
+       my $inflight = gcf_inflight($self) or return;
+       cat_async_step($self, $inflight) while (scalar(@$inflight));
 }
 
 sub batch_prepare ($) {
@@ -253,7 +277,6 @@ sub batch_prepare ($) {
        } else {
                _sock_cmd($self, 'batch');
        }
-       $self->{inflight} = [];
 }
 
 sub _cat_file_cb {
@@ -271,7 +294,7 @@ sub cat_file {
 
 sub check_async_step ($$) {
        my ($ck, $inflight) = @_;
-       die 'BUG: inflight empty or odd' if scalar(@$inflight) < 3;
+       croak 'BUG: inflight empty or odd' if scalar(@$inflight) < 3;
        my ($req, $cb, $arg) = @$inflight[0, 1, 2];
        chomp(my $line = $ck->{sock}->my_readline);
        my ($hex, $type, $size) = split(/ /, $line);
@@ -291,8 +314,7 @@ sub check_async_wait ($) {
        my ($self) = @_;
        return cat_async_wait($self) if $self->{-bc};
        my $ck = $self->{ck} or return;
-       return $ck->close if !$ck->{sock};
-       my $inflight = $ck->{inflight} or return;
+       my $inflight = gcf_inflight($ck) or return;
        check_async_step($ck, $inflight) while (scalar(@$inflight));
 }
 
@@ -312,7 +334,6 @@ sub check_async_begin ($) {
        } else {
                _sock_cmd($self = ck($self), 'batch-check', 1);
        }
-       $self->{inflight} = [];
 }
 
 sub write_all {
@@ -337,12 +358,13 @@ sub check_async ($$$$) {
        my $inflight;
        if ($self->{-bc}) { # likely as time goes on
 batch_command:
-               $inflight = $self->{inflight} // cat_async_begin($self);
+               $inflight = gcf_inflight($self) // cat_async_begin($self);
                substr($oid, 0, 0) = 'info ';
                write_all($self, "$oid\n", \&cat_async_step, $inflight);
        } else { # accounts for git upgrades while we're running:
                my $ck = $self->{ck}; # undef OK, maybe set in check_async_begin
-               $inflight = $ck->{inflight} // check_async_begin($self);
+               $inflight = ($ck ? gcf_inflight($ck) : undef)
+                                // check_async_begin($self);
                goto batch_command if $self->{-bc};
                write_all($self->{ck}, "$oid\n", \&check_async_step, $inflight);
        }
@@ -417,8 +439,8 @@ sub date_parse {
 }
 
 sub _active ($) {
-       scalar(@{$_[0]->{inflight} // []}) ||
-               ($_[0]->{ck} && scalar(@{$_[0]->{ck}->{inflight} // []}))
+       scalar(@{gcf_inflight($_[0]) // []}) ||
+               ($_[0]->{ck} && scalar(@{gcf_inflight($_[0]->{ck}) // []}))
 }
 
 # check_async and cat_async may trigger the other, so ensure they're
@@ -493,13 +515,13 @@ sub pub_urls {
 sub cat_async_begin {
        my ($self) = @_;
        cleanup($self) if $self->alternates_changed;
-       die 'BUG: already in async' if $self->{inflight};
+       die 'BUG: already in async' if gcf_inflight($self);
        batch_prepare($self);
 }
 
 sub cat_async ($$$;$) {
        my ($self, $oid, $cb, $arg) = @_;
-       my $inflight = $self->{inflight} // cat_async_begin($self);
+       my $inflight = gcf_inflight($self) // cat_async_begin($self);
        substr($oid, 0, 0) = 'contents ' if $self->{-bc};
        write_all($self, $oid."\n", \&cat_async_step, $inflight);
        push(@$inflight, $oid, $cb, $arg);
@@ -596,8 +618,7 @@ sub cleanup_if_unlinked {
 
 sub event_step {
        my ($self) = @_;
-       $self->close if !$self->{sock}; # process died while requeued
-       my $inflight = $self->{inflight};
+       my $inflight = gcf_inflight($self);
        if ($inflight && @$inflight) {
                $self->cat_async_step($inflight);
                return $self->close unless $self->{sock};
@@ -619,18 +640,10 @@ sub watch_async ($) {
 
 sub close {
        my ($self) = @_;
-       if (my $q = $self->{inflight}) { # abort inflight requests
-               while (@$q) {
-                       my ($req, $cb, $arg) = splice(@$q, 0, 3);
-                       $req = $$req if ref($req);
-                       $self->{-bc} and $req =~ s/\A(?:contents|info) //;
-                       $req =~ s/ .*//; # drop git_dir for Gcf2Client
-                       eval { $cb->(undef, $req, undef, undef, $arg) };
-                       warn "E: (in abort) $req: $@" if $@;
-               }
-       }
+       my $sock = $self->{sock};
        delete @$self{qw(-bc err_c inflight)};
        delete($self->{epwatch}) ? $self->SUPER::close : delete($self->{sock});
+       $sock->close if $sock; # calls gcf_drain via awaitpid
 }
 
 package PublicInbox::GitCheck; # only for git <2.36

diff --git a/lib/PublicInbox/GitAsyncCat.pm b/lib/PublicInbox/GitAsyncCat.pm
index f8b2a9fc688c880be970a16dd7e721203c163e59..09744b3461b31e3af2eef14c5ffddee209be2104 100644 (file)
--- a/lib/PublicInbox/GitAsyncCat.pm
+++ b/lib/PublicInbox/GitAsyncCat.pm
@@ -40,7 +40,7 @@ sub ibx_async_prefetch {
        my ($ibx, $oid, $cb, $arg) = @_;
        my $git = $ibx->git;
        if (!defined($ibx->{topdir}) && $GCF2C) {
-               if (!@{$GCF2C->{inflight} // []}) {
+               if (!@{$GCF2C->gcf_inflight // []}) {
                        $oid .= " $git->{git_dir}\n";
                        return $GCF2C->gcf2_async($oid, $cb, $arg); # true
                }

diff --git a/lib/PublicInbox/IO.pm b/lib/PublicInbox/IO.pm
index fcebac59f329126c66bc7bb0604d12b52df8a20a..6593dcdf8a5ae1047e080c8509093993de75768e 100644 (file)
--- a/lib/PublicInbox/IO.pm
+++ b/lib/PublicInbox/IO.pm
@@ -15,8 +15,10 @@ use Errno qw(EINTR EAGAIN);
 
 sub waitcb { # awaitpid callback
        my ($pid, $errref, $cb, @args) = @_;
+       $errref //= \my $workaround_await_pids_clobbered;
        $$errref = $?; # sets .cerr for _close
-       $cb->($pid, @args) if $cb;
+       $cb->($pid, @args) if $cb; # may clobber $?
+       $? = $$errref;
 }
 
 sub attach_pid {
@@ -33,6 +35,11 @@ sub attached_pid {
        ${${*$io}{pi_io_reap} // []}[1];
 }
 
+sub owner_pid {
+       my ($io) = @_;
+       ${${*$io}{pi_io_reap} // [-1]}[0];
+}
+
 # caller cares about error result if they call close explicitly
 # reap->[2] may be set before this is called via waitcb
 sub close {
@@ -40,8 +47,12 @@ sub close {
        my $ret = $io->SUPER::close;
        my $reap = delete ${*$io}{pi_io_reap};
        return $ret unless $reap && $reap->[0] == $$;
-       ${$reap->[2]} // (my $w = awaitpid($reap->[1])); # sets [2]
-       ($? = ${$reap->[2]}) ? '' : $ret;
+       if (defined ${$reap->[2]}) { # reap_pids already reaped asynchronously
+               $? = ${$reap->[2]};
+       } else { # wait synchronously
+               my $w = awaitpid($reap->[1]);
+       }
+       $? ? '' : $ret; # use $?, AWAIT_PIDS may be cleared on ->Reset (FIXME?)
 }
 
 sub DESTROY {