]> Sergey Matveev's repositories - public-inbox.git/blob - lib/PublicInbox/IPC.pm
avoid calling waitpid from children in DESTROY
[public-inbox.git] / lib / PublicInbox / IPC.pm
1 # Copyright (C) 2020 all contributors <meta@public-inbox.org>
2 # License: AGPL-3.0+ <https://www.gnu.org/licenses/agpl-3.0.txt>
3
4 # base class for remote IPC calls, requires Storable
5 # TODO: this ought to be usable in SearchIdxShard
6 package PublicInbox::IPC;
7 use strict;
8 use v5.10.1;
9 use Socket qw(AF_UNIX SOCK_STREAM);
10 use Carp qw(confess croak);
11 use PublicInbox::Sigfd;
12 my ($enc, $dec);
13 # ->imports at BEGIN turns serial_*_with_object into custom ops on 5.14+
14 # and eliminate method call overhead
15 BEGIN {
16         eval {
17                 require Sereal::Encoder;
18                 require Sereal::Decoder;
19                 Sereal::Encoder->import('sereal_encode_with_object');
20                 Sereal::Decoder->import('sereal_decode_with_object');
21                 ($enc, $dec) = (Sereal::Encoder->new, Sereal::Decoder->new);
22         };
23 };
24
25 if ($enc && $dec) { # should be custom ops
26         *freeze = sub ($) { sereal_encode_with_object $enc, $_[0] };
27         *thaw = sub ($) { sereal_decode_with_object $dec, $_[0], my $ret };
28 } else {
29         eval { # some distros have Storable as a separate package from Perl
30                 require Storable;
31                 Storable->import(qw(freeze thaw));
32                 $enc = 1;
33         } // warn("Storable (part of Perl) missing: $@\n");
34 }
35
36 sub _get_rec ($) {
37         my ($sock) = @_;
38         local $/ = "\n";
39         defined(my $len = <$sock>) or return;
40         chop($len) eq "\n" or croak "no LF byte in $len";
41         defined(my $r = read($sock, my $buf, $len)) or croak "read error: $!";
42         $r == $len or croak "short read: $r != $len";
43         thaw($buf);
44 }
45
46 sub _send_rec ($$) {
47         my ($sock, $ref) = @_;
48         my $buf = freeze($ref);
49         print $sock length($buf), "\n", $buf or croak "print: $!";
50 }
51
52 sub ipc_return ($$$) {
53         my ($s2, $ret, $exc) = @_;
54         _send_rec($s2, $exc ? bless(\$exc, 'PublicInbox::IPC::Die') : $ret);
55 }
56
57 sub ipc_worker_loop ($$) {
58         my ($self, $s2) = @_;
59         while (my $rec = _get_rec($s2)) {
60                 my ($wantarray, $sub, @args) = @$rec;
61                 if (!defined($wantarray)) { # no waiting if client doesn't care
62                         eval { $self->$sub(@args) };
63                         eval { warn "die: $@ (from nowait $sub)\n" } if $@;
64                 } elsif ($wantarray) {
65                         my @ret = eval { $self->$sub(@args) };
66                         ipc_return($s2, \@ret, $@);
67                 } else {
68                         my $ret = eval { $self->$sub(@args) };
69                         ipc_return($s2, \$ret, $@);
70                 }
71         }
72 }
73
74 sub ipc_worker_spawn {
75         my ($self, $ident, $oldset) = @_;
76         return unless $enc;
77         my $pid = $self->{-ipc_worker_pid};
78         confess "BUG: already spawned PID:$pid" if $pid;
79         confess "BUG: already have worker socket" if $self->{-ipc_sock};
80         my ($s1, $s2);
81         socketpair($s1, $s2, AF_UNIX, SOCK_STREAM, 0) or die "socketpair: $!";
82         my $sigset = $oldset // PublicInbox::Sigfd::block_signals();
83         my $parent = $$;
84         $self->ipc_atfork_parent;
85         defined($pid = fork) or die "fork: $!";
86         if ($pid == 0) {
87                 eval { PublicInbox::DS->Reset };
88                 $self->{-ipc_parent_pid} = $parent;
89                 close $s1 or die "close(\$s1): $!";
90                 $s2->autoflush(1);
91                 $SIG{$_} = 'IGNORE' for (qw(TERM INT QUIT));
92                 local $0 = $ident;
93                 PublicInbox::Sigfd::sig_setmask($oldset);
94                 $self->ipc_atfork_child;
95                 eval { ipc_worker_loop($self, $s2) };
96                 die "worker $ident PID:$$ died: $@\n" if $@;
97                 exit;
98         }
99         PublicInbox::Sigfd::sig_setmask($sigset) unless $oldset;
100         close $s2 or die "close(\$s2): $!";
101         $s1->autoflush(1);
102         $self->{-ipc_sock} = $s1;
103         $self->{-ipc_worker_pid} = $pid;
104 }
105
106 sub ipc_worker_reap { # dwaitpid callback
107         my ($self, $pid) = @_;
108         warn "PID:$pid died with \$?=$?\n" if $?;
109 }
110
111 # for base class, override in superclasses
112 sub ipc_atfork_parent {}
113 sub ipc_atfork_child {}
114
115 sub ipc_worker_exit {
116         my (undef, $code) = @_;
117         exit($code);
118 }
119
120 sub ipc_worker_stop {
121         my ($self) = @_;
122         my $pid;
123         my $s1 = delete $self->{-ipc_sock} or do {
124                 $pid = delete $self->{-ipc_worker_pid} and
125                         die "unexpected PID:$pid without ipc_sock";
126                 return;
127         };
128         $pid = delete $self->{-ipc_worker_pid} or die "no PID?";
129         _send_rec($s1, [ undef, 'ipc_worker_exit', 0 ]);
130         shutdown($s1, 2) or die "shutdown(\$s1) for PID:$pid";
131         eval {
132                 my $reap = $self->can('ipc_worker_reap');
133                 PublicInbox::DS::dwaitpid($pid, $reap, $self);
134         };
135         if ($@) {
136                 my $wp = waitpid($pid, 0);
137                 $pid == $wp or die "waitpid($pid) returned $wp: \$?=$?";
138                 $self->ipc_worker_reap($pid);
139         }
140 }
141
142 # use this if we have multiple readers reading curl or "pigz -dc"
143 # and writing to the same store
144 sub ipc_lock_init {
145         my ($self, $f) = @_;
146         require PublicInbox::Lock;
147         $self->{-ipc_lock} //= bless { lock_path => $f }, 'PublicInbox::Lock'
148 }
149
150 sub ipc_do {
151         my ($self, $sub, @args) = @_;
152         if (my $s1 = $self->{-ipc_sock}) {
153                 my $ipc_lock = $self->{-ipc_lock};
154                 my $lock = $ipc_lock ? $ipc_lock->lock_for_scope : undef;
155                 _send_rec($s1, [ wantarray, $sub, @args ]);
156                 return unless defined(wantarray);
157                 my $ret = _get_rec($s1) // die "no response on $sub";
158                 die $$ret if ref($ret) eq 'PublicInbox::IPC::Die';
159                 wantarray ? @$ret : $$ret;
160         } else {
161                 $self->$sub(@args);
162         }
163 }
164
165 1;