* [PATCH] WIP async barrier for l2m
@ 2024-04-16 18:29 Eric Wong
0 siblings, 0 replies; only message in thread
From: Eric Wong @ 2024-04-16 18:29 UTC (permalink / raw)
To: spew
---
lib/PublicInbox/EOFpipe.pm | 7 ++++---
lib/PublicInbox/LeiToMail.pm | 20 +++++++++++++-------
lib/PublicInbox/LeiXSearch.pm | 11 ++++++++---
3 files changed, 25 insertions(+), 13 deletions(-)
diff --git a/lib/PublicInbox/EOFpipe.pm b/lib/PublicInbox/EOFpipe.pm
index 3474874f..f85451fb 100644
--- a/lib/PublicInbox/EOFpipe.pm
+++ b/lib/PublicInbox/EOFpipe.pm
@@ -7,8 +7,8 @@ use parent qw(PublicInbox::DS);
use PublicInbox::Syscall qw(EPOLLIN EPOLLONESHOT $F_SETPIPE_SZ);
sub new {
- my (undef, $rd, $cb) = @_;
- my $self = bless { cb => $cb }, __PACKAGE__;
+ my (undef, $rd, @cb_args) = @_;
+ my $self = bless { cb_args => \@cb_args }, __PACKAGE__;
# 4096: page size
fcntl($rd, $F_SETPIPE_SZ, 4096) if $F_SETPIPE_SZ;
$self->SUPER::new($rd, EPOLLIN|EPOLLONESHOT);
@@ -17,7 +17,8 @@ sub new {
sub event_step {
my ($self) = @_;
if ($self->do_read(my $buf, 1) == 0) { # auto-closed
- $self->{cb}->();
+ my ($cb, @args) = @{$self->{cb_args}};
+ $cb->(@args);
}
}
diff --git a/lib/PublicInbox/LeiToMail.pm b/lib/PublicInbox/LeiToMail.pm
index 593547f6..a990d918 100644
--- a/lib/PublicInbox/LeiToMail.pm
+++ b/lib/PublicInbox/LeiToMail.pm
@@ -14,7 +14,7 @@ use PublicInbox::Import;
use IO::Handle; # ->autoflush
use Fcntl qw(SEEK_SET SEEK_END O_CREAT O_EXCL O_WRONLY);
use PublicInbox::Syscall qw(rename_noreplace);
-use autodie qw(open seek close);
+use autodie qw(pipe open seek close);
use Carp qw(croak);
my %kw2char = ( # Maildir characters
@@ -605,7 +605,7 @@ sub _pre_augment_mbox {
$lei->{dedupe} && $lei->{dedupe}->can('reset_dedupe');
}
if ($self->{zsfx} = PublicInbox::MboxReader::zsfx($dst)) {
- pipe(my ($r, $w)) or die "pipe: $!";
+ pipe(my $r, my $w);
$lei->{zpipe} = [ $r, $w ];
$lei->{ovv}->{lock_path} and
die 'BUG: unexpected {ovv}->{lock_path}';
@@ -721,15 +721,21 @@ sub do_augment { # slow, runs in wq worker
# fast (spawn compressor or mkdir), runs in same process as pre_augment
sub post_augment {
- my ($self, $lei, @args) = @_;
+ my ($self, $lei, $post_augment_done) = @_;
$self->{-au_noted}++ and $lei->qerr("# writing to $self->{dst} ...");
- # FIXME: this synchronous wait can be slow w/ parallel callers
- my $wait = $lei->{opt}->{'import-before'} ?
- $lei->{sto}->wq_do('barrier') : 0;
# _post_augment_mbox
my $m = $self->can("_post_augment_$self->{base_type}") or return;
- $m->($self, $lei, @args);
+
+ # --import-before is only for lei-(q|lcat), not lei-convert
+ return $m->($self, $lei) if !$lei->{opt}->{'import-before'};
+
+ require PublicInbox::EOFpipe;
+ my @io = @$lei{qw(2 sock)};
+ pipe(my $r, $io[2]);
+ my $e = PublicInbox::EOFpipe->new($r, $m, $self, $lei);
+ $e->{-lei_post_augment_done} = $post_augment_done;
+ $lei->{sto}->wq_io_do('barrier', \@io);
}
# called by every single l2m worker process
diff --git a/lib/PublicInbox/LeiXSearch.pm b/lib/PublicInbox/LeiXSearch.pm
index 5a5a1adc..060b1065 100644
--- a/lib/PublicInbox/LeiXSearch.pm
+++ b/lib/PublicInbox/LeiXSearch.pm
@@ -22,6 +22,7 @@ use PublicInbox::ContentHash qw(git_sha);
use POSIX qw(strftime);
use autodie qw(close open read seek truncate);
use PublicInbox::Syscall qw($F_SETPIPE_SZ);
+use PublicInbox::OnDestroy;
sub new {
my ($class) = @_;
@@ -428,10 +429,8 @@ sub query_done { # EOF callback for main daemon
$lei->dclose;
}
-sub do_post_augment {
+sub post_augment_done { # via on_destroy in top-level lei-daemon
my ($lei) = @_;
- my $l2m = $lei->{l2m} or return; # client disconnected
- eval { $l2m->post_augment($lei) };
my $err = $@;
if ($err) {
if (my $lxs = delete $lei->{lxs}) {
@@ -447,6 +446,12 @@ sub do_post_augment {
close(delete $lei->{au_done}); # trigger wait_startq if start_mua didn't
}
+sub do_post_augment {
+ my ($lei) = @_;
+ my $l2m = $lei->{l2m} or return; # client disconnected
+ $l2m->post_augment($lei, on_destroy(\&post_augment_done, $lei));
+}
+
sub incr_post_augment { # called whenever an l2m shard finishes augment
my ($lei) = @_;
my $l2m = $lei->{l2m} or return; # client disconnected
^ permalink raw reply related [flat|nested] only message in thread
only message in thread, other threads:[~2024-04-16 18:29 UTC | newest]
Thread overview: (only message) (download: mbox.gz / follow: Atom feed)
-- links below jump to the message on this page --
2024-04-16 18:29 [PATCH] WIP async barrier for l2m Eric Wong
This is a public inbox, see mirroring instructions
for how to clone and mirror all data and code used for this inbox;
as well as URLs for read-only IMAP folder(s) and NNTP newsgroup(s).