Cap our read buffers before AIO file operations to 1MB, this will pre…

…vent

memory over-use on systems with slow disks for PUT, or buffered upload temp
files. Assuming you system is fast enough to keep up with the writes this
will not harm upload speeds.

CHANGES

@@ -1,3 +1,8 @@
+    -- Cap our read buffers before AIO file operations to 1MB, this will prevent
+       memory over-use on systems with slow disks for PUT, or buffered upload temp
+       files. Assuming you system is fast enough to keep up with the writes this
+       will not harm upload speeds.
+
 1.75: 2010-04-02
 
     -- Stop doubleforking on daemonization. Seems useless and triggered a bug

lib/Perlbal/ClientHTTP.pm

@@ -247,6 +247,9 @@ sub handle_put_chunked {
                 return;
             }
 
+            # Reading too far ahead of our AIO subsystem will cause us to buffer it in memory.
+            $self->watch_read(0) if $self->{read_ahead} >= 1024 * 1024; # arbitrary
+            # ->put_writeout clears {read_ahead}, so we run it after we need that
             $self->put_writeout if $self->{read_ahead} >= 8192; # arbitrary
         },
         on_disconnect => sub {
@@ -333,6 +336,9 @@ sub event_read_put {
     $self->{content_length_remain} -= $clen;
 
     if ($self->{content_length_remain}) {
+        # Reading too far ahead of our AIO subsystem will cause us to buffer it in memory.
+        $self->watch_read(0) if $self->{read_ahead} >= 1024 * 1024; # arbitrary
+        # ->put_writeout clears {read_ahead}, so we run it after we need that
         $self->put_writeout if $self->{read_ahead} >= 8192; # arbitrary
     } else {
         # now, if we've filled the content of this put, we're done
@@ -426,6 +432,9 @@ sub put_writeout {
     $self->{read_buf}   = [];
     $self->{read_ahead} = 0;
 
+    # After copying out and clearing the buffer, turn reads back on again to fill up another buffer.
+    $self->watch_read(1) if $self->{content_length_remain} || $self->{chunked_upload_state};
+
     # okay, file is open, write some data
     $self->{put_in_progress} = 1;
 

lib/Perlbal/ClientProxy.pm

@@ -1084,6 +1084,8 @@ sub continue_buffered_upload {
 # write data to disk
 sub buffered_upload_update {
     my Perlbal::ClientProxy $self = shift;
+    # Reading too far ahead of our AIO subsystem will cause us to buffer it in memory.
+    $self->watch_read(0) if $self->{read_ahead} >= 1024 * 1024; # arbitrary
     return if $self->{is_writing};
     return unless $self->{is_buffering} && $self->{read_ahead};
 
@@ -1124,9 +1126,13 @@ sub buffered_upload_update {
     }
 
     # at this point, we want to do some writing
-    my $bref = shift(@{$self->{read_buf}});
+    my $bref = \join("", map { $$_ } @{$self->{read_buf}});
+    $self->{read_buf} = []; # clear these out
+    $self->{read_ahead} = 0;
     my $len = length $$bref;
-    $self->{read_ahead} -= $len;
+
+    # After copying out and clearing the buffer, turn reads back on again to fill up another buffer.
+    $self->watch_read(1) if $self->{content_length_remain} || $self->{chunked_upload_state};
 
     # so at this point we have a valid filename and file handle and should write out
     # the buffer that we have