X-Git-Url: http://www.git.stargrave.org/?a=blobdiff_plain;f=lib%2FPublicInbox%2FNNTP.pm;h=23be7754445b01686a6df0e209ef539cd285553b;hb=4f0b09919ae9c8823bf6c1fa1452bc27945952a3;hp=b07e1845c2321962bb7962a926cc04032ce10aff;hpb=b830cf90a2d9787315320aa4f4640f97fb1bea87;p=public-inbox.git diff --git a/lib/PublicInbox/NNTP.pm b/lib/PublicInbox/NNTP.pm index b07e1845..23be7754 100644 --- a/lib/PublicInbox/NNTP.pm +++ b/lib/PublicInbox/NNTP.pm @@ -1,5 +1,5 @@ -# Copyright (C) 2015 all contributors -# License: AGPLv3 or later (https://www.gnu.org/licenses/agpl-3.0.txt) +# Copyright (C) 2015-2018 all contributors +# License: AGPL-3.0+ # # Each instance of this represents a NNTP client socket package PublicInbox::NNTP; @@ -9,12 +9,14 @@ use base qw(Danga::Socket); use fields qw(nntpd article rbuf ng long_res); use PublicInbox::Search; use PublicInbox::Msgmap; +use PublicInbox::MID qw(mid_escape); use PublicInbox::Git; require PublicInbox::EvCleanup; use Email::Simple; use POSIX qw(strftime); use Time::HiRes qw(clock_gettime CLOCK_MONOTONIC); -use URI::Escape qw(uri_escape_utf8); +use Digest::SHA qw(sha1_hex); +use Time::Local qw(timegm timelocal); use constant { r501 => '501 command syntax error', r221 => '221 Header follows', @@ -38,8 +40,6 @@ my %DISABLED; # = map { $_ => 1 } qw(xover list_overview_fmt newnews xhdr); my $EXPMAP; # fd -> [ idle_time, $self ] my $expt; our $EXPTIME = 180; # 3 minutes -my $WEAKEN = {}; # string(nntpd) -> nntpd -my $weakt; my $nextt; my $nextq = []; @@ -64,16 +64,6 @@ sub update_idle_time ($) { defined $fd and $EXPMAP->{$fd} = [ now(), $self ]; } -# reduce FD pressure by closing some "git cat-file --batch" processes -# and unused FDs for msgmap and Xapian indices -sub weaken_groups () { - $weakt = undef; - foreach my $nntpd (values %$WEAKEN) { - $_->weaken_all foreach (@{$nntpd->{grouplist}}); - } - $WEAKEN = {}; -} - sub expire_old () { my $now = now(); my $exp = $EXPTIME; @@ -92,15 +82,11 @@ sub expire_old () { $EXPMAP = \%new; if ($nr) { $expt = PublicInbox::EvCleanup::later(*expire_old); - weaken_groups(); } else { $expt = undef; # noop to kick outselves out of the loop ASAP so descriptors # really get closed PublicInbox::EvCleanup::asap(sub {}); - - # grace period for reaping resources - $weakt ||= PublicInbox::EvCleanup::later(*weaken_groups); } } @@ -113,7 +99,6 @@ sub new ($$$) { $self->{rbuf} = ''; $self->watch_read(1); update_idle_time($self); - $WEAKEN->{"$nntpd"} = $nntpd; $expt ||= PublicInbox::EvCleanup::later(*expire_old); $self; } @@ -142,6 +127,7 @@ sub process_line ($$) { my $res = eval { $req->($self, @args) }; my $err = $@; if ($err && !$self->{closed}) { + local $/ = "\n"; chomp($l); err($self, 'error from: %s (%s)', $l, $err); $res = '503 program fault - command not performed'; @@ -253,7 +239,6 @@ sub cmd_listgroup ($;$) { sub parse_time ($$;$) { my ($date, $time, $gmt) = @_; - use Time::Local qw(); my ($hh, $mm, $ss) = unpack('A2A2A2', $time); if (defined $gmt) { $gmt =~ /\A(?:UTC|GMT)\z/i or die "GM invalid: $gmt"; @@ -265,15 +250,15 @@ sub parse_time ($$;$) { ($YYYY, $MM, $DD) = unpack('A4A2A2', $date); } else { # legacy clients send YYMMDD ($YYYY, $MM, $DD) = unpack('A2A2A2', $date); - if ($YYYY > strftime('%y', @now)) { - my $cur_year = $now[5] + 1900; + my $cur_year = $now[5] + 1900; + if ($YYYY > $cur_year) { $YYYY += int($cur_year / 1000) * 1000 - 100; } } if ($gmt) { - Time::Local::timegm($ss, $mm, $hh, $DD, $MM - 1, $YYYY); + timegm($ss, $mm, $hh, $DD, $MM - 1, $YYYY); } else { - Time::Local::timelocal($ss, $mm, $hh, $DD, $MM - 1, $YYYY); + timelocal($ss, $mm, $hh, $DD, $MM - 1, $YYYY); } } @@ -302,7 +287,6 @@ sub wildmat2re (;$) { return $_[0] = qr/.*/ if (!defined $_[0] || $_[0] eq '*'); my %keep; my $salt = rand; - use Digest::SHA qw(sha1_hex); my $tmp = $_[0]; $tmp =~ s#(?query($ts, $opts); + my $res = $srch->query_ts($ts, $opts); my $msgs = $res->{msgs}; if (my $nr = scalar @$msgs) { more($self, '<' . @@ -437,7 +421,7 @@ sub set_nntp_headers { $hdr->header_set('Xref', xref($ng, $n)); header_append($hdr, 'List-Post', "{-primary_address}>"); if (my $url = $ng->base_url) { - $mid = uri_escape_utf8($mid); + $mid = mid_escape($mid); header_append($hdr, 'Archived-At', "<$url$mid/>"); header_append($hdr, 'List-Archive', "<$url>"); } @@ -606,9 +590,10 @@ sub long_response ($$$$) { my $err; do { - eval { $cb->(\$beg) }; + eval { $cb->(\$beg, \$lim) }; } until (($err = $@) || $self->{closed} || - ++$beg > $end || !--$lim || $self->{write_buf_size}); + ++$beg > $end || --$lim < 0 || + $self->{write_buf_size}); if ($err || $self->{closed}) { $self->{long_res} = undef; @@ -625,7 +610,7 @@ sub long_response ($$$$) { update_idle_time($self); $self->watch_read(1); } - } elsif (!$lim || $self->{write_buf_size}) { + } elsif ($lim < 0 || $self->{write_buf_size}) { # no recursion, schedule another call ASAP # but only after all pending writes are done update_idle_time($self); @@ -731,11 +716,12 @@ sub hdr_searchmsg ($$$$) { more($self, $xhdr ? r221 : r225); my $off = 0; long_response($self, $beg, $end, sub { - my ($i) = @_; + my ($i, $lim) = @_; my $res = $srch->query_xover($beg, $end, $off); my $msgs = $res->{msgs}; my $nr = scalar @$msgs or return; $off += $nr; + $$lim -= $nr; my $tmp = ''; foreach my $s (@$msgs) { $tmp .= $s->num . ' ' . $s->$field . "\r\n"; @@ -869,11 +855,12 @@ sub cmd_xover ($;$) { my $srch = $self->{ng}->search; my $off = 0; long_response($self, $beg, $end, sub { - my ($i) = @_; + my ($i, $lim) = @_; my $res = $srch->query_xover($beg, $end, $off); my $msgs = $res->{msgs}; my $nr = scalar @$msgs or return; $off += $nr; + $$lim -= $nr; # OVERVIEW.FMT more($self, join("\r\n", map { @@ -911,7 +898,7 @@ sub more ($$) { sub do_write ($$) { my ($self, $data) = @_; my $done = $self->write($data); - die if $self->{closed}; + return if $self->{closed}; # Do not watch for readability if we have data in the queue, # instead re-enable watching for readability when we can @@ -962,10 +949,12 @@ sub event_write { sub event_read { my ($self) = @_; use constant LINE_MAX => 512; # RFC 977 section 2.3 - my $r = 1; - my $buf = $self->read(LINE_MAX) or return $self->close; - $self->{rbuf} .= $$buf; + if (index($self->{rbuf}, "\n") < 0) { + my $buf = $self->read(LINE_MAX) or return $self->close; + $self->{rbuf} .= $$buf; + } + my $r = 1; while ($r > 0 && $self->{rbuf} =~ s/\A\s*([^\r\n]+)\r?\n//) { my $line = $1; return $self->close if $line =~ /[[:cntrl:]]/s; @@ -998,10 +987,19 @@ sub watch_read { $rv; } +sub not_idle_long ($$) { + my ($self, $now) = @_; + defined(my $fd = $self->{fd}) or return; + my $ary = $EXPMAP->{$fd} or return; + my $exp_at = $ary->[0] + $EXPTIME; + $exp_at > $now; +} + # for graceful shutdown in PublicInbox::Daemon: -sub busy () { - my ($self) = @_; - ($self->{rbuf} ne '' || $self->{long_res} || $self->{write_buf_size}); +sub busy { + my ($self, $now) = @_; + ($self->{rbuf} ne '' || $self->{long_res} || $self->{write_buf_size} || + not_idle_long($self, $now)); } 1;