lib/PublicInbox/DS.pm | 14 ++++++-------- lib/PublicInbox/EvCleanup.pm | 2 +- lib/PublicInbox/HTTP.pm | 13 +++++-------- lib/PublicInbox/HTTPD/Async.pm | 2 +- lib/PublicInbox/NNTP.pm | 16 ++++++---------- diff --git a/lib/PublicInbox/DS.pm b/lib/PublicInbox/DS.pm index f1b7bab7ce0b3fc09eced454003b6be402002e6b..d07620a87215892a846ac0112ea6ad271bed5700 100644 --- a/lib/PublicInbox/DS.pm +++ b/lib/PublicInbox/DS.pm @@ -402,8 +402,6 @@ Carp::cluck("undef sock and/or fd in PublicInbox::DS->new. sock=" . ($sock || "") . ", fd=" . ($fd || "")) unless $sock && $fd; - $self->{wbuf} = []; - my $ev = $self->{event_watch} = POLLERR|POLLHUP|POLLNVAL; _InitPoller(); @@ -454,7 +452,7 @@ # we need to flush our write buffer, as there may # be self-referential closures (sub { $client->close }) # preventing the object from being destroyed - @{$self->{wbuf}} = (); + delete $self->{wbuf}; # if we're using epoll, we have to remove this from our epoll fd so we stop getting # notifications about it @@ -483,8 +481,8 @@ # returns 1 if done, 0 if incomplete sub flush_write ($) { my ($self) = @_; + my $wbuf = $self->{wbuf} or return 1; my $sock = $self->{sock} or return 1; - my $wbuf = $self->{wbuf}; while (my $bref = $wbuf->[0]) { my $ref = ref($bref); @@ -512,6 +510,7 @@ $bref->(); } } # while @$wbuf + delete $self->{wbuf}; $self->watch_write(0); 1; # all done } @@ -538,8 +537,7 @@ # hurt by this lie. my $sock = $self->{sock} or return 1; my $ref = ref $data; my $bref = $ref ? $data : \$data; - my $wbuf = $self->{wbuf}; - if (@$wbuf) { # already buffering, can't write more... + if (my $wbuf = $self->{wbuf}) { # already buffering, can't write more... push @$wbuf, $bref; return 0; } elsif ($ref eq 'CODE') { @@ -552,10 +550,10 @@ if (defined $written) { return 1 if $written == $to_write; $self->{wbuf_off} = $written; - push @$wbuf, $bref; + $self->{wbuf} = [ $bref ]; return flush_write($self); # try until EAGAIN } elsif ($! == EAGAIN) { - push @$wbuf, $bref; + $self->{wbuf} = [ $bref ]; $self->watch_write(1); } else { $self->close; diff --git a/lib/PublicInbox/EvCleanup.pm b/lib/PublicInbox/EvCleanup.pm index c64e2388512508cfea87624007adce585eadfc24..bd4dda117192f9dfa107070fff63890fea8cf1c9 100644 --- a/lib/PublicInbox/EvCleanup.pm +++ b/lib/PublicInbox/EvCleanup.pm @@ -29,7 +29,7 @@ # always writable, since PublicInbox::EvCleanup::event_step # never drains wbuf. We can avoid wasting a hash slot by # stuffing the read-end of the pipe into the never-to-be-touched # wbuf - push @{$self->{wbuf}}, $r; + $self->{wbuf} = $r; $self; } diff --git a/lib/PublicInbox/HTTP.pm b/lib/PublicInbox/HTTP.pm index dff5928681a9d904a281bb42f0f89fc5faa381e2..9a43069f1eda9eb31a2eead6a903dc04f4ca6d7a 100644 --- a/lib/PublicInbox/HTTP.pm +++ b/lib/PublicInbox/HTTP.pm @@ -67,11 +67,8 @@ sub event_step { # called by PublicInbox::DS my ($self) = @_; - my $wbuf = $self->{wbuf}; - if (@$wbuf) { - $self->write(undef); - return if !$self->{sock} || scalar(@$wbuf); - } + return unless $self->flush_write && $self->{sock}; + # only read more requests if we've drained the write buffer, # otherwise we can be buffering infinitely w/o backpressure @@ -268,7 +265,7 @@ if (defined $buf) { $write->($buf); # may close in PublicInbox::DS::write if ($self->{sock}) { my $next = $self->{pull}; - if (scalar @{$self->{wbuf}}) { + if ($self->{wbuf}) { $self->write($next); } else { PublicInbox::EvCleanup::asap($next); @@ -323,7 +320,7 @@ use constant MSG_MORE => ($^O eq 'linux') ? 0x8000 : 0; sub more ($$) { my $self = $_[0]; return unless $self->{sock}; - if (MSG_MORE && !scalar(@{$self->{wbuf}})) { + if (MSG_MORE && !$self->{wbuf}) { my $n = send($self->{sock}, $_[1], MSG_MORE); if (defined $n) { my $nlen = length($_[1]) - $n; @@ -490,7 +487,7 @@ # for graceful shutdown in PublicInbox::Daemon: sub busy () { my ($self) = @_; - ($self->{rbuf} ne '' || $self->{env} || scalar(@{$self->{wbuf}})); + ($self->{rbuf} ne '' || $self->{env} || $self->{wbuf}); } 1; diff --git a/lib/PublicInbox/HTTPD/Async.pm b/lib/PublicInbox/HTTPD/Async.pm index 261a01e0fa477f0cd1b200c137c3c6ab25286fe8..46ea188c0127c8a0791015b1d475861f6aff9a39 100644 --- a/lib/PublicInbox/HTTPD/Async.pm +++ b/lib/PublicInbox/HTTPD/Async.pm @@ -46,7 +46,7 @@ my $r = sysread($self->{sock}, $$bref, 8192); if ($r) { $fh->write($$bref); if ($http->{sock}) { # !closed - if (scalar @{$http->{wbuf}}) { + if ($http->{wbuf}) { $self->watch_read(0); $http->write(restart_read_cb($self)); } diff --git a/lib/PublicInbox/NNTP.pm b/lib/PublicInbox/NNTP.pm index 0a473e42729707058abc6e676eff5e56278024b3..d9097cc792b1231c9236181c6a95611d507e905e 100644 --- a/lib/PublicInbox/NNTP.pm +++ b/lib/PublicInbox/NNTP.pm @@ -644,7 +644,7 @@ } else { out($self, " deferred[$fd] aborted - %0.6f", now() - $t0); } - } elsif ($more) { # scalar @{$self->{wbuf}}: + } elsif ($more) { # $self->{wbuf}: # no recursion, schedule another call ASAP # but only after all pending writes are done update_idle_time($self); @@ -950,7 +950,7 @@ use constant MSG_MORE => ($^O eq 'linux') ? 0x8000 : 0; sub do_more ($$) { my ($self, $data) = @_; - if (MSG_MORE && !scalar(@{$self->{wbuf}})) { + if (MSG_MORE && !$self->{wbuf}) { my $n = send($self->{sock}, $data, MSG_MORE); if (defined $n) { my $dlen = length($data); @@ -963,15 +963,11 @@ } sub event_step { my ($self) = @_; - return unless $self->{sock}; - my $wbuf = $self->{wbuf}; - if (@$wbuf) { - update_idle_time($self); - $self->write(undef); - return if !$self->{sock} || scalar(@$wbuf); - } + return unless $self->flush_write && $self->{sock}; return if $self->{long_res}; + + update_idle_time($self); # only read more requests if we've drained the write buffer, # otherwise we can be buffering infinitely w/o backpressure @@ -1035,7 +1031,7 @@ # for graceful shutdown in PublicInbox::Daemon: sub busy { my ($self, $now) = @_; ($self->{rbuf} ne '' || $self->{long_res} || - scalar(@{$self->{wbuf}}) || not_idle_long($self, $now)); + $self->{wbuf} || not_idle_long($self, $now)); } 1;