From mboxrd@z Thu Jan 1 00:00:00 1970 Return-Path: X-Spam-Checker-Version: SpamAssassin 3.4.2 (2018-09-13) on dcvr.yhbt.net X-Spam-Level: X-Spam-ASN: X-Spam-Status: No, score=-4.2 required=3.0 tests=ALL_TRUSTED,BAYES_00, DKIM_SIGNED,DKIM_VALID,DKIM_VALID_AU,DKIM_VALID_EF shortcircuit=no autolearn=ham autolearn_force=no version=3.4.2 Received: from localhost (dcvr.yhbt.net [127.0.0.1]) by dcvr.yhbt.net (Postfix) with ESMTP id 56C831F59D; Tue, 19 Jul 2022 22:42:53 +0000 (UTC) DKIM-Signature: v=1; a=rsa-sha256; c=relaxed/simple; d=80x24.org; s=selector1; t=1658270573; bh=L/I+lWbsdmJlOhrk42N6HRZp0TzOp6MmSeinbiV9Iww=; h=From:To:Subject:Date:In-Reply-To:References:From; b=Isthop0ZuM0KqxGg7H5YpZoZ83AyfMaieVmpaj/GNQ90RzdZHcFW0Y1s3IjByQX+o z70MssF4griy6tmT4/1hTorh1EUVwmlLd87pvxn3yUEwenTk9L6pX8cYraN4PC9w7q D1TnpMPgvv/oeF3RsVsI+YtECvnrOEzGfbPeUl4c= From: Eric Wong To: meta@public-inbox.org Subject: [PATCH 1/2] lei: avoid deadlock on inotify/EVFILT_VNODE wakeups Date: Tue, 19 Jul 2022 22:42:52 +0000 Message-Id: <20220719224253.3218929-2-e@80x24.org> In-Reply-To: <20220719224253.3218929-1-e@80x24.org> References: <20220719224253.3218929-1-e@80x24.org> MIME-Version: 1.0 Content-Transfer-Encoding: 8bit List-Id: Enqueuing "note-event" requests from the DS event loop must not wait on workers being able to drain the queue quickly enough. Thus we make the SOCK_SEQPACKET writes nonblocking and rely on the lei-daemon event loop to enqueue writes. This is a unique problem for "note-event" since it reuses workers in between commands, while most lei commands currently fork off new workers. --- MANIFEST | 1 + lib/PublicInbox/IPC.pm | 26 +++++++++++++++-- lib/PublicInbox/LeiNoteEvent.pm | 5 ++-- lib/PublicInbox/WQBlocked.pm | 49 +++++++++++++++++++++++++++++++++ 4 files changed, 77 insertions(+), 4 deletions(-) create mode 100644 lib/PublicInbox/WQBlocked.pm diff --git a/MANIFEST b/MANIFEST index 2cbe66b7..923f5147 100644 --- a/MANIFEST +++ b/MANIFEST @@ -337,6 +337,7 @@ lib/PublicInbox/V2Writable.pm lib/PublicInbox/View.pm lib/PublicInbox/ViewDiff.pm lib/PublicInbox/ViewVCS.pm +lib/PublicInbox/WQBlocked.pm lib/PublicInbox/WQWorker.pm lib/PublicInbox/WWW.pm lib/PublicInbox/WWW.pod diff --git a/lib/PublicInbox/IPC.pm b/lib/PublicInbox/IPC.pm index 67e86a43..74862673 100644 --- a/lib/PublicInbox/IPC.pm +++ b/lib/PublicInbox/IPC.pm @@ -1,4 +1,4 @@ -# Copyright (C) 2020-2021 all contributors +# Copyright (C) all contributors # License: AGPL-3.0+ # base class for remote IPC calls and workqueues, requires Storable or Sereal @@ -43,7 +43,7 @@ if ($enc && $dec) { # should be custom ops } my $recv_cmd = PublicInbox::Spawn->can('recv_cmd4'); -my $send_cmd = PublicInbox::Spawn->can('send_cmd4') // do { +our $send_cmd = PublicInbox::Spawn->can('send_cmd4') // do { require PublicInbox::CmdIPC4; $recv_cmd //= PublicInbox::CmdIPC4->can('recv_cmd4'); PublicInbox::CmdIPC4->can('send_cmd4'); @@ -348,6 +348,24 @@ sub wq_do { } } +sub prepare_nonblock { + ($_[0]->{-wq_s1} // die 'BUG: no {-wq_s1}')->blocking(0); + $_[0]->{-reap_async} or die 'BUG: {-reap_async} needed for nonblock'; + require PublicInbox::WQBlocked; +} + +sub wq_nonblock_do { # always async + my ($self, $sub, @args) = @_; + my $buf = ipc_freeze([$sub, @args]); + if ($self->{wqb}) { # saturated once, assume saturated forever + $self->{wqb}->flush_send($buf); + } else { + $send_cmd->($self->{-wq_s1}, [], $buf, MSG_EOR) // + ($!{EAGAIN} ? PublicInbox::WQBlocked->new($self, $buf) + : croak("sendmsg: $!")); + } +} + sub _wq_worker_start ($$$$) { my ($self, $oldset, $fields, $one) = @_; my ($bcast1, $bcast2); @@ -405,6 +423,10 @@ sub wq_workers_start { sub wq_close { my ($self) = @_; + if (my $wqb = delete $self->{wqb}) { + $self->{-reap_async} or die 'BUG: {-reap_async} unset'; + $wqb->enq_close; + } delete @$self{qw(-wq_s1 -wq_s2)} or return; return if $self->{-reap_async}; my @pids = keys %{$self->{-wq_workers}}; diff --git a/lib/PublicInbox/LeiNoteEvent.pm b/lib/PublicInbox/LeiNoteEvent.pm index db387633..93f80116 100644 --- a/lib/PublicInbox/LeiNoteEvent.pm +++ b/lib/PublicInbox/LeiNoteEvent.pm @@ -58,7 +58,7 @@ sub eml_event ($$$$) { } } -sub maildir_event { # via wq_io_do +sub maildir_event { # via wq_nonblock_do my ($self, $fn, $vmd, $state) = @_; if (my $eml = PublicInbox::InboxWritable::eml_from_path($fn)) { eml_event($self, $eml, $vmd, $state); @@ -93,6 +93,7 @@ sub lei_note_event { my ($op_c, $ops) = $lei->workers_start($wq, $jobs); $lei->wait_wq_events($op_c, $ops); note_event_arm_done($lei); + $wq->prepare_nonblock; $lei->{lne} = $wq; }; if ($folder =~ /\Amaildir:/i) { @@ -101,7 +102,7 @@ sub lei_note_event { return if index($fl, 'T') >= 0; my $kw = PublicInbox::MdirReader::flags2kw($fl); my $vmd = { kw => $kw, sync_info => [ $folder, \$bn ] }; - $self->wq_do('maildir_event', $fn, $vmd, $state); + $self->wq_nonblock_do('maildir_event', $fn, $vmd, $state); } # else: TODO: imap } diff --git a/lib/PublicInbox/WQBlocked.pm b/lib/PublicInbox/WQBlocked.pm new file mode 100644 index 00000000..fbb43600 --- /dev/null +++ b/lib/PublicInbox/WQBlocked.pm @@ -0,0 +1,49 @@ +# Copyright (C) all contributors +# License: AGPL-3.0+ + +# non-blocking workqueues, currently used by LeiNoteEvent to track renames +package PublicInbox::WQBlocked; +use v5.12; +use parent qw(PublicInbox::DS); +use PublicInbox::Syscall qw(EPOLLOUT EPOLLONESHOT); +use PublicInbox::IPC; +use Carp (); +use Socket qw(MSG_EOR); + +sub new { + my ($cls, $wq, $buf) = @_; + my $self = bless { msgq => [$buf], }, $cls; + $wq->{wqb} = $self->SUPER::new($wq->{-wq_s1}, EPOLLOUT|EPOLLONESHOT); +} + +sub flush_send { + my ($self) = @_; + push(@{$self->{msgq}}, $_[1]) if defined($_[1]); + while (defined(my $buf = shift @{$self->{msgq}})) { + if (ref($buf) eq 'CODE') { + $buf->($self); # could be \&PublicInbox::DS::close + } else { + my $wq_s1 = $self->{sock}; + my $n = $PublicInbox::IPC::send_cmd->($wq_s1, [], $buf, + MSG_EOR); + next if defined($n); + Carp::croak("sendmsg: $!") unless $!{EAGAIN}; + PublicInbox::DS::epwait($wq_s1, EPOLLOUT|EPOLLONESHOT); + unshift @{$self->{msgq}}, $buf; + last; # wait for ->event_step + } + } +} + +sub enq_close { flush_send($_[0], $_[0]->can('close')) } + +sub event_step { # called on EPOLLOUT wakeup + my ($self) = @_; + eval { flush_send($self) } if $self->{sock}; + if ($@) { + warn $@; + $self->close; + } +} + +1;