1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
| | # Copyright (C) all contributors <meta@public-inbox.org>
# License: AGPL-3.0+ <https://www.gnu.org/licenses/agpl-3.0.txt>
# supports reaping of children tied to a pipe or socket
package PublicInbox::IO;
use v5.12;
use parent qw(IO::Handle Exporter);
use PublicInbox::DS qw(awaitpid);
our @EXPORT_OK = qw(poll_in read_all try_cat write_file);
use Carp qw(croak);
use IO::Poll qw(POLLIN);
use Errno qw(EINTR EAGAIN);
use PublicInbox::OnDestroy;
# don't autodie in top-level for Perl 5.16.3 (and maybe newer versions)
# we have our own ->close, so we scope autodie into each sub
sub waitcb { # awaitpid callback
my ($pid, $errref, $cb, @args) = @_;
$$errref = $?; # sets .cerr for _close
$cb->($pid, @args) if $cb; # may clobber $?
}
sub attach_pid {
my ($io, $pid, @cb_arg) = @_;
bless $io, __PACKAGE__;
# we share $err (and not $self) with awaitpid to avoid a ref cycle
my $e = \(my $err);
${*$io}{pi_io_reap} = [ $PublicInbox::OnDestroy::fork_gen, $pid, $e ];
awaitpid($pid, \&waitcb, $e, @cb_arg);
$io;
}
sub attached_pid {
my ($io) = @_;
${${*$io}{pi_io_reap} // []}[1];
}
sub can_reap {
my ($io) = @_;
${${*$io}{pi_io_reap} // [-1]}[0] == $PublicInbox::OnDestroy::fork_gen;
}
# caller cares about error result if they call close explicitly
# reap->[2] may be set before this is called via waitcb
sub close {
my ($io) = @_;
my $ret = $io->SUPER::close;
my $reap = delete ${*$io}{pi_io_reap};
return $ret if ($reap->[0] // -1) != $PublicInbox::OnDestroy::fork_gen;
if (defined ${$reap->[2]}) { # reap_pids already reaped asynchronously
$? = ${$reap->[2]};
} else { # wait synchronously
my $w = awaitpid($reap->[1]);
}
$? ? '' : $ret;
}
sub DESTROY {
my ($io) = @_;
my $reap = delete ${*$io}{pi_io_reap};
if (($reap->[0] // -1) == $PublicInbox::OnDestroy::fork_gen) {
$io->SUPER::close;
${$reap->[2]} // awaitpid($reap->[1]);
}
$io->SUPER::DESTROY;
}
sub write_file ($$@) { # mode, filename, LIST (for print)
use autodie qw(open close);
open(my $fh, shift, shift);
print $fh @_;
defined(wantarray) && !wantarray ? $fh : close $fh;
}
sub poll_in ($;$) {
IO::Poll::_poll($_[1] // -1, fileno($_[0]), my $ev = POLLIN);
}
sub read_all ($;$$$) { # pass $len=0 to read until EOF for :utf8 handles
use autodie qw(read);
my ($io, $len, $bref, $off) = @_;
$bref //= \(my $buf);
$off //= 0;
my $r = 0;
if (my $left = $len //= -s $io) { # known size (binmode :raw/:unix)
do { # retry for binmode :unix
$r = read($io, $$bref, $left, $off += $r) or croak(
"read($io) premature EOF ($left/$len remain)");
} while ($left -= $r);
} else { # read until EOF
while (($r = read($io, $$bref, 65536, $off += $r))) {}
}
wantarray ? split(/^/sm, $$bref) : $$bref
}
sub try_cat ($) {
my ($path) = @_;
open(my $fh, '<', $path) or return '';
read_all $fh;
}
# TODO: move existing HTTP/IMAP/NNTP/POP3 uses of rbuf here
sub my_bufread {
my ($io, $len) = @_;
my $rbuf = ${*$io}{pi_io_rbuf} //= \(my $new = '');
my $left = $len - length($$rbuf);
my $r;
while ($left > 0) {
$r = sysread($io, $$rbuf, $left, length($$rbuf));
if ($r) {
$left -= $r;
} elsif (defined($r)) { # EOF
return 0;
} else {
next if ($! == EAGAIN and poll_in($io));
next if $! == EINTR; # may be set by sysread or poll_in
return; # unrecoverable error
}
}
my $no_pad = substr($$rbuf, 0, $len, '');
delete(${*$io}{pi_io_rbuf}) if $$rbuf eq '';
\$no_pad;
}
# always uses "\n"
sub my_readline {
my ($io) = @_;
my $rbuf = ${*$io}{pi_io_rbuf} //= \(my $new = '');
while (1) {
if ((my $n = index($$rbuf, "\n")) >= 0) {
my $ret = substr($$rbuf, 0, $n + 1, '');
delete(${*$io}{pi_io_rbuf}) if $$rbuf eq '';
return $ret;
}
my $r = sysread($io, $$rbuf, 65536, length($$rbuf));
if (!defined($r)) {
next if ($! == EAGAIN and poll_in($io));
next if $! == EINTR; # may be set by sysread or poll_in
return; # unrecoverable error
} elsif ($r == 0) { # return whatever's left on EOF
delete(${*$io}{pi_io_rbuf});
return $$rbuf;
} # else { continue
}
}
sub has_rbuf {
my ($io) = @_;
defined(${*$io}{pi_io_rbuf});
}
1;
|