lei: consolidate the bulk of the IPC code

The backends for "lei add-external --mirror", "lei convert", and
"lei import" all share a similar pattern for spawning background
workers.  Hoist out the common parts to slim down our code base
a bit.

The LeiXSearch and LeiToMail workers for "lei q" remains a the
odd duck due to the deep pipelining and parallelization.

diff --git a/lib/PublicInbox/LEI.pm b/lib/PublicInbox/LEI.pm
index 1e4c36d021712f26e4faebe803061f30c7d7da3a..0b4bc20e64a18544b4509026a9845d8b1ee15b95 100644 (file)
--- a/lib/PublicInbox/LEI.pm
+++ b/lib/PublicInbox/LEI.pm
@@ -468,6 +468,25 @@ sub lei_atfork_child {
        $current_lei = $persist ? undef : $self; # for SIG{__WARN__}
 }
 
+sub workers_start {
+       my ($lei, $wq, $ident, $jobs, $ops) = @_;
+       $ops = {
+               '!' => [ $lei->can('fail_handler'), $lei ],
+               '|' => [ $lei->can('sigpipe_handler'), $lei ],
+               'x_it' => [ $lei->can('x_it'), $lei ],
+               'child_error' => [ $lei->can('child_error'), $lei ],
+               %$ops
+       };
+       require PublicInbox::PktOp;
+       ($lei->{pkt_op_c}, $lei->{pkt_op_p}) = PublicInbox::PktOp->pair($ops);
+       $wq->wq_workers_start($ident, $jobs, $lei->oldset, { lei => $lei });
+       delete $lei->{pkt_op_p};
+       my $op = delete $lei->{pkt_op_c};
+       $lei->event_step_init;
+       # oneshot needs $op, daemon-mode uses DS->EventLoop to handle $op
+       $lei->{oneshot} ? $op : undef;
+}
+
 sub _help {
        require PublicInbox::LeiHelp;
        PublicInbox::LeiHelp::call($_[0], $_[1], \%CMD, \%OPTDESC);

diff --git a/lib/PublicInbox/LeiAuth.pm b/lib/PublicInbox/LeiAuth.pm
index 6593ba515db4bfd82ecc7b65e82c65635b139c7a..7acb9900a3bdcf0c9c7947d8147f2cc276397b61 100644 (file)
--- a/lib/PublicInbox/LeiAuth.pm
+++ b/lib/PublicInbox/LeiAuth.pm
@@ -43,24 +43,13 @@ sub auth_eof {
 sub auth_start {
        my ($self, $lei, $post_auth_cb, @args) = @_;
        $lei->_lei_cfg(1); # workers may need to read config
-       my $ops = {
-               '!' => [ $lei->can('fail_handler'), $lei ],
-               '|' => [ $lei->can('sigpipe_handler'), $lei ],
-               'x_it' => [ $lei->can('x_it'), $lei ],
-               'child_error' => [ $lei->can('child_error'), $lei ],
+       my $op = $lei->workers_start($self, 'auth', 1, {
                'nrd_merge' => [ \&nrd_merge, $lei ],
                '' => [ \&auth_eof, $lei, $post_auth_cb, @args ],
-       };
-       ($lei->{pkt_op_c}, $lei->{pkt_op_p}) = PublicInbox::PktOp->pair($ops);
-       $self->wq_workers_start('lei_auth', 1, $lei->oldset, {lei => $lei});
-       my $op = delete $lei->{pkt_op_c};
-       delete $lei->{pkt_op_p};
+       });
        $self->wq_io_do('do_auth', []);
        $self->wq_close(1);
-       $lei->event_step_init; # wait for shutdowns
-       if ($lei->{oneshot}) {
-               while ($op->{sock}) { $op->event_step }
-       }
+       while ($op && $op->{sock}) { $op->event_step }
 }
 
 sub ipc_atfork_child {

diff --git a/lib/PublicInbox/LeiConvert.pm b/lib/PublicInbox/LeiConvert.pm
index 78fd5e177dcebf47c48a837eb990808700e6220e..ba375772369065d725c08ec3d0cc845a68225b18 100644 (file)
--- a/lib/PublicInbox/LeiConvert.pm
+++ b/lib/PublicInbox/LeiConvert.pm
@@ -8,7 +8,6 @@ use v5.10.1;
 use parent qw(PublicInbox::IPC);
 use PublicInbox::Eml;
 use PublicInbox::InboxWritable qw(eml_from_path);
-use PublicInbox::PktOp;
 use PublicInbox::LeiStore;
 use PublicInbox::LeiOverview;
 
@@ -59,26 +58,15 @@ sub do_convert { # via wq_do
        delete $self->{wcb}; # commit
 }
 
-sub convert_start {
+sub convert_start { # LeiAuth->auth_start callback
        my ($lei) = @_;
-       my $ops = {
-               '!' => [ $lei->can('fail_handler'), $lei ],
-               '|' => [ $lei->can('sigpipe_handler'), $lei ],
-               'x_it' => [ $lei->can('x_it'), $lei ],
-               'child_error' => [ $lei->can('child_error'), $lei ],
-               '' => [ $lei->can('dclose'), $lei ],
-       };
-       ($lei->{pkt_op_c}, $lei->{pkt_op_p}) = PublicInbox::PktOp->pair($ops);
        my $self = $lei->{cnv};
-       $self->wq_workers_start('lei_convert', 1, $lei->oldset, {lei => $lei});
-       my $op = delete $lei->{pkt_op_c};
-       delete $lei->{pkt_op_p};
+       my $op = $lei->workers_start($self, 'lei_convert', 1, {
+               '' => [ $lei->can('dclose'), $lei ]
+       });
        $self->wq_io_do('do_convert', []);
        $self->wq_close(1);
-       $lei->event_step_init; # wait for shutdowns
-       if ($lei->{oneshot}) {
-               while ($op->{sock}) { $op->event_step }
-       }
+       while ($op && $op->{sock}) { $op->event_step }
 }
 
 sub call { # the main "lei convert" method

diff --git a/lib/PublicInbox/LeiImport.pm b/lib/PublicInbox/LeiImport.pm
index 62a2a412456542b9e128d74690978dd32734a675..68cab12c819323787f4341cfc02de9d9d8538b46 100644 (file)
--- a/lib/PublicInbox/LeiImport.pm
+++ b/lib/PublicInbox/LeiImport.pm
@@ -8,7 +8,6 @@ use v5.10.1;
 use parent qw(PublicInbox::IPC);
 use PublicInbox::Eml;
 use PublicInbox::InboxWritable qw(eml_from_path);
-use PublicInbox::PktOp;
 
 sub _import_eml { # MboxReader callback
        my ($eml, $sto, $set_kw) = @_;
@@ -31,13 +30,6 @@ sub import_done { # EOF callback for main daemon
 
 sub import_start {
        my ($lei) = @_;
-       my $ops = {
-               '!' => [ $lei->can('fail_handler'), $lei ],
-               'x_it' => [ $lei->can('x_it'), $lei ],
-               'child_error' => [ $lei->can('child_error'), $lei ],
-               '' => [ \&import_done, $lei ],
-       };
-       ($lei->{pkt_op_c}, $lei->{pkt_op_p}) = PublicInbox::PktOp->pair($ops);
        my $self = $lei->{imp};
        my $j = $lei->{opt}->{jobs} // scalar(@{$self->{inputs}}) || 1;
        if (my $nrd = $lei->{nrd}) {
@@ -46,18 +38,15 @@ sub import_start {
                my $nproc = $self->detect_nproc;
                $j = $nproc if $j > $nproc;
        }
-       $self->wq_workers_start('lei_import', $j, $lei->oldset, {lei => $lei});
-       my $op = delete $lei->{pkt_op_c};
-       delete $lei->{pkt_op_p};
+       my $op = $lei->workers_start($self, 'lei_import', $j, {
+               '' => [ \&import_done, $lei ],
+       });
        $self->wq_io_do('import_stdin', []) if $self->{0};
        for my $input (@{$self->{inputs}}) {
                $self->wq_io_do('import_path_url', [], $input);
        }
        $self->wq_close(1);
-       $lei->event_step_init; # wait for shutdowns
-       if ($lei->{oneshot}) {
-               while ($op->{sock}) { $op->event_step }
-       }
+       while ($op && $op->{sock}) { $op->event_step }
 }
 
 sub call { # the main "lei import" method

diff --git a/lib/PublicInbox/LeiMirror.pm b/lib/PublicInbox/LeiMirror.pm
index c5153148362578f6027e76224bd0adaed564743d..f8ca1ee55d7b4e585306356739dcae527a5944f9 100644 (file)
--- a/lib/PublicInbox/LeiMirror.pm
+++ b/lib/PublicInbox/LeiMirror.pm
@@ -8,7 +8,6 @@ use v5.10.1;
 use parent qw(PublicInbox::IPC);
 use IO::Uncompress::Gunzip qw(gunzip $GunzipError);
 use PublicInbox::Spawn qw(popen_rd spawn);
-use PublicInbox::PktOp;
 
 sub do_finish_mirror { # dwaitpid callback
        my ($arg, $pid) = @_;
@@ -279,22 +278,12 @@ sub start {
        require PublicInbox::Inbox;
        require PublicInbox::Admin;
        require PublicInbox::InboxWritable;
-       my $ops = {
-               '!' => [ $lei->can('fail_handler'), $lei ],
-               'x_it' => [ $lei->can('x_it'), $lei ],
-               'child_error' => [ $lei->can('child_error'), $lei ],
-               '' => [ \&mirror_done, $lei ],
-       };
-       ($lei->{pkt_op_c}, $lei->{pkt_op_p}) = PublicInbox::PktOp->pair($ops);
-       $self->wq_workers_start('lei_mirror', 1, $lei->oldset, {lei => $lei});
-       my $op = delete $lei->{pkt_op_c};
-       delete $lei->{pkt_op_p};
+       my $op = $lei->workers_start($self, 'lei_mirror', 1, {
+               '' => [ \&mirror_done, $lei ]
+       });
        $self->wq_io_do('do_mirror', []);
        $self->wq_close(1);
-       $lei->event_step_init; # wait for shutdowns
-       if ($lei->{oneshot}) {
-               while ($op->{sock}) { $op->event_step }
-       }
+       while ($op && $op->{sock}) { $op->event_step }
 }
 
 sub ipc_atfork_child {