user/dev discussion of public-inbox itself
 help / color / mirror / code / Atom feed
From: Eric Wong <e@80x24.org>
To: meta@public-inbox.org
Subject: [PATCH 01/16] lei: switch to use SEQPACKET socketpair instead of pipe
Date: Tue,  2 Feb 2021 11:46:47 +0000	[thread overview]
Message-ID: <20210202114702.29886-2-e@80x24.org> (raw)
In-Reply-To: <20210202114702.29886-1-e@80x24.org>

This will allow us to use larger messages and do progress
reporting to accumulate in the main daemon.
---
 MANIFEST                      |  2 +-
 lib/PublicInbox/LEI.pm        |  8 ++--
 lib/PublicInbox/LeiXSearch.pm | 27 ++++++------
 lib/PublicInbox/OpPipe.pm     | 41 ------------------
 lib/PublicInbox/PktOp.pm      | 79 +++++++++++++++++++++++++++++++++++
 5 files changed, 98 insertions(+), 59 deletions(-)
 delete mode 100644 lib/PublicInbox/OpPipe.pm
 create mode 100644 lib/PublicInbox/PktOp.pm

diff --git a/MANIFEST b/MANIFEST
index 017dc7f2..bcb9d08e 100644
--- a/MANIFEST
+++ b/MANIFEST
@@ -205,9 +205,9 @@ lib/PublicInbox/NNTPD.pm
 lib/PublicInbox/NNTPdeflate.pm
 lib/PublicInbox/NewsWWW.pm
 lib/PublicInbox/OnDestroy.pm
-lib/PublicInbox/OpPipe.pm
 lib/PublicInbox/Over.pm
 lib/PublicInbox/OverIdx.pm
+lib/PublicInbox/PktOp.pm
 lib/PublicInbox/ProcessPipe.pm
 lib/PublicInbox/Qspawn.pm
 lib/PublicInbox/Reply.pm
diff --git a/lib/PublicInbox/LEI.pm b/lib/PublicInbox/LEI.pm
index 17ad18b9..737db1e1 100644
--- a/lib/PublicInbox/LEI.pm
+++ b/lib/PublicInbox/LEI.pm
@@ -306,7 +306,7 @@ sub qerr ($;@) { $_[0]->{opt}->{quiet} or err(shift, @_) }
 sub fail ($$;$) {
 	my ($self, $buf, $exit_code) = @_;
 	err($self, $buf) if defined $buf;
-	syswrite($self->{op_pipe}, '!') if $self->{op_pipe}; # fail_handler
+	send($self->{pkt_op}, '!', MSG_EOR) if $self->{pkt_op}; # fail_handler
 	x_it($self, ($exit_code // 1) << 8);
 	undef;
 }
@@ -369,14 +369,14 @@ sub io_restore ($$) {
 sub note_sigpipe { # triggers sigpipe_handler
 	my ($self, $fd) = @_;
 	close(delete($self->{$fd})); # explicit close silences Perl warning
-	syswrite($self->{op_pipe}, '|') if $self->{op_pipe};
+	send($self->{pkt_op}, '|', MSG_EOR) if $self->{pkt_op};
 	x_it($self, 13);
 }
 
 sub atfork_child_wq {
 	my ($self, $wq) = @_;
 	io_restore($self, $wq);
-	-p $self->{op_pipe} or die 'BUG: {op_pipe} expected';
+	-S $self->{pkt_op} or die 'BUG: {pkt_op} expected';
 	io_restore($self->{l2m}, $wq);
 	%PATH2CFG = ();
 	undef $errors_log;
@@ -408,7 +408,7 @@ sub atfork_parent_wq {
 	$self->{env} = $env;
 	delete @$lei{qw(3 -lei_store cfg old_1 pgr lxs)}; # keep l2m
 	my @io = (delete(@$lei{qw(0 1 2)}),
-			io_extract($lei, qw(sock op_pipe startq)));
+			io_extract($lei, qw(sock pkt_op startq)));
 	my $l2m = $lei->{l2m};
 	if ($l2m && $l2m != $wq) { # $wq == lxs
 		if (my $wq_s1 = $l2m->{-wq_s1}) {
diff --git a/lib/PublicInbox/LeiXSearch.pm b/lib/PublicInbox/LeiXSearch.pm
index f630e79a..e577ab09 100644
--- a/lib/PublicInbox/LeiXSearch.pm
+++ b/lib/PublicInbox/LeiXSearch.pm
@@ -9,10 +9,11 @@ use strict;
 use v5.10.1;
 use parent qw(PublicInbox::LeiSearch PublicInbox::IPC);
 use PublicInbox::DS qw(dwaitpid);
-use PublicInbox::OpPipe;
+use PublicInbox::PktOp;
 use PublicInbox::Import;
 use File::Temp 0.19 (); # 0.19 for ->newdir
 use File::Spec ();
+use Socket qw(MSG_EOR);
 use PublicInbox::Search qw(xap_terms);
 use PublicInbox::Spawn qw(popen_rd spawn which);
 use PublicInbox::MID qw(mids);
@@ -353,7 +354,8 @@ sub query_prepare { # called by wq_do
 	delete $lei->{l2m}->{-wq_s1};
 	eval { $lei->{l2m}->do_augment($lei) };
 	$lei->fail($@) if $@;
-	syswrite($lei->{op_pipe}, '.') == 1 or die "do_post_augment trigger: $!"
+	send($lei->{pkt_op}, '.', MSG_EOR) == 1 or
+		die "do_post_augment trigger: $!"
 }
 
 sub fail_handler ($;$$) {
@@ -380,20 +382,19 @@ sub do_query {
 		fcntl($lei->{startq}, 1031, 4096) if $^O eq 'linux';
 		$zpipe = $l2m->pre_augment($lei);
 	}
-	pipe(my $done, $lei->{op_pipe}) or die "pipe $!";
+	my $in_loop = exists $lei->{sock};
+	my $ops = {
+		'|' => [ \&sigpipe_handler, $lei ],
+		'!' => [ \&fail_handler, $lei ],
+		'.' => [ \&do_post_augment, $lei, $zpipe, $au_done ],
+		'' => [ \&query_done, $lei ],
+	};
+	(my $op, $lei->{pkt_op}) = PublicInbox::PktOp->pair($ops, $in_loop);
 	my ($lei_ipc, @io) = $lei->atfork_parent_wq($self);
-	delete($lei->{op_pipe});
+	delete($lei->{pkt_op});
 
 	$lei->event_step_init; # wait for shutdowns
-	my $done_op = {
-		'' => [ \&query_done, $lei ],
-		'|' => [ \&sigpipe_handler, $lei ],
-		'!' => [ \&fail_handler, $lei ]
-	};
-	my $in_loop = exists $lei->{sock};
-	$done = PublicInbox::OpPipe->new($done, $done_op, $in_loop);
 	if ($l2m) {
-		$done_op->{'.'} = [ \&do_post_augment, $lei, $zpipe, $au_done ];
 		$self->wq_do('query_prepare', \@io, $lei_ipc);
 		$io[1] = $zpipe->[1] if $zpipe;
 	}
@@ -401,7 +402,7 @@ sub do_query {
 	$self->wq_close(1);
 	unless ($in_loop) {
 		# for the $lei_ipc->atfork_child_wq PIPE handler:
-		while ($done->{sock}) { $done->event_step }
+		while ($op->{sock}) { $op->event_step }
 	}
 }
 
diff --git a/lib/PublicInbox/OpPipe.pm b/lib/PublicInbox/OpPipe.pm
deleted file mode 100644
index 295a8aa5..00000000
--- a/lib/PublicInbox/OpPipe.pm
+++ /dev/null
@@ -1,41 +0,0 @@
-# Copyright (C) 2021 all contributors <meta@public-inbox.org>
-# License: AGPL-3.0+ <https://www.gnu.org/licenses/agpl-3.0.txt>
-
-# bytecode dispatch pipe, reads a byte, runs a sub
-# byte => [ sub, @operands ]
-package PublicInbox::OpPipe;
-use strict;
-use v5.10.1;
-use parent qw(PublicInbox::DS);
-use PublicInbox::Syscall qw(EPOLLIN);
-
-sub new {
-	my ($cls, $rd, $op_map, $in_loop) = @_;
-	my $self = bless { sock => $rd, op_map => $op_map }, $cls;
-	# 1031: F_SETPIPE_SZ, 4096: page size
-	fcntl($rd, 1031, 4096) if $^O eq 'linux';
-	if ($in_loop) { # iff using DS->EventLoop
-		$rd->blocking(0);
-		$self->SUPER::new($rd, EPOLLIN);
-	}
-	$self;
-}
-
-sub event_step {
-	my ($self) = @_;
-	my $rd = $self->{sock};
-	my $byte;
-	until (defined(sysread($rd, $byte, 1))) {
-		return if $!{EAGAIN};
-		next if $!{EINTR};
-		die "read \$rd: $!";
-	}
-	my $op = $self->{op_map}->{$byte} or die "BUG: unknown byte `$byte'";
-	if ($byte eq '') { # close on EOF
-		$rd->blocking ? delete($self->{sock}) : $self->close;
-	}
-	my ($sub, @args) = @$op;
-	$sub->(@args);
-}
-
-1;
diff --git a/lib/PublicInbox/PktOp.pm b/lib/PublicInbox/PktOp.pm
new file mode 100644
index 00000000..d5b95a73
--- /dev/null
+++ b/lib/PublicInbox/PktOp.pm
@@ -0,0 +1,79 @@
+# Copyright (C) 2021 all contributors <meta@public-inbox.org>
+# License: AGPL-3.0+ <https://www.gnu.org/licenses/agpl-3.0.txt>
+
+# op dispatch socket, reads a message, runs a sub
+# There may be multiple producers, but (for now) only one consumer
+# Used for lei_xsearch and maybe other things
+# "literal" => [ sub, @operands ]
+# /regexp/ => [ sub, @operands ]
+package PublicInbox::PktOp;
+use strict;
+use v5.10.1;
+use parent qw(PublicInbox::DS);
+use Errno qw(EAGAIN EINTR);
+use PublicInbox::Syscall qw(EPOLLIN EPOLLET);
+use Socket qw(AF_UNIX MSG_EOR SOCK_SEQPACKET);
+
+sub new {
+	my ($cls, $r, $ops, $in_loop) = @_;
+	my $self = bless { sock => $r, ops => $ops, re => [] }, $cls;
+	if (ref($ops) eq 'ARRAY') {
+		my %ops;
+		for my $op (@$ops) {
+			if (ref($op->[0])) {
+				push @{$self->{re}}, $op;
+			} else {
+				$ops{$op->[0]} = $op->[1];
+			}
+		}
+		$self->{ops} = \%ops;
+	}
+	if ($in_loop) { # iff using DS->EventLoop
+		$r->blocking(0);
+		$self->SUPER::new($r, EPOLLIN|EPOLLET);
+	}
+	$self;
+}
+
+# returns a blessed object as the consumer, and a GLOB/IO for the producer
+sub pair {
+	my ($cls, $ops, $in_loop) = @_;
+	my ($c, $p);
+	socketpair($c, $p, AF_UNIX, SOCK_SEQPACKET, 0) or die "socketpair: $!";
+	(new($cls, $c, $ops, $in_loop), $p);
+}
+
+sub close {
+	my ($self) = @_;
+	my $c = $self->{sock} or return;
+	$c->blocking ? delete($self->{sock}) : $self->SUPER::close;
+}
+
+sub event_step {
+	my ($self) = @_;
+	my $c = $self->{sock};
+	my $msg;
+	do {
+		my $n = recv($c, $msg, 128, 0);
+		unless (defined $n) {
+			return if $! == EAGAIN;
+			next if $! == EINTR;
+			$self->close;
+			die "recv: $!";
+		}
+		my $op = $self->{ops}->{$msg};
+		unless ($op) {
+			for my $re_op (@{$self->{re}}) {
+				$msg =~ $re_op->[0] or next;
+				$op = $re_op->[1];
+				last;
+			}
+		}
+		die "BUG: unknown message: `$msg'" unless $op;
+		my ($sub, @args) = @$op;
+		$sub->(@args);
+		return $self->close if $msg eq ''; # close on EOF
+	} while (1);
+}
+
+1;

  reply	other threads:[~2021-02-02 11:47 UTC|newest]

Thread overview: 17+ messages / expand[flat|nested]  mbox.gz  Atom feed  top
2021-02-02 11:46 [PATCH 00/16] lei: -I/--include and more Eric Wong
2021-02-02 11:46 ` Eric Wong [this message]
2021-02-02 11:46 ` [PATCH 02/16] lei_query: default to 10000 messages as documented Eric Wong
2021-02-02 11:46 ` [PATCH 03/16] lei q: emit progress and counting via PktOp Eric Wong
2021-02-02 11:46 ` [PATCH 04/16] lei q: support --only, --include and --exclude Eric Wong
2021-02-02 11:46 ` [PATCH 05/16] lei: complete: do not complete non-arg options w/ help text Eric Wong
2021-02-02 11:46 ` [PATCH 06/16] lei: q: shell completion for --(include|exclude|only) Eric Wong
2021-02-02 11:46 ` [PATCH 07/16] lei_xsearch: truncate curl stderr after reading it Eric Wong
2021-02-02 11:46 ` [PATCH 08/16] lib: explicitly distinguish oneshot use Eric Wong
2021-02-02 11:46 ` [PATCH 09/16] lei q: do not leave temporary files after oneshot exit Eric Wong
2021-02-02 11:46 ` [PATCH 10/16] cmd_ipc4: fix comments and formatting Eric Wong
2021-02-02 11:46 ` [PATCH 11/16] pktop: fix potential undefined var Eric Wong
2021-02-02 11:46 ` [PATCH 12/16] lei_xsearch: ensure curl.err and tail(1) cleanup happens Eric Wong
2021-02-02 11:46 ` [PATCH 13/16] doc: lei-q: note "-a" and link to Xapian QueryParser Eric Wong
2021-02-02 11:47 ` [PATCH 14/16] lei_overview: avoid unnecessary {l2m} delete Eric Wong
2021-02-02 11:47 ` [PATCH 15/16] lei q: tidy up progress reporting Eric Wong
2021-02-02 11:47 ` [PATCH 16/16] lei q: support --jobs [SEARCHERS],[WRITERS] Eric Wong

Reply instructions:

You may reply publicly to this message via plain-text email
using any one of the following methods:

* Save the following mbox file, import it into your mail client,
  and reply-to-all from there: mbox

  Avoid top-posting and favor interleaved quoting:
  https://en.wikipedia.org/wiki/Posting_style#Interleaved_style

  List information: https://public-inbox.org/README

* Reply using the --to, --cc, and --in-reply-to
  switches of git-send-email(1):

  git send-email \
    --in-reply-to=20210202114702.29886-2-e@80x24.org \
    --to=e@80x24.org \
    --cc=meta@public-inbox.org \
    --subject='Re: [PATCH 01/16] lei: switch to use SEQPACKET socketpair instead of pipe' \
    /path/to/YOUR_REPLY

  https://kernel.org/pub/software/scm/git/docs/git-send-email.html

* If your mail client supports setting the In-Reply-To header
  via mailto: links, try the mailto: link

user/dev discussion of public-inbox itself

This inbox may be cloned and mirrored by anyone:

	git clone --mirror https://public-inbox.org/meta
	git clone --mirror http://czquwvybam4bgbro.onion/meta
	git clone --mirror http://hjrcffqmbrq6wope.onion/meta
	git clone --mirror http://ou63pmih66umazou.onion/meta

	# If you have public-inbox 1.1+ installed, you may
	# initialize and index your mirror using the following commands:
	public-inbox-init -V1 meta meta/ https://public-inbox.org/meta \
		meta@public-inbox.org
	public-inbox-index meta

Example config snippet for mirrors.
Newsgroups are available over NNTP:
	nntp://news.public-inbox.org/inbox.comp.mail.public-inbox.meta
	nntp://7fh6tueqddpjyxjmgtdiueylzoqt6pt7hec3pukyptlmohoowvhde4yd.onion/inbox.comp.mail.public-inbox.meta
	nntp://ie5yzdi7fg72h7s4sdcztq5evakq23rdt33mfyfcddc5u3ndnw24ogqd.onion/inbox.comp.mail.public-inbox.meta
	nntp://4uok3hntl7oi7b4uf4rtfwefqeexfzil2w6kgk2jn5z2f764irre7byd.onion/inbox.comp.mail.public-inbox.meta
	nntp://news.gmane.io/gmane.mail.public-inbox.general
 note: .onion URLs require Tor: https://www.torproject.org/

code repositories for project(s) associated with this inbox:

	https://80x24.org/public-inbox.git

AGPL code for this site: git clone https://public-inbox.org/public-inbox.git