* [PATCH 0/4] bundle Danga::Socket and Sys::Syscall
@ 2019-05-05 0:52 Eric Wong
2019-05-05 0:52 ` [PATCH 1/4] " Eric Wong
` (4 more replies)
0 siblings, 5 replies; 12+ messages in thread
From: Eric Wong @ 2019-05-05 0:52 UTC (permalink / raw)
To: meta
This is probably our rarest and most esoteric dependencies
at the moment, so bundle them, add some features, and drop
unused ones. It'll also give me an excuse to play with more
recent Linux kernel developments :> More on this in [1/4]
Eric Wong (4):
bundle Danga::Socket and Sys::Syscall
listener: use EPOLLEXCLUSIVE for listen sockets
DS: remove unused fields and functions
DS: drop profiling support
INSTALL | 4 -
MANIFEST | 2 +
TODO | 5 +-
lib/PublicInbox/DS.pm | 1051 +++++++++++++++++++++++++++++
lib/PublicInbox/Daemon.pm | 8 +-
lib/PublicInbox/EvCleanup.pm | 12 +-
lib/PublicInbox/GitHTTPBackend.pm | 2 +-
lib/PublicInbox/HTTP.pm | 12 +-
lib/PublicInbox/HTTPD/Async.pm | 4 +-
lib/PublicInbox/Listener.pm | 4 +-
lib/PublicInbox/NNTP.pm | 6 +-
lib/PublicInbox/ParentPipe.pm | 2 +-
lib/PublicInbox/Qspawn.pm | 4 +-
lib/PublicInbox/Syscall.pm | 329 +++++++++
t/git-http-backend.t | 2 +-
t/httpd-corner.t | 2 +-
t/httpd-unix.t | 2 +-
t/httpd.t | 2 +-
t/nntp.t | 2 +-
t/nntpd.t | 2 +-
t/v2mirror.t | 2 +-
t/v2writable.t | 4 +-
22 files changed, 1419 insertions(+), 44 deletions(-)
create mode 100644 lib/PublicInbox/DS.pm
create mode 100644 lib/PublicInbox/Syscall.pm
--
EW
^ permalink raw reply [flat|nested] 12+ messages in thread
* [PATCH 1/4] bundle Danga::Socket and Sys::Syscall
2019-05-05 0:52 [PATCH 0/4] bundle Danga::Socket and Sys::Syscall Eric Wong
@ 2019-05-05 0:52 ` Eric Wong
2019-05-05 4:56 ` [PATCH 5/4] DS: workaround IO::Kqueue EINTR (mis-)handling Eric Wong
2019-05-08 9:07 ` [PATCH 6/4] DS: handle EINTR in IO::Poll path, too Eric Wong
2019-05-05 0:52 ` [PATCH 2/4] listener: use EPOLLEXCLUSIVE for listen sockets Eric Wong
` (3 subsequent siblings)
4 siblings, 2 replies; 12+ messages in thread
From: Eric Wong @ 2019-05-05 0:52 UTC (permalink / raw)
To: meta
These modules are unmaintained upstream at the moment, but I'll
be able to help with the intended maintainer once/if CPAN
ownership is transferred. OTOH, we've been waiting for that
transfer for several years, now...
Changes I intend to make:
* EPOLLEXCLUSIVE for Linux
* remove unused fields wasting memory
* kqueue bugfixes e.g. https://rt.cpan.org/Ticket/Display.html?id=116615
* accept4 support
And some lower priority experiments:
* switch to EV_ONESHOT / EPOLLONESHOT (incompatible changes)
* nginx-style buffering to tmpfile instead of string array
* sendfile off tmpfile buffers
* io_uring maybe?
---
INSTALL | 4 -
MANIFEST | 2 +
TODO | 2 +-
lib/PublicInbox/DS.pm | 1334 +++++++++++++++++++++++++++++
lib/PublicInbox/Daemon.pm | 8 +-
lib/PublicInbox/EvCleanup.pm | 12 +-
lib/PublicInbox/GitHTTPBackend.pm | 2 +-
lib/PublicInbox/HTTP.pm | 12 +-
lib/PublicInbox/HTTPD/Async.pm | 4 +-
lib/PublicInbox/Listener.pm | 2 +-
lib/PublicInbox/NNTP.pm | 6 +-
lib/PublicInbox/ParentPipe.pm | 2 +-
lib/PublicInbox/Qspawn.pm | 4 +-
lib/PublicInbox/Syscall.pm | 326 +++++++
t/git-http-backend.t | 2 +-
t/httpd-corner.t | 2 +-
t/httpd-unix.t | 2 +-
t/httpd.t | 2 +-
t/nntp.t | 2 +-
t/nntpd.t | 2 +-
t/v2mirror.t | 2 +-
t/v2writable.t | 4 +-
22 files changed, 1698 insertions(+), 40 deletions(-)
create mode 100644 lib/PublicInbox/DS.pm
create mode 100644 lib/PublicInbox/Syscall.pm
diff --git a/INSTALL b/INSTALL
index 9470d83..3c0b910 100644
--- a/INSTALL
+++ b/INSTALL
@@ -73,10 +73,6 @@ Numerous optional modules are likely to be useful as well:
rpm: perl-DBD-SQLite
(for NNTP service or gzipped mbox over HTTP)
- - Danga::Socket deb: libdanga-socket-perl
- rpm: perl-Danga-Socket
- (for bundled HTTP and NNTP servers)
-
- Net::Server deb: libnet-server-perl
rpm: perl-Net-Server
(for HTTP/NNTP servers as standalone daemons,
diff --git a/MANIFEST b/MANIFEST
index ed8ff49..afe5ae1 100644
--- a/MANIFEST
+++ b/MANIFEST
@@ -64,6 +64,7 @@ lib/PublicInbox/AltId.pm
lib/PublicInbox/Cgit.pm
lib/PublicInbox/Config.pm
lib/PublicInbox/ContentId.pm
+lib/PublicInbox/DS.pm
lib/PublicInbox/Daemon.pm
lib/PublicInbox/Emergency.pm
lib/PublicInbox/EvCleanup.pm
@@ -117,6 +118,7 @@ lib/PublicInbox/Spamcheck.pm
lib/PublicInbox/Spamcheck/Spamc.pm
lib/PublicInbox/Spawn.pm
lib/PublicInbox/SpawnPP.pm
+lib/PublicInbox/Syscall.pm
lib/PublicInbox/Unsubscribe.pm
lib/PublicInbox/UserContent.pm
lib/PublicInbox/V2Writable.pm
diff --git a/TODO b/TODO
index 7a3bb6b..372f733 100644
--- a/TODO
+++ b/TODO
@@ -54,7 +54,7 @@ all need to be considered for everything we introduce)
* portability to FreeBSD (and other Free Software *BSDs)
ugh... https://rt.cpan.org/Ticket/Display.html?id=116615
- (IO::KQueue is broken with Danga::Socket)
+ (IO::KQueue is broken with Danga::Socket / PublicInbox::DS)
* EPOLLEXCLUSIVE for listen socket fairness across -httpd/nntpd
worker processes.
diff --git a/lib/PublicInbox/DS.pm b/lib/PublicInbox/DS.pm
new file mode 100644
index 0000000..543d3fd
--- /dev/null
+++ b/lib/PublicInbox/DS.pm
@@ -0,0 +1,1334 @@
+# This library is free software; you can redistribute it and/or modify
+# it under the same terms as Perl itself.
+#
+# This license differs from the rest of public-inbox
+#
+# This is a fork of the (for now) unmaintained Danga::Socket 1.61.
+# Unused features will be removed, and updates will be made to take
+# advantage of newer kernels
+
+package PublicInbox::DS;
+use strict;
+use bytes;
+use POSIX ();
+use Time::HiRes ();
+
+my $opt_bsd_resource = eval "use BSD::Resource; 1;";
+
+use vars qw{$VERSION};
+$VERSION = "1.61";
+
+use warnings;
+no warnings qw(deprecated);
+
+use PublicInbox::Syscall qw(:epoll);
+
+use fields ('sock', # underlying socket
+ 'fd', # numeric file descriptor
+ 'write_buf', # arrayref of scalars, scalarrefs, or coderefs to write
+ 'write_buf_offset', # offset into first array of write_buf to start writing at
+ 'write_buf_size', # total length of data in all write_buf items
+ 'write_set_watch', # bool: true if we internally set watch_write rather than by a subclass
+ 'read_push_back', # arrayref of "pushed-back" read data the application didn't want
+ 'closed', # bool: socket is closed
+ 'corked', # bool: socket is corked
+ 'event_watch', # bitmask of events the client is interested in (POLLIN,OUT,etc.)
+ 'peer_v6', # bool: cached; if peer is an IPv6 address
+ 'peer_ip', # cached stringified IP address of $sock
+ 'peer_port', # cached port number of $sock
+ 'local_ip', # cached stringified IP address of local end of $sock
+ 'local_port', # cached port number of local end of $sock
+ 'writer_func', # subref which does writing. must return bytes written (or undef) and set $! on errors
+ );
+
+use Errno qw(EINPROGRESS EWOULDBLOCK EISCONN ENOTSOCK
+ EPIPE EAGAIN EBADF ECONNRESET ENOPROTOOPT);
+use Socket qw(IPPROTO_TCP);
+use Carp qw(croak confess);
+
+use constant TCP_CORK => ($^O eq "linux" ? 3 : 0); # FIXME: not hard-coded (Linux-specific too)
+use constant DebugLevel => 0;
+
+use constant POLLIN => 1;
+use constant POLLOUT => 4;
+use constant POLLERR => 8;
+use constant POLLHUP => 16;
+use constant POLLNVAL => 32;
+
+our $HAVE_KQUEUE = eval { require IO::KQueue; 1 };
+
+our (
+ $HaveEpoll, # Flag -- is epoll available? initially undefined.
+ $HaveKQueue,
+ %DescriptorMap, # fd (num) -> PublicInbox::DS object
+ %PushBackSet, # fd (num) -> PublicInbox::DS (fds with pushed back read data)
+ $Epoll, # Global epoll fd (for epoll mode only)
+ $KQueue, # Global kqueue fd (for kqueue mode only)
+ @ToClose, # sockets to close when event loop is done
+ %OtherFds, # A hash of "other" (non-PublicInbox::DS) file
+ # descriptors for the event loop to track.
+
+ $PostLoopCallback, # subref to call at the end of each loop, if defined (global)
+ %PLCMap, # fd (num) -> PostLoopCallback (per-object)
+
+ $LoopTimeout, # timeout of event loop in milliseconds
+ $DoProfile, # if on, enable profiling
+ %Profiling, # what => [ utime, stime, calls ]
+ $DoneInit, # if we've done the one-time module init yet
+ @Timers, # timers
+ );
+
+Reset();
+
+#####################################################################
+### C L A S S M E T H O D S
+#####################################################################
+
+=head2 C<< CLASS->Reset() >>
+
+Reset all state
+
+=cut
+sub Reset {
+ %DescriptorMap = ();
+ %PushBackSet = ();
+ @ToClose = ();
+ %OtherFds = ();
+ $LoopTimeout = -1; # no timeout by default
+ $DoProfile = 0;
+ %Profiling = ();
+ @Timers = ();
+
+ $PostLoopCallback = undef;
+ %PLCMap = ();
+ $DoneInit = 0;
+
+ POSIX::close($Epoll) if defined $Epoll && $Epoll >= 0;
+ POSIX::close($KQueue) if defined $KQueue && $KQueue >= 0;
+
+ *EventLoop = *FirstTimeEventLoop;
+}
+
+=head2 C<< CLASS->HaveEpoll() >>
+
+Returns a true value if this class will use IO::Epoll for async IO.
+
+=cut
+sub HaveEpoll {
+ _InitPoller();
+ return $HaveEpoll;
+}
+
+=head2 C<< CLASS->WatchedSockets() >>
+
+Returns the number of file descriptors which are registered with the global
+poll object.
+
+=cut
+sub WatchedSockets {
+ return scalar keys %DescriptorMap;
+}
+*watched_sockets = *WatchedSockets;
+
+=head2 C<< CLASS->EnableProfiling() >>
+
+Turns profiling on, clearing current profiling data.
+
+=cut
+sub EnableProfiling {
+ if ($opt_bsd_resource) {
+ %Profiling = ();
+ $DoProfile = 1;
+ return 1;
+ }
+ return 0;
+}
+
+=head2 C<< CLASS->DisableProfiling() >>
+
+Turns off profiling, but retains data up to this point
+
+=cut
+sub DisableProfiling {
+ $DoProfile = 0;
+}
+
+=head2 C<< CLASS->ProfilingData() >>
+
+Returns reference to a hash of data in format:
+
+ ITEM => [ utime, stime, #calls ]
+
+=cut
+sub ProfilingData {
+ return \%Profiling;
+}
+
+=head2 C<< CLASS->ToClose() >>
+
+Return the list of sockets that are awaiting close() at the end of the
+current event loop.
+
+=cut
+sub ToClose { return @ToClose; }
+
+=head2 C<< CLASS->OtherFds( [%fdmap] ) >>
+
+Get/set the hash of file descriptors that need processing in parallel with
+the registered PublicInbox::DS objects.
+
+=cut
+sub OtherFds {
+ my $class = shift;
+ if ( @_ ) { %OtherFds = @_ }
+ return wantarray ? %OtherFds : \%OtherFds;
+}
+
+=head2 C<< CLASS->AddOtherFds( [%fdmap] ) >>
+
+Add fds to the OtherFds hash for processing.
+
+=cut
+sub AddOtherFds {
+ my $class = shift;
+ %OtherFds = ( %OtherFds, @_ ); # FIXME investigate what happens on dupe fds
+ return wantarray ? %OtherFds : \%OtherFds;
+}
+
+=head2 C<< CLASS->SetLoopTimeout( $timeout ) >>
+
+Set the loop timeout for the event loop to some value in milliseconds.
+
+A timeout of 0 (zero) means poll forever. A timeout of -1 means poll and return
+immediately.
+
+=cut
+sub SetLoopTimeout {
+ return $LoopTimeout = $_[1] + 0;
+}
+
+=head2 C<< CLASS->DebugMsg( $format, @args ) >>
+
+Print the debugging message specified by the C<sprintf>-style I<format> and
+I<args>
+
+=cut
+sub DebugMsg {
+ my ( $class, $fmt, @args ) = @_;
+ chomp $fmt;
+ printf STDERR ">>> $fmt\n", @args;
+}
+
+=head2 C<< CLASS->AddTimer( $seconds, $coderef ) >>
+
+Add a timer to occur $seconds from now. $seconds may be fractional, but timers
+are not guaranteed to fire at the exact time you ask for.
+
+Returns a timer object which you can call C<< $timer->cancel >> on if you need to.
+
+=cut
+sub AddTimer {
+ my $class = shift;
+ my ($secs, $coderef) = @_;
+
+ my $fire_time = Time::HiRes::time() + $secs;
+
+ my $timer = bless [$fire_time, $coderef], "PublicInbox::DS::Timer";
+
+ if (!@Timers || $fire_time >= $Timers[-1][0]) {
+ push @Timers, $timer;
+ return $timer;
+ }
+
+ # Now, where do we insert? (NOTE: this appears slow, algorithm-wise,
+ # but it was compared against calendar queues, heaps, naive push/sort,
+ # and a bunch of other versions, and found to be fastest with a large
+ # variety of datasets.)
+ for (my $i = 0; $i < @Timers; $i++) {
+ if ($Timers[$i][0] > $fire_time) {
+ splice(@Timers, $i, 0, $timer);
+ return $timer;
+ }
+ }
+
+ die "Shouldn't get here.";
+}
+
+=head2 C<< CLASS->DescriptorMap() >>
+
+Get the hash of PublicInbox::DS objects keyed by the file descriptor (fileno) they
+are wrapping.
+
+Returns a hash in list context or a hashref in scalar context.
+
+=cut
+sub DescriptorMap {
+ return wantarray ? %DescriptorMap : \%DescriptorMap;
+}
+*descriptor_map = *DescriptorMap;
+*get_sock_ref = *DescriptorMap;
+
+sub _InitPoller
+{
+ return if $DoneInit;
+ $DoneInit = 1;
+
+ if ($HAVE_KQUEUE) {
+ $KQueue = IO::KQueue->new();
+ $HaveKQueue = $KQueue >= 0;
+ if ($HaveKQueue) {
+ *EventLoop = *KQueueEventLoop;
+ }
+ }
+ elsif (PublicInbox::Syscall::epoll_defined()) {
+ $Epoll = eval { epoll_create(1024); };
+ $HaveEpoll = defined $Epoll && $Epoll >= 0;
+ if ($HaveEpoll) {
+ *EventLoop = *EpollEventLoop;
+ }
+ }
+
+ if (!$HaveEpoll && !$HaveKQueue) {
+ require IO::Poll;
+ *EventLoop = *PollEventLoop;
+ }
+}
+
+=head2 C<< CLASS->EventLoop() >>
+
+Start processing IO events. In most daemon programs this never exits. See
+C<PostLoopCallback> below for how to exit the loop.
+
+=cut
+sub FirstTimeEventLoop {
+ my $class = shift;
+
+ _InitPoller();
+
+ if ($HaveEpoll) {
+ EpollEventLoop($class);
+ } elsif ($HaveKQueue) {
+ KQueueEventLoop($class);
+ } else {
+ PollEventLoop($class);
+ }
+}
+
+## profiling-related data/functions
+our ($Prof_utime0, $Prof_stime0);
+sub _pre_profile {
+ ($Prof_utime0, $Prof_stime0) = getrusage();
+}
+
+sub _post_profile {
+ # get post information
+ my ($autime, $astime) = getrusage();
+
+ # calculate differences
+ my $utime = $autime - $Prof_utime0;
+ my $stime = $astime - $Prof_stime0;
+
+ foreach my $k (@_) {
+ $Profiling{$k} ||= [ 0.0, 0.0, 0 ];
+ $Profiling{$k}->[0] += $utime;
+ $Profiling{$k}->[1] += $stime;
+ $Profiling{$k}->[2]++;
+ }
+}
+
+# runs timers and returns milliseconds for next one, or next event loop
+sub RunTimers {
+ return $LoopTimeout unless @Timers;
+
+ my $now = Time::HiRes::time();
+
+ # Run expired timers
+ while (@Timers && $Timers[0][0] <= $now) {
+ my $to_run = shift(@Timers);
+ $to_run->[1]->($now) if $to_run->[1];
+ }
+
+ return $LoopTimeout unless @Timers;
+
+ # convert time to an even number of milliseconds, adding 1
+ # extra, otherwise floating point fun can occur and we'll
+ # call RunTimers like 20-30 times, each returning a timeout
+ # of 0.0000212 seconds
+ my $timeout = int(($Timers[0][0] - $now) * 1000) + 1;
+
+ # -1 is an infinite timeout, so prefer a real timeout
+ return $timeout if $LoopTimeout == -1;
+
+ # otherwise pick the lower of our regular timeout and time until
+ # the next timer
+ return $LoopTimeout if $LoopTimeout < $timeout;
+ return $timeout;
+}
+
+### The epoll-based event loop. Gets installed as EventLoop if IO::Epoll loads
+### okay.
+sub EpollEventLoop {
+ my $class = shift;
+
+ foreach my $fd ( keys %OtherFds ) {
+ if (epoll_ctl($Epoll, EPOLL_CTL_ADD, $fd, EPOLLIN) == -1) {
+ warn "epoll_ctl(): failure adding fd=$fd; $! (", $!+0, ")\n";
+ }
+ }
+
+ while (1) {
+ my @events;
+ my $i;
+ my $timeout = RunTimers();
+
+ # get up to 1000 events
+ my $evcount = epoll_wait($Epoll, 1000, $timeout, \@events);
+ EVENT:
+ for ($i=0; $i<$evcount; $i++) {
+ my $ev = $events[$i];
+
+ # it's possible epoll_wait returned many events, including some at the end
+ # 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.
+ my PublicInbox::DS $pob = $DescriptorMap{$ev->[0]};
+ my $code;
+ my $state = $ev->[1];
+
+ # if we didn't find a Perlbal::Socket subclass for that fd, try other
+ # pseudo-registered (above) fds.
+ if (! $pob) {
+ if (my $code = $OtherFds{$ev->[0]}) {
+ $code->($state);
+ } else {
+ my $fd = $ev->[0];
+ warn "epoll() returned fd $fd w/ state $state for which we have no mapping. removing.\n";
+ POSIX::close($fd);
+ epoll_ctl($Epoll, EPOLL_CTL_DEL, $fd, 0);
+ }
+ next;
+ }
+
+ DebugLevel >= 1 && $class->DebugMsg("Event: fd=%d (%s), state=%d \@ %s\n",
+ $ev->[0], ref($pob), $ev->[1], time);
+
+ if ($DoProfile) {
+ my $class = ref $pob;
+
+ # call profiling action on things that need to be done
+ if ($state & EPOLLIN && ! $pob->{closed}) {
+ _pre_profile();
+ $pob->event_read;
+ _post_profile("$class-read");
+ }
+
+ if ($state & EPOLLOUT && ! $pob->{closed}) {
+ _pre_profile();
+ $pob->event_write;
+ _post_profile("$class-write");
+ }
+
+ if ($state & (EPOLLERR|EPOLLHUP)) {
+ if ($state & EPOLLERR && ! $pob->{closed}) {
+ _pre_profile();
+ $pob->event_err;
+ _post_profile("$class-err");
+ }
+ if ($state & EPOLLHUP && ! $pob->{closed}) {
+ _pre_profile();
+ $pob->event_hup;
+ _post_profile("$class-hup");
+ }
+ }
+
+ next;
+ }
+
+ # standard non-profiling codepat
+ $pob->event_read if $state & EPOLLIN && ! $pob->{closed};
+ $pob->event_write if $state & EPOLLOUT && ! $pob->{closed};
+ if ($state & (EPOLLERR|EPOLLHUP)) {
+ $pob->event_err if $state & EPOLLERR && ! $pob->{closed};
+ $pob->event_hup if $state & EPOLLHUP && ! $pob->{closed};
+ }
+ }
+ return unless PostEventLoop();
+ }
+ exit 0;
+}
+
+### The fallback IO::Poll-based event loop. Gets installed as EventLoop if
+### IO::Epoll fails to load.
+sub PollEventLoop {
+ my $class = shift;
+
+ my PublicInbox::DS $pob;
+
+ while (1) {
+ my $timeout = RunTimers();
+
+ # the following sets up @poll as a series of ($poll,$event_mask)
+ # items, then uses IO::Poll::_poll, implemented in XS, which
+ # modifies the array in place with the even elements being
+ # replaced with the event masks that occured.
+ my @poll;
+ foreach my $fd ( keys %OtherFds ) {
+ push @poll, $fd, POLLIN;
+ }
+ while ( my ($fd, $sock) = each %DescriptorMap ) {
+ push @poll, $fd, $sock->{event_watch};
+ }
+
+ # if nothing to poll, either end immediately (if no timeout)
+ # or just keep calling the callback
+ unless (@poll) {
+ select undef, undef, undef, ($timeout / 1000);
+ return unless PostEventLoop();
+ next;
+ }
+
+ my $count = IO::Poll::_poll($timeout, @poll);
+ unless ($count) {
+ return unless PostEventLoop();
+ next;
+ }
+
+ # Fetch handles with read events
+ while (@poll) {
+ my ($fd, $state) = splice(@poll, 0, 2);
+ next unless $state;
+
+ $pob = $DescriptorMap{$fd};
+
+ if (!$pob) {
+ if (my $code = $OtherFds{$fd}) {
+ $code->($state);
+ }
+ next;
+ }
+
+ $pob->event_read if $state & POLLIN && ! $pob->{closed};
+ $pob->event_write if $state & POLLOUT && ! $pob->{closed};
+ $pob->event_err if $state & POLLERR && ! $pob->{closed};
+ $pob->event_hup if $state & POLLHUP && ! $pob->{closed};
+ }
+
+ return unless PostEventLoop();
+ }
+
+ exit 0;
+}
+
+### The kqueue-based event loop. Gets installed as EventLoop if IO::KQueue works
+### okay.
+sub KQueueEventLoop {
+ my $class = shift;
+
+ foreach my $fd (keys %OtherFds) {
+ $KQueue->EV_SET($fd, IO::KQueue::EVFILT_READ(), IO::KQueue::EV_ADD());
+ }
+
+ while (1) {
+ my $timeout = RunTimers();
+ my @ret = $KQueue->kevent($timeout);
+
+ foreach my $kev (@ret) {
+ my ($fd, $filter, $flags, $fflags) = @$kev;
+ my PublicInbox::DS $pob = $DescriptorMap{$fd};
+ if (!$pob) {
+ if (my $code = $OtherFds{$fd}) {
+ $code->($filter);
+ } else {
+ warn "kevent() returned fd $fd for which we have no mapping. removing.\n";
+ POSIX::close($fd); # close deletes the kevent entry
+ }
+ next;
+ }
+
+ DebugLevel >= 1 && $class->DebugMsg("Event: fd=%d (%s), flags=%d \@ %s\n",
+ $fd, ref($pob), $flags, time);
+
+ $pob->event_read if $filter == IO::KQueue::EVFILT_READ() && !$pob->{closed};
+ $pob->event_write if $filter == IO::KQueue::EVFILT_WRITE() && !$pob->{closed};
+ if ($flags == IO::KQueue::EV_EOF() && !$pob->{closed}) {
+ if ($fflags) {
+ $pob->event_err;
+ } else {
+ $pob->event_hup;
+ }
+ }
+ }
+ return unless PostEventLoop();
+ }
+
+ exit(0);
+}
+
+=head2 C<< CLASS->SetPostLoopCallback( CODEREF ) >>
+
+Sets post loop callback function. Pass a subref and it will be
+called every time the event loop finishes.
+
+Return 1 (or any true value) from the sub to make the loop continue, 0 or false
+and it will exit.
+
+The callback function will be passed two parameters: \%DescriptorMap, \%OtherFds.
+
+=cut
+sub SetPostLoopCallback {
+ my ($class, $ref) = @_;
+
+ if (ref $class) {
+ # per-object callback
+ my PublicInbox::DS $self = $class;
+ if (defined $ref && ref $ref eq 'CODE') {
+ $PLCMap{$self->{fd}} = $ref;
+ } else {
+ delete $PLCMap{$self->{fd}};
+ }
+ } else {
+ # global callback
+ $PostLoopCallback = (defined $ref && ref $ref eq 'CODE') ? $ref : undef;
+ }
+}
+
+# Internal function: run the post-event callback, send read events
+# for pushed-back data, and close pending connections. returns 1
+# if event loop should continue, or 0 to shut it all down.
+sub PostEventLoop {
+ # fire read events for objects with pushed-back read data
+ my $loop = 1;
+ while ($loop) {
+ $loop = 0;
+ foreach my $fd (keys %PushBackSet) {
+ my PublicInbox::DS $pob = $PushBackSet{$fd};
+
+ # a previous event_read invocation could've closed a
+ # connection that we already evaluated in "keys
+ # %PushBackSet", so skip ones that seem to have
+ # disappeared. this is expected.
+ next unless $pob;
+
+ die "ASSERT: the $pob socket has no read_push_back" unless @{$pob->{read_push_back}};
+ next unless (! $pob->{closed} &&
+ $pob->{event_watch} & POLLIN);
+ $loop = 1;
+ $pob->event_read;
+ }
+ }
+
+ # now we can close sockets that wanted to close during our event processing.
+ # (we didn't want to close them during the loop, as we didn't want fd numbers
+ # being reused and confused during the event loop)
+ while (my $sock = shift @ToClose) {
+ my $fd = fileno($sock);
+
+ # close the socket. (not a PublicInbox::DS close)
+ $sock->close;
+
+ # and now we can finally remove the fd from the map. see
+ # comment above in _cleanup.
+ delete $DescriptorMap{$fd};
+ }
+
+
+ # by default we keep running, unless a postloop callback (either per-object
+ # or global) cancels it
+ my $keep_running = 1;
+
+ # per-object post-loop-callbacks
+ for my $plc (values %PLCMap) {
+ $keep_running &&= $plc->(\%DescriptorMap, \%OtherFds);
+ }
+
+ # now we're at the very end, call callback if defined
+ if (defined $PostLoopCallback) {
+ $keep_running &&= $PostLoopCallback->(\%DescriptorMap, \%OtherFds);
+ }
+
+ return $keep_running;
+}
+
+#####################################################################
+### PublicInbox::DS-the-object code
+#####################################################################
+
+=head2 OBJECT METHODS
+
+=head2 C<< CLASS->new( $socket ) >>
+
+Create a new PublicInbox::DS subclass object for the given I<socket> which will
+react to events on it during the C<EventLoop>.
+
+This is normally (always?) called from your subclass via:
+
+ $class->SUPER::new($socket);
+
+=cut
+sub new {
+ my PublicInbox::DS $self = shift;
+ $self = fields::new($self) unless ref $self;
+
+ my $sock = shift;
+
+ $self->{sock} = $sock;
+ my $fd = fileno($sock);
+
+ Carp::cluck("undef sock and/or fd in PublicInbox::DS->new. sock=" . ($sock || "") . ", fd=" . ($fd || ""))
+ unless $sock && $fd;
+
+ $self->{fd} = $fd;
+ $self->{write_buf} = [];
+ $self->{write_buf_offset} = 0;
+ $self->{write_buf_size} = 0;
+ $self->{closed} = 0;
+ $self->{corked} = 0;
+ $self->{read_push_back} = [];
+
+ $self->{event_watch} = POLLERR|POLLHUP|POLLNVAL;
+
+ _InitPoller();
+
+ if ($HaveEpoll) {
+ epoll_ctl($Epoll, EPOLL_CTL_ADD, $fd, $self->{event_watch})
+ and die "couldn't add epoll watch for $fd\n";
+ }
+ elsif ($HaveKQueue) {
+ # Add them to the queue but disabled for now
+ $KQueue->EV_SET($fd, IO::KQueue::EVFILT_READ(),
+ IO::KQueue::EV_ADD() | IO::KQueue::EV_DISABLE());
+ $KQueue->EV_SET($fd, IO::KQueue::EVFILT_WRITE(),
+ IO::KQueue::EV_ADD() | IO::KQueue::EV_DISABLE());
+ }
+
+ Carp::cluck("PublicInbox::DS::new blowing away existing descriptor map for fd=$fd ($DescriptorMap{$fd})")
+ if $DescriptorMap{$fd};
+
+ $DescriptorMap{$fd} = $self;
+ return $self;
+}
+
+
+#####################################################################
+### I N S T A N C E M E T H O D S
+#####################################################################
+
+=head2 C<< $obj->tcp_cork( $boolean ) >>
+
+Turn TCP_CORK on or off depending on the value of I<boolean>.
+
+=cut
+sub tcp_cork {
+ my PublicInbox::DS $self = $_[0];
+ my $val = $_[1];
+
+ # make sure we have a socket
+ return unless $self->{sock};
+ return if $val == $self->{corked};
+
+ my $rv;
+ if (TCP_CORK) {
+ $rv = setsockopt($self->{sock}, IPPROTO_TCP, TCP_CORK,
+ pack("l", $val ? 1 : 0));
+ } else {
+ # FIXME: implement freebsd *PUSH sockopts
+ $rv = 1;
+ }
+
+ # if we failed, close (if we're not already) and warn about the error
+ if ($rv) {
+ $self->{corked} = $val;
+ } else {
+ if ($! == EBADF || $! == ENOTSOCK) {
+ # internal state is probably corrupted; warn and then close if
+ # we're not closed already
+ warn "setsockopt: $!";
+ $self->close('tcp_cork_failed');
+ } elsif ($! == ENOPROTOOPT || $!{ENOTSOCK} || $!{EOPNOTSUPP}) {
+ # TCP implementation doesn't support corking, so just ignore it
+ # or we're trying to tcp-cork a non-socket (like a socketpair pipe
+ # which is acting like a socket, which Perlbal does for child
+ # processes acting like inetd-like web servers)
+ } else {
+ # some other error; we should never hit here, but if we do, die
+ die "setsockopt: $!";
+ }
+ }
+}
+
+=head2 C<< $obj->steal_socket() >>
+
+Basically returns our socket and makes it so that we don't try to close it,
+but we do remove it from epoll handlers. THIS CLOSES $self. It is the same
+thing as calling close, except it gives you the socket to use.
+
+=cut
+sub steal_socket {
+ my PublicInbox::DS $self = $_[0];
+ return if $self->{closed};
+
+ # cleanup does most of the work of closing this socket
+ $self->_cleanup();
+
+ # now undef our internal sock and fd structures so we don't use them
+ my $sock = $self->{sock};
+ $self->{sock} = undef;
+ return $sock;
+}
+
+=head2 C<< $obj->close( [$reason] ) >>
+
+Close the socket. The I<reason> argument will be used in debugging messages.
+
+=cut
+sub close {
+ my PublicInbox::DS $self = $_[0];
+ return if $self->{closed};
+
+ # print out debugging info for this close
+ if (DebugLevel) {
+ my ($pkg, $filename, $line) = caller;
+ my $reason = $_[1] || "";
+ warn "Closing \#$self->{fd} due to $pkg/$filename/$line ($reason)\n";
+ }
+
+ # this does most of the work of closing us
+ $self->_cleanup();
+
+ # defer closing the actual socket until the event loop is done
+ # processing this round of events. (otherwise we might reuse fds)
+ if ($self->{sock}) {
+ push @ToClose, $self->{sock};
+ $self->{sock} = undef;
+ }
+
+ return 0;
+}
+
+### METHOD: _cleanup()
+### Called by our closers so we can clean internal data structures.
+sub _cleanup {
+ my PublicInbox::DS $self = $_[0];
+
+ # we're effectively closed; we have no fd and sock when we leave here
+ $self->{closed} = 1;
+
+ # we need to flush our write buffer, as there may
+ # be self-referential closures (sub { $client->close })
+ # preventing the object from being destroyed
+ $self->{write_buf} = [];
+
+ # uncork so any final data gets sent. only matters if the person closing
+ # us forgot to do it, but we do it to be safe.
+ $self->tcp_cork(0);
+
+ # if we're using epoll, we have to remove this from our epoll fd so we stop getting
+ # notifications about it
+ if ($HaveEpoll && $self->{fd}) {
+ if (epoll_ctl($Epoll, EPOLL_CTL_DEL, $self->{fd}, $self->{event_watch}) != 0) {
+ # dump_error prints a backtrace so we can try to figure out why this happened
+ $self->dump_error("epoll_ctl(): failure deleting fd=$self->{fd} during _cleanup(); $! (" . ($!+0) . ")");
+ }
+ }
+
+ # now delete from mappings. this fd no longer belongs to us, so we don't want
+ # to get alerts for it if it becomes writable/readable/etc.
+ delete $PushBackSet{$self->{fd}};
+ delete $PLCMap{$self->{fd}};
+
+ # we explicitly don't delete from DescriptorMap here until we
+ # actually close the socket, as we might be in the middle of
+ # processing an epoll_wait/etc that returned hundreds of fds, one
+ # of which is not yet processed and is what we're closing. if we
+ # keep it in DescriptorMap, then the event harnesses can just
+ # looked at $pob->{closed} and ignore it. but if it's an
+ # un-accounted for fd, then it (understandably) freak out a bit
+ # and emit warnings, thinking their state got off.
+
+ # and finally get rid of our fd so we can't use it anywhere else
+ $self->{fd} = undef;
+}
+
+=head2 C<< $obj->sock() >>
+
+Returns the underlying IO::Handle for the object.
+
+=cut
+sub sock {
+ my PublicInbox::DS $self = shift;
+ return $self->{sock};
+}
+
+=head2 C<< $obj->set_writer_func( CODEREF ) >>
+
+Sets a function to use instead of C<syswrite()> when writing data to the socket.
+
+=cut
+sub set_writer_func {
+ my PublicInbox::DS $self = shift;
+ my $wtr = shift;
+ Carp::croak("Not a subref") unless !defined $wtr || UNIVERSAL::isa($wtr, "CODE");
+ $self->{writer_func} = $wtr;
+}
+
+=head2 C<< $obj->write( $data ) >>
+
+Write the specified data to the underlying handle. I<data> may be scalar,
+scalar ref, code ref (to run when there), or undef just to kick-start.
+Returns 1 if writes all went through, or 0 if there are writes in queue. If
+it returns 1, caller should stop waiting for 'writable' events)
+
+=cut
+sub write {
+ my PublicInbox::DS $self;
+ my $data;
+ ($self, $data) = @_;
+
+ # nobody should be writing to closed sockets, but caller code can
+ # do two writes within an event, have the first fail and
+ # disconnect the other side (whose destructor then closes the
+ # calling object, but it's still in a method), and then the
+ # now-dead object does its second write. that is this case. we
+ # just lie and say it worked. it'll be dead soon and won't be
+ # hurt by this lie.
+ return 1 if $self->{closed};
+
+ my $bref;
+
+ # just queue data if there's already a wait
+ my $need_queue;
+
+ if (defined $data) {
+ $bref = ref $data ? $data : \$data;
+ if ($self->{write_buf_size}) {
+ push @{$self->{write_buf}}, $bref;
+ $self->{write_buf_size} += ref $bref eq "SCALAR" ? length($$bref) : 1;
+ return 0;
+ }
+
+ # this flag says we're bypassing the queue system, knowing we're the
+ # only outstanding write, and hoping we don't ever need to use it.
+ # if so later, though, we'll need to queue
+ $need_queue = 1;
+ }
+
+ WRITE:
+ while (1) {
+ return 1 unless $bref ||= $self->{write_buf}[0];
+
+ my $len;
+ eval {
+ $len = length($$bref); # this will die if $bref is a code ref, caught below
+ };
+ if ($@) {
+ if (UNIVERSAL::isa($bref, "CODE")) {
+ unless ($need_queue) {
+ $self->{write_buf_size}--; # code refs are worth 1
+ shift @{$self->{write_buf}};
+ }
+ $bref->();
+
+ # code refs are just run and never get reenqueued
+ # (they're one-shot), so turn off the flag indicating the
+ # outstanding data needs queueing.
+ $need_queue = 0;
+
+ undef $bref;
+ next WRITE;
+ }
+ die "Write error: $@ <$bref>";
+ }
+
+ my $to_write = $len - $self->{write_buf_offset};
+ my $written;
+ if (my $wtr = $self->{writer_func}) {
+ $written = $wtr->($bref, $to_write, $self->{write_buf_offset});
+ } else {
+ $written = syswrite($self->{sock}, $$bref, $to_write, $self->{write_buf_offset});
+ }
+
+ if (! defined $written) {
+ if ($! == EPIPE) {
+ return $self->close("EPIPE");
+ } elsif ($! == EAGAIN) {
+ # since connection has stuff to write, it should now be
+ # interested in pending writes:
+ if ($need_queue) {
+ push @{$self->{write_buf}}, $bref;
+ $self->{write_buf_size} += $len;
+ }
+ $self->{write_set_watch} = 1 unless $self->{event_watch} & POLLOUT;
+ $self->watch_write(1);
+ return 0;
+ } elsif ($! == ECONNRESET) {
+ return $self->close("ECONNRESET");
+ }
+
+ DebugLevel >= 1 && $self->debugmsg("Closing connection ($self) due to write error: $!\n");
+
+ return $self->close("write_error");
+ } elsif ($written != $to_write) {
+ DebugLevel >= 2 && $self->debugmsg("Wrote PARTIAL %d bytes to %d",
+ $written, $self->{fd});
+ if ($need_queue) {
+ push @{$self->{write_buf}}, $bref;
+ $self->{write_buf_size} += $len;
+ }
+ # since connection has stuff to write, it should now be
+ # interested in pending writes:
+ $self->{write_buf_offset} += $written;
+ $self->{write_buf_size} -= $written;
+ $self->on_incomplete_write;
+ return 0;
+ } elsif ($written == $to_write) {
+ DebugLevel >= 2 && $self->debugmsg("Wrote ALL %d bytes to %d (nq=%d)",
+ $written, $self->{fd}, $need_queue);
+ $self->{write_buf_offset} = 0;
+
+ if ($self->{write_set_watch}) {
+ $self->watch_write(0);
+ $self->{write_set_watch} = 0;
+ }
+
+ # this was our only write, so we can return immediately
+ # since we avoided incrementing the buffer size or
+ # putting it in the buffer. we also know there
+ # can't be anything else to write.
+ return 1 if $need_queue;
+
+ $self->{write_buf_size} -= $written;
+ shift @{$self->{write_buf}};
+ undef $bref;
+ next WRITE;
+ }
+ }
+}
+
+sub on_incomplete_write {
+ my PublicInbox::DS $self = shift;
+ $self->{write_set_watch} = 1 unless $self->{event_watch} & POLLOUT;
+ $self->watch_write(1);
+}
+
+=head2 C<< $obj->push_back_read( $buf ) >>
+
+Push back I<buf> (a scalar or scalarref) into the read stream. Useful if you read
+more than you need to and want to return this data on the next "read".
+
+=cut
+sub push_back_read {
+ my PublicInbox::DS $self = shift;
+ my $buf = shift;
+ push @{$self->{read_push_back}}, ref $buf ? $buf : \$buf;
+ $PushBackSet{$self->{fd}} = $self;
+}
+
+=head2 C<< $obj->read( $bytecount ) >>
+
+Read at most I<bytecount> bytes from the underlying handle; returns scalar
+ref on read, or undef on connection closed.
+
+=cut
+sub read {
+ my PublicInbox::DS $self = shift;
+ return if $self->{closed};
+ my $bytes = shift;
+ my $buf;
+ my $sock = $self->{sock};
+
+ if (@{$self->{read_push_back}}) {
+ $buf = shift @{$self->{read_push_back}};
+ my $len = length($$buf);
+
+ if ($len <= $bytes) {
+ delete $PushBackSet{$self->{fd}} unless @{$self->{read_push_back}};
+ return $buf;
+ } else {
+ # if the pushed back read is too big, we have to split it
+ my $overflow = substr($$buf, $bytes);
+ $buf = substr($$buf, 0, $bytes);
+ unshift @{$self->{read_push_back}}, \$overflow;
+ return \$buf;
+ }
+ }
+
+ # if this is too high, perl quits(!!). reports on mailing lists
+ # don't seem to point to a universal answer. 5MB worked for some,
+ # crashed for others. 1MB works for more people. let's go with 1MB
+ # for now. :/
+ my $req_bytes = $bytes > 1048576 ? 1048576 : $bytes;
+
+ my $res = sysread($sock, $buf, $req_bytes, 0);
+ DebugLevel >= 2 && $self->debugmsg("sysread = %d; \$! = %d", $res, $!);
+
+ if (! $res && $! != EWOULDBLOCK) {
+ # catches 0=conn closed or undef=error
+ DebugLevel >= 2 && $self->debugmsg("Fd \#%d read hit the end of the road.", $self->{fd});
+ return undef;
+ }
+
+ return \$buf;
+}
+
+=head2 (VIRTUAL) C<< $obj->event_read() >>
+
+Readable event handler. Concrete deriviatives of PublicInbox::DS should
+provide an implementation of this. The default implementation will die if
+called.
+
+=cut
+sub event_read { die "Base class event_read called for $_[0]\n"; }
+
+=head2 (VIRTUAL) C<< $obj->event_err() >>
+
+Error event handler. Concrete deriviatives of PublicInbox::DS should
+provide an implementation of this. The default implementation will die if
+called.
+
+=cut
+sub event_err { die "Base class event_err called for $_[0]\n"; }
+
+=head2 (VIRTUAL) C<< $obj->event_hup() >>
+
+'Hangup' event handler. Concrete deriviatives of PublicInbox::DS should
+provide an implementation of this. The default implementation will die if
+called.
+
+=cut
+sub event_hup { die "Base class event_hup called for $_[0]\n"; }
+
+=head2 C<< $obj->event_write() >>
+
+Writable event handler. Concrete deriviatives of PublicInbox::DS may wish to
+provide an implementation of this. The default implementation calls
+C<write()> with an C<undef>.
+
+=cut
+sub event_write {
+ my $self = shift;
+ $self->write(undef);
+}
+
+=head2 C<< $obj->watch_read( $boolean ) >>
+
+Turn 'readable' event notification on or off.
+
+=cut
+sub watch_read {
+ my PublicInbox::DS $self = shift;
+ return if $self->{closed} || !$self->{sock};
+
+ my $val = shift;
+ my $event = $self->{event_watch};
+
+ $event &= ~POLLIN if ! $val;
+ $event |= POLLIN if $val;
+
+ # If it changed, set it
+ if ($event != $self->{event_watch}) {
+ if ($HaveKQueue) {
+ $KQueue->EV_SET($self->{fd}, IO::KQueue::EVFILT_READ(),
+ $val ? IO::KQueue::EV_ENABLE() : IO::KQueue::EV_DISABLE());
+ }
+ elsif ($HaveEpoll) {
+ epoll_ctl($Epoll, EPOLL_CTL_MOD, $self->{fd}, $event)
+ and $self->dump_error("couldn't modify epoll settings for $self->{fd} " .
+ "from $self->{event_watch} -> $event: $! (" . ($!+0) . ")");
+ }
+ $self->{event_watch} = $event;
+ }
+}
+
+=head2 C<< $obj->watch_write( $boolean ) >>
+
+Turn 'writable' event notification on or off.
+
+=cut
+sub watch_write {
+ my PublicInbox::DS $self = shift;
+ return if $self->{closed} || !$self->{sock};
+
+ my $val = shift;
+ my $event = $self->{event_watch};
+
+ $event &= ~POLLOUT if ! $val;
+ $event |= POLLOUT if $val;
+
+ if ($val && caller ne __PACKAGE__) {
+ # A subclass registered interest, it's now responsible for this.
+ $self->{write_set_watch} = 0;
+ }
+
+ # If it changed, set it
+ if ($event != $self->{event_watch}) {
+ if ($HaveKQueue) {
+ $KQueue->EV_SET($self->{fd}, IO::KQueue::EVFILT_WRITE(),
+ $val ? IO::KQueue::EV_ENABLE() : IO::KQueue::EV_DISABLE());
+ }
+ elsif ($HaveEpoll) {
+ epoll_ctl($Epoll, EPOLL_CTL_MOD, $self->{fd}, $event)
+ and $self->dump_error("couldn't modify epoll settings for $self->{fd} " .
+ "from $self->{event_watch} -> $event: $! (" . ($!+0) . ")");
+ }
+ $self->{event_watch} = $event;
+ }
+}
+
+=head2 C<< $obj->dump_error( $message ) >>
+
+Prints to STDERR a backtrace with information about this socket and what lead
+up to the dump_error call.
+
+=cut
+sub dump_error {
+ my $i = 0;
+ my @list;
+ while (my ($file, $line, $sub) = (caller($i++))[1..3]) {
+ push @list, "\t$file:$line called $sub\n";
+ }
+
+ warn "ERROR: $_[1]\n" .
+ "\t$_[0] = " . $_[0]->as_string . "\n" .
+ join('', @list);
+}
+
+=head2 C<< $obj->debugmsg( $format, @args ) >>
+
+Print the debugging message specified by the C<sprintf>-style I<format> and
+I<args>.
+
+=cut
+sub debugmsg {
+ my ( $self, $fmt, @args ) = @_;
+ confess "Not an object" unless ref $self;
+
+ chomp $fmt;
+ printf STDERR ">>> $fmt\n", @args;
+}
+
+
+=head2 C<< $obj->peer_ip_string() >>
+
+Returns the string describing the peer's IP
+
+=cut
+sub peer_ip_string {
+ my PublicInbox::DS $self = shift;
+ return _undef("peer_ip_string undef: no sock") unless $self->{sock};
+ return $self->{peer_ip} if defined $self->{peer_ip};
+
+ my $pn = getpeername($self->{sock});
+ return _undef("peer_ip_string undef: getpeername") unless $pn;
+
+ my ($port, $iaddr) = eval {
+ if (length($pn) >= 28) {
+ return Socket6::unpack_sockaddr_in6($pn);
+ } else {
+ return Socket::sockaddr_in($pn);
+ }
+ };
+
+ if ($@) {
+ $self->{peer_port} = "[Unknown peerport '$@']";
+ return "[Unknown peername '$@']";
+ }
+
+ $self->{peer_port} = $port;
+
+ if (length($iaddr) == 4) {
+ return $self->{peer_ip} = Socket::inet_ntoa($iaddr);
+ } else {
+ $self->{peer_v6} = 1;
+ return $self->{peer_ip} = Socket6::inet_ntop(Socket6::AF_INET6(),
+ $iaddr);
+ }
+}
+
+=head2 C<< $obj->peer_addr_string() >>
+
+Returns the string describing the peer for the socket which underlies this
+object in form "ip:port"
+
+=cut
+sub peer_addr_string {
+ my PublicInbox::DS $self = shift;
+ my $ip = $self->peer_ip_string
+ or return undef;
+ return $self->{peer_v6} ?
+ "[$ip]:$self->{peer_port}" :
+ "$ip:$self->{peer_port}";
+}
+
+=head2 C<< $obj->local_ip_string() >>
+
+Returns the string describing the local IP
+
+=cut
+sub local_ip_string {
+ my PublicInbox::DS $self = shift;
+ return _undef("local_ip_string undef: no sock") unless $self->{sock};
+ return $self->{local_ip} if defined $self->{local_ip};
+
+ my $pn = getsockname($self->{sock});
+ return _undef("local_ip_string undef: getsockname") unless $pn;
+
+ my ($port, $iaddr) = Socket::sockaddr_in($pn);
+ $self->{local_port} = $port;
+
+ return $self->{local_ip} = Socket::inet_ntoa($iaddr);
+}
+
+=head2 C<< $obj->local_addr_string() >>
+
+Returns the string describing the local end of the socket which underlies this
+object in form "ip:port"
+
+=cut
+sub local_addr_string {
+ my PublicInbox::DS $self = shift;
+ my $ip = $self->local_ip_string;
+ return $ip ? "$ip:$self->{local_port}" : undef;
+}
+
+
+=head2 C<< $obj->as_string() >>
+
+Returns a string describing this socket.
+
+=cut
+sub as_string {
+ my PublicInbox::DS $self = shift;
+ my $rw = "(" . ($self->{event_watch} & POLLIN ? 'R' : '') .
+ ($self->{event_watch} & POLLOUT ? 'W' : '') . ")";
+ my $ret = ref($self) . "$rw: " . ($self->{closed} ? "closed" : "open");
+ my $peer = $self->peer_addr_string;
+ if ($peer) {
+ $ret .= " to " . $self->peer_addr_string;
+ }
+ return $ret;
+}
+
+sub _undef {
+ return undef unless $ENV{DS_DEBUG};
+ my $msg = shift || "";
+ warn "PublicInbox::DS: $msg\n";
+ return undef;
+}
+
+package PublicInbox::DS::Timer;
+# [$abs_float_firetime, $coderef];
+sub cancel {
+ $_[0][1] = undef;
+}
+
+1;
+
+=head1 AUTHORS (Danga::Socket)
+
+Brad Fitzpatrick <brad@danga.com> - author
+
+Michael Granger <ged@danga.com> - docs, testing
+
+Mark Smith <junior@danga.com> - contributor, heavy user, testing
+
+Matt Sergeant <matt@sergeant.org> - kqueue support, docs, timers, other bits
diff --git a/lib/PublicInbox/Daemon.pm b/lib/PublicInbox/Daemon.pm
index 48051f4..68ba987 100644
--- a/lib/PublicInbox/Daemon.pm
+++ b/lib/PublicInbox/Daemon.pm
@@ -12,7 +12,7 @@ use Cwd qw/abs_path/;
use Time::HiRes qw(clock_gettime CLOCK_MONOTONIC);
STDOUT->autoflush(1);
STDERR->autoflush(1);
-require Danga::Socket;
+require PublicInbox::DS;
require POSIX;
require PublicInbox::Listener;
require PublicInbox::ParentPipe;
@@ -172,14 +172,14 @@ sub worker_quit {
# killing again terminates immediately:
exit unless @listeners;
- $_->close foreach @listeners; # call Danga::Socket::close
+ $_->close foreach @listeners; # call PublicInbox::DS::close
@listeners = ();
$reason->close if ref($reason) eq 'PublicInbox::ParentPipe';
my $proc_name;
my $warn = 0;
# drop idle connections and try to quit gracefully
- Danga::Socket->SetPostLoopCallback(sub {
+ PublicInbox::DS->SetPostLoopCallback(sub {
my ($dmap, undef) = @_;
my $n = 0;
my $now = clock_gettime(CLOCK_MONOTONIC);
@@ -486,7 +486,7 @@ sub daemon_loop ($$) {
PublicInbox::Listener->new($_, $post_accept)
} @listeners;
PublicInbox::EvCleanup::enable();
- Danga::Socket->EventLoop;
+ PublicInbox::DS->EventLoop;
$parent_pipe = undef;
}
diff --git a/lib/PublicInbox/EvCleanup.pm b/lib/PublicInbox/EvCleanup.pm
index 1a2bdb2..b2f8c08 100644
--- a/lib/PublicInbox/EvCleanup.pm
+++ b/lib/PublicInbox/EvCleanup.pm
@@ -1,11 +1,11 @@
# Copyright (C) 2016-2018 all contributors <meta@public-inbox.org>
# License: AGPL-3.0+ <https://www.gnu.org/licenses/agpl-3.0.txt>
-# event cleanups (currently for Danga::Socket)
+# event cleanups (currently for PublicInbox::DS)
package PublicInbox::EvCleanup;
use strict;
use warnings;
-use base qw(Danga::Socket);
+use base qw(PublicInbox::DS);
use fields qw(rd);
my $ENABLED;
@@ -38,7 +38,7 @@ sub _run_all ($) {
$_->() foreach @$run;
}
-# ensure Danga::Socket::ToClose fires after timers fire
+# ensure PublicInbox::DS::ToClose fires after timers fire
sub _asap_close () { $asapq->[1] ||= _asap_timer() }
sub _run_asap () { _run_all($asapq) }
@@ -52,7 +52,7 @@ sub _run_later () {
_asap_close();
}
-# Called by Danga::Socket
+# Called by PublicInbox::DS
sub event_write {
my ($self) = @_;
$self->watch_write(0);
@@ -74,13 +74,13 @@ sub asap ($) {
sub next_tick ($) {
my ($cb) = @_;
push @{$nextq->[0]}, $cb;
- $nextq->[1] ||= Danga::Socket->AddTimer(0, *_run_next);
+ $nextq->[1] ||= PublicInbox::DS->AddTimer(0, *_run_next);
}
sub later ($) {
my ($cb) = @_;
push @{$laterq->[0]}, $cb;
- $laterq->[1] ||= Danga::Socket->AddTimer(60, *_run_later);
+ $laterq->[1] ||= PublicInbox::DS->AddTimer(60, *_run_later);
}
END {
diff --git a/lib/PublicInbox/GitHTTPBackend.pm b/lib/PublicInbox/GitHTTPBackend.pm
index 57944a0..0941104 100644
--- a/lib/PublicInbox/GitHTTPBackend.pm
+++ b/lib/PublicInbox/GitHTTPBackend.pm
@@ -67,7 +67,7 @@ sub err ($@) {
sub drop_client ($) {
if (my $io = $_[0]->{'psgix.io'}) {
- $io->close; # this is Danga::Socket::close
+ $io->close; # this is PublicInbox::DS::close
}
}
diff --git a/lib/PublicInbox/HTTP.pm b/lib/PublicInbox/HTTP.pm
index e73bd81..11bd241 100644
--- a/lib/PublicInbox/HTTP.pm
+++ b/lib/PublicInbox/HTTP.pm
@@ -10,7 +10,7 @@
package PublicInbox::HTTP;
use strict;
use warnings;
-use base qw(Danga::Socket);
+use base qw(PublicInbox::DS);
use fields qw(httpd env rbuf input_left remote_addr remote_port forward pull);
use bytes (); # only for bytes::length
use Fcntl qw(:seek);
@@ -63,7 +63,7 @@ sub new ($$$) {
$self;
}
-sub event_read { # called by Danga::Socket
+sub event_read { # called by PublicInbox::DS
my ($self) = @_;
return event_read_input($self) if defined $self->{env};
@@ -148,7 +148,7 @@ sub app_dispatch {
sysseek($input, 0, SEEK_SET) or
die "BUG: psgi.input seek failed: $!";
}
- # note: NOT $self->{sock}, we want our close (+ Danga::Socket::close),
+ # note: NOT $self->{sock}, we want our close (+ PublicInbox::DS::close),
# to do proper cleanup:
$env->{'psgix.io'} = $self; # only for ->close
my $res = Plack::Util::run_app($self->{httpd}->{app}, $env);
@@ -256,7 +256,7 @@ sub getline_cb ($$$) {
if ($forward) {
my $buf = eval { $forward->getline };
if (defined $buf) {
- $write->($buf); # may close in Danga::Socket::write
+ $write->($buf); # may close in PublicInbox::DS::write
unless ($self->{closed}) {
my $next = $self->{pull};
if ($self->{write_buf_size}) {
@@ -320,7 +320,7 @@ sub more ($$) {
my $nlen = length($_[1]) - $n;
return 1 if $nlen == 0; # all done!
- # Danga::Socket::write queues the unwritten substring:
+ # PublicInbox::DS::write queues the unwritten substring:
return $self->write(substr($_[1], $n, $nlen));
}
}
@@ -465,7 +465,7 @@ sub quit {
$self->close;
}
-# callbacks for Danga::Socket
+# callbacks for PublicInbox::DS
sub event_hup { $_[0]->close }
sub event_err { $_[0]->close }
diff --git a/lib/PublicInbox/HTTPD/Async.pm b/lib/PublicInbox/HTTPD/Async.pm
index a647f10..dbe8a84 100644
--- a/lib/PublicInbox/HTTPD/Async.pm
+++ b/lib/PublicInbox/HTTPD/Async.pm
@@ -8,7 +8,7 @@
package PublicInbox::HTTPD::Async;
use strict;
use warnings;
-use base qw(Danga::Socket);
+use base qw(PublicInbox::DS);
use fields qw(cb cleanup);
require PublicInbox::EvCleanup;
@@ -45,7 +45,7 @@ sub main_cb ($$$) {
my $r = sysread($self->{sock}, $$bref, 8192);
if ($r) {
$fh->write($$bref);
- unless ($http->{closed}) { # Danga::Socket sets this
+ unless ($http->{closed}) { # PublicInbox::DS sets this
if ($http->{write_buf_size}) {
$self->watch_read(0);
$http->write(restart_read_cb($self));
diff --git a/lib/PublicInbox/Listener.pm b/lib/PublicInbox/Listener.pm
index 52894cb..d1f0d2e 100644
--- a/lib/PublicInbox/Listener.pm
+++ b/lib/PublicInbox/Listener.pm
@@ -5,7 +5,7 @@
package PublicInbox::Listener;
use strict;
use warnings;
-use base 'Danga::Socket';
+use base 'PublicInbox::DS';
use Socket qw(SOL_SOCKET SO_KEEPALIVE IPPROTO_TCP TCP_NODELAY);
use fields qw(post_accept);
require IO::Handle;
diff --git a/lib/PublicInbox/NNTP.pm b/lib/PublicInbox/NNTP.pm
index 13591e5..f756e92 100644
--- a/lib/PublicInbox/NNTP.pm
+++ b/lib/PublicInbox/NNTP.pm
@@ -5,7 +5,7 @@
package PublicInbox::NNTP;
use strict;
use warnings;
-use base qw(Danga::Socket);
+use base qw(PublicInbox::DS);
use fields qw(nntpd article rbuf ng long_res);
use PublicInbox::Search;
use PublicInbox::Msgmap;
@@ -936,7 +936,7 @@ sub do_more ($$) {
do_write($self, $data);
}
-# callbacks for Danga::Socket
+# callbacks for PublicInbox::DS
sub event_hup { $_[0]->close }
sub event_err { $_[0]->close }
@@ -989,7 +989,7 @@ sub check_read {
} else {
# no pipelined requests available, let the kernel know
# to wake us up if there's more
- $self->watch_read(1); # Danga::Socket::watch_read
+ $self->watch_read(1); # PublicInbox::DS::watch_read
}
}
diff --git a/lib/PublicInbox/ParentPipe.pm b/lib/PublicInbox/ParentPipe.pm
index 4f7ee15..25f13a8 100644
--- a/lib/PublicInbox/ParentPipe.pm
+++ b/lib/PublicInbox/ParentPipe.pm
@@ -4,7 +4,7 @@
package PublicInbox::ParentPipe;
use strict;
use warnings;
-use base qw(Danga::Socket);
+use base qw(PublicInbox::DS);
use fields qw(cb);
sub new ($$$) {
diff --git a/lib/PublicInbox/Qspawn.pm b/lib/PublicInbox/Qspawn.pm
index 79cdae7..9aede10 100644
--- a/lib/PublicInbox/Qspawn.pm
+++ b/lib/PublicInbox/Qspawn.pm
@@ -12,9 +12,9 @@
# operate in. This can be useful to ensure smaller inboxes can
# be cloned while cloning of large inboxes is maxed out.
#
-# This does not depend on Danga::Socket or any other external
+# This does not depend on PublicInbox::DS or any other external
# scheduling mechanism, you just need to call start() and finish()
-# appropriately. However, public-inbox-httpd (which uses Danga::Socket)
+# appropriately. However, public-inbox-httpd (which uses PublicInbox::DS)
# will be able to schedule this based on readability of stdout from
# the spawned process. See GitHTTPBackend.pm and SolverGit.pm for
# usage examples. It does not depend on any form of threading.
diff --git a/lib/PublicInbox/Syscall.pm b/lib/PublicInbox/Syscall.pm
new file mode 100644
index 0000000..cf70045
--- /dev/null
+++ b/lib/PublicInbox/Syscall.pm
@@ -0,0 +1,326 @@
+# This is a fork of the (for now) unmaintained Sys::Syscall 0.25,
+# specifically the Debian libsys-syscall-perl 0.25-6 version to
+# fix upstream regressions in 0.25.
+#
+# This license differs from the rest of public-inbox
+#
+# This module is Copyright (c) 2005 Six Apart, Ltd.
+# Copyright (C) 2019 all contributors <meta@public-inbox.org>
+#
+# All rights reserved.
+#
+# You may distribute under the terms of either the GNU General Public
+# License or the Artistic License, as specified in the Perl README file.
+package PublicInbox::Syscall;
+use strict;
+use POSIX qw(ENOSYS SEEK_CUR);
+use Config;
+
+require Exporter;
+use vars qw(@ISA @EXPORT_OK %EXPORT_TAGS $VERSION);
+
+$VERSION = "0.25";
+@ISA = qw(Exporter);
+@EXPORT_OK = qw(sendfile epoll_ctl epoll_create epoll_wait
+ EPOLLIN EPOLLOUT EPOLLERR EPOLLHUP EPOLLRDBAND
+ EPOLL_CTL_ADD EPOLL_CTL_DEL EPOLL_CTL_MOD);
+%EXPORT_TAGS = (epoll => [qw(epoll_ctl epoll_create epoll_wait
+ EPOLLIN EPOLLOUT EPOLLERR EPOLLHUP EPOLLRDBAND
+ EPOLL_CTL_ADD EPOLL_CTL_DEL EPOLL_CTL_MOD)],
+ sendfile => [qw(sendfile)],
+ );
+
+use constant EPOLLIN => 1;
+use constant EPOLLOUT => 4;
+use constant EPOLLERR => 8;
+use constant EPOLLHUP => 16;
+use constant EPOLLRDBAND => 128;
+use constant EPOLL_CTL_ADD => 1;
+use constant EPOLL_CTL_DEL => 2;
+use constant EPOLL_CTL_MOD => 3;
+
+our $loaded_syscall = 0;
+
+sub _load_syscall {
+ # props to Gaal for this!
+ return if $loaded_syscall++;
+ my $clean = sub {
+ delete @INC{qw<syscall.ph asm/unistd.ph bits/syscall.ph
+ _h2ph_pre.ph sys/syscall.ph>};
+ };
+ $clean->(); # don't trust modules before us
+ my $rv = eval { require 'syscall.ph'; 1 } || eval { require 'sys/syscall.ph'; 1 };
+ $clean->(); # don't require modules after us trust us
+ return $rv;
+}
+
+our ($sysname, $nodename, $release, $version, $machine) = POSIX::uname();
+
+our (
+ $SYS_epoll_create,
+ $SYS_epoll_ctl,
+ $SYS_epoll_wait,
+ $SYS_sendfile,
+ $SYS_readahead,
+ );
+
+our $no_deprecated = 0;
+
+if ($^O eq "linux") {
+ # whether the machine requires 64-bit numbers to be on 8-byte
+ # boundaries.
+ my $u64_mod_8 = 0;
+
+ # if we're running on an x86_64 kernel, but a 32-bit process,
+ # we need to use the i386 syscall numbers.
+ if ($machine eq "x86_64" && $Config{ptrsize} == 4) {
+ $machine = "i386";
+ }
+
+ # Similarly for mips64 vs mips
+ if ($machine eq "mips64" && $Config{ptrsize} == 4) {
+ $machine = "mips";
+ }
+
+ if ($machine =~ m/^i[3456]86$/) {
+ $SYS_epoll_create = 254;
+ $SYS_epoll_ctl = 255;
+ $SYS_epoll_wait = 256;
+ $SYS_sendfile = 187; # or 64: 239
+ $SYS_readahead = 225;
+ } elsif ($machine eq "x86_64") {
+ $SYS_epoll_create = 213;
+ $SYS_epoll_ctl = 233;
+ $SYS_epoll_wait = 232;
+ $SYS_sendfile = 40;
+ $SYS_readahead = 187;
+ } elsif ($machine =~ m/^parisc/) {
+ $SYS_epoll_create = 224;
+ $SYS_epoll_ctl = 225;
+ $SYS_epoll_wait = 226;
+ $SYS_sendfile = 122; # sys_sendfile64=209
+ $SYS_readahead = 207;
+ $u64_mod_8 = 1;
+ } elsif ($machine =~ m/^ppc64/) {
+ $SYS_epoll_create = 236;
+ $SYS_epoll_ctl = 237;
+ $SYS_epoll_wait = 238;
+ $SYS_sendfile = 186; # (sys32_sendfile). sys32_sendfile64=226 (64 bit processes: sys_sendfile64=186)
+ $SYS_readahead = 191; # both 32-bit and 64-bit vesions
+ $u64_mod_8 = 1;
+ } elsif ($machine eq "ppc") {
+ $SYS_epoll_create = 236;
+ $SYS_epoll_ctl = 237;
+ $SYS_epoll_wait = 238;
+ $SYS_sendfile = 186; # sys_sendfile64=226
+ $SYS_readahead = 191;
+ $u64_mod_8 = 1;
+ } elsif ($machine =~ m/^s390/) {
+ $SYS_epoll_create = 249;
+ $SYS_epoll_ctl = 250;
+ $SYS_epoll_wait = 251;
+ $SYS_sendfile = 187; # sys_sendfile64=223
+ $SYS_readahead = 222;
+ $u64_mod_8 = 1;
+ } elsif ($machine eq "ia64") {
+ $SYS_epoll_create = 1243;
+ $SYS_epoll_ctl = 1244;
+ $SYS_epoll_wait = 1245;
+ $SYS_sendfile = 1187;
+ $SYS_readahead = 1216;
+ $u64_mod_8 = 1;
+ } elsif ($machine eq "alpha") {
+ # natural alignment, ints are 32-bits
+ $SYS_sendfile = 370; # (sys_sendfile64)
+ $SYS_epoll_create = 407;
+ $SYS_epoll_ctl = 408;
+ $SYS_epoll_wait = 409;
+ $SYS_readahead = 379;
+ $u64_mod_8 = 1;
+ } elsif ($machine eq "aarch64") {
+ $SYS_epoll_create = 20; # (sys_epoll_create1)
+ $SYS_epoll_ctl = 21;
+ $SYS_epoll_wait = 22; # (sys_epoll_pwait)
+ $SYS_sendfile = 71; # (sys_sendfile64)
+ $SYS_readahead = 213;
+ $u64_mod_8 = 1;
+ $no_deprecated = 1;
+ } elsif ($machine =~ m/arm(v\d+)?.*l/) {
+ # ARM OABI
+ $SYS_epoll_create = 250;
+ $SYS_epoll_ctl = 251;
+ $SYS_epoll_wait = 252;
+ $SYS_sendfile = 187;
+ $SYS_readahead = 225;
+ $u64_mod_8 = 1;
+ } elsif ($machine =~ m/^mips64/) {
+ $SYS_sendfile = 5039;
+ $SYS_epoll_create = 5207;
+ $SYS_epoll_ctl = 5208;
+ $SYS_epoll_wait = 5209;
+ $SYS_readahead = 5179;
+ $u64_mod_8 = 1;
+ } elsif ($machine =~ m/^mips/) {
+ $SYS_sendfile = 4207;
+ $SYS_epoll_create = 4248;
+ $SYS_epoll_ctl = 4249;
+ $SYS_epoll_wait = 4250;
+ $SYS_readahead = 4223;
+ $u64_mod_8 = 1;
+ } else {
+ # as a last resort, try using the *.ph files which may not
+ # exist or may be wrong
+ _load_syscall();
+ $SYS_epoll_create = eval { &SYS_epoll_create; } || 0;
+ $SYS_epoll_ctl = eval { &SYS_epoll_ctl; } || 0;
+ $SYS_epoll_wait = eval { &SYS_epoll_wait; } || 0;
+ $SYS_readahead = eval { &SYS_readahead; } || 0;
+ }
+
+ if ($u64_mod_8) {
+ *epoll_wait = \&epoll_wait_mod8;
+ *epoll_ctl = \&epoll_ctl_mod8;
+ } else {
+ *epoll_wait = \&epoll_wait_mod4;
+ *epoll_ctl = \&epoll_ctl_mod4;
+ }
+}
+
+elsif ($^O eq "freebsd") {
+ if ($ENV{FREEBSD_SENDFILE}) {
+ # this is still buggy and in development
+ $SYS_sendfile = 393; # old is 336
+ }
+}
+
+############################################################################
+# sendfile functions
+############################################################################
+
+unless ($SYS_sendfile) {
+ _load_syscall();
+ $SYS_sendfile = eval { &SYS_sendfile; } || 0;
+}
+
+sub sendfile_defined { return $SYS_sendfile ? 1 : 0; }
+
+if ($^O eq "linux" && $SYS_sendfile) {
+ *sendfile = \&sendfile_linux;
+} elsif ($^O eq "freebsd" && $SYS_sendfile) {
+ *sendfile = \&sendfile_freebsd;
+} else {
+ *sendfile = \&sendfile_noimpl;
+}
+
+sub sendfile_noimpl {
+ $! = ENOSYS;
+ return -1;
+}
+
+# C: ssize_t sendfile(int out_fd, int in_fd, off_t *offset, size_t count)
+# Perl: sendfile($write_fd, $read_fd, $max_count) --> $actually_sent
+sub sendfile_linux {
+ return syscall(
+ $SYS_sendfile,
+ $_[0] + 0, # fd
+ $_[1] + 0, # fd
+ 0, # don't keep track of offset. callers can lseek and keep track.
+ $_[2] + 0 # count
+ );
+}
+
+sub sendfile_freebsd {
+ my $offset = POSIX::lseek($_[1]+0, 0, SEEK_CUR) + 0;
+ my $ct = $_[2] + 0;
+ my $sbytes_buf = "\0" x 8;
+ my $rv = syscall(
+ $SYS_sendfile,
+ $_[1] + 0, # fd (from)
+ $_[0] + 0, # socket (to)
+ $offset,
+ $ct,
+ 0, # struct sf_hdtr *hdtr
+ $sbytes_buf, # off_t *sbytes
+ 0); # flags
+ return $rv if $rv < 0;
+
+
+ my $set = unpack("L", $sbytes_buf);
+ POSIX::lseek($_[1]+0, SEEK_CUR, $set);
+ return $set;
+}
+
+
+############################################################################
+# epoll functions
+############################################################################
+
+sub epoll_defined { return $SYS_epoll_create ? 1 : 0; }
+
+# ARGS: (size) -- but in modern Linux 2.6, the
+# size doesn't even matter (radix tree now, not hash)
+sub epoll_create {
+ return -1 unless defined $SYS_epoll_create;
+ my $epfd = eval { syscall($SYS_epoll_create, $no_deprecated ? 0 : ($_[0]||100)+0) };
+ return -1 if $@;
+ return $epfd;
+}
+
+# epoll_ctl wrapper
+# ARGS: (epfd, op, fd, events_mask)
+sub epoll_ctl_mod4 {
+ syscall($SYS_epoll_ctl, $_[0]+0, $_[1]+0, $_[2]+0, pack("LLL", $_[3], $_[2], 0));
+}
+sub epoll_ctl_mod8 {
+ syscall($SYS_epoll_ctl, $_[0]+0, $_[1]+0, $_[2]+0, pack("LLLL", $_[3], 0, $_[2], 0));
+}
+
+# epoll_wait wrapper
+# ARGS: (epfd, maxevents, timeout (milliseconds), arrayref)
+# arrayref: values modified to be [$fd, $event]
+our $epoll_wait_events;
+our $epoll_wait_size = 0;
+sub epoll_wait_mod4 {
+ # resize our static buffer if requested size is bigger than we've ever done
+ if ($_[1] > $epoll_wait_size) {
+ $epoll_wait_size = $_[1];
+ $epoll_wait_events = "\0" x 12 x $epoll_wait_size;
+ }
+ my $ct = syscall($SYS_epoll_wait, $_[0]+0, $epoll_wait_events, $_[1]+0, $_[2]+0);
+ for (0..$ct-1) {
+ @{$_[3]->[$_]}[1,0] = unpack("LL", substr($epoll_wait_events, 12*$_, 8));
+ }
+ return $ct;
+}
+
+sub epoll_wait_mod8 {
+ # resize our static buffer if requested size is bigger than we've ever done
+ if ($_[1] > $epoll_wait_size) {
+ $epoll_wait_size = $_[1];
+ $epoll_wait_events = "\0" x 16 x $epoll_wait_size;
+ }
+ my $ct;
+ if ($no_deprecated) {
+ $ct = syscall($SYS_epoll_wait, $_[0]+0, $epoll_wait_events, $_[1]+0, $_[2]+0, undef);
+ } else {
+ $ct = syscall($SYS_epoll_wait, $_[0]+0, $epoll_wait_events, $_[1]+0, $_[2]+0);
+ }
+ for (0..$ct-1) {
+ # 16 byte epoll_event structs, with format:
+ # 4 byte mask [idx 1]
+ # 4 byte padding (we put it into idx 2, useless)
+ # 8 byte data (first 4 bytes are fd, into idx 0)
+ @{$_[3]->[$_]}[1,2,0] = unpack("LLL", substr($epoll_wait_events, 16*$_, 12));
+ }
+ return $ct;
+}
+
+1;
+
+=head1 WARRANTY
+
+This is free software. IT COMES WITHOUT WARRANTY OF ANY KIND.
+
+=head1 AUTHORS
+
+Brad Fitzpatrick <brad@danga.com>
diff --git a/t/git-http-backend.t b/t/git-http-backend.t
index 4e51f2b..b616e82 100644
--- a/t/git-http-backend.t
+++ b/t/git-http-backend.t
@@ -11,7 +11,7 @@ use Cwd qw(getcwd);
my $git_dir = $ENV{GIANT_GIT_DIR};
plan 'skip_all' => 'GIANT_GIT_DIR not defined' unless $git_dir;
-foreach my $mod (qw(Danga::Socket BSD::Resource
+foreach my $mod (qw(PublicInbox::DS BSD::Resource
Plack::Util Plack::Builder
HTTP::Date HTTP::Status Net::HTTP)) {
eval "require $mod";
diff --git a/t/httpd-corner.t b/t/httpd-corner.t
index aa0698d..49c5d1f 100644
--- a/t/httpd-corner.t
+++ b/t/httpd-corner.t
@@ -7,7 +7,7 @@ use warnings;
use Test::More;
use Time::HiRes qw(gettimeofday tv_interval);
-foreach my $mod (qw(Plack::Util Plack::Builder Danga::Socket
+foreach my $mod (qw(Plack::Util Plack::Builder PublicInbox::DS
HTTP::Date HTTP::Status IPC::Run)) {
eval "require $mod";
plan skip_all => "$mod missing for httpd-corner.t" if $@;
diff --git a/t/httpd-unix.t b/t/httpd-unix.t
index 0a93f20..627adfa 100644
--- a/t/httpd-unix.t
+++ b/t/httpd-unix.t
@@ -5,7 +5,7 @@ use strict;
use warnings;
use Test::More;
-foreach my $mod (qw(Plack::Util Plack::Builder Danga::Socket
+foreach my $mod (qw(Plack::Util Plack::Builder PublicInbox::DS
HTTP::Date HTTP::Status)) {
eval "require $mod";
plan skip_all => "$mod missing for httpd-unix.t" if $@;
diff --git a/t/httpd.t b/t/httpd.t
index 44df164..45cbcbf 100644
--- a/t/httpd.t
+++ b/t/httpd.t
@@ -4,7 +4,7 @@ use strict;
use warnings;
use Test::More;
-foreach my $mod (qw(Plack::Util Plack::Builder Danga::Socket
+foreach my $mod (qw(Plack::Util Plack::Builder PublicInbox::DS
HTTP::Date HTTP::Status)) {
eval "require $mod";
plan skip_all => "$mod missing for httpd.t" if $@;
diff --git a/t/nntp.t b/t/nntp.t
index 6df7db8..c39a05f 100644
--- a/t/nntp.t
+++ b/t/nntp.t
@@ -5,7 +5,7 @@ use warnings;
use Test::More;
use Data::Dumper;
-foreach my $mod (qw(DBD::SQLite Search::Xapian Danga::Socket)) {
+foreach my $mod (qw(DBD::SQLite Search::Xapian PublicInbox::DS)) {
eval "require $mod";
plan skip_all => "$mod missing for nntp.t" if $@;
}
diff --git a/t/nntpd.t b/t/nntpd.t
index 6b13f81..ecfd74f 100644
--- a/t/nntpd.t
+++ b/t/nntpd.t
@@ -3,7 +3,7 @@
use strict;
use warnings;
use Test::More;
-foreach my $mod (qw(DBD::SQLite Search::Xapian Danga::Socket)) {
+foreach my $mod (qw(DBD::SQLite Search::Xapian PublicInbox::DS)) {
eval "require $mod";
plan skip_all => "$mod missing for nntpd.t" if $@;
}
diff --git a/t/v2mirror.t b/t/v2mirror.t
index ef9a540..eaf9e61 100644
--- a/t/v2mirror.t
+++ b/t/v2mirror.t
@@ -7,7 +7,7 @@ require './t/common.perl';
require_git(2.6);
# Integration tests for HTTP cloning + mirroring
-foreach my $mod (qw(Plack::Util Plack::Builder Danga::Socket
+foreach my $mod (qw(Plack::Util Plack::Builder PublicInbox::DS
HTTP::Date HTTP::Status Search::Xapian DBD::SQLite
IPC::Run)) {
eval "require $mod";
diff --git a/t/v2writable.t b/t/v2writable.t
index f171417..06b2251 100644
--- a/t/v2writable.t
+++ b/t/v2writable.t
@@ -134,8 +134,8 @@ SKIP: {
use Net::NNTP;
use IO::Socket;
use Socket qw(SO_KEEPALIVE IPPROTO_TCP TCP_NODELAY);
- eval { require Danga::Socket };
- skip "Danga::Socket missing $@", 2 if $@;
+ eval { require PublicInbox::DS };
+ skip "PublicInbox::DS missing $@", 2 if $@;
my $err = "$mainrepo/stderr.log";
my $out = "$mainrepo/stdout.log";
my %opts = (
--
EW
^ permalink raw reply related [flat|nested] 12+ messages in thread
* [PATCH 2/4] listener: use EPOLLEXCLUSIVE for listen sockets
2019-05-05 0:52 [PATCH 0/4] bundle Danga::Socket and Sys::Syscall Eric Wong
2019-05-05 0:52 ` [PATCH 1/4] " Eric Wong
@ 2019-05-05 0:52 ` Eric Wong
2019-05-05 0:52 ` [PATCH 3/4] DS: remove unused fields and functions Eric Wong
` (2 subsequent siblings)
4 siblings, 0 replies; 12+ messages in thread
From: Eric Wong @ 2019-05-05 0:52 UTC (permalink / raw)
To: meta
Since our listen sockets are non-blocking and we may run
multiple httpd|nntpd processes; we need a way to avoid
thundering herds when there are multiple httpd|nntpd worker
processes.
EPOLLEXCLUSIVE was added just for that in Linux 4.5
---
TODO | 3 ---
lib/PublicInbox/DS.pm | 22 ++++++++++++++++------
lib/PublicInbox/Listener.pm | 2 +-
lib/PublicInbox/Syscall.pm | 7 +++++--
4 files changed, 22 insertions(+), 12 deletions(-)
diff --git a/TODO b/TODO
index 372f733..ac255b8 100644
--- a/TODO
+++ b/TODO
@@ -56,9 +56,6 @@ all need to be considered for everything we introduce)
ugh... https://rt.cpan.org/Ticket/Display.html?id=116615
(IO::KQueue is broken with Danga::Socket / PublicInbox::DS)
-* EPOLLEXCLUSIVE for listen socket fairness across -httpd/nntpd
- worker processes.
-
* improve documentation
* linkify thread skeletons better
diff --git a/lib/PublicInbox/DS.pm b/lib/PublicInbox/DS.pm
index 543d3fd..3ccc275 100644
--- a/lib/PublicInbox/DS.pm
+++ b/lib/PublicInbox/DS.pm
@@ -78,6 +78,8 @@ our (
@Timers, # timers
);
+# this may be set to zero with old kernels
+our $EPOLLEXCLUSIVE = EPOLLEXCLUSIVE;
Reset();
#####################################################################
@@ -666,11 +668,9 @@ This is normally (always?) called from your subclass via:
=cut
sub new {
- my PublicInbox::DS $self = shift;
+ my ($self, $sock, $exclusive) = @_;
$self = fields::new($self) unless ref $self;
- my $sock = shift;
-
$self->{sock} = $sock;
my $fd = fileno($sock);
@@ -685,13 +685,23 @@ sub new {
$self->{corked} = 0;
$self->{read_push_back} = [];
- $self->{event_watch} = POLLERR|POLLHUP|POLLNVAL;
+ my $ev = $self->{event_watch} = POLLERR|POLLHUP|POLLNVAL;
_InitPoller();
if ($HaveEpoll) {
- epoll_ctl($Epoll, EPOLL_CTL_ADD, $fd, $self->{event_watch})
- and die "couldn't add epoll watch for $fd\n";
+ if ($exclusive) {
+ $ev = $self->{event_watch} = EPOLLIN|EPOLLERR|EPOLLHUP|$EPOLLEXCLUSIVE;
+ }
+retry:
+ if (epoll_ctl($Epoll, EPOLL_CTL_ADD, $fd, $ev)) {
+ if ($!{EINVAL} && ($ev & $EPOLLEXCLUSIVE)) {
+ $EPOLLEXCLUSIVE = 0; # old kernel
+ $ev = $self->{event_watch} = EPOLLIN|EPOLLERR|EPOLLHUP;
+ goto retry;
+ }
+ die "couldn't add epoll watch for $fd: $!\n";
+ }
}
elsif ($HaveKQueue) {
# Add them to the queue but disabled for now
diff --git a/lib/PublicInbox/Listener.pm b/lib/PublicInbox/Listener.pm
index d1f0d2e..a75a6fd 100644
--- a/lib/PublicInbox/Listener.pm
+++ b/lib/PublicInbox/Listener.pm
@@ -17,7 +17,7 @@ sub new ($$$) {
listen($s, 1024);
IO::Handle::blocking($s, 0);
my $self = fields::new($class);
- $self->SUPER::new($s); # calls epoll_create for the first socket
+ $self->SUPER::new($s, 1); # calls epoll_create for the first socket
$self->watch_read(1);
$self->{post_accept} = $cb;
$self
diff --git a/lib/PublicInbox/Syscall.pm b/lib/PublicInbox/Syscall.pm
index cf70045..9194364 100644
--- a/lib/PublicInbox/Syscall.pm
+++ b/lib/PublicInbox/Syscall.pm
@@ -23,10 +23,12 @@ $VERSION = "0.25";
@ISA = qw(Exporter);
@EXPORT_OK = qw(sendfile epoll_ctl epoll_create epoll_wait
EPOLLIN EPOLLOUT EPOLLERR EPOLLHUP EPOLLRDBAND
- EPOLL_CTL_ADD EPOLL_CTL_DEL EPOLL_CTL_MOD);
+ EPOLL_CTL_ADD EPOLL_CTL_DEL EPOLL_CTL_MOD
+ EPOLLEXCLUSIVE);
%EXPORT_TAGS = (epoll => [qw(epoll_ctl epoll_create epoll_wait
EPOLLIN EPOLLOUT EPOLLERR EPOLLHUP EPOLLRDBAND
- EPOLL_CTL_ADD EPOLL_CTL_DEL EPOLL_CTL_MOD)],
+ EPOLL_CTL_ADD EPOLL_CTL_DEL EPOLL_CTL_MOD
+ EPOLLEXCLUSIVE)],
sendfile => [qw(sendfile)],
);
@@ -35,6 +37,7 @@ use constant EPOLLOUT => 4;
use constant EPOLLERR => 8;
use constant EPOLLHUP => 16;
use constant EPOLLRDBAND => 128;
+use constant EPOLLEXCLUSIVE => (1 << 28);
use constant EPOLL_CTL_ADD => 1;
use constant EPOLL_CTL_DEL => 2;
use constant EPOLL_CTL_MOD => 3;
--
EW
^ permalink raw reply related [flat|nested] 12+ messages in thread
* [PATCH 3/4] DS: remove unused fields and functions
2019-05-05 0:52 [PATCH 0/4] bundle Danga::Socket and Sys::Syscall Eric Wong
2019-05-05 0:52 ` [PATCH 1/4] " Eric Wong
2019-05-05 0:52 ` [PATCH 2/4] listener: use EPOLLEXCLUSIVE for listen sockets Eric Wong
@ 2019-05-05 0:52 ` Eric Wong
2019-05-05 0:52 ` [PATCH 4/4] DS: drop profiling support Eric Wong
2019-05-08 19:18 ` [PATCH 0/4] Danga::Socket bundling cleanups Eric Wong
4 siblings, 0 replies; 12+ messages in thread
From: Eric Wong @ 2019-05-05 0:52 UTC (permalink / raw)
To: meta
More will likely be dropped in the future, but drop the obvious
ones we aren't using, for now; especially since some of them are
set at ->new time and unavoidable.
This saves 579 bytes per-client on my 64-bit Debian stable
system as measured by Devel::Size::total_size from
PublicInbox::HTTP::event_read. This adds up in C10K or C100K
situations.
Things we drop are:
* corked - MSG_MORE requires fewer syscalls
* read_push_back - tried to use it, ate CPU with slow clients
* IP/port fields - accept() already returns what we care about
---
lib/PublicInbox/DS.pm | 199 ------------------------------------------
1 file changed, 199 deletions(-)
diff --git a/lib/PublicInbox/DS.pm b/lib/PublicInbox/DS.pm
index 3ccc275..f181eee 100644
--- a/lib/PublicInbox/DS.pm
+++ b/lib/PublicInbox/DS.pm
@@ -29,15 +29,8 @@ use fields ('sock', # underlying socket
'write_buf_offset', # offset into first array of write_buf to start writing at
'write_buf_size', # total length of data in all write_buf items
'write_set_watch', # bool: true if we internally set watch_write rather than by a subclass
- 'read_push_back', # arrayref of "pushed-back" read data the application didn't want
'closed', # bool: socket is closed
- 'corked', # bool: socket is corked
'event_watch', # bitmask of events the client is interested in (POLLIN,OUT,etc.)
- 'peer_v6', # bool: cached; if peer is an IPv6 address
- 'peer_ip', # cached stringified IP address of $sock
- 'peer_port', # cached port number of $sock
- 'local_ip', # cached stringified IP address of local end of $sock
- 'local_port', # cached port number of local end of $sock
'writer_func', # subref which does writing. must return bytes written (or undef) and set $! on errors
);
@@ -46,7 +39,6 @@ use Errno qw(EINPROGRESS EWOULDBLOCK EISCONN ENOTSOCK
use Socket qw(IPPROTO_TCP);
use Carp qw(croak confess);
-use constant TCP_CORK => ($^O eq "linux" ? 3 : 0); # FIXME: not hard-coded (Linux-specific too)
use constant DebugLevel => 0;
use constant POLLIN => 1;
@@ -61,7 +53,6 @@ our (
$HaveEpoll, # Flag -- is epoll available? initially undefined.
$HaveKQueue,
%DescriptorMap, # fd (num) -> PublicInbox::DS object
- %PushBackSet, # fd (num) -> PublicInbox::DS (fds with pushed back read data)
$Epoll, # Global epoll fd (for epoll mode only)
$KQueue, # Global kqueue fd (for kqueue mode only)
@ToClose, # sockets to close when event loop is done
@@ -93,7 +84,6 @@ Reset all state
=cut
sub Reset {
%DescriptorMap = ();
- %PushBackSet = ();
@ToClose = ();
%OtherFds = ();
$LoopTimeout = -1; # no timeout by default
@@ -598,27 +588,6 @@ sub SetPostLoopCallback {
# for pushed-back data, and close pending connections. returns 1
# if event loop should continue, or 0 to shut it all down.
sub PostEventLoop {
- # fire read events for objects with pushed-back read data
- my $loop = 1;
- while ($loop) {
- $loop = 0;
- foreach my $fd (keys %PushBackSet) {
- my PublicInbox::DS $pob = $PushBackSet{$fd};
-
- # a previous event_read invocation could've closed a
- # connection that we already evaluated in "keys
- # %PushBackSet", so skip ones that seem to have
- # disappeared. this is expected.
- next unless $pob;
-
- die "ASSERT: the $pob socket has no read_push_back" unless @{$pob->{read_push_back}};
- next unless (! $pob->{closed} &&
- $pob->{event_watch} & POLLIN);
- $loop = 1;
- $pob->event_read;
- }
- }
-
# now we can close sockets that wanted to close during our event processing.
# (we didn't want to close them during the loop, as we didn't want fd numbers
# being reused and confused during the event loop)
@@ -682,8 +651,6 @@ sub new {
$self->{write_buf_offset} = 0;
$self->{write_buf_size} = 0;
$self->{closed} = 0;
- $self->{corked} = 0;
- $self->{read_push_back} = [];
my $ev = $self->{event_watch} = POLLERR|POLLHUP|POLLNVAL;
@@ -723,49 +690,6 @@ retry:
### I N S T A N C E M E T H O D S
#####################################################################
-=head2 C<< $obj->tcp_cork( $boolean ) >>
-
-Turn TCP_CORK on or off depending on the value of I<boolean>.
-
-=cut
-sub tcp_cork {
- my PublicInbox::DS $self = $_[0];
- my $val = $_[1];
-
- # make sure we have a socket
- return unless $self->{sock};
- return if $val == $self->{corked};
-
- my $rv;
- if (TCP_CORK) {
- $rv = setsockopt($self->{sock}, IPPROTO_TCP, TCP_CORK,
- pack("l", $val ? 1 : 0));
- } else {
- # FIXME: implement freebsd *PUSH sockopts
- $rv = 1;
- }
-
- # if we failed, close (if we're not already) and warn about the error
- if ($rv) {
- $self->{corked} = $val;
- } else {
- if ($! == EBADF || $! == ENOTSOCK) {
- # internal state is probably corrupted; warn and then close if
- # we're not closed already
- warn "setsockopt: $!";
- $self->close('tcp_cork_failed');
- } elsif ($! == ENOPROTOOPT || $!{ENOTSOCK} || $!{EOPNOTSUPP}) {
- # TCP implementation doesn't support corking, so just ignore it
- # or we're trying to tcp-cork a non-socket (like a socketpair pipe
- # which is acting like a socket, which Perlbal does for child
- # processes acting like inetd-like web servers)
- } else {
- # some other error; we should never hit here, but if we do, die
- die "setsockopt: $!";
- }
- }
-}
-
=head2 C<< $obj->steal_socket() >>
Basically returns our socket and makes it so that we don't try to close it,
@@ -828,10 +752,6 @@ sub _cleanup {
# preventing the object from being destroyed
$self->{write_buf} = [];
- # uncork so any final data gets sent. only matters if the person closing
- # us forgot to do it, but we do it to be safe.
- $self->tcp_cork(0);
-
# if we're using epoll, we have to remove this from our epoll fd so we stop getting
# notifications about it
if ($HaveEpoll && $self->{fd}) {
@@ -843,7 +763,6 @@ sub _cleanup {
# now delete from mappings. this fd no longer belongs to us, so we don't want
# to get alerts for it if it becomes writable/readable/etc.
- delete $PushBackSet{$self->{fd}};
delete $PLCMap{$self->{fd}};
# we explicitly don't delete from DescriptorMap here until we
@@ -1020,19 +939,6 @@ sub on_incomplete_write {
$self->watch_write(1);
}
-=head2 C<< $obj->push_back_read( $buf ) >>
-
-Push back I<buf> (a scalar or scalarref) into the read stream. Useful if you read
-more than you need to and want to return this data on the next "read".
-
-=cut
-sub push_back_read {
- my PublicInbox::DS $self = shift;
- my $buf = shift;
- push @{$self->{read_push_back}}, ref $buf ? $buf : \$buf;
- $PushBackSet{$self->{fd}} = $self;
-}
-
=head2 C<< $obj->read( $bytecount ) >>
Read at most I<bytecount> bytes from the underlying handle; returns scalar
@@ -1046,22 +952,6 @@ sub read {
my $buf;
my $sock = $self->{sock};
- if (@{$self->{read_push_back}}) {
- $buf = shift @{$self->{read_push_back}};
- my $len = length($$buf);
-
- if ($len <= $bytes) {
- delete $PushBackSet{$self->{fd}} unless @{$self->{read_push_back}};
- return $buf;
- } else {
- # if the pushed back read is too big, we have to split it
- my $overflow = substr($$buf, $bytes);
- $buf = substr($$buf, 0, $bytes);
- unshift @{$self->{read_push_back}}, \$overflow;
- return \$buf;
- }
- }
-
# if this is too high, perl quits(!!). reports on mailing lists
# don't seem to point to a universal answer. 5MB worked for some,
# crashed for others. 1MB works for more people. let's go with 1MB
@@ -1216,91 +1106,6 @@ sub debugmsg {
printf STDERR ">>> $fmt\n", @args;
}
-
-=head2 C<< $obj->peer_ip_string() >>
-
-Returns the string describing the peer's IP
-
-=cut
-sub peer_ip_string {
- my PublicInbox::DS $self = shift;
- return _undef("peer_ip_string undef: no sock") unless $self->{sock};
- return $self->{peer_ip} if defined $self->{peer_ip};
-
- my $pn = getpeername($self->{sock});
- return _undef("peer_ip_string undef: getpeername") unless $pn;
-
- my ($port, $iaddr) = eval {
- if (length($pn) >= 28) {
- return Socket6::unpack_sockaddr_in6($pn);
- } else {
- return Socket::sockaddr_in($pn);
- }
- };
-
- if ($@) {
- $self->{peer_port} = "[Unknown peerport '$@']";
- return "[Unknown peername '$@']";
- }
-
- $self->{peer_port} = $port;
-
- if (length($iaddr) == 4) {
- return $self->{peer_ip} = Socket::inet_ntoa($iaddr);
- } else {
- $self->{peer_v6} = 1;
- return $self->{peer_ip} = Socket6::inet_ntop(Socket6::AF_INET6(),
- $iaddr);
- }
-}
-
-=head2 C<< $obj->peer_addr_string() >>
-
-Returns the string describing the peer for the socket which underlies this
-object in form "ip:port"
-
-=cut
-sub peer_addr_string {
- my PublicInbox::DS $self = shift;
- my $ip = $self->peer_ip_string
- or return undef;
- return $self->{peer_v6} ?
- "[$ip]:$self->{peer_port}" :
- "$ip:$self->{peer_port}";
-}
-
-=head2 C<< $obj->local_ip_string() >>
-
-Returns the string describing the local IP
-
-=cut
-sub local_ip_string {
- my PublicInbox::DS $self = shift;
- return _undef("local_ip_string undef: no sock") unless $self->{sock};
- return $self->{local_ip} if defined $self->{local_ip};
-
- my $pn = getsockname($self->{sock});
- return _undef("local_ip_string undef: getsockname") unless $pn;
-
- my ($port, $iaddr) = Socket::sockaddr_in($pn);
- $self->{local_port} = $port;
-
- return $self->{local_ip} = Socket::inet_ntoa($iaddr);
-}
-
-=head2 C<< $obj->local_addr_string() >>
-
-Returns the string describing the local end of the socket which underlies this
-object in form "ip:port"
-
-=cut
-sub local_addr_string {
- my PublicInbox::DS $self = shift;
- my $ip = $self->local_ip_string;
- return $ip ? "$ip:$self->{local_port}" : undef;
-}
-
-
=head2 C<< $obj->as_string() >>
Returns a string describing this socket.
@@ -1311,10 +1116,6 @@ sub as_string {
my $rw = "(" . ($self->{event_watch} & POLLIN ? 'R' : '') .
($self->{event_watch} & POLLOUT ? 'W' : '') . ")";
my $ret = ref($self) . "$rw: " . ($self->{closed} ? "closed" : "open");
- my $peer = $self->peer_addr_string;
- if ($peer) {
- $ret .= " to " . $self->peer_addr_string;
- }
return $ret;
}
--
EW
^ permalink raw reply related [flat|nested] 12+ messages in thread
* [PATCH 4/4] DS: drop profiling support
2019-05-05 0:52 [PATCH 0/4] bundle Danga::Socket and Sys::Syscall Eric Wong
` (2 preceding siblings ...)
2019-05-05 0:52 ` [PATCH 3/4] DS: remove unused fields and functions Eric Wong
@ 2019-05-05 0:52 ` Eric Wong
2019-05-08 19:18 ` [PATCH 0/4] Danga::Socket bundling cleanups Eric Wong
4 siblings, 0 replies; 12+ messages in thread
From: Eric Wong @ 2019-05-05 0:52 UTC (permalink / raw)
To: meta
There's other ways to profile and we don't need to add runtime
branches to do this.
---
lib/PublicInbox/DS.pm | 94 -------------------------------------------
1 file changed, 94 deletions(-)
diff --git a/lib/PublicInbox/DS.pm b/lib/PublicInbox/DS.pm
index f181eee..7bd5d42 100644
--- a/lib/PublicInbox/DS.pm
+++ b/lib/PublicInbox/DS.pm
@@ -13,8 +13,6 @@ use bytes;
use POSIX ();
use Time::HiRes ();
-my $opt_bsd_resource = eval "use BSD::Resource; 1;";
-
use vars qw{$VERSION};
$VERSION = "1.61";
@@ -63,8 +61,6 @@ our (
%PLCMap, # fd (num) -> PostLoopCallback (per-object)
$LoopTimeout, # timeout of event loop in milliseconds
- $DoProfile, # if on, enable profiling
- %Profiling, # what => [ utime, stime, calls ]
$DoneInit, # if we've done the one-time module init yet
@Timers, # timers
);
@@ -87,8 +83,6 @@ sub Reset {
@ToClose = ();
%OtherFds = ();
$LoopTimeout = -1; # no timeout by default
- $DoProfile = 0;
- %Profiling = ();
@Timers = ();
$PostLoopCallback = undef;
@@ -122,40 +116,6 @@ sub WatchedSockets {
}
*watched_sockets = *WatchedSockets;
-=head2 C<< CLASS->EnableProfiling() >>
-
-Turns profiling on, clearing current profiling data.
-
-=cut
-sub EnableProfiling {
- if ($opt_bsd_resource) {
- %Profiling = ();
- $DoProfile = 1;
- return 1;
- }
- return 0;
-}
-
-=head2 C<< CLASS->DisableProfiling() >>
-
-Turns off profiling, but retains data up to this point
-
-=cut
-sub DisableProfiling {
- $DoProfile = 0;
-}
-
-=head2 C<< CLASS->ProfilingData() >>
-
-Returns reference to a hash of data in format:
-
- ITEM => [ utime, stime, #calls ]
-
-=cut
-sub ProfilingData {
- return \%Profiling;
-}
-
=head2 C<< CLASS->ToClose() >>
Return the list of sockets that are awaiting close() at the end of the
@@ -306,28 +266,6 @@ sub FirstTimeEventLoop {
}
}
-## profiling-related data/functions
-our ($Prof_utime0, $Prof_stime0);
-sub _pre_profile {
- ($Prof_utime0, $Prof_stime0) = getrusage();
-}
-
-sub _post_profile {
- # get post information
- my ($autime, $astime) = getrusage();
-
- # calculate differences
- my $utime = $autime - $Prof_utime0;
- my $stime = $astime - $Prof_stime0;
-
- foreach my $k (@_) {
- $Profiling{$k} ||= [ 0.0, 0.0, 0 ];
- $Profiling{$k}->[0] += $utime;
- $Profiling{$k}->[1] += $stime;
- $Profiling{$k}->[2]++;
- }
-}
-
# runs timers and returns milliseconds for next one, or next event loop
sub RunTimers {
return $LoopTimeout unless @Timers;
@@ -404,38 +342,6 @@ sub EpollEventLoop {
DebugLevel >= 1 && $class->DebugMsg("Event: fd=%d (%s), state=%d \@ %s\n",
$ev->[0], ref($pob), $ev->[1], time);
- if ($DoProfile) {
- my $class = ref $pob;
-
- # call profiling action on things that need to be done
- if ($state & EPOLLIN && ! $pob->{closed}) {
- _pre_profile();
- $pob->event_read;
- _post_profile("$class-read");
- }
-
- if ($state & EPOLLOUT && ! $pob->{closed}) {
- _pre_profile();
- $pob->event_write;
- _post_profile("$class-write");
- }
-
- if ($state & (EPOLLERR|EPOLLHUP)) {
- if ($state & EPOLLERR && ! $pob->{closed}) {
- _pre_profile();
- $pob->event_err;
- _post_profile("$class-err");
- }
- if ($state & EPOLLHUP && ! $pob->{closed}) {
- _pre_profile();
- $pob->event_hup;
- _post_profile("$class-hup");
- }
- }
-
- next;
- }
-
# standard non-profiling codepat
$pob->event_read if $state & EPOLLIN && ! $pob->{closed};
$pob->event_write if $state & EPOLLOUT && ! $pob->{closed};
--
EW
^ permalink raw reply related [flat|nested] 12+ messages in thread
* [PATCH 5/4] DS: workaround IO::Kqueue EINTR (mis-)handling
2019-05-05 0:52 ` [PATCH 1/4] " Eric Wong
@ 2019-05-05 4:56 ` Eric Wong
2019-05-08 9:07 ` [PATCH 6/4] DS: handle EINTR in IO::Poll path, too Eric Wong
1 sibling, 0 replies; 12+ messages in thread
From: Eric Wong @ 2019-05-05 4:56 UTC (permalink / raw)
To: meta
IO::Kqueue seems unmaintained, so workaround a long-standing
bug where it falls over on signals:
https://rt.cpan.org/Ticket/Display.html?id=116615
---
TODO | 4 ----
lib/PublicInbox/DS.pm | 10 +++++++++-
2 files changed, 9 insertions(+), 5 deletions(-)
diff --git a/TODO b/TODO
index ac255b8..d947b0f 100644
--- a/TODO
+++ b/TODO
@@ -52,10 +52,6 @@ all need to be considered for everything we introduce)
cf. https://public-inbox.org/git/20160814012706.GA18784@starla/
-* portability to FreeBSD (and other Free Software *BSDs)
- ugh... https://rt.cpan.org/Ticket/Display.html?id=116615
- (IO::KQueue is broken with Danga::Socket / PublicInbox::DS)
-
* improve documentation
* linkify thread skeletons better
diff --git a/lib/PublicInbox/DS.pm b/lib/PublicInbox/DS.pm
index 7bd5d42..ea09fc9 100644
--- a/lib/PublicInbox/DS.pm
+++ b/lib/PublicInbox/DS.pm
@@ -428,7 +428,15 @@ sub KQueueEventLoop {
while (1) {
my $timeout = RunTimers();
- my @ret = $KQueue->kevent($timeout);
+ my @ret = eval { $KQueue->kevent($timeout) };
+ if (my $err = $@) {
+ # workaround https://rt.cpan.org/Ticket/Display.html?id=116615
+ if ($err =~ /Interrupted system call/) {
+ @ret = ();
+ } else {
+ die $err;
+ }
+ }
foreach my $kev (@ret) {
my ($fd, $filter, $flags, $fflags) = @$kev;
--
^ permalink raw reply related [flat|nested] 12+ messages in thread
* [PATCH 6/4] DS: handle EINTR in IO::Poll path, too
2019-05-05 0:52 ` [PATCH 1/4] " Eric Wong
2019-05-05 4:56 ` [PATCH 5/4] DS: workaround IO::Kqueue EINTR (mis-)handling Eric Wong
@ 2019-05-08 9:07 ` Eric Wong
1 sibling, 0 replies; 12+ messages in thread
From: Eric Wong @ 2019-05-08 9:07 UTC (permalink / raw)
To: meta
IO::Poll::_poll returns -1, which is "true" to Perl.
cf. https://rt.cpan.org/Ticket/Display.html?id=129484
---
Not sure if anybody is using platforms w/o epoll or kqueue;
but I encountered this on FreeBSD since I forgot to install
p5-IO-KQueue (IO::Kqueue).
Maybe GNU/Hurd users can benefit :>
lib/PublicInbox/DS.pm | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/lib/PublicInbox/DS.pm b/lib/PublicInbox/DS.pm
index ea09fc9..5dd1bb7 100644
--- a/lib/PublicInbox/DS.pm
+++ b/lib/PublicInbox/DS.pm
@@ -386,7 +386,7 @@ sub PollEventLoop {
}
my $count = IO::Poll::_poll($timeout, @poll);
- unless ($count) {
+ unless ($count >= 0) {
return unless PostEventLoop();
next;
}
--
EW
^ permalink raw reply related [flat|nested] 12+ messages in thread
* [PATCH 0/4] Danga::Socket bundling cleanups
2019-05-05 0:52 [PATCH 0/4] bundle Danga::Socket and Sys::Syscall Eric Wong
` (3 preceding siblings ...)
2019-05-05 0:52 ` [PATCH 4/4] DS: drop profiling support Eric Wong
@ 2019-05-08 19:18 ` Eric Wong
2019-05-08 19:18 ` [PATCH 1/4] build: do not manify DS and Syscall pods Eric Wong
` (3 more replies)
4 siblings, 4 replies; 12+ messages in thread
From: Eric Wong @ 2019-05-08 19:18 UTC (permalink / raw)
To: meta
Dropping some unused stuff, and a bugfix for an error path we never hit.
(all bugfixes are queued for the future maintainer via
bug-Danga-Socket@rt.cpan.org )
Eric Wong (4):
build: do not manify DS and Syscall pods
syscall: drop readahead wrapper
DS: drop unused "_undef" sub
DS: epoll: fix misordered EPOLL_CTL_DEL call
Makefile.PL | 10 ++++++++++
lib/PublicInbox/DS.pm | 9 +--------
lib/PublicInbox/Syscall.pm | 14 --------------
3 files changed, 11 insertions(+), 22 deletions(-)
The "danga-bundle" is up to 10 patches, now; and dogfooded
on public-inbox.org for several days without problems.
Will merge to "master" soon:
bundle Danga::Socket and Sys::Syscall
listener: use EPOLLEXCLUSIVE for listen sockets
DS: remove unused fields and functions
DS: drop profiling support
DS: workaround IO::Kqueue EINTR (mis-)handling
DS: handle EINTR in IO::Poll path, too
build: do not manify DS and Syscall pods
syscall: drop readahead wrapper
DS: drop unused "_undef" sub
DS: epoll: fix misordered EPOLL_CTL_DEL call
^ permalink raw reply [flat|nested] 12+ messages in thread
* [PATCH 1/4] build: do not manify DS and Syscall pods
2019-05-08 19:18 ` [PATCH 0/4] Danga::Socket bundling cleanups Eric Wong
@ 2019-05-08 19:18 ` Eric Wong
2019-05-08 19:18 ` [PATCH 2/4] syscall: drop readahead wrapper Eric Wong
` (2 subsequent siblings)
3 siblings, 0 replies; 12+ messages in thread
From: Eric Wong @ 2019-05-08 19:18 UTC (permalink / raw)
To: meta
We don't need to increase our install footprint with
documentation from our internals (which will surely
change).
---
Makefile.PL | 10 ++++++++++
1 file changed, 10 insertions(+)
diff --git a/Makefile.PL b/Makefile.PL
index e00c015..3bb0072 100644
--- a/Makefile.PL
+++ b/Makefile.PL
@@ -8,6 +8,15 @@ chomp(my @manifest = (<$m>));
my @EXE_FILES = grep(m!^script/!, @manifest);
my $PM_FILES = join(' ', grep(m!^lib/.*\.pm$!, @manifest));
+# Don't waste user's disk space by installing some pods from
+# imported code or internal use only
+my %man3 = map {; # semi-colon tells Perl this is a BLOCK (and not EXPR)
+ my $base = $_;
+ my $mod = $base;
+ $mod =~ s/\.\w+\z//;
+ "lib/PublicInbox/$_" => "blib/man3/PublicInbox::$mod.3"
+} qw(Git.pm Import.pm WWW.pod);
+
WriteMakefile(
NAME => 'PublicInbox',
VERSION => '1.1.0-pre1',
@@ -31,6 +40,7 @@ WriteMakefile(
# We have more test dependencies, but do not force
# users to install them. See INSTALL
},
+ MAN3PODS => \%man3,
);
sub MY::postamble {
--
EW
^ permalink raw reply related [flat|nested] 12+ messages in thread
* [PATCH 2/4] syscall: drop readahead wrapper
2019-05-08 19:18 ` [PATCH 0/4] Danga::Socket bundling cleanups Eric Wong
2019-05-08 19:18 ` [PATCH 1/4] build: do not manify DS and Syscall pods Eric Wong
@ 2019-05-08 19:18 ` Eric Wong
2019-05-08 19:18 ` [PATCH 3/4] DS: drop unused "_undef" sub Eric Wong
2019-05-08 19:18 ` [PATCH 4/4] DS: epoll: fix misordered EPOLL_CTL_DEL call Eric Wong
3 siblings, 0 replies; 12+ messages in thread
From: Eric Wong @ 2019-05-08 19:18 UTC (permalink / raw)
To: meta
No backwards compatibility to worry about for us; and fadvise
is superior anyways.
---
lib/PublicInbox/Syscall.pm | 14 --------------
1 file changed, 14 deletions(-)
diff --git a/lib/PublicInbox/Syscall.pm b/lib/PublicInbox/Syscall.pm
index 9194364..4ef64cc 100644
--- a/lib/PublicInbox/Syscall.pm
+++ b/lib/PublicInbox/Syscall.pm
@@ -64,7 +64,6 @@ our (
$SYS_epoll_ctl,
$SYS_epoll_wait,
$SYS_sendfile,
- $SYS_readahead,
);
our $no_deprecated = 0;
@@ -90,47 +89,40 @@ if ($^O eq "linux") {
$SYS_epoll_ctl = 255;
$SYS_epoll_wait = 256;
$SYS_sendfile = 187; # or 64: 239
- $SYS_readahead = 225;
} elsif ($machine eq "x86_64") {
$SYS_epoll_create = 213;
$SYS_epoll_ctl = 233;
$SYS_epoll_wait = 232;
$SYS_sendfile = 40;
- $SYS_readahead = 187;
} elsif ($machine =~ m/^parisc/) {
$SYS_epoll_create = 224;
$SYS_epoll_ctl = 225;
$SYS_epoll_wait = 226;
$SYS_sendfile = 122; # sys_sendfile64=209
- $SYS_readahead = 207;
$u64_mod_8 = 1;
} elsif ($machine =~ m/^ppc64/) {
$SYS_epoll_create = 236;
$SYS_epoll_ctl = 237;
$SYS_epoll_wait = 238;
$SYS_sendfile = 186; # (sys32_sendfile). sys32_sendfile64=226 (64 bit processes: sys_sendfile64=186)
- $SYS_readahead = 191; # both 32-bit and 64-bit vesions
$u64_mod_8 = 1;
} elsif ($machine eq "ppc") {
$SYS_epoll_create = 236;
$SYS_epoll_ctl = 237;
$SYS_epoll_wait = 238;
$SYS_sendfile = 186; # sys_sendfile64=226
- $SYS_readahead = 191;
$u64_mod_8 = 1;
} elsif ($machine =~ m/^s390/) {
$SYS_epoll_create = 249;
$SYS_epoll_ctl = 250;
$SYS_epoll_wait = 251;
$SYS_sendfile = 187; # sys_sendfile64=223
- $SYS_readahead = 222;
$u64_mod_8 = 1;
} elsif ($machine eq "ia64") {
$SYS_epoll_create = 1243;
$SYS_epoll_ctl = 1244;
$SYS_epoll_wait = 1245;
$SYS_sendfile = 1187;
- $SYS_readahead = 1216;
$u64_mod_8 = 1;
} elsif ($machine eq "alpha") {
# natural alignment, ints are 32-bits
@@ -138,14 +130,12 @@ if ($^O eq "linux") {
$SYS_epoll_create = 407;
$SYS_epoll_ctl = 408;
$SYS_epoll_wait = 409;
- $SYS_readahead = 379;
$u64_mod_8 = 1;
} elsif ($machine eq "aarch64") {
$SYS_epoll_create = 20; # (sys_epoll_create1)
$SYS_epoll_ctl = 21;
$SYS_epoll_wait = 22; # (sys_epoll_pwait)
$SYS_sendfile = 71; # (sys_sendfile64)
- $SYS_readahead = 213;
$u64_mod_8 = 1;
$no_deprecated = 1;
} elsif ($machine =~ m/arm(v\d+)?.*l/) {
@@ -154,21 +144,18 @@ if ($^O eq "linux") {
$SYS_epoll_ctl = 251;
$SYS_epoll_wait = 252;
$SYS_sendfile = 187;
- $SYS_readahead = 225;
$u64_mod_8 = 1;
} elsif ($machine =~ m/^mips64/) {
$SYS_sendfile = 5039;
$SYS_epoll_create = 5207;
$SYS_epoll_ctl = 5208;
$SYS_epoll_wait = 5209;
- $SYS_readahead = 5179;
$u64_mod_8 = 1;
} elsif ($machine =~ m/^mips/) {
$SYS_sendfile = 4207;
$SYS_epoll_create = 4248;
$SYS_epoll_ctl = 4249;
$SYS_epoll_wait = 4250;
- $SYS_readahead = 4223;
$u64_mod_8 = 1;
} else {
# as a last resort, try using the *.ph files which may not
@@ -177,7 +164,6 @@ if ($^O eq "linux") {
$SYS_epoll_create = eval { &SYS_epoll_create; } || 0;
$SYS_epoll_ctl = eval { &SYS_epoll_ctl; } || 0;
$SYS_epoll_wait = eval { &SYS_epoll_wait; } || 0;
- $SYS_readahead = eval { &SYS_readahead; } || 0;
}
if ($u64_mod_8) {
--
EW
^ permalink raw reply related [flat|nested] 12+ messages in thread
* [PATCH 3/4] DS: drop unused "_undef" sub
2019-05-08 19:18 ` [PATCH 0/4] Danga::Socket bundling cleanups Eric Wong
2019-05-08 19:18 ` [PATCH 1/4] build: do not manify DS and Syscall pods Eric Wong
2019-05-08 19:18 ` [PATCH 2/4] syscall: drop readahead wrapper Eric Wong
@ 2019-05-08 19:18 ` Eric Wong
2019-05-08 19:18 ` [PATCH 4/4] DS: epoll: fix misordered EPOLL_CTL_DEL call Eric Wong
3 siblings, 0 replies; 12+ messages in thread
From: Eric Wong @ 2019-05-08 19:18 UTC (permalink / raw)
To: meta
No longer used since we removed the *_ip_string fields
---
lib/PublicInbox/DS.pm | 7 -------
1 file changed, 7 deletions(-)
diff --git a/lib/PublicInbox/DS.pm b/lib/PublicInbox/DS.pm
index 5dd1bb7..c03bd5d 100644
--- a/lib/PublicInbox/DS.pm
+++ b/lib/PublicInbox/DS.pm
@@ -1033,13 +1033,6 @@ sub as_string {
return $ret;
}
-sub _undef {
- return undef unless $ENV{DS_DEBUG};
- my $msg = shift || "";
- warn "PublicInbox::DS: $msg\n";
- return undef;
-}
-
package PublicInbox::DS::Timer;
# [$abs_float_firetime, $coderef];
sub cancel {
--
EW
^ permalink raw reply related [flat|nested] 12+ messages in thread
* [PATCH 4/4] DS: epoll: fix misordered EPOLL_CTL_DEL call
2019-05-08 19:18 ` [PATCH 0/4] Danga::Socket bundling cleanups Eric Wong
` (2 preceding siblings ...)
2019-05-08 19:18 ` [PATCH 3/4] DS: drop unused "_undef" sub Eric Wong
@ 2019-05-08 19:18 ` Eric Wong
3 siblings, 0 replies; 12+ messages in thread
From: Eric Wong @ 2019-05-08 19:18 UTC (permalink / raw)
To: meta
Any operations on an fd after POSIX::close() are invalid, so
epoll_ctl will fail. Worse off, in a multi-threaded Perl, the
fd may be reused by another thread and EPOLL_CTL_DEL can hit the
wrong file description as a result.
cf. https://rt.cpan.org/Ticket/Display.html?id=129487
---
lib/PublicInbox/DS.pm | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/lib/PublicInbox/DS.pm b/lib/PublicInbox/DS.pm
index c03bd5d..779215c 100644
--- a/lib/PublicInbox/DS.pm
+++ b/lib/PublicInbox/DS.pm
@@ -333,8 +333,8 @@ sub EpollEventLoop {
} else {
my $fd = $ev->[0];
warn "epoll() returned fd $fd w/ state $state for which we have no mapping. removing.\n";
- POSIX::close($fd);
epoll_ctl($Epoll, EPOLL_CTL_DEL, $fd, 0);
+ POSIX::close($fd);
}
next;
}
--
EW
^ permalink raw reply related [flat|nested] 12+ messages in thread
end of thread, other threads:[~2019-05-08 19:18 UTC | newest]
Thread overview: 12+ messages (download: mbox.gz / follow: Atom feed)
-- links below jump to the message on this page --
2019-05-05 0:52 [PATCH 0/4] bundle Danga::Socket and Sys::Syscall Eric Wong
2019-05-05 0:52 ` [PATCH 1/4] " Eric Wong
2019-05-05 4:56 ` [PATCH 5/4] DS: workaround IO::Kqueue EINTR (mis-)handling Eric Wong
2019-05-08 9:07 ` [PATCH 6/4] DS: handle EINTR in IO::Poll path, too Eric Wong
2019-05-05 0:52 ` [PATCH 2/4] listener: use EPOLLEXCLUSIVE for listen sockets Eric Wong
2019-05-05 0:52 ` [PATCH 3/4] DS: remove unused fields and functions Eric Wong
2019-05-05 0:52 ` [PATCH 4/4] DS: drop profiling support Eric Wong
2019-05-08 19:18 ` [PATCH 0/4] Danga::Socket bundling cleanups Eric Wong
2019-05-08 19:18 ` [PATCH 1/4] build: do not manify DS and Syscall pods Eric Wong
2019-05-08 19:18 ` [PATCH 2/4] syscall: drop readahead wrapper Eric Wong
2019-05-08 19:18 ` [PATCH 3/4] DS: drop unused "_undef" sub Eric Wong
2019-05-08 19:18 ` [PATCH 4/4] DS: epoll: fix misordered EPOLL_CTL_DEL call Eric Wong
Code repositories for project(s) associated with this public inbox
https://80x24.org/public-inbox.git
This is a public inbox, see mirroring instructions
for how to clone and mirror all data and code used for this inbox;
as well as URLs for read-only IMAP folder(s) and NNTP newsgroup(s).