[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..f987b26e5ca23bb82d489ca28323f4d8a26271a2 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,
@@ -116,9 +130,16 @@ 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_)
+  request_headers_sent_ = request_headers_sent;
+  write_state_ = WAITING_FOR_FLUSH;
+  if (write_end_of_stream_) {
+    if (!request_headers_sent) {

[xunjieli, 2016/06/10 16:30:49]

Should we not check |write_end_stream_|? Currently if it is a GET request and
|delay_headers_until_flush_| is set, we delay the headers until flush() in Java.
I wonder whether we should do the same.

[mef, 2016/06/10 18:36:36]

But that would never come if auto flush is enabled, right?

[xunjieli, 2016/06/10 19:00:01]

You are right. Hopefully we will get rid of auto flush soon, so we don't run
into this case.

+      // 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();
 }
 
@@ -162,10 +183,17 @@ void CronetBidirectionalStream::OnDataRead(int bytes_read) {
 void CronetBidirectionalStream::OnDataSent() {
   DCHECK(environment_->IsOnNetworkThread());
   DCHECK(write_state_ == WRITING);

[xunjieli, 2016/06/10 16:30:49]

DCHECK_EQ?

[mef, 2016/06/10 18:36:36]

Done.

-  write_state_ = WAITING_FOR_WRITE;
-  delegate_->OnDataSent(write_buffer_->data());
-  // Free the write buffer.
-  write_buffer_ = nullptr;
+  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_) {
+    SendFlushingWriteData();

[xunjieli, 2016/06/10 16:30:50]

Maybe invoke FlushOnNetworkThread() directly? This is match the Java
implementation more closely. Having one fewer method is easier to manage.

[mef, 2016/06/10 18:36:36]

I think they are different.
FlushOnNetworkThread is posted from app call to Flush() at any time.
SendFlushingWriteData() is called when we are ready to send data that was
accumulated for flushing.

+    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,68 @@ 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_;
+  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);
+  pending_write_data_->write_buffer_len_list.push_back(buffer_size);
   write_end_of_stream_ = end_of_stream;
+  if (!disable_auto_flush_)
+    FlushOnNetworkThread();
+}
+
+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_) {

[xunjieli, 2016/06/10 16:30:50]

Also need to check |flushing_write_data_| ?

[mef, 2016/06/10 18:36:36]

Hrm, I think if |flushing_write_data_| is not null, then it means that
FlushOnNetworkThread was called before, and headers will be sent with the 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_) {
+    std::move(pending_write_data_->write_buffer_list.begin(),
+              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 (!sending_write_data_)

[xunjieli, 2016/06/10 19:00:01]

Can this check be just "if (write_state_ != WRITING)"?

[mef, 2016/06/10 19:11:28]

Done.

+    SendFlushingWriteData();
+}
 
-  write_buffer_ = write_buffer;
-  bidi_stream_->SendData(write_buffer_.get(), buffer_size, end_of_stream);
+void CronetBidirectionalStream::SendFlushingWriteData() {
+  DCHECK(environment_->IsOnNetworkThread());
+  DCHECK(!sending_write_data_);
+  if (write_state_ != WAITING_FOR_FLUSH) {

[xunjieli, 2016/06/10 16:30:50]

Flush() can be called anytime after OnStreamReady() (even after stream is
destroyed), so we need to handle that.

In FlushOnNetworkThread, maybe check that write_state_ == WAITING_FOR_FLUSH, if
not, just accumulate the pending write data in the flushing write data?

Suggest combining this method with FlushOnNetworkThread() method. It might be
slightly easier to reason. 

[mef, 2016/06/10 18:36:36]

SendFlushingWriteData is similar to nativeWritevData, it never gets called
directly by the app, only from FlushOnNetworkThread or onDataSent when previous
set of buffers has flushed.

[xunjieli, 2016/06/10 19:00:01]

I learned from Andrei that we should re-use routine where possible during the
review of the Android version of this change. It looks like we can combine
SendFlushingWriteData() with the logic in FlushOnNetworkThread(). In
OnDataSent(), we can invoke FlushOnNetworkThread() directly.
FlushOnNetworkThread() will send pending data and/or pending flush data
accordingly. I think it might be a cleaner interface that way.

[xunjieli, 2016/06/10 19:04:38]

In other words, iI am wondering if we can get rid of SendFlushingWriteData(),
and move the logic to FlushOnNetworkThread(). The SendFlushingWriteData() method
name is a bit unclear, and it seems that it is doing what FlushOnNetworkThread()
should do. If it is, making it into a separate method might be confusing.

[mef, 2016/06/10 19:11:28]

I don't think we can though. FlushOnNetworkThread moves all pending write data
into flushing_write_data_ and SendFlushingWriteData sends flushing_write_data_
if any.

We don't want to flush pending data from onDataSent as it maybe incomplete (e.g.
app written 2 buffers out of 5 and hasn't called Flush yet).

[mef, 2016/06/10 20:03:48]

Per offline discussion this is current behavior on Android, and should be ok, so
I've changed iOS implementation to match.

+    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;
+  }
+  if (!flushing_write_data_)
+    return;
+  write_state_ = WRITING;
+  sending_write_data_.swap(flushing_write_data_);
+  bidi_stream_->SendvData(sending_write_data_->write_buffer_list,
+                          sending_write_data_->write_buffer_len_list,
+                          write_end_of_stream_);
 }
 
 void CronetBidirectionalStream::CancelOnNetworkThread() {