]> Sergey Matveev's repositories - public-inbox.git/blob - lib/PublicInbox/GitAsyncCat.pm
imap+nntp: share COMPRESS implementation
[public-inbox.git] / lib / PublicInbox / GitAsyncCat.pm
1 # Copyright (C) 2020-2021 all contributors <meta@public-inbox.org>
2 # License: AGPL-3.0+ <https://www.gnu.org/licenses/agpl-3.0.txt>
3 #
4 # internal class used by PublicInbox::Git + PublicInbox::DS
5 # This parses the output pipe of "git cat-file --batch"
6 package PublicInbox::GitAsyncCat;
7 use strict;
8 use parent qw(PublicInbox::DS Exporter);
9 use POSIX qw(WNOHANG);
10 use PublicInbox::Syscall qw(EPOLLIN EPOLLET);
11 our @EXPORT = qw(ibx_async_cat ibx_async_prefetch);
12 use PublicInbox::Git ();
13
14 our $GCF2C; # singleton PublicInbox::Gcf2Client
15
16 sub close {
17         my ($self) = @_;
18         if (my $git = delete $self->{git}) {
19                 $git->async_abort;
20         }
21         $self->SUPER::close; # PublicInbox::DS::close
22 }
23
24 sub event_step {
25         my ($self) = @_;
26         my $git = $self->{git} or return;
27         return $self->close if ($git->{in} // 0) != ($self->{sock} // 1);
28         my $inflight = $git->{inflight};
29         if ($inflight && @$inflight) {
30                 $git->cat_async_step($inflight);
31
32                 # child death?
33                 if (($git->{in} // 0) != ($self->{sock} // 1)) {
34                         $self->close;
35                 } elsif (@$inflight || exists $git->{rbuf}) {
36                         # ok, more to do, requeue for fairness
37                         $self->requeue;
38                 }
39         } elsif ((my $pid = waitpid($git->{pid}, WNOHANG)) > 0) {
40                 # May happen if the child process is killed by a BOFH
41                 # (or segfaults)
42                 delete $git->{pid};
43                 warn "E: git $pid exited with \$?=$?\n";
44                 $self->close;
45         }
46 }
47
48 sub ibx_async_cat ($$$$) {
49         my ($ibx, $oid, $cb, $arg) = @_;
50         my $git = $ibx->git;
51         # {topdir} means ExtSearch (likely [extindex "all"]) with potentially
52         # 100K alternates.  git(1) has a proposed patch for 100K alternates:
53         # <https://lore.kernel.org/git/20210624005806.12079-1-e@80x24.org/>
54         if (!defined($ibx->{topdir}) && ($GCF2C //= eval {
55                 require PublicInbox::Gcf2Client;
56                 PublicInbox::Gcf2Client::new();
57         } // 0)) { # 0: do not retry if libgit2 or Inline::C are missing
58                 $GCF2C->gcf2_async(\"$oid $git->{git_dir}\n", $cb, $arg);
59                 \undef;
60         } else { # read-only end of git-cat-file pipe
61                 $git->cat_async($oid, $cb, $arg);
62                 $git->{async_cat} //= do {
63                         my $self = bless { git => $git }, __PACKAGE__;
64                         $git->{in}->blocking(0);
65                         $self->SUPER::new($git->{in}, EPOLLIN|EPOLLET);
66                         \undef; # this is a true ref()
67                 };
68         }
69 }
70
71 # this is safe to call inside $cb, but not guaranteed to enqueue
72 # returns true if successful, undef if not.
73 sub ibx_async_prefetch {
74         my ($ibx, $oid, $cb, $arg) = @_;
75         my $git = $ibx->git;
76         if (!defined($ibx->{topdir}) && $GCF2C) {
77                 if (!$GCF2C->{wbuf}) {
78                         $oid .= " $git->{git_dir}\n";
79                         return $GCF2C->gcf2_async(\$oid, $cb, $arg); # true
80                 }
81         } elsif ($git->{async_cat} && (my $inflight = $git->{inflight})) {
82                 # we could use MAX_INFLIGHT here w/o the halving,
83                 # but lets not allow one client to monopolize a git process
84                 if (@$inflight < int(PublicInbox::Git::MAX_INFLIGHT/2)) {
85                         print { $git->{out} } $oid, "\n" or
86                                                 $git->fail("write error: $!");
87                         return push(@$inflight, $oid, $cb, $arg);
88                 }
89         }
90         undef;
91 }
92
93 1;