[Cronet] Coalesce small buffers into single QUIC packet in GRPC on iOS.

components/cronet/ios/cronet_bidirectional_stream.cc

Index: components/cronet/ios/cronet_bidirectional_stream.cc
diff --git a/components/cronet/ios/cronet_bidirectional_stream.cc b/components/cronet/ios/cronet_bidirectional_stream.cc
index d3c881b472e185f481329563b53caed36a1dfe95..dcaf9af6b6959bf0f5ec75fbbc268a571cc3870a 100644
--- a/components/cronet/ios/cronet_bidirectional_stream.cc
+++ b/components/cronet/ios/cronet_bidirectional_stream.cc
@@ -32,13 +32,47 @@
 
 namespace cronet {
 
+CronetBidirectionalStream::WriteBuffers::WriteBuffers() {}
+
+CronetBidirectionalStream::WriteBuffers::~WriteBuffers() {}
+
+void CronetBidirectionalStream::WriteBuffers::Clear() {
+  write_buffer_list.clear();
+  write_buffer_len_list.clear();
+}
+
+void CronetBidirectionalStream::WriteBuffers::AppendBuffer(
+    const scoped_refptr<net::IOBuffer>& buffer,
+    int buffer_size) {
+  write_buffer_list.push_back(buffer);
+  write_buffer_len_list.push_back(buffer_size);
+}
+
+void CronetBidirectionalStream::WriteBuffers::MoveTo(WriteBuffers* target) {
+  std::move(write_buffer_list.begin(), write_buffer_list.end(),
+            std::back_inserter(target->write_buffer_list));
+  std::move(write_buffer_len_list.begin(), write_buffer_len_list.end(),
+            std::back_inserter(target->write_buffer_len_list));
+  Clear();
+}
+
+bool CronetBidirectionalStream::WriteBuffers::Empty() const {
+  return write_buffer_list.empty();
+}
+
 CronetBidirectionalStream::CronetBidirectionalStream(
     CronetEnvironment* environment,
     Delegate* delegate)
     : read_state_(NOT_STARTED),
       write_state_(NOT_STARTED),
       write_end_of_stream_(false),
+      request_headers_sent_(false),
+      disable_auto_flush_(false),
+      delay_headers_until_flush_(false),
       environment_(environment),
+      pending_write_data_(new WriteBuffers()),
+      flushing_write_data_(new WriteBuffers()),
+      sending_write_data_(new WriteBuffers()),
       delegate_(delegate) {}
 
 CronetBidirectionalStream::~CronetBidirectionalStream() {
@@ -98,6 +132,12 @@ bool CronetBidirectionalStream::WriteData(const char* buffer,
   return true;
 }
 
+void CronetBidirectionalStream::Flush() {
+  environment_->PostToNetworkThread(
+      FROM_HERE, base::Bind(&CronetBidirectionalStream::FlushOnNetworkThread,
+                            base::Unretained(this)));
+}
+
 void CronetBidirectionalStream::Cancel() {
   environment_->PostToNetworkThread(
       FROM_HERE, base::Bind(&CronetBidirectionalStream::CancelOnNetworkThread,
@@ -115,17 +155,24 @@ void CronetBidirectionalStream::Destroy() {
 
 void CronetBidirectionalStream::OnStreamReady(bool request_headers_sent) {
   DCHECK(environment_->IsOnNetworkThread());
-  DCHECK(write_state_ == STARTED);
-  write_state_ = WAITING_FOR_WRITE;
-  if (write_end_of_stream_)
+  DCHECK_EQ(STARTED, write_state_);
+  request_headers_sent_ = request_headers_sent;
+  write_state_ = WAITING_FOR_FLUSH;
+  if (write_end_of_stream_) {
+    if (!request_headers_sent) {
+      // If there is no data to write, then just send headers explicitly.
+      bidi_stream_->SendRequestHeaders();
+      request_headers_sent_ = true;
+    }
     write_state_ = WRITING_DONE;
+  }
   delegate_->OnStreamReady();
 }
 
 void CronetBidirectionalStream::OnHeadersReceived(
     const net::SpdyHeaderBlock& response_headers) {
   DCHECK(environment_->IsOnNetworkThread());
-  DCHECK(read_state_ == STARTED);
+  DCHECK_EQ(STARTED, read_state_);
   read_state_ = WAITING_FOR_READ;
   // Get http status code from response headers.
   int http_status_code = 0;
@@ -148,7 +195,7 @@ void CronetBidirectionalStream::OnHeadersReceived(
 
 void CronetBidirectionalStream::OnDataRead(int bytes_read) {
   DCHECK(environment_->IsOnNetworkThread());
-  DCHECK(read_state_ == READING);
+  DCHECK_EQ(READING, read_state_);
   read_state_ = WAITING_FOR_READ;
   delegate_->OnDataRead(read_buffer_->data(), bytes_read);
 
@@ -161,12 +208,19 @@ void CronetBidirectionalStream::OnDataRead(int bytes_read) {
 
 void CronetBidirectionalStream::OnDataSent() {
   DCHECK(environment_->IsOnNetworkThread());
-  DCHECK(write_state_ == WRITING);
-  write_state_ = WAITING_FOR_WRITE;
-  delegate_->OnDataSent(write_buffer_->data());
-  // Free the write buffer.
-  write_buffer_ = nullptr;
-  if (write_end_of_stream_)
+  DCHECK_EQ(WRITING, write_state_);
+  write_state_ = WAITING_FOR_FLUSH;
+  for (const scoped_refptr<net::IOBuffer>& buffer :
+       sending_write_data_->buffers()) {
+    delegate_->OnDataSent(buffer->data());
+  }
+  sending_write_data_->Clear();
+  // Send data flushed while other data was sending.
+  if (!flushing_write_data_->Empty()) {
+    SendFlushingWriteData();
+    return;
+  }
+  if (write_end_of_stream_ && pending_write_data_->Empty())
     write_state_ = WRITING_DONE;
   MaybeOnSucceded();

[kapishnikov, 2016/06/16 16:16:47]

For future CL: I think MaybeOnSucceded() can go inside the 'if' body.

 }
@@ -195,7 +249,7 @@ void CronetBidirectionalStream::StartOnNetworkThread(
       std::move(request_info), environment_->GetURLRequestContext()
                                    ->http_transaction_factory()
                                    ->GetSession(),
-      /*send_request_headers_automatically=*/true, this));
+      !delay_headers_until_flush_, this));
   DCHECK(read_state_ == NOT_STARTED && write_state_ == NOT_STARTED);
   read_state_ = write_state_ = STARTED;
 }
@@ -234,19 +288,52 @@ void CronetBidirectionalStream::WriteDataOnNetworkThread(
     bool end_of_stream) {
   DCHECK(environment_->IsOnNetworkThread());
   DCHECK(write_buffer);
-  DCHECK(!write_buffer_);
-  if (write_state_ != WAITING_FOR_WRITE) {
-    DLOG(ERROR) << "Unexpected Write Data in write_state " << write_state_;
+  DCHECK(!write_end_of_stream_);
+  if (!bidi_stream_ || write_end_of_stream_) {
+    DLOG(ERROR) << "Unexpected Flush Data in write_state " << write_state_;
     // Invoke OnFailed unless it is already invoked.
     if (write_state_ != ERROR)
       OnFailed(net::ERR_UNEXPECTED);
     return;
   }
-  write_state_ = WRITING;
+  pending_write_data_->AppendBuffer(write_buffer, buffer_size);
   write_end_of_stream_ = end_of_stream;
+  if (!disable_auto_flush_)
+    FlushOnNetworkThread();
+}
 
-  write_buffer_ = write_buffer;
-  bidi_stream_->SendData(write_buffer_.get(), buffer_size, end_of_stream);
+void CronetBidirectionalStream::FlushOnNetworkThread() {
+  DCHECK(environment_->IsOnNetworkThread());
+  if (!bidi_stream_)
+    return;
+  // If there is no data to flush, may need to send headers.
+  if (pending_write_data_->Empty()) {
+    if (!request_headers_sent_) {
+      request_headers_sent_ = true;
+      bidi_stream_->SendRequestHeaders();
+    }
+    return;
+  }
+  // If request headers are not sent yet, they will be sent with the data.
+  if (!request_headers_sent_)
+    request_headers_sent_ = true;
+
+  // Move pending data to the flushing list.
+  pending_write_data_->MoveTo(flushing_write_data_.get());
+  DCHECK(pending_write_data_->Empty());
+  if (write_state_ != WRITING)
+    SendFlushingWriteData();
+}
+
+void CronetBidirectionalStream::SendFlushingWriteData() {
+  // If previous send is not done, or there is nothing to flush, then exit.
+  if (write_state_ == WRITING || flushing_write_data_->Empty())
+    return;
+  DCHECK(sending_write_data_->Empty());
+  write_state_ = WRITING;
+  flushing_write_data_->MoveTo(sending_write_data_.get());
+  bidi_stream_->SendvData(sending_write_data_->buffers(),
+                          sending_write_data_->lengths(), write_end_of_stream_);

[xunjieli, 2016/06/20 15:24:24]

While I was doing the same change on Android side, I noticed that there is a
problem with using |write_end_of_stream_| here. The bug is currently present in
both Android and iOS.

If we have written EndOfStream, but we have non-empty |pending_write_data_|. We
can incorrectly set EndOfStream = true for |flushing_write_data_|.

To reproduce, let's say we have 3 buffers to write. Write all three buffers at
once just after stream is ready, but only call flush after the second write.
stream.write(buffer1, /*eos*=/false);
stream.write(buffer2, /*eos*=/false);
stream.flush();
stream.write(buffer3, /*eos*=/true);

SendFlushingWriteData() for the first flush() can happen after the final write,
so |write_end_of_stream_| is true. SendFlushingWriteData() will end up sending
two buffers with EndOfStream = true. The next time we try to flush the last
buffer, it will be just no-op because writing is done :(

I reproduced on Android side, and I am working on a fix. 

 }
 
 void CronetBidirectionalStream::CancelOnNetworkThread() {