ds: lazy-initialize wbuf

We don't need write buffering unless we encounter slow clients
requesting large responses.  So don't waste a hash slot or
(empty) arrayref for it.

5 files changed, 19 insertions, 28 deletions

diff --git a/lib/PublicInbox/DS.pm b/lib/PublicInbox/DS.pm
index f1b7bab7..d07620a8 100644
--- a/lib/PublicInbox/DS.pm
+++ b/lib/PublicInbox/DS.pm
@@ -402,8 +402,6 @@ sub new {
     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 @@ sub close {
     # 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 @@ sub close {
 # 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 @@ sub flush_write ($) {
         }
     } # while @$wbuf
 
+    delete $self->{wbuf};
     $self->watch_write(0);
     1; # all done
 }
@@ -538,8 +537,7 @@ sub write {
     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 @@ sub write {
         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 c64e2388..bd4dda11 100644
--- a/lib/PublicInbox/EvCleanup.pm
+++ b/lib/PublicInbox/EvCleanup.pm
@@ -29,7 +29,7 @@ sub once_init () {
         # 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 dff59286..9a43069f 100644
--- a/lib/PublicInbox/HTTP.pm
+++ b/lib/PublicInbox/HTTP.pm
@@ -67,11 +67,8 @@ sub new ($$$) {
 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 @@ sub getline_cb ($$$) {
                         $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 @@ sub close {
 # 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 261a01e0..46ea188c 100644
--- a/lib/PublicInbox/HTTPD/Async.pm
+++ b/lib/PublicInbox/HTTPD/Async.pm
@@ -46,7 +46,7 @@ sub main_cb ($$$) {
                 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 0a473e42..d9097cc7 100644
--- a/lib/PublicInbox/NNTP.pm
+++ b/lib/PublicInbox/NNTP.pm
@@ -644,7 +644,7 @@ sub long_response ($$) {
                                 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 do_more ($$) {
 
 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 @@ sub not_idle_long ($$) {
 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;