diff options
author | Eric Wong <e@80x24.org> | 2021-02-22 08:22:55 -0300 |
---|---|---|
committer | Eric Wong <e@80x24.org> | 2021-02-22 18:14:20 -0400 |
commit | eda916393e1d7c203225279df84ccf692953ab57 (patch) | |
tree | 7362cf4f5e60ecca71cbe67fa5f9986844128220 | |
parent | 83b7a243361b79c26e566430d53f6c610b0e6c82 (diff) | |
download | public-inbox-eda916393e1d7c203225279df84ccf692953ab57.tar.gz |
lei_to_mail will be able to use this, too.
-rw-r--r-- | lib/PublicInbox/LeiAuth.pm | 37 | ||||
-rw-r--r-- | lib/PublicInbox/LeiImport.pm | 39 |
2 files changed, 42 insertions, 34 deletions
diff --git a/lib/PublicInbox/LeiAuth.pm b/lib/PublicInbox/LeiAuth.pm index 5d321be2..d329eadb 100644 --- a/lib/PublicInbox/LeiAuth.pm +++ b/lib/PublicInbox/LeiAuth.pm @@ -8,7 +8,6 @@ use strict; use v5.10.1; use parent qw(PublicInbox::IPC); use PublicInbox::PktOp qw(pkt_do); -use PublicInbox::NetReader; sub net_merge { my ($lei, $net_new) = @_; @@ -28,6 +27,42 @@ sub do_auth { # called via wq_io_do net_merge($lei, $net); # tell lei-daemon updated auth info } +sub do_auth_atfork { # used by IPC WQ workers + my ($self, $wq) = @_; + return if $wq->{-wq_worker_nr} != 0; + my $lei = $wq->{lei}; + my $net = $self->{net}; + my $mics = $net->imap_common_init($lei); + net_merge($lei, $net); + $net->{mics_cached} = $mics; +} + +sub net_merge_done1 { # bump merge-count in top-level lei-daemon + my ($wq) = @_; + return if ++$wq->{nr_net_merge_done} != $wq->{-wq_nr_workers}; + $wq->net_merge_complete; # defined per wq-class (e.g. LeiImport) +} + +sub net_merge_all { # called via wq_broadcast + my ($wq, $net_new) = @_; + my $net = $wq->{lei}->{net}; + %$net = (%$net, %$net_new); + pkt_do($wq->{lei}->{pkt_op_p}, 'net_merge_done1') or + die "pkt_op_do net_merge_done1: $!"; +} + +# called by top-level lei-daemon when first worker is done with auth +sub net_merge_continue { + my ($wq, $net_new) = @_; + $wq->wq_broadcast('net_merge_all', $net_new); +} + +sub op_merge { # prepares PktOp->pair ops + my ($self, $ops, $wq) = @_; + $ops->{net_merge} = [ \&net_merge_continue, $wq ]; + $ops->{net_merge_done1} = [ \&net_merge_done1, $wq ]; +} + sub do_finish_auth { # dwaitpid callback my ($arg, $pid) = @_; my ($self, $lei, $post_auth_cb, @args) = @$arg; diff --git a/lib/PublicInbox/LeiImport.pm b/lib/PublicInbox/LeiImport.pm index 5e2e61af..bc37c628 100644 --- a/lib/PublicInbox/LeiImport.pm +++ b/lib/PublicInbox/LeiImport.pm @@ -29,20 +29,7 @@ sub import_done { # EOF callback for main daemon $imp->wq_wait_old(\&import_done_wait, $lei); } -sub net_merge_all { # via wq_broadcast - my ($self, $net_new) = @_; - my $net = $self->{lei}->{net}; - %$net = (%$net, %$net_new); - pkt_do($self->{lei}->{pkt_op_p}, 'net_merge_done1') or - die "pkt_op_do net_merge_done1: $!"; -} - -sub net_merge_continue { # first worker is done with auth - my ($self, $net_new) = @_; - $self->wq_broadcast('net_merge_all', $net_new); -} - -sub net_merge_complete { +sub net_merge_complete { # callback used by LeiAuth my ($self) = @_; for my $input (@{$self->{inputs}}) { $self->wq_io_do('import_path_url', [], $input); @@ -50,13 +37,6 @@ sub net_merge_complete { $self->wq_close(1); } -sub net_merge_done1 { - my ($self) = @_; - my $lei = $self->{lei}; - return if ++$lei->{nr_net_merge_done} != $self->{-wq_nr_workers}; - net_merge_complete($self); -} - sub import_start { my ($lei) = @_; my $self = $lei->{imp}; @@ -68,15 +48,11 @@ sub import_start { $j = $nproc if $j > $nproc; } my $ops = { '' => [ \&import_done, $lei ] }; - my $auth = $lei->{auth}; - if ($auth) { - $ops->{net_merge} = [ \&net_merge_continue, $self ]; - $ops->{net_merge_done1} = [ \&net_merge_done1, $self ]; - } + $lei->{auth}->op_merge($ops, $self) if $lei->{auth}; $self->{-wq_nr_workers} = $j // 1; # locked my $op = $lei->workers_start($self, 'lei_import', undef, $ops); $self->wq_io_do('import_stdin', []) if $self->{0}; - net_merge_complete($self) if !$auth; + net_merge_complete($self) unless $lei->{auth}; while ($op && $op->{sock}) { $op->event_step } } @@ -147,12 +123,7 @@ sub ipc_atfork_child { delete $lei->{imp}; # drop circular ref $lei->lei_atfork_child; $self->SUPER::ipc_atfork_child; - my $net = $lei->{net}; - if ($net && $self->{-wq_worker_nr} == 0) { - my $mics = $net->imap_common_init($lei); - PublicInbox::LeiAuth::net_merge($lei, $net); - $net->{mics_cached} = $mics; - } + $lei->{auth}->do_auth_atfork($self) if $lei->{auth}; undef; } @@ -222,4 +193,6 @@ sub import_stdin { _import_fh($lei, delete $self->{0}, '<stdin>', $lei->{opt}->{'format'}); } +no warnings 'once'; # the following works even when LeiAuth is lazy-loaded +*net_merge_all = \&PublicInbox::LeiAuth::net_merge_all; 1; |