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-Status: No, score=-4.0 required=3.0 tests=ALL_TRUSTED,BAYES_00 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 678C01F55B for ; Thu, 21 May 2020 07:15:32 +0000 (UTC) From: Eric Wong To: meta@public-inbox.org Subject: [PATCH] inboxidle: new class to detect inbox changes Date: Thu, 21 May 2020 07:15:32 +0000 Message-Id: <20200521071532.1970-1-e@yhbt.net> MIME-Version: 1.0 Content-Transfer-Encoding: 8bit List-Id: This will be used to implement IMAP IDLE, first. Eventually, it may be used to trigger other things: * incremental internal updates for manifest.js.gz * restart `git cat-file' processes on pack index unlink * IMAP IDLE-like long-polling HTTP endpoint And maybe more things we haven't thought of, yet. It uses Linux::Inotify2 or IO::KQueue depending on what packages are installed and what the kernel supports. It falls back to nanosecond-aware Time::HiRes::stat() (available with Perl 5.10.0+) on systems lacking Linux::Inotify2 and IO::KQueue. In the future, a pure Perl alternative to Linux::Inotify2 may be supplied for users of architectures we already support signalfd and epoll on. --- MANIFEST | 5 +++ lib/PublicInbox/FakeInotify.pm | 39 ++++++++++++++++++ lib/PublicInbox/In2Tie.pm | 17 ++++++++ lib/PublicInbox/Inbox.pm | 20 ++++++++++ lib/PublicInbox/InboxIdle.pm | 55 ++++++++++++++++++++++++++ lib/PublicInbox/KQNotify.pm | 60 ++++++++++++++++++++++++++++ lib/PublicInbox/Lock.pm | 7 +++- t/inbox_idle.t | 72 ++++++++++++++++++++++++++++++++++ 8 files changed, 274 insertions(+), 1 deletion(-) create mode 100644 lib/PublicInbox/FakeInotify.pm create mode 100644 lib/PublicInbox/In2Tie.pm create mode 100644 lib/PublicInbox/InboxIdle.pm create mode 100644 lib/PublicInbox/KQNotify.pm create mode 100644 t/inbox_idle.t diff --git a/MANIFEST b/MANIFEST index 24f95faa..9aaf9a2d 100644 --- a/MANIFEST +++ b/MANIFEST @@ -108,6 +108,7 @@ lib/PublicInbox/Emergency.pm lib/PublicInbox/Eml.pm lib/PublicInbox/EmlContentFoo.pm lib/PublicInbox/ExtMsg.pm +lib/PublicInbox/FakeInotify.pm lib/PublicInbox/Feed.pm lib/PublicInbox/Filter/Base.pm lib/PublicInbox/Filter/Gmane.pm @@ -125,8 +126,11 @@ lib/PublicInbox/HTTPD/Async.pm lib/PublicInbox/HlMod.pm lib/PublicInbox/Hval.pm lib/PublicInbox/Import.pm +lib/PublicInbox/In2Tie.pm lib/PublicInbox/Inbox.pm +lib/PublicInbox/InboxIdle.pm lib/PublicInbox/InboxWritable.pm +lib/PublicInbox/KQNotify.pm lib/PublicInbox/Linkify.pm lib/PublicInbox/Listener.pm lib/PublicInbox/Lock.pm @@ -259,6 +263,7 @@ t/httpd.t t/hval.t t/import.t t/inbox.t +t/inbox_idle.t t/index-git-times.t t/indexlevels-mirror-v1.t t/indexlevels-mirror.t diff --git a/lib/PublicInbox/FakeInotify.pm b/lib/PublicInbox/FakeInotify.pm new file mode 100644 index 00000000..bd610463 --- /dev/null +++ b/lib/PublicInbox/FakeInotify.pm @@ -0,0 +1,39 @@ +# Copyright (C) 2020 all contributors +# License: AGPL-3.0+ + +# for systems lacking Linux::Inotify2 or IO::KQueue, just emulates +# enough of Linux::Inotify2 +package PublicInbox::FakeInotify; +use strict; +use Time::HiRes qw(stat); +my $IN_CLOSE = 0x08 | 0x10; # match Linux inotify + +sub new { bless { watch => {} }, __PACKAGE__ } + +# behaves like Linux::Inotify2->watch +sub watch { + my ($self, $path, $mask, $cb) = @_; + my @st = stat($path) or return; + $self->{watch}->{"$path\0$mask"} = [ @st, $cb ]; +} + +# behaves like non-blocking Linux::Inotify2->poll +sub poll { + my ($self) = @_; + my $watch = $self->{watch} or return; + for my $x (keys %$watch) { + my ($path, $mask) = split(/\0/, $x, 2); + my @now = stat($path) or next; + my $prv = $watch->{$x}; + my $cb = $prv->[-1]; + # 10: ctime, 7: size + if ($prv->[10] != $now[10]) { + if (($mask & $IN_CLOSE) == $IN_CLOSE) { + eval { $cb->() }; + } + } + @$prv = (@now, $cb); + } +} + +1; diff --git a/lib/PublicInbox/In2Tie.pm b/lib/PublicInbox/In2Tie.pm new file mode 100644 index 00000000..db1dc104 --- /dev/null +++ b/lib/PublicInbox/In2Tie.pm @@ -0,0 +1,17 @@ +# Copyright (C) 2020 all contributors +# License: AGPL-3.0+ + +# used to ensure PublicInbox::DS can call fileno() as a function +# on Linux::Inotify2 objects +package PublicInbox::In2Tie; +use strict; + +sub TIEHANDLE { + my ($class, $in2) = @_; + bless \$in2, $class; # a scalar reference to an existing reference +} + +# this calls Linux::Inotify2::fileno +sub FILENO { ${$_[0]}->fileno } + +1; diff --git a/lib/PublicInbox/Inbox.pm b/lib/PublicInbox/Inbox.pm index 002b980f..2f83b9ae 100644 --- a/lib/PublicInbox/Inbox.pm +++ b/lib/PublicInbox/Inbox.pm @@ -390,4 +390,24 @@ sub altid_map ($) { } // {}; } +# $obj must respond to >inbox_changed, which takes Inbox ($self) as an arg +sub subscribe_unlock { + my ($self, $ident, $obj) = @_; + $self->{over_subs}->{$ident} = $obj; +} + +sub unsubscribe_unlock { + my ($self, $ident) = @_; + delete $self->{over_subs}->{$ident}; +} + +# called by inotify +sub on_unlock { + my ($self) = @_; + my $subs = $self->{over_subs} or return; + for (values %$subs) { + eval { $_->on_inbox_unlock($self) }; + } +} + 1; diff --git a/lib/PublicInbox/InboxIdle.pm b/lib/PublicInbox/InboxIdle.pm new file mode 100644 index 00000000..095a801c --- /dev/null +++ b/lib/PublicInbox/InboxIdle.pm @@ -0,0 +1,55 @@ +# Copyright (C) 2020 all contributors +# License: AGPL-3.0+ + +package PublicInbox::InboxIdle; +use strict; +use base qw(PublicInbox::DS); +use fields qw(pi_config inot); +use Symbol qw(gensym); +use PublicInbox::Syscall qw(EPOLLIN EPOLLET); +my $IN_CLOSE = 0x08 | 0x10; # match Linux inotify +my $ino_cls; +if ($^O eq 'linux' && eval { require Linux::Inotify2; 1 }) { + $IN_CLOSE = Linux::Inotify2::IN_CLOSE(); + $ino_cls = 'Linux::Inotify2'; +} elsif (eval { require PublicInbox::KQNotify }) { + $IN_CLOSE = PublicInbox::KQNotify::IN_CLOSE(); + $ino_cls = 'PublicInbox::KQNotify'; +} +require PublicInbox::In2Tie if $ino_cls; + +sub in2_arm ($$) { # PublicInbox::Config::each_inbox callback + my ($ibx, $inot) = @_; + my $path = "$ibx->{inboxdir}/"; + $path .= $ibx->version >= 2 ? 'inbox.lock' : 'ssoma.lock'; + $inot->watch($path, $IN_CLOSE, sub { $ibx->on_unlock }); + # TODO: detect deleted packs (and possibly other files) +} + +sub new { + my ($class, $pi_config) = @_; + my $self = fields::new($class); + my $inot; + if ($ino_cls) { + $inot = $ino_cls->new or die "E: $ino_cls->new: $!"; + my $sock = gensym; + tie *$sock, 'PublicInbox::In2Tie', $inot; + $inot->blocking(0); + $inot->on_overflow(undef); # broadcasts everything on overflow + $self->SUPER::new($sock, EPOLLIN | EPOLLET); + } else { + require PublicInbox::FakeInotify; + $inot = PublicInbox::FakeInotify->new; + } + $self->{inot} = $inot; + $pi_config->each_inbox(\&in2_arm, $inot); + $self; +} + +sub event_step { + my ($self) = @_; + eval { $self->{inot}->poll }; # Linux::Inotify2::poll + warn "$self->{inot}->poll err: $@\n" if $@; +} + +1; diff --git a/lib/PublicInbox/KQNotify.pm b/lib/PublicInbox/KQNotify.pm new file mode 100644 index 00000000..3cf9c0f5 --- /dev/null +++ b/lib/PublicInbox/KQNotify.pm @@ -0,0 +1,60 @@ +# Copyright (C) 2020 all contributors +# License: AGPL-3.0+ + +# implements the small subset of Linux::Inotify2 functionality we use +# using IO::KQueue on *BSD systems. +package PublicInbox::KQNotify; +use strict; +use IO::KQueue; +use PublicInbox::DSKQXS; # wraps IO::KQueue for fork-safe DESTROY + +# only true as far as public-inbox is concerned with .lock files: +sub IN_CLOSE () { NOTE_WRITE } +#sub IN_CLOSE () { 0x200 } # NOTE_CLOSE_WRITE (FreeBSD 11+ only) + +sub new { + my ($class) = @_; + bless { dskq => PublicInbox::DSKQXS->new, watch => {} }, $class; +} + +sub watch { + my ($self, $path, $mask, $cb) = @_; + open(my $fh, '<', $path) or return; + my $ident = fileno($fh); + $self->{dskq}->{kq}->EV_SET($ident, # ident + EVFILT_VNODE, # filter + EV_ADD | EV_CLEAR, # flags + $mask, # fflags + 0, 0); # data, udata + if ($mask == IN_CLOSE) { + $self->{watch}->{$ident} = [ $fh, $cb ]; + } else { + die "TODO Not implemented: $mask"; + } +} + +# emulate Linux::Inotify::fileno +sub fileno { ${$_[0]->{dskq}->{kq}} } + +# noop for Linux::Inotify2 compatibility. Unlike inotify, +# kqueue doesn't seem to overflow since it's limited by the number of +# open FDs the process has +sub on_overflow {} + +# noop for Linux::Inotify2 compatibility, we use `0' timeout for ->kevent +sub blocking {} + +# behave like Linux::Inotify2::poll +sub poll { + my ($self) = @_; + my @kevents = $self->{dskq}->{kq}->kevent(0); + for my $kev (@kevents) { + my $ident = $kev->[KQ_IDENT]; + my $mask = $kev->[KQ_FFLAGS]; + if (($mask & IN_CLOSE) == IN_CLOSE) { + eval { $self->{watch}->{$ident}->[1]->() }; + } + } +} + +1; diff --git a/lib/PublicInbox/Lock.pm b/lib/PublicInbox/Lock.pm index 032841ed..5a55c9d3 100644 --- a/lib/PublicInbox/Lock.pm +++ b/lib/PublicInbox/Lock.pm @@ -14,7 +14,7 @@ sub lock_acquire { my ($self) = @_; croak 'already locked' if $self->{lockfh}; my $lock_path = $self->{lock_path} or return; - sysopen(my $lockfh, $lock_path, O_WRONLY|O_CREAT) or + sysopen(my $lockfh, $lock_path, O_TRUNC|O_WRONLY|O_CREAT) or die "failed to open lock $lock_path: $!\n"; flock($lockfh, LOCK_EX) or die "lock failed: $!\n"; $self->{lockfh} = $lockfh; @@ -24,6 +24,11 @@ sub lock_release { my ($self) = @_; return unless $self->{lock_path}; my $lockfh = delete $self->{lockfh} or croak 'not locked'; + + # NetBSD 8.1 and OpenBSD 6.5 (and maybe other versions/*BSDs) lack + # NOTE_CLOSE_WRITE from FreeBSD 11+, so trigger NOTE_WRITE, instead + syswrite($lockfh, '.') if $^O ne 'linux'; + flock($lockfh, LOCK_UN) or die "unlock failed: $!\n"; close $lockfh or die "close failed: $!\n"; } diff --git a/t/inbox_idle.t b/t/inbox_idle.t new file mode 100644 index 00000000..6bd56113 --- /dev/null +++ b/t/inbox_idle.t @@ -0,0 +1,72 @@ +#!perl -w +# Copyright (C) 2020 all contributors +# License: AGPL-3.0+ +use Test::More; +use PublicInbox::TestCommon; +use PublicInbox::Config; +require_mods(qw(DBD::SQLite)); +require PublicInbox::SearchIdx; +use_ok 'PublicInbox::InboxIdle'; +require_git('2.6'); +use PublicInbox::InboxWritable; +my ($tmpdir, $for_destroy) = tmpdir(); + +for my $V (1, 2) { + my $inboxdir = "$tmpdir/$V"; + mkdir $inboxdir or BAIL_OUT("mkdir: $!"); + my %opt = ( + inboxdir => $inboxdir, + name => 'inbox-idle', + version => $V, + -primary_address => 'test@example.com', + indexlevel => 'basic', + ); + my $ibx = PublicInbox::Inbox->new({ %opt }); + $ibx = PublicInbox::InboxWritable->new($ibx); + my $obj = InboxIdleTestObj->new; + $ibx->init_inbox(0); + my $im = $ibx->importer(0); + if ($V == 1) { + my $sidx = PublicInbox::SearchIdx->new($ibx, 1); + $sidx->_xdb_acquire; + $sidx->set_indexlevel; + $sidx->_xdb_release; # allow watching on lockfile + } + my $pi_config = PublicInbox::Config->new(\<each_inbox(sub { shift->subscribe_unlock($ident, $obj) }); + my $ii = PublicInbox::InboxIdle->new($pi_config); + ok($ii, 'InboxIdle created'); + SKIP: { + skip('inotify or kqueue missing', 1) unless $ii->{sock}; + ok(fileno($ii->{sock}) >= 0, 'fileno() gave valid FD'); + } + ok($im->add(eml_load('t/utf8.eml')), "$V added"); + $im->done; + PublicInbox::SearchIdx->new($ibx)->index_sync if $V == 1; + $ii->event_step; + is(scalar @{$obj->{called}}, 1, 'called on unlock'); + $pi_config->each_inbox(sub { shift->unsubscribe_unlock($ident) }); + ok($im->add(eml_load('t/data/0001.patch')), "$V added #2"); + $im->done; + PublicInbox::SearchIdx->new($ibx)->index_sync if $V == 1; + $ii->event_step; + is(scalar @{$obj->{called}}, 1, 'not called when unsubbed'); + $ii->close; +} + +done_testing; + +package InboxIdleTestObj; +use strict; + +sub new { bless {}, shift } + +sub on_inbox_unlock { + my ($self, $ibx) = @_; + push @{$self->{called}}, $ibx; +}