[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..3a9e4e0932192583929a97a2daaba69134254276 100644
--- a/components/cronet/ios/cronet_bidirectional_stream.cc
+++ b/components/cronet/ios/cronet_bidirectional_stream.cc
@@ -32,12 +32,19 @@
 
 namespace cronet {
 
+CronetBidirectionalStream::PendingWriteData::PendingWriteData() {}
+
+CronetBidirectionalStream::PendingWriteData::~PendingWriteData() {}
+
 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),
       delegate_(delegate) {}
 
@@ -98,6 +105,13 @@ bool CronetBidirectionalStream::WriteData(const char* buffer,
   return true;
 }
 
+bool CronetBidirectionalStream::Flush() {
+  environment_->PostToNetworkThread(
+      FROM_HERE, base::Bind(&CronetBidirectionalStream::FlushOnNetworkThread,
+                            base::Unretained(this)));
+  return true;
+}
+
 void CronetBidirectionalStream::Cancel() {
   environment_->PostToNetworkThread(
       FROM_HERE, base::Bind(&CronetBidirectionalStream::CancelOnNetworkThread,
@@ -115,17 +129,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(write_state_, STARTED);
+  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(read_state_, STARTED);

[xunjieli, 2016/06/10 20:31:37]

nit: the expected state should be the first arg.
DCHECK_EQ(STARTED, read_state_);

[mef, 2016/06/10 21:15:14]

Hrm, I think that's true for ASSERT_EQ, but I see DCHECK_EQ both ways. I don't
mind changing though.

[xunjieli, 2016/06/10 22:22:09]

I have been doing that way. But if there are places in the code base that use it
in the other way, then I am not sure. Looking at the header file, DCHECK_EQ does
put the variable in the second argument:
https://cs.chromium.org/chromium/src/base/logging.h?rcl=0&l=746.

[kapishnikov, 2016/06/13 21:21:57]

Found this one: https://bugs.chromium.org/p/chromium/issues/detail?id=58409#c13.
The order of DCHECK_EQ arguments is "expected", "actual".

[mef, 2016/06/14 00:20:38]

Done.

   read_state_ = WAITING_FOR_READ;
   // Get http status code from response headers.
   int http_status_code = 0;
@@ -148,7 +169,7 @@ void CronetBidirectionalStream::OnHeadersReceived(
 
 void CronetBidirectionalStream::OnDataRead(int bytes_read) {
   DCHECK(environment_->IsOnNetworkThread());
-  DCHECK(read_state_ == READING);
+  DCHECK_EQ(read_state_, READING);
   read_state_ = WAITING_FOR_READ;
   delegate_->OnDataRead(read_buffer_->data(), bytes_read);
 
@@ -161,11 +182,18 @@ 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;
+  DCHECK_EQ(write_state_, WRITING);
+  write_state_ = WAITING_FOR_FLUSH;
+  for (scoped_refptr<net::IOBuffer> buffer :
+       sending_write_data_->write_buffer_list) {
+    delegate_->OnDataSent(buffer->data());
+  }
+  sending_write_data_.reset();
+  // Send data flushed while other data was sending.
+  if (flushing_write_data_) {
+    FlushOnNetworkThread();
+    return;
+  }
   if (write_end_of_stream_)
     write_state_ = WRITING_DONE;
   MaybeOnSucceded();
@@ -195,7 +223,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 +262,55 @@ void CronetBidirectionalStream::WriteDataOnNetworkThread(
     bool end_of_stream) {
   DCHECK(environment_->IsOnNetworkThread());

[kapishnikov, 2016/06/13 21:21:57]

Can we add DCHECK here to check that write_end_of_stream_ is 'false'? Maybe we
should call onFailed()?

[mef, 2016/06/14 00:20:37]

Done.

   DCHECK(write_buffer);
-  DCHECK(!write_buffer_);
-  if (write_state_ != WAITING_FOR_WRITE) {
-    DLOG(ERROR) << "Unexpected Write Data in write_state " << write_state_;
+  if (!bidi_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;
+  if (!pending_write_data_)
+    pending_write_data_.reset(new PendingWriteData());
+  pending_write_data_->write_buffer_list.push_back(write_buffer);

[kapishnikov, 2016/06/13 21:21:57]

Can we create a method inside PendingWriteData struct and combine
write_buffer_list.push_back() and write_buffer_len_list.push_back() in this
method? E.g.,
pending_write_data_->AppendBuffer(write_buffer, buffer_size).

I think it will make it more object-oriented.

[mef, 2016/06/14 00:20:38]

Done.

+  pending_write_data_->write_buffer_len_list.push_back(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_) {
+    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.
+  request_headers_sent_ = true;
+  // Move pending data to the flushing list.
+  if (flushing_write_data_) {

[xunjieli, 2016/06/10 20:31:37]

I think it is slightly confusing that |flushing_write_data_| and
|pending_write_data_| can be in three states: (1) null, (2) not-null but empty,
(3) not-null and not empty.

It might be easier if we have two std::vector<WrappedBufferWithSize>, which can
only be empty or not empty, but never null. Can create a WrappedBufferWithSize
which is a subclass of net::WrappedBuffer, but with a size.

WDYT?

[mef, 2016/06/10 21:15:14]

That's an interesting obeservation. With FlushOnNetworkThread doing moving data
intp flushing vector anyway, I wonder whether we actually need 2 separate lists.
I'll think about it.

[mef, 2016/06/14 00:20:38]

Done. I think. :)

+    std::move(pending_write_data_->write_buffer_list.begin(),

[kapishnikov, 2016/06/13 21:21:57]

Maybe a new method in PendingWriteData struct? E.g.,
flushing_write_data_->MoveDataBuffers(pending_write_data_).

[mef, 2016/06/14 00:20:37]

I think we don't need to separate pending_write_data from flusing_write_data as
we always flush everything flushed before explicit Flush or onDataSent.

+              pending_write_data_->write_buffer_list.end(),
+              std::back_inserter(flushing_write_data_->write_buffer_list));
+    std::move(pending_write_data_->write_buffer_len_list.begin(),
+              pending_write_data_->write_buffer_len_list.end(),
+              std::back_inserter(flushing_write_data_->write_buffer_len_list));
+  } else {
+    std::swap(flushing_write_data_, pending_write_data_);
+  }
+  // If previous send is not done, or there is nothing to flush, then exit.
+  if ((write_state_ == WRITING) || !flushing_write_data_)

[kapishnikov, 2016/06/13 21:21:56]

I think the content of flushing_write_data_ can never be NULL here. Should we
add DCHECK instead?

[mef, 2016/06/14 00:20:38]

Done.

+    return;
+  write_state_ = WRITING;
+  sending_write_data_.swap(flushing_write_data_);

[kapishnikov, 2016/06/13 21:21:57]

Should we add DCHECK that the content of sending_write_data_ is NULL? I assume
that should always be true here.

[mef, 2016/06/14 00:20:38]

Done.

+  bidi_stream_->SendvData(sending_write_data_->write_buffer_list,
+                          sending_write_data_->write_buffer_len_list,
+                          write_end_of_stream_);
 }
 
 void CronetBidirectionalStream::CancelOnNetworkThread() {