gdata: Convert GDataFileSystem::AddUploadFile() to asynchronous.

chrome/browser/chromeos/gdata/gdata_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/chromeos/gdata/gdata_uploader.h"
 
 #include <algorithm>
 
 #include "base/bind.h"
 #include "base/callback.h"
@@ skipping 34 matching lines @@
 int GDataUploader::UploadFile(scoped_ptr<UploadFileInfo> upload_file_info) {
   DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
   DCHECK(upload_file_info.get());
   DCHECK_EQ(upload_file_info->upload_id, -1);
   DCHECK(!upload_file_info->file_path.empty());
   DCHECK_NE(upload_file_info->file_size, 0);
   DCHECK(!upload_file_info->gdata_path.empty());
   DCHECK(!upload_file_info->title.empty());
   DCHECK(!upload_file_info->content_type.empty());
 
+  base::AutoLock lock(lock_);
+
   const int upload_id = next_upload_id_++;
   upload_file_info->upload_id = upload_id;
   // Add upload_file_info to our internal map and take ownership.
   pending_uploads_[upload_id] = upload_file_info.release();
 
   UploadFileInfo* info = GetUploadFileInfo(upload_id);
   DVLOG(1) << "Uploading file: " << info->DebugString();
 
   // Create a FileStream to make sure the file can be opened successfully.
   info->file_stream = new net::FileStream(NULL);
 
   // Create buffer to hold upload data.
   info->buf_len = std::min(info->file_size, kUploadChunkSize);
   info->buf = new net::IOBuffer(info->buf_len);
 
   OpenFile(info);
   return upload_id;
 }
 
 void GDataUploader::UpdateUpload(int upload_id,
                                  content::DownloadItem* download) {

[satorux1, 2012/06/14 21:32:34]

on what thread is this function called? In other places, we are mostly on UI
thread, but we don't have a DCHECK here

could you add DCHECK(BrowserThread::CurrentlyOn(...))?

[hshi1, 2012/06/14 21:40:17]

Done.

[satorux1, 2012/06/14 21:47:25]

Hmm, then seems to me that all functions run on UI thread. If so, we don't need
a lock?

+  base::AutoLock lock(lock_);
+
   UploadFileInfo* upload_file_info = GetUploadFileInfo(upload_id);
   if (!upload_file_info)
     return;
 
   const int64 file_size = download->GetReceivedBytes();
 
   // Update file_size and all_bytes_present.
   DVLOG(1) << "Updating file size from " << upload_file_info->file_size
            << " to " << file_size
            << (download->AllDataSaved() ? " (AllDataSaved)" : " (In-progress)");
@@ skipping 41 matching lines @@
   UploadFileInfo* upload_info = GetUploadFileInfo(upload_id);
   // We return the start_range as the count of uploaded bytes since that is the
   // start of the next or currently uploading chunk.
   // TODO(asanka): Use a finer grained progress value than this. We end up
   //               reporting progress in kUploadChunkSize increments.
   return upload_info ? upload_info->start_range : 0;
 }
 
 UploadFileInfo* GDataUploader::GetUploadFileInfo(int upload_id) const {
   DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
+  lock_.AssertAcquired();
 
   UploadFileInfoMap::const_iterator it = pending_uploads_.find(upload_id);
   DVLOG_IF(1, it == pending_uploads_.end()) << "No upload found for id "
                                             << upload_id;
   return it != pending_uploads_.end() ? it->second : NULL;
 }
 
 void GDataUploader::OpenFile(UploadFileInfo* upload_file_info) {
   // Open the file asynchronously.
   const int rv = upload_file_info->file_stream->Open(
       upload_file_info->file_path,
       base::PLATFORM_FILE_OPEN |
       base::PLATFORM_FILE_READ |
       base::PLATFORM_FILE_ASYNC,
       base::Bind(&GDataUploader::OpenCompletionCallback,
                  uploader_factory_.GetWeakPtr(),
                  upload_file_info->upload_id));
   DCHECK_EQ(net::ERR_IO_PENDING, rv);
 }
 
 void GDataUploader::OpenCompletionCallback(int upload_id, int result) {
   DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
+  base::AutoLock lock(lock_);
 
   UploadFileInfo* upload_file_info = GetUploadFileInfo(upload_id);
   if (!upload_file_info)
     return;
 
   // The file may actually not exist yet, as the downloads system downloads
   // to a temp location and then renames the file. If this is the case, we
   // just retry opening the file later.
   if (result != net::OK) {
     DCHECK_EQ(result, net::ERR_FILE_NOT_FOUND);
@@ skipping 32 matching lines @@
       base::Bind(&GDataUploader::OnUploadLocationReceived,
                  uploader_factory_.GetWeakPtr(),
                  upload_file_info->upload_id));
 }
 
 void GDataUploader::OnUploadLocationReceived(
     int upload_id,
     GDataErrorCode code,
     const GURL& upload_location) {
   DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
+  base::AutoLock lock(lock_);
 
   UploadFileInfo* upload_file_info = GetUploadFileInfo(upload_id);
   if (!upload_file_info)
     return;
 
   DVLOG(1) << "Got upload location [" << upload_location.spec()
            << "] for [" << upload_file_info->title << "]";
 
   if (code != HTTP_SUCCESS) {
     // TODO(achuith): Handle error codes from Google Docs server.
     UploadFailed(upload_file_info, base::PLATFORM_FILE_ERROR_ABORT);
     return;
   }
 
   upload_file_info->upload_location = upload_location;
 
   // Start the upload from the beginning of the file.
   UploadNextChunk(upload_file_info);
 }
 
 void GDataUploader::UploadNextChunk(UploadFileInfo* upload_file_info) {
   DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
+  lock_.AssertAcquired();
+
   // Check that |upload_file_info| is in pending_uploads_.
   DCHECK(upload_file_info == GetUploadFileInfo(upload_file_info->upload_id));
   DVLOG(1) << "Number of pending uploads=" << pending_uploads_.size();
 
   // Determine number of bytes to read for this upload iteration, which cannot
   // exceed size of buf i.e. buf_len.
   const int64 bytes_remaining = upload_file_info->SizeRemaining();
   const int bytes_to_read = std::min(upload_file_info->SizeRemaining(),
                                      upload_file_info->buf_len);
 
@@ skipping 20 matching lines @@
                  bytes_to_read));
 }
 
 void GDataUploader::ReadCompletionCallback(
     int upload_id,
     int bytes_to_read,
     int bytes_read) {
   // The Read is asynchronously executed on BrowserThread::UI, where
   // Read() was called.
   DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
+  base::AutoLock lock(lock_);
+
   DVLOG(1) << "ReadCompletionCallback bytes read=" << bytes_read;
 
   UploadFileInfo* upload_file_info = GetUploadFileInfo(upload_id);
   if (!upload_file_info)
     return;
 
   // TODO(achuith): Handle this error.
   DCHECK_EQ(bytes_to_read, bytes_read);
   DCHECK_GT(bytes_read, 0) << "Error reading from file "
                            << upload_file_info->file_path.value();
@@ skipping 14 matching lines @@
       base::Bind(&GDataUploader::OnResumeUploadResponseReceived,
                  uploader_factory_.GetWeakPtr(),
                  upload_file_info->upload_id));
 }
 
 void GDataUploader::OnResumeUploadResponseReceived(
     int upload_id,
     const ResumeUploadResponse& response,
     scoped_ptr<DocumentEntry> entry) {
   DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
+  base::AutoLock lock(lock_);
 
   UploadFileInfo* upload_file_info = GetUploadFileInfo(upload_id);
   if (!upload_file_info)
     return;
 
   if (response.code == HTTP_CREATED) {
     DVLOG(1) << "Successfully created uploaded file=["
              << upload_file_info->title;
 
     // Done uploading.
     upload_file_info->entry = entry.Pass();
     if (!upload_file_info->completion_callback.is_null()) {
       upload_file_info->completion_callback.Run(base::PLATFORM_FILE_OK,
                                                 upload_file_info);
     }
-    // TODO(achuith): DeleteUpload() here and let clients call
-    // GDataFileSystem::AddUploadedFile.
+
+    // Remove |upload_file_info| from the UploadFileInfoMap. The UploadFileInfo
+    // object will be deleted later upon transfer completion.
+    RemoveUpload(upload_file_info);
     return;
   }
 
   // If code is 308 (RESUME_INCOMPLETE) and range_received is what has been
   // previously uploaded (i.e. = upload_file_info->end_range), proceed to
   // upload the next chunk.
   if (response.code != HTTP_RESUME_INCOMPLETE ||
       response.start_range_received != 0 ||
       response.end_range_received != upload_file_info->end_range) {
     // TODO(achuith): Handle error cases, e.g.
@@ skipping 13 matching lines @@
   DVLOG(1) << "Received range " << response.start_range_received
            << "-" << response.end_range_received
            << " for [" << upload_file_info->title << "]";
 
   // Continue uploading.
   UploadNextChunk(upload_file_info);
 }
 
 void GDataUploader::MoveFileToCache(UploadFileInfo* upload_file_info) {
   DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
+  lock_.AssertAcquired();
+
   if (upload_file_info->entry == NULL)
     return;
 
   DVLOG(1) << "MoveFileToCache " << upload_file_info->file_path.value();
   file_system_->AddUploadedFile(
       upload_file_info->gdata_path.DirName(),
       upload_file_info->entry.get(),
       upload_file_info->file_path,
-      GDataCache::FILE_OPERATION_MOVE);
-  DeleteUpload(upload_file_info);
+      GDataCache::FILE_OPERATION_MOVE,
+      base::Bind(&GDataUploader::OnAddUploadFileComplete,
+                 uploader_factory_.GetWeakPtr(),
+                 upload_file_info));
+
+  // Remove |upload_file_info| from the UploadFileInfoMap. The UploadFileInfo
+  // object will be deleted in OnAddUploadFileComplete.
+  RemoveUpload(upload_file_info);
+}
+
+void GDataUploader::OnAddUploadFileComplete(UploadFileInfo* upload_file_info) {
+  DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
+  delete upload_file_info;
 }
 
 void GDataUploader::UploadFailed(UploadFileInfo* upload_file_info,
                                  base::PlatformFileError error) {
   DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
+  lock_.AssertAcquired();
+
   LOG(ERROR) << "Upload failed " << upload_file_info->DebugString();
   if (!upload_file_info->completion_callback.is_null()) {
     upload_file_info->completion_callback.Run(error,
                                               upload_file_info);
   }
-  DeleteUpload(upload_file_info);
-}
 
-void GDataUploader::DeleteUpload(UploadFileInfo* upload_file_info) {
-  DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
-
-  DVLOG(1) << "Deleting upload " << upload_file_info->gdata_path.value();
-  pending_uploads_.erase(upload_file_info->upload_id);
-
-  // The file stream is closed by the destructor asynchronously.
-  delete upload_file_info->file_stream;
+  RemoveUpload(upload_file_info);
   delete upload_file_info;
 }
 
+void GDataUploader::RemoveUpload(UploadFileInfo* upload_file_info) {
+  DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
+  lock_.AssertAcquired();
+
+  DVLOG(1) << "Removing upload " << upload_file_info->gdata_path.value();
+  pending_uploads_.erase(upload_file_info->upload_id);
+}
+
 }  // namespace gdata