support streaming data form fh to client

Use an explicit AnyEvent::Handle similar to websocket proxying.

Needs some special care to make sure we apply backpressure correctly to
avoid caching too much data. Note that because of AnyEvent restrictions,
specifying a "fh" to point to a file or a packet-based socket may result
in unwanted behaviour[0].

[0]: https://metacpan.org/pod/AnyEvent::Handle#DESCRIPTION

Signed-off-by: Stefan Reiter <s.reiter@proxmox.com>

diff --git a/PVE/APIServer/AnyEvent.pm b/PVE/APIServer/AnyEvent.pm
index 60a2a1c5f64628acd0cb1ea6beadc07cc565de18..f3db0e37d1ab8cf82e28d3229ada23e87a5e6edc 100644 (file)
--- a/PVE/APIServer/AnyEvent.pm
+++ b/PVE/APIServer/AnyEvent.pm
@@ -188,8 +188,93 @@ sub finish_response {
     }
 }
 
+sub response_stream {
+    my ($self, $reqstate, $stream_fh) = @_;
+
+    # disable timeout, we don't know how big the data is
+    $reqstate->{hdl}->timeout(0);
+
+    my $buf_size = 4*1024*1024;
+
+    my $on_read;
+    $on_read = sub {
+       my ($hdl) = @_;
+       my $reqhdl = $reqstate->{hdl};
+       return if !$reqhdl;
+
+       my $wbuf_len = length($reqhdl->{wbuf});
+       my $rbuf_len = length($hdl->{rbuf});
+       # TODO: Take into account $reqhdl->{wbuf_max} ? Right now
+       # that's unbounded, so just assume $buf_size
+       my $to_read = $buf_size - $wbuf_len;
+       $to_read = $rbuf_len if $rbuf_len < $to_read;
+       if ($to_read > 0) {
+           my $data = substr($hdl->{rbuf}, 0, $to_read, '');
+           $reqhdl->push_write($data);
+           $rbuf_len -= $to_read;
+       } elsif ($hdl->{_eof}) {
+           # workaround: AnyEvent gives us a fake EPIPE if we don't consume
+           # any data when called at EOF, so unregister ourselves - data is
+           # flushed by on_eof anyway
+           # see: https://sources.debian.org/src/libanyevent-perl/7.170-2/lib/AnyEvent/Handle.pm/#L1329
+           $hdl->on_read();
+           return;
+       }
+
+       # apply backpressure so we don't accept any more data into
+       # buffer if the client isn't downloading fast enough
+       # note: read_size can double upon read, and we also need to
+       # account for one more read after start_read, so *4
+       if ($rbuf_len + $hdl->{read_size}*4 > $buf_size) {
+           # stop reading until write buffer is empty
+           $hdl->on_read();
+           my $prev_on_drain = $reqhdl->{on_drain};
+           $reqhdl->on_drain(sub {
+               my ($wrhdl) = @_;
+               # on_drain called because write buffer is empty, continue reading
+               $hdl->on_read($on_read);
+               if ($prev_on_drain) {
+                   $wrhdl->on_drain($prev_on_drain);
+                   $prev_on_drain->($wrhdl);
+               }
+           });
+       }
+    };
+
+    $reqstate->{proxyhdl} = AnyEvent::Handle->new(
+       fh => $stream_fh,
+       rbuf_max => $buf_size,
+       timeout => 0,
+       on_read => $on_read,
+       on_eof => sub {
+           my ($hdl) = @_;
+           eval {
+               if (my $reqhdl = $reqstate->{hdl}) {
+                   $self->log_aborted_request($reqstate);
+                   # write out any remaining data
+                   $reqhdl->push_write($hdl->{rbuf}) if length($hdl->{rbuf}) > 0;
+                   $hdl->{rbuf} = "";
+                   $reqhdl->push_shutdown();
+                   $self->finish_response($reqstate);
+               }
+           };
+           if (my $err = $@) { syslog('err', "$err"); }
+           $on_read = undef;
+       },
+       on_error => sub {
+           my ($hdl, $fatal, $message) = @_;
+           eval {
+               $self->log_aborted_request($reqstate, $message);
+               $self->client_do_disconnect($reqstate);
+           };
+           if (my $err = $@) { syslog('err', "$err"); }
+           $on_read = undef;
+       },
+    );
+}
+
 sub response {
-    my ($self, $reqstate, $resp, $mtime, $nocomp, $delay) = @_;
+    my ($self, $reqstate, $resp, $mtime, $nocomp, $delay, $stream_fh) = @_;
 
     #print "$$: send response: " . Dumper($resp);
 
@@ -231,7 +316,7 @@ sub response {
     $resp->header('Server' => "pve-api-daemon/3.0");
 
     my $content_length;
-    if ($content) {
+    if ($content && !$stream_fh) {
 
        $content_length = length($content);
 
@@ -258,11 +343,16 @@ sub response {
     #print "SEND(without content) $res\n" if $self->{debug};
 
     $res .= "\015\012";
-    $res .= $content if $content;
+    $res .= $content if $content && !$stream_fh;
 
     $self->log_request($reqstate, $reqstate->{request});
 
-    if ($delay && $delay > 0) {
+    if ($stream_fh) {
+       # write headers and preamble...
+       $reqstate->{hdl}->push_write($res);
+       # ...then stream data via an AnyEvent::Handle
+       $self->response_stream($reqstate, $stream_fh);
+    } elsif ($delay && $delay > 0) {
        my $w; $w = AnyEvent->timer(after => $delay, cb => sub {
            undef $w; # delete reference
            $reqstate->{hdl}->push_write($res);
@@ -322,6 +412,13 @@ sub send_file_start {
            if (ref($download) eq 'HASH') {
                $fh = $download->{fh};
                $mime = $download->{'content-type'};
+
+               if ($download->{stream}) {
+                   my $header = HTTP::Headers->new(Content_Type => $mime);
+                   my $resp = HTTP::Response->new(200, "OK", $header);
+                   $self->response($reqstate, $resp, undef, 1, 0, $fh);
+                   return;
+               }
            } else {
                my $filename = $download;
                $fh = IO::File->new($filename, '<') ||