Implement retry flow on DriveUploader.

chrome/browser/google_apis/drive_uploader.cc

 // Copyright (c) 2012 The Chromium Authors. All rights reserved.
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
 #include "chrome/browser/google_apis/drive_uploader.h"
 
 #include <algorithm>
 
 #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"
 #include "content/public/browser/browser_thread.h"
 #include "content/public/browser/power_save_blocker.h"
 #include "net/base/file_stream.h"
 #include "net/base/net_errors.h"
 
 using content::BrowserThread;
 
 namespace {
 
 // Google Documents List API requires uploading in chunks of 512kB.
 const int64 kUploadChunkSize = 512 * 1024;
 
 // Opens |path| with |file_stream| and returns the file size.
 // If failed, returns an error code in a negative value.
 int64 OpenFileStreamAndGetSizeOnBlockingPool(net::FileStream* file_stream,
                                              const FilePath& path) {
   int result = file_stream->OpenSync(
       path, base::PLATFORM_FILE_OPEN | base::PLATFORM_FILE_READ);
   if (result != net::OK)
     return result;
   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 {
 
 // Structure containing current upload information of file, passed between
 // DriveServiceInterface methods and callbacks.
 struct DriveUploader::UploadFileInfo {
   UploadFileInfo(scoped_refptr<base::SequencedTaskRunner> task_runner,
                  UploadMode upload_mode,
                  const GURL& initial_upload_location,
                  const FilePath& drive_path,
                  const FilePath& local_path,
                  const std::string& title,
                  const std::string& content_type,
                  const std::string& etag,
                  const UploadCompletionCallback& callback)
       : upload_mode(upload_mode),
         initial_upload_location(initial_upload_location),
         drive_path(drive_path),
         file_path(local_path),
         title(title),
         content_type(content_type),
         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,
             "Upload in progress")) {
   }
 
   ~UploadFileInfo() {
     blocking_task_runner->DeleteSoon(FROM_HERE, file_stream.release());
   }
 
   // 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.
   std::string DebugString() const {
     return "title=[" + title +
            "], file_path=[" + file_path.AsUTF8Unsafe() +
            "], content_type=[" + content_type +
            "], content_length=[" + base::UintToString(content_length) +
            "], drive_path=[" + drive_path.AsUTF8Unsafe() +
            "]";
@@ skipping 22 matching lines @@
   // Callback to be invoked once the upload has finished.
   const UploadCompletionCallback completion_callback;
 
   // Location URL where file is to be uploaded to, returned from
   // InitiateUpload. Used for the subsequent ResumeUpload requests.
   GURL upload_location;
 
   // 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
   // of the stream is held in UI thread. At the point when this UploadFileInfo
   // is destroyed, the ownership of the stream is passed to the worker pool.
   // TODO(kinaba): We should switch to async API of FileStream once
   // crbug.com/164312 is fixed.
   scoped_ptr<net::FileStream> file_stream;
 
   // 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;
 
   // Blocks system suspend while upload is in progress.
   scoped_ptr<content::PowerSaveBlocker> power_save_blocker;
 };
 
 DriveUploader::DriveUploader(DriveServiceInterface* drive_service)
     : drive_service_(drive_service),
       ALLOW_THIS_IN_INITIALIZER_LIST(weak_ptr_factory_(this)) {
@@ skipping 174 matching lines @@
   DCHECK_EQ(bytes_to_read, bytes_read);
   DVLOG(1) << "ReadCompletionCallback bytes read=" << bytes_read;
 
   if (bytes_read < 0) {
     LOG(ERROR) << "Error reading from file "
                << upload_file_info->file_path.value();
     UploadFailed(upload_file_info.Pass(), DRIVE_UPLOAD_ERROR_ABORT);
     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,
                          start_position,
                          end_position,
                          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,
                  weak_ptr_factory_.GetWeakPtr(),
                  base::Passed(&upload_file_info)));
 }
 
 void DriveUploader::OnUploadRangeResponseReceived(
     scoped_ptr<UploadFileInfo> upload_file_info,
     const UploadRangeResponse& response,
@@ skipping 13 matching lines @@
                                               entry.Pass());
     return;
   }
 
   // ETag mismatch.
   if (response.code == HTTP_PRECONDITION) {
     UploadFailed(upload_file_info.Pass(), DRIVE_UPLOAD_ERROR_CONFLICT);
     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);
     return;
   }
 
   DVLOG(1) << "Received range " << response.start_position_received
            << "-" << response.end_position_received
            << " 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,
                                  DriveUploadError error) {
   DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
 
   LOG(ERROR) << "Upload failed " << upload_file_info->DebugString();
 
   upload_file_info->completion_callback.Run(error,
                                             upload_file_info->drive_path,
                                             upload_file_info->file_path,
                                             scoped_ptr<ResourceEntry>());
 }
 
 }  // namespace google_apis