Implement retry flow on DriveUploader.

chrome/browser/google_apis/drive_uploader.cc

Index: chrome/browser/google_apis/drive_uploader.cc
diff --git a/chrome/browser/google_apis/drive_uploader.cc b/chrome/browser/google_apis/drive_uploader.cc
index 9be960f48138b0a7623d3bd2e3bedf337c25b9b2..7d9c89e5cc212cf215076fb123c97edc67363ab5 100644
--- a/chrome/browser/google_apis/drive_uploader.cc
+++ b/chrome/browser/google_apis/drive_uploader.cc
@@ -9,7 +9,9 @@
 #include "base/bind.h"
 #include "base/callback.h"
 #include "base/message_loop.h"
+#include "base/rand_util.h"
 #include "base/string_number_conversions.h"
+#include "base/time.h"
 #include "chrome/browser/google_apis/drive_service_interface.h"
 #include "chrome/browser/google_apis/drive_upload_mode.h"
 #include "chrome/browser/google_apis/gdata_wapi_parser.h"
@@ -36,6 +38,29 @@ int64 OpenFileStreamAndGetSizeOnBlockingPool(net::FileStream* file_stream,
   return file_stream->Available();
 }
 
+base::TimeDelta GetWaitDuration(int trial_count) {
+  // The waiting duration for the exponential back is
+  // 1st trial: 1sec + rand millisecs
+  // 2nd trial: 2secs + rand millisecs
+  // 3rd trial: 4secs + rand millisecs
+  // 4th trial: 8secs + rand millisecs
+  //   :
+  // nth trial: 2^(n-1) + rand millisecs
+  // Please see also
+  // https://developers.google.com/drive/manage-uploads#exp-backoff
+  // for more details.

[satorux1, 2013/02/07 05:48:32]

I thought we had similar code elsewhere. zork@ what do you think?

[Zachary Kuznia, 2013/02/07 06:26:36]

This should call into the scheduler instead.  That will both handle retry logic
for throttled cases, and suspend the operation when the user turns off drive or
switches to LTE.

The CL that makes the uploader wrapped by the scheduler is in progress.  I'll
update and reupload it presently.

[hidehiko, 2013/02/07 06:53:26]

So, all retrying logic will be in DriveScheduler somehow soon, right?
Sounds great, but then I have two questions:

1) DriveUploader is a state machine based on Chrome's message queue, and it is
completely independent from the DriveScheduler at the moment. How will they
communicate with each other? In particular, I'm interested in how to manage the
uploading state and to re-construct it when http-connection is failed, because
I'm about sending another CL to change it to more efficient way.

2) DriveUploader is used not only from chrome/browser/chromeos/drive/... but
also chrome/browser/sync_file_system/... What's the plan for it?

+  DCHECK_GT(trial_count, 0);
+
+  // Note: The range for RandInt is inclusive.
+  return base::TimeDelta::FromSeconds(1 << (trial_count - 1))
+      + base::TimeDelta::FromMilliseconds(base::RandInt(0, 9));
+}
+
+// Returns true if the status_code is 5xx SERVER ERROR, otherwise false.
+bool IsHttp5xxServerError(int status_code) {
+  return status_code >= 500 && status_code < 600;
+}
+
 }  // namespace
 
 namespace google_apis {
@@ -61,9 +86,10 @@ struct DriveUploader::UploadFileInfo {
         etag(etag),
         completion_callback(callback),
         content_length(0),
-        next_send_position(0),
         file_stream(new net::FileStream(NULL)),
         buf(new net::IOBuffer(kUploadChunkSize)),
+        buf_position(0),
+        buf_size(0),
         blocking_task_runner(task_runner),
         power_save_blocker(content::PowerSaveBlocker::Create(
             content::PowerSaveBlocker::kPowerSaveBlockPreventAppSuspension,
@@ -76,8 +102,8 @@ struct DriveUploader::UploadFileInfo {
 
   // Bytes left to upload.
   int64 SizeRemaining()  const {
-    DCHECK(content_length >= next_send_position);
-    return content_length - next_send_position;
+    DCHECK(content_length >= buf_position + buf_size);
+    return content_length - (buf_position + buf_size);
   }
 
   // Useful for printf debugging.
@@ -120,9 +146,6 @@ struct DriveUploader::UploadFileInfo {
   // Header content-Length.
   int64 content_length;
 
-  // The start position of the contents to be sent as the next upload chunk.
-  int64 next_send_position;
-
   // For opening and reading from physical file.
   //
   // File operations are posted to |blocking_task_runner|, while the ownership
@@ -135,6 +158,16 @@ struct DriveUploader::UploadFileInfo {
   // Holds current content to be uploaded.
   const scoped_refptr<net::IOBuffer> buf;
 
+  // The start position of the |buf| in the file.
+  int64 buf_position;
+
+  // The size of available data in |buf|.
+  int64 buf_size;
+
+  // The number of trials for the current buffer data.
+  // This is used to calculate exponential backoff waiting duration.
+  int trial_count;
+
   // Runner for net::FileStream tasks.
   const scoped_refptr<base::SequencedTaskRunner> blocking_task_runner;
 
@@ -329,10 +362,27 @@ void DriveUploader::ReadCompletionCallback(
     return;
   }
 
-  int64 start_position = upload_file_info->next_send_position;
-  upload_file_info->next_send_position += bytes_read;
-  int64 end_position = upload_file_info->next_send_position;
+  // Update the buffer's position.
+  // First, move the start position of the buffer by previous buffer size,
+  // and then store the current buffer size.
+  upload_file_info->buf_position += upload_file_info->buf_size;
+  upload_file_info->buf_size = bytes_read;
+
+  // Try to send all the data in the buffer.
+  int64 start_position = upload_file_info->buf_position;
+  ResumeUpload(upload_file_info.Pass(), start_position);
+}
 
+void DriveUploader::ResumeUpload(scoped_ptr<UploadFileInfo> upload_file_info,
+                                 int64 start_position) {
+  DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
+  int64 end_position =
+      upload_file_info->buf_position + upload_file_info->buf_size;
+  DCHECK_GE(start_position, upload_file_info->buf_position);
+  DCHECK_LE(start_position, end_position);
+  int64 buffer_offset = start_position - upload_file_info->buf_position;
+
+  // Try to send all the data in the buffer.
   UploadFileInfo* info_ptr = upload_file_info.get();
   drive_service_->ResumeUpload(
       ResumeUploadParams(info_ptr->upload_mode,
@@ -341,6 +391,7 @@ void DriveUploader::ReadCompletionCallback(
                          info_ptr->content_length,
                          info_ptr->content_type,
                          info_ptr->buf,
+                         buffer_offset,
                          info_ptr->upload_location,
                          info_ptr->drive_path),
       base::Bind(&DriveUploader::OnUploadRangeResponseReceived,
@@ -374,19 +425,30 @@ void DriveUploader::OnUploadRangeResponseReceived(
     return;
   }
 
+  if (IsHttp5xxServerError(response.code)) {
+    // The upload is interrupted. So retry with after waiting for a short
+    // period.
+    ResumeInterruptedUpload(upload_file_info.Pass());
+    return;
+  }
+
   // If code is 308 (RESUME_INCOMPLETE) and range_received is what has been
   // previously uploaded (i.e. = upload_file_info->end_position), proceed to
   // upload the next chunk.
   if (response.code != HTTP_RESUME_INCOMPLETE ||
       response.start_position_received != 0 ||
-      response.end_position_received != upload_file_info->next_send_position) {
+      response.end_position_received < upload_file_info->buf_position ||
+      response.end_position_received >
+          upload_file_info->buf_position + upload_file_info->buf_size) {
     // TODO(achuith): Handle error cases, e.g.
     // - when previously uploaded data wasn't received by Google Docs server,
     //   i.e. when end_position_received < upload_file_info->end_position
     LOG(ERROR) << "UploadNextChunk http code=" << response.code
         << ", start_position_received=" << response.start_position_received
         << ", end_position_received=" << response.end_position_received
-        << ", expected end range=" << upload_file_info->next_send_position;
+        << ", sending buffer = [" << upload_file_info->buf_position
+        << ":" << upload_file_info->buf_position + upload_file_info->buf_size
+        << ")";
     UploadFailed(upload_file_info.Pass(),
                  response.code == HTTP_FORBIDDEN ?
                      DRIVE_UPLOAD_ERROR_NO_SPACE : DRIVE_UPLOAD_ERROR_ABORT);
@@ -398,7 +460,56 @@ void DriveUploader::OnUploadRangeResponseReceived(
            << " for [" << upload_file_info->title << "]";
 
   // Continue uploading.
-  UploadNextChunk(upload_file_info.Pass());
+  if (response.end_position_received <
+          upload_file_info->buf_position + upload_file_info->buf_size) {
+    // There is the remaining data in the buffer, so resend it.
+    // PostTask is necessary here because we have to finish the callback
+    // before calling ResumeUpload due to the implementation of
+    // OperationRegistry. (http://crbug.com/134814)
+    MessageLoop::current()->PostTask(
+        FROM_HERE,
+        base::Bind(&DriveUploader::ResumeUpload,
+                   weak_ptr_factory_.GetWeakPtr(),
+                   base::Passed(&upload_file_info),
+                   response.end_position_received));
+  } else {
+    // The whole data in the current buffer is successfully sent.
+    // Send the next chunk.
+    UploadNextChunk(upload_file_info.Pass());
+  }
+}
+
+void DriveUploader::ResumeInterruptedUpload(
+    scoped_ptr<UploadFileInfo> upload_file_info) {
+  DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
+
+  // Calculate the waiting duration based on the number of trial.
+  // See GetWaitDuration for more details.
+  ++upload_file_info->trial_count;
+  base::TimeDelta wait_duration =
+      GetWaitDuration(upload_file_info->trial_count);
+
+  MessageLoop::current()->PostDelayedTask(
+      FROM_HERE,
+      base::Bind(&DriveUploader::ResumeInterruptedUploadAfterWait,
+                 weak_ptr_factory_.GetWeakPtr(),
+                 base::Passed(&upload_file_info)),
+      wait_duration);
+}
+
+void DriveUploader::ResumeInterruptedUploadAfterWait(
+    scoped_ptr<UploadFileInfo> upload_file_info) {
+  DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
+
+  UploadFileInfo* info_ptr = upload_file_info.get();
+  drive_service_->GetUploadStatus(
+      info_ptr->upload_mode,
+      info_ptr->drive_path,
+      info_ptr->upload_location,
+      info_ptr->content_length,
+      base::Bind(&DriveUploader::OnUploadRangeResponseReceived,
+                 weak_ptr_factory_.GetWeakPtr(),
+                 base::Passed(&upload_file_info)));
 }
 
 void DriveUploader::UploadFailed(scoped_ptr<UploadFileInfo> upload_file_info,