ipc: switch wq to use the event loop

[public-inbox.git] / lib / PublicInbox / LeiToMail.pm

diff --git a/lib/PublicInbox/LeiToMail.pm b/lib/PublicInbox/LeiToMail.pm
index 3dcce9e71687f4628152277aa98be83b63238e92..1f6c2a3b69da18265cfc45ef9173d341f467eae5 100644 (file)
--- a/lib/PublicInbox/LeiToMail.pm
+++ b/lib/PublicInbox/LeiToMail.pm
@@ -11,11 +11,17 @@ use PublicInbox::Lock;
 use PublicInbox::ProcessPipe;
 use PublicInbox::Spawn qw(which spawn popen_rd);
 use PublicInbox::LeiDedupe;
+use PublicInbox::OnDestroy;
+use PublicInbox::Git;
+use PublicInbox::GitAsyncCat;
 use Symbol qw(gensym);
 use IO::Handle; # ->autoflush
 use Fcntl qw(SEEK_SET SEEK_END O_CREAT O_EXCL O_WRONLY);
 use Errno qw(EEXIST ESPIPE ENOENT);
-use PublicInbox::Git;
+
+# struggles with short-lived repos, Gcf2Client makes little sense with lei;
+# but we may use in-process libgit2 in the future.
+$PublicInbox::GitAsyncCat::GCF2C = 0;
 
 my %kw2char = ( # Maildir characters
        draft => 'D',
@@ -226,9 +232,7 @@ sub decompress_src ($$$) {
 
 sub dup_src ($) {
        my ($in) = @_;
-       # fileno needed because wq_set_recv_modes only used ">&=" for {1}
-       # and Perl blindly trusts that to reject the '+' (readability flag)
-       open my $dup, '+>>&=', fileno($in) or die "dup: $!";
+       open my $dup, '+>>&', $in or die "dup: $!";
        $dup;
 }
 
@@ -250,9 +254,9 @@ sub _mbox_write_cb ($$) {
        my $dedupe = $lei->{dedupe};
        $dedupe->prepare_dedupe;
        sub { # for git_to_mail
-               my ($buf, $smsg) = @_;
+               my ($buf, $smsg, $eml) = @_;
                return unless $out;
-               my $eml = PublicInbox::Eml->new($buf);
+               $eml //= PublicInbox::Eml->new($buf);
                if (!$dedupe->is_dup($eml, $smsg->{blob})) {
                        $buf = $eml2mbox->($eml, $smsg);
                        my $lk = $ovv->lock_for_scope;
@@ -285,18 +289,23 @@ sub _augment_file { # _maildir_each_file cb
 # _maildir_each_file callback, \&CORE::unlink doesn't work with it
 sub _unlink { unlink($_[0]) }
 
+sub _rand () {
+       state $seq = 0;
+       sprintf('%x,%x,%x,%x', rand(0xffffffff), time, $$, ++$seq);
+}
+
 sub _buf2maildir {
        my ($dst, $buf, $smsg) = @_;
        my $kw = $smsg->{kw} // [];
        my $sfx = join('', sort(map { $kw2char{$_} // () } @$kw));
        my $rand = ''; # chosen by die roll :P
        my ($tmp, $fh, $final);
-       my $common = $smsg->{blob};
+       my $common = $smsg->{blob} // _rand;
        if (defined(my $pct = $smsg->{pct})) { $common .= "=$pct" }
        do {
                $tmp = $dst.'tmp/'.$rand.$common;
        } while (!sysopen($fh, $tmp, O_CREAT|O_EXCL|O_WRONLY) &&
-               $! == EEXIST && ($rand = int(rand 0x7fffffff).','));
+               $! == EEXIST && ($rand = _rand.','));
        if (print $fh $$buf and close($fh)) {
                # ignore new/ and write only to cur/, otherwise MUAs
                # with R/W access to the Maildir will end up doing
@@ -307,7 +316,7 @@ sub _buf2maildir {
                do {
                        $final = $dst.$rand.$common.':2,'.$sfx;
                } while (!link($tmp, $final) && $! == EEXIST &&
-                       ($rand = int(rand 0x7fffffff).','));
+                       ($rand = _rand.','));
                unlink($tmp) or warn "W: failed to unlink $tmp: $!\n";
        } else {
                my $err = $!;
@@ -319,12 +328,13 @@ sub _buf2maildir {
 sub _maildir_write_cb ($$) {
        my ($self, $lei) = @_;
        my $dedupe = $lei->{dedupe};
-       $dedupe->prepare_dedupe;
+       $dedupe->prepare_dedupe if $dedupe;
        my $dst = $lei->{ovv}->{dst};
        sub { # for git_to_mail
-               my ($buf, $smsg) = @_;
+               my ($buf, $smsg, $eml) = @_;
+               $buf //= \($eml->as_string);
                return _buf2maildir($dst, $buf, $smsg) if !$dedupe;
-               my $eml = PublicInbox::Eml->new($$buf); # copy buf
+               $eml //= PublicInbox::Eml->new($$buf); # copy buf
                return if $dedupe->is_dup($eml, $smsg->{blob});
                undef $eml;
                _buf2maildir($dst, $buf, $smsg);
@@ -398,6 +408,7 @@ sub _pre_augment_mbox {
                        $! == ENOENT or die "unlink($dst): $!";
                }
                open my $out, $mode, $dst or die "open($dst): $!";
+               $lei->{old_1} = $lei->{1};
                $lei->{1} = $out;
        }
        # Perl does SEEK_END even with O_APPEND :<
@@ -454,30 +465,25 @@ sub post_augment { # fast (spawn compressor or mkdir), runs in main daemon
 
 sub write_mail { # via ->wq_do
        my ($self, $git_dir, $smsg, $lei) = @_;
-       my $not_done = delete $self->{4}; # write end of {each_smsg_done}
+       my $not_done = delete $self->{$lei->{each_smsg_not_done}};
        my $wcb = $self->{wcb} //= do { # first message
                my %sig = $lei->atfork_child_wq($self);
                @SIG{keys %sig} = values %sig; # not local
-               $lei->{dedupe}->prepare_dedupe;
                $self->write_cb($lei);
        };
        my $git = $self->{"$$\0$git_dir"} //= PublicInbox::Git->new($git_dir);
-       $git->cat_async($smsg->{blob}, \&git_to_mail, [$wcb, $smsg, $not_done]);
-}
-
-sub ipc_atfork_prepare {
-       my ($self) = @_;
-       # (qry_status_wr, stdout|mbox, stderr, 3: sock, 4: each_smsg_done_wr)
-       $self->wq_set_recv_modes(qw[+<&= >&= >&= +<&= >&=]);
-       $self->SUPER::ipc_atfork_prepare; # PublicInbox::IPC
+       git_async_cat($git, $smsg->{blob}, \&git_to_mail,
+                               [$wcb, $smsg, $not_done]);
 }
 
-sub DESTROY {
+sub wq_atexit_child {
        my ($self) = @_;
-       for my $pid_git (grep(/\A$$\0/, keys %$self)) {
-               $self->{$pid_git}->async_wait_all;
+       delete $self->{wcb};
+       for my $git (delete @$self{grep(/\A$$\0/, keys %$self)}) {
+               $git->async_wait_all;
        }
-       $self->SUPER::DESTROY; # PublicInbox::IPC
+       $SIG{__WARN__} = 'DEFAULT';
+       $SIG{PIPE} = 'DEFAULT';
 }
 
 1;