Thing

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