X-Git-Url: http://www.git.stargrave.org/?a=blobdiff_plain;f=lib%2FPublicInbox%2FNNTP.pm;h=9973fcaf149eb3a4b637a148286303f702e06a96;hb=7c83d3e706811095cedab0bf62ac530d7b0f3a5a;hp=468a22f529673b2aa0043fdb2d9b92237312b272;hpb=6b1ee7ed8032e277a84bbe2a343f2c318a0defb8;p=public-inbox.git diff --git a/lib/PublicInbox/NNTP.pm b/lib/PublicInbox/NNTP.pm index 468a22f5..9973fcaf 100644 --- a/lib/PublicInbox/NNTP.pm +++ b/lib/PublicInbox/NNTP.pm @@ -1,4 +1,4 @@ -# Copyright (C) 2015-2018 all contributors +# Copyright (C) 2015-2019 all contributors # License: AGPL-3.0+ # # Each instance of this represents a NNTP client socket @@ -6,7 +6,7 @@ package PublicInbox::NNTP; use strict; use warnings; use base qw(PublicInbox::DS); -use fields qw(nntpd article rbuf ng long_res); +use fields qw(nntpd article ng); use PublicInbox::Search; use PublicInbox::Msgmap; use PublicInbox::MID qw(mid_escape); @@ -38,30 +38,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 $nextt; - -my $nextq = []; -sub next_tick () { - $nextt = undef; - my $q = $nextq; - $nextq = []; - foreach my $nntp (@$q) { - # for request && response protocols, always finish writing - # before finishing reading: - if (my $long_cb = $nntp->{long_res}) { - $nntp->write($long_cb); - } else { - # pipelined request, we bypassed socket-readiness - # checks to get here: - event_step($nntp); - } - } -} - -sub requeue ($) { - push @$nextq, $_[0]; - $nextt ||= PublicInbox::EvCleanup::asap(*next_tick); -} sub update_idle_time ($) { my ($self) = @_; @@ -78,30 +54,37 @@ sub expire_old () { while (my ($fd, $v) = each %$EXPMAP) { my ($idle_time, $nntp) = @$v; if ($idle_time < $old) { - $nntp->close; # idempotent + if (!$nntp->shutdn) { + ++$nr; + $new{$fd} = $v; + } } else { ++$nr; $new{$fd} = $v; } } $EXPMAP = \%new; - if ($nr) { - $expt = PublicInbox::EvCleanup::later(*expire_old); - } else { - $expt = undef; - # noop to kick outselves out of the loop ASAP so descriptors - # really get closed - PublicInbox::EvCleanup::asap(sub {}); - } + $expt = PublicInbox::EvCleanup::later(*expire_old) if $nr; } +sub greet ($) { $_[0]->write($_[0]->{nntpd}->{greet}) }; + sub new ($$$) { my ($class, $sock, $nntpd) = @_; my $self = fields::new($class); - $self->SUPER::new($sock, EPOLLIN | EPOLLONESHOT); + my $ev = EPOLLIN; + my $wbuf; + if (ref($sock) eq 'IO::Socket::SSL' && !$sock->accept_SSL) { + $ev = PublicInbox::TLS::epollbit() or return CORE::close($sock); + $wbuf = [ \&PublicInbox::DS::accept_tls_step, \&greet ]; + } + $self->SUPER::new($sock, $ev | EPOLLONESHOT); $self->{nntpd} = $nntpd; - res($self, '201 ' . $nntpd->{servername} . ' ready - post via email'); - $self->{rbuf} = ''; + if ($wbuf) { + $self->{wbuf} = $wbuf; + } else { + greet($self); + } update_idle_time($self); $expt ||= PublicInbox::EvCleanup::later(*expire_old); $self; @@ -400,7 +383,7 @@ sub cmd_post ($) { sub cmd_quit ($) { my ($self) = @_; res($self, '205 closing connection - goodbye!'); - $self->close; + $self->shutdn; undef; } @@ -617,8 +600,7 @@ sub get_range ($$) { } sub long_response ($$) { - my ($self, $cb) = @_; - die "BUG: nested long response" if $self->{long_res}; + my ($self, $cb) = @_; # cb returns true if more, false if done my $fd = fileno($self->{sock}); defined $fd or return; @@ -626,36 +608,38 @@ sub long_response ($$) { # clients should not be sending us stuff and making us do more # work while we are stream a response to them my $t0 = now(); - $self->{long_res} = sub { + my $long_cb; # DANGER: self-referential + $long_cb = sub { + # wbuf is unset or empty, here; $cb may add to it my $more = eval { $cb->() }; if ($@ || !$self->{sock}) { # something bad happened... - delete $self->{long_res}; - + $long_cb = undef; + my $diff = now() - $t0; if ($@) { err($self, "%s during long response[$fd] - %0.6f", - $@, now() - $t0); - } - if ($self->{sock}) { - update_idle_time($self); - requeue($self); - } else { - out($self, " deferred[$fd] aborted - %0.6f", - now() - $t0); + $@, $diff); } + out($self, " deferred[$fd] aborted - %0.6f", $diff); + $self->close; } elsif ($more) { # $self->{wbuf}: + update_idle_time($self); + # no recursion, schedule another call ASAP # but only after all pending writes are done - update_idle_time($self); - requeue($self); + my $wbuf = $self->{wbuf} ||= []; + push @$wbuf, $long_cb; + + # wbuf may be populated by $cb, no need to rearm if so: + $self->requeue if scalar(@$wbuf) == 1; } else { # all done! - delete $self->{long_res}; + $long_cb = undef; res($self, '.'); out($self, " deferred[$fd] done - %0.6f", now() - $t0); - requeue($self); + $self->requeue unless $self->{wbuf}; } }; - $self->{long_res}->(); # kick off! + $self->write($long_cb); # kick off! undef; } @@ -898,6 +882,19 @@ sub cmd_xover ($;$) { }); } +sub cmd_starttls ($) { + my ($self) = @_; + my $sock = $self->{sock} or return; + # RFC 4642 2.2.1 + (ref($sock) eq 'IO::Socket::SSL') and return '502 Command unavailable'; + my $opt = $self->{nntpd}->{accept_tls} or + return '580 can not initiate TLS negotiation'; + res($self, '382 Continue with TLS negotiation'); + $self->{sock} = IO::Socket::SSL->start_SSL($sock, %$opt); + $self->requeue if PublicInbox::DS::accept_tls_step($self); + undef; +} + sub cmd_xpath ($$) { my ($self, $mid) = @_; return r501 unless $mid =~ /\A<(.+)>\z/; @@ -944,7 +941,7 @@ sub event_step { # otherwise we can be buffering infinitely w/o backpressure use constant LINE_MAX => 512; # RFC 977 section 2.3 - my $rbuf = \($self->{rbuf}); + my $rbuf = $self->{rbuf} // (\(my $x = '')); my $r = 1; if (index($$rbuf, "\n") < 0) { @@ -957,19 +954,19 @@ sub event_step { my $t0 = now(); my $fd = fileno($self->{sock}); $r = eval { process_line($self, $line) }; - my $d = $self->{long_res} ? - " deferred[$fd]" : ''; - out($self, "[$fd] %s - %0.6f$d", $line, now() - $t0); + my $pending = $self->{wbuf} ? ' pending' : ''; + out($self, "[$fd] %s - %0.6f$pending", $line, now() - $t0); } return $self->close if $r < 0; my $len = bytes::length($$rbuf); return $self->close if ($len >= LINE_MAX); + $self->rbuf_idle($rbuf); update_idle_time($self); # maybe there's more pipelined data, or we'll have # to register it for socket-readiness notifications - requeue($self) unless ($self->{long_res} || $self->{wbuf}); + $self->requeue unless $self->{wbuf}; } sub not_idle_long ($$) { @@ -983,8 +980,7 @@ sub not_idle_long ($$) { # for graceful shutdown in PublicInbox::Daemon: sub busy { my ($self, $now) = @_; - ($self->{rbuf} ne '' || $self->{long_res} || - $self->{wbuf} || not_idle_long($self, $now)); + ($self->{rbuf} || $self->{wbuf} || not_idle_long($self, $now)); } 1;