# (tmpio = [ GLOB, offset, [ length ] ])
package PublicInbox::DS;
use strict;
+use v5.10.1;
+use parent qw(Exporter);
use bytes;
-use POSIX qw(WNOHANG);
+use POSIX qw(WNOHANG sigprocmask SIG_SETMASK);
use IO::Handle qw();
use Fcntl qw(SEEK_SET :DEFAULT O_APPEND);
use Time::HiRes qw(clock_gettime CLOCK_MONOTONIC);
-use parent qw(Exporter);
-our @EXPORT_OK = qw(now msg_more);
-use 5.010_001;
use Scalar::Util qw(blessed);
use PublicInbox::Syscall qw(:epoll);
use PublicInbox::Tmpfile;
use Errno qw(EAGAIN EINVAL);
-use Carp qw(confess carp);
+use Carp qw(carp);
+our @EXPORT_OK = qw(now msg_more dwaitpid);
my $nextq; # queue for next_tick
my $wait_pids; # list of [ pid, callback, callback_arg ]
=cut
sub Reset {
+ $in_loop = undef; # first in case DESTROY callbacks use this
%DescriptorMap = ();
- $in_loop = $wait_pids = $later_queue = $reap_armed = undef;
+ $wait_pids = $later_queue = $reap_armed = undef;
$EXPMAP = {};
$nextq = $ToClose = $later_timer = $exp_timer = undef;
$LoopTimeout = -1; # no timeout by default
sub next_tick () {
my $q = $nextq or return;
$nextq = undef;
- for (@$q) {
+ for my $obj (@$q) {
# we avoid "ref" on blessed refs to workaround a Perl 5.16.3 leak:
# https://rt.perl.org/Public/Bug/Display.html?id=114340
- if (blessed($_)) {
- $_->event_step;
+ if (blessed($obj)) {
+ $obj->event_step;
} else {
- $_->();
+ $obj->();
}
}
}
($LoopTimeout < 0 || $LoopTimeout >= $timeout) ? $timeout : $LoopTimeout;
}
+sub sig_setmask { sigprocmask(SIG_SETMASK, @_) or die "sigprocmask: $!" }
+
+sub block_signals () {
+ my $oldset = POSIX::SigSet->new;
+ my $newset = POSIX::SigSet->new;
+ $newset->fillset or die "fillset: $!";
+ sig_setmask($newset, $oldset);
+ $oldset;
+}
+
# We can't use waitpid(-1) safely here since it can hit ``, system(),
# and other things. So we scan the $wait_pids list, which is hopefully
# not too big. We keep $wait_pids small by not calling dwaitpid()
$reap_armed = undef;
my $tmp = $wait_pids or return;
$wait_pids = undef;
+ my $oldset = block_signals();
foreach my $ary (@$tmp) {
my ($pid, $cb, $arg) = @$ary;
my $ret = waitpid($pid, WNOHANG);
if ($ret == 0) {
push @$wait_pids, $ary; # autovivifies @$wait_pids
- } elsif ($cb) {
- eval { $cb->($arg, $pid) };
+ } elsif ($ret == $pid) {
+ if ($cb) {
+ eval { $cb->($arg, $pid) };
+ warn "E: dwaitpid($pid) in_loop: $@" if $@;
+ }
+ } else {
+ warn "waitpid($pid, WNOHANG) = $ret, \$!=$!, \$?=$?";
}
}
- # we may not be done, yet, and could've missed/masked a SIGCHLD:
- $reap_armed //= requeue(\&reap_pids) if $wait_pids;
+ sig_setmask($oldset);
}
# reentrant SIGCHLD handler (since reap_pids is not reentrant)
$Epoll //= _InitPoller();
local $in_loop = 1;
my @events;
+ my $obj; # guard stack-not-refcounted w/ Carp + @DB::args
do {
my $timeout = RunTimers();
# that ones in the front triggered unregister-interest actions. if we
# can't find the %sock entry, it's because we're no longer interested
# in that event.
- $DescriptorMap{$fd}->event_step;
+ $obj = $DescriptorMap{$fd};
+ $obj->event_step;
}
} while (PostEventLoop());
_run_later();
$ev &= ~EPOLLEXCLUSIVE;
goto retry;
}
- die "couldn't add epoll watch for $fd: $!\n";
+ die "EPOLL_CTL_ADD $self/$sock/$fd: $!";
}
- confess("DescriptorMap{$fd} defined ($DescriptorMap{$fd})")
+ croak("FD:$fd in use by $DescriptorMap{$fd} (for $self/$sock)")
if defined($DescriptorMap{$fd});
$DescriptorMap{$fd} = $self;
# notifications about it
my $fd = fileno($sock);
epoll_ctl($Epoll, EPOLL_CTL_DEL, $fd, 0) and
- confess("EPOLL_CTL_DEL: $!");
+ croak("EPOLL_CTL_DEL($self/$sock): $!");
# we explicitly don't delete from DescriptorMap here until we
# actually close the socket, as we might be in the middle of
sub epwait ($$) {
my ($sock, $ev) = @_;
epoll_ctl($Epoll, EPOLL_CTL_MOD, fileno($sock), $ev) and
- confess("EPOLL_CTL_MOD $!");
+ croak("EPOLL_CTL_MOD($sock): $!");
}
# return true if complete, false if incomplete (or failure)
}
}
-# must be called with eval, PublicInbox::DS may not be loaded (see t/qspawn.t)
-sub dwaitpid ($$$) {
- die "Not in EventLoop\n" unless $in_loop;
- push @$wait_pids, [ @_ ]; # [ $pid, $cb, $arg ]
-
- # We could've just missed our SIGCHLD, cover it, here:
- enqueue_reap();
+sub dwaitpid ($;$$) {
+ my ($pid, $cb, $arg) = @_;
+ if ($in_loop) {
+ push @$wait_pids, [ $pid, $cb, $arg ];
+ # We could've just missed our SIGCHLD, cover it, here:
+ enqueue_reap();
+ } else {
+ my $ret = waitpid($pid, 0);
+ if ($ret == $pid) {
+ if ($cb) {
+ eval { $cb->($arg, $pid) };
+ carp "E: dwaitpid($pid) !in_loop: $@" if $@;
+ }
+ } else {
+ carp "waitpid($pid, 0) = $ret, \$!=$!, \$?=$?";
+ }
+ }
}
sub _run_later () {