net: add optimization for SourceStream knowing it is complete

In http://crbug.com/957175, wanderview@ added an optimization for
BlobReader where it would add an extra hop when finishing a stream.
Many blobs size mojo::DataPipe to exactly fit the blob size, and so
this can be quite common.

As SourceStreamToDataPipe is going to be used in this CL
(https://chromium-review.googlesource.com/c/chromium/src/+/1830360),
that optimization needs to be carried forward.

Bug: 1012869,957175
Change-Id: I9f7fc006b633a248fe3b726ac78d1fe4773472a6
Reviewed-on: https://chromium-review.googlesource.com/c/chromium/src/+/1862253
Reviewed-by: Eric Roman <eroman@chromium.org>
Reviewed-by: Kinuko Yasuda <kinuko@chromium.org>
Reviewed-by: Zentaro Kavanagh <zentaro@chromium.org>
Commit-Queue: enne <enne@chromium.org>
Cr-Commit-Position: refs/heads/master@{#707920}

Author: quisquous

chromeos/printing/ppd_line_reader.cc

@@ -47,18 +47,21 @@ class StringSourceStream : public net::SourceStream {
   int Read(net::IOBuffer* dest_buffer,
            int buffer_size,
            net::CompletionOnceCallback) override {
-    int read_size = src_.size() - read_ofs_;
-    if (read_size > buffer_size) {
-      read_size = buffer_size;
-    }
+    if (buffer_size < 0)
+      return net::ERR_INVALID_ARGUMENT;
+    if (!MayHaveMoreBytes())
+      return net::OK;
+    const size_t read_size =
+        std::min(src_.size() - read_ofs_, static_cast<size_t>(buffer_size));
     memcpy(dest_buffer->data(), src_.data() + read_ofs_, read_size);
     read_ofs_ += read_size;
     return read_size;
   }
   std::string Description() const override { return ""; }
+  bool MayHaveMoreBytes() const override { return read_ofs_ < src_.size(); }
 
  private:
-  int read_ofs_ = 0;
+  size_t read_ofs_ = 0;
   const std::string& src_;
 };
 

net/filter/filter_source_stream.cc

@@ -73,6 +73,10 @@ std::string FilterSourceStream::Description() const {
   return next_type_string + "," + GetTypeAsString();
 }
 
+bool FilterSourceStream::MayHaveMoreBytes() const {
+  return !upstream_end_reached_;
+}
+
 FilterSourceStream::SourceType FilterSourceStream::ParseEncodingType(
     const std::string& encoding) {
   if (encoding.empty()) {

net/filter/filter_source_stream.h

@@ -32,11 +32,12 @@ class NET_EXPORT_PRIVATE FilterSourceStream : public SourceStream {
 
   ~FilterSourceStream() override;
 
+  // SourceStream implementation.
   int Read(IOBuffer* read_buffer,
            int read_buffer_size,
            CompletionOnceCallback callback) override;
-
   std::string Description() const override;
+  bool MayHaveMoreBytes() const override;
 
   static SourceType ParseEncodingType(const std::string& encoding);
 

net/filter/fuzzed_source_stream.cc

@@ -73,6 +73,10 @@ std::string FuzzedSourceStream::Description() const {
   return "";
 }
 
+bool FuzzedSourceStream::MayHaveMoreBytes() const {
+  return !end_returned_;
+}
+
 void FuzzedSourceStream::OnReadComplete(CompletionOnceCallback callback,
                                         const std::string& fuzzed_data,
                                         scoped_refptr<IOBuffer> read_buf,

net/filter/fuzzed_source_stream.h

@@ -32,6 +32,7 @@ class FuzzedSourceStream : public SourceStream {
            int buffer_size,
            CompletionOnceCallback callback) override;
   std::string Description() const override;
+  bool MayHaveMoreBytes() const override;
 
  private:
   void OnReadComplete(CompletionOnceCallback callback,

net/filter/mock_source_stream.cc

@@ -12,12 +12,7 @@
 
 namespace net {
 
-MockSourceStream::MockSourceStream()
-    : SourceStream(SourceStream::TYPE_NONE),
-      read_one_byte_at_a_time_(false),
-      awaiting_completion_(false),
-      dest_buffer_(nullptr),
-      dest_buffer_size_(0) {}
+MockSourceStream::MockSourceStream() : SourceStream(SourceStream::TYPE_NONE) {}
 
 MockSourceStream::~MockSourceStream() {
   DCHECK(!awaiting_completion_);
@@ -53,6 +48,12 @@ std::string MockSourceStream::Description() const {
   return "";
 }
 
+bool MockSourceStream::MayHaveMoreBytes() const {
+  if (always_report_has_more_bytes_)
+    return true;
+  return !results_.empty();
+}
+
 MockSourceStream::QueuedResult::QueuedResult(const char* data,
                                              int len,
                                              Error error,

net/filter/mock_source_stream.h

@@ -35,6 +35,7 @@ class MockSourceStream : public SourceStream {
            int buffer_size,
            CompletionOnceCallback callback) override;
   std::string Description() const override;
+  bool MayHaveMoreBytes() const override;
 
   // Enqueues a result to be returned by |Read|. This method does not make a
   // copy of |data|, so |data| must outlive this object. If |mode| is SYNC,
@@ -55,6 +56,10 @@ class MockSourceStream : public SourceStream {
     read_one_byte_at_a_time_ = read_one_byte_at_a_time;
   }
 
+  void set_always_report_has_more_bytes(bool always_report_has_more_bytes) {
+    always_report_has_more_bytes_ = always_report_has_more_bytes;
+  }
+
   // Returns true if a read is waiting to be completed.
   bool awaiting_completion() const { return awaiting_completion_; }
 
@@ -68,12 +73,13 @@ class MockSourceStream : public SourceStream {
     const Mode mode;
   };
 
-  bool read_one_byte_at_a_time_;
+  bool read_one_byte_at_a_time_ = false;
+  bool always_report_has_more_bytes_ = true;
   base::queue<QueuedResult> results_;
-  bool awaiting_completion_;
+  bool awaiting_completion_ = false;
   scoped_refptr<IOBuffer> dest_buffer_;
   CompletionOnceCallback callback_;
-  int dest_buffer_size_;
+  int dest_buffer_size_ = 0;
 
   DISALLOW_COPY_AND_ASSIGN(MockSourceStream);
 };

net/filter/source_stream.h

@@ -54,6 +54,12 @@ class NET_EXPORT_PRIVATE SourceStream {
   // logging.
   virtual std::string Description() const = 0;
 
+  // Returns true if there may be more bytes to read in this source stream.
+  // This is not a guarantee that there are more bytes (in the case that
+  // the stream doesn't know).  However, if this returns false, then the stream
+  // is guaranteed to be complete.
+  virtual bool MayHaveMoreBytes() const = 0;
+
   SourceType type() const { return type_; }
 
  private:

net/tools/content_decoder_tool/content_decoder_tool.cc

@@ -49,6 +49,8 @@ class StdinSourceStream : public SourceStream {
 
   std::string Description() const override { return ""; }
 
+  bool MayHaveMoreBytes() const override { return true; }
+
  private:
   std::istream* input_stream_;
 

net/url_request/url_request_job.cc

@@ -67,6 +67,8 @@ class URLRequestJob::URLRequestJobSourceStream : public SourceStream {
 
   std::string Description() const override { return std::string(); }
 
+  bool MayHaveMoreBytes() const override { return true; }
+
  private:
   // It is safe to keep a raw pointer because |job_| owns the last stream which
   // indirectly owns |this|. Therefore, |job_| will not be destroyed when |this|

services/network/public/cpp/data_pipe_to_source_stream.cc

@@ -31,6 +31,10 @@ std::string DataPipeToSourceStream::Description() const {
   return "DataPipe";
 }
 
+bool DataPipeToSourceStream::MayHaveMoreBytes() const {
+  return !complete_;
+}
+
 int DataPipeToSourceStream::Read(net::IOBuffer* buf,
                                  int buf_size,
                                  net::CompletionOnceCallback callback) {
@@ -99,6 +103,7 @@ void DataPipeToSourceStream::OnReadable(MojoResult unused) {
 }
 
 void DataPipeToSourceStream::FinishReading() {
+  complete_ = true;
   handle_watcher_.Cancel();
   body_.reset();
 }

services/network/public/cpp/data_pipe_to_source_stream.h

@@ -19,10 +19,12 @@ class COMPONENT_EXPORT(NETWORK_CPP) DataPipeToSourceStream final
   explicit DataPipeToSourceStream(mojo::ScopedDataPipeConsumerHandle body);
   ~DataPipeToSourceStream() override;
 
+  // net::SourceStream implementation.
   int Read(net::IOBuffer* buf,
            int buf_size,
            net::CompletionOnceCallback callback) override;
   std::string Description() const override;
+  bool MayHaveMoreBytes() const override;
 
  private:
   void OnReadable(MojoResult result);
@@ -32,6 +34,7 @@ class COMPONENT_EXPORT(NETWORK_CPP) DataPipeToSourceStream final
   mojo::SimpleWatcher handle_watcher_;
 
   bool inside_read_ = false;
+  bool complete_ = false;
 
   scoped_refptr<net::IOBuffer> output_buf_;
   int output_buf_size_ = 0;

services/network/public/cpp/data_pipe_to_source_stream_unittest.cc

@@ -96,13 +96,19 @@ class DataPipeToSourceStreamTest
         WriteToPipe();
         rv = callback.WaitForResult();
       }
-      if (rv == net::OK)
+      if (rv == net::OK) {
+        EXPECT_FALSE(adapter_->MayHaveMoreBytes());
         break;
-      if (rv < net::OK)
+      }
+      if (rv < net::OK) {
+        EXPECT_FALSE(adapter_->MayHaveMoreBytes());
         return rv;
+      }
       EXPECT_GT(rv, net::OK);
       output->append(output_buffer_->data(), rv);
+      EXPECT_TRUE(adapter_->MayHaveMoreBytes());
     }
+    EXPECT_FALSE(adapter_->MayHaveMoreBytes());
     return 0;
   }
 

services/network/public/cpp/source_stream_to_data_pipe.cc

@@ -70,10 +70,18 @@ void SourceStreamToDataPipe::DidRead(int result) {
     OnComplete(result);
     return;
   }
+
   dest_ = pending_write_->Complete(result);
-  pending_write_ = nullptr;
   transferred_bytes_ += result;
 
+  // Don't hop through an extra ReadMore just to find out there's no more data.
+  if (!source_->MayHaveMoreBytes()) {
+    OnComplete(net::OK);
+    return;
+  }
+
+  pending_write_ = nullptr;
+
   base::SequencedTaskRunnerHandle::Get()->PostTask(
       FROM_HERE, base::BindOnce(&SourceStreamToDataPipe::ReadMore,
                                 weak_factory_.GetWeakPtr()));

services/network/public/cpp/source_stream_to_data_pipe_unittest.cc

@@ -196,4 +196,20 @@ TEST_P(SourceStreamToDataPipeTest, ConsumerClosed) {
   EXPECT_EQ(*CallbackResult(), net::ERR_ABORTED);
 }
 
+TEST_P(SourceStreamToDataPipeTest, MayHaveMoreBytes) {
+  const char message[] = "Hello, world!";
+
+  // Test that having the SourceStream properly report when !MayHaveMoreBytes
+  // shortcuts extra work and still reports things properly.
+  Init();
+  source()->set_always_report_has_more_bytes(false);
+  // Unlike other test reads (see "Simple" test), there is only one result here.
+  source()->AddReadResult(message, sizeof(message) - 1, net::OK,
+                          GetParam().mode);
+  adapter()->Start(callback());
+
+  std::string output;
+  EXPECT_EQ(ReadPipe(&output), net::OK);
+  EXPECT_EQ(output, message);
+}
 }  // namespace network