chrome/browser/chromeos/drive/sync_client.cc

   1 // Copyright (c) 2012 The Chromium Authors. All rights reserved.
   2 // Use of this source code is governed by a BSD-style license that can be
   3 // found in the LICENSE file.
   4
   5 #include "chrome/browser/chromeos/drive/sync_client.h"
   6
   7 #include <vector>
   8
   9 #include "base/bind.h"
  10 #include "base/message_loop/message_loop_proxy.h"
  11 #include "chrome/browser/chromeos/drive/drive.pb.h"
  12 #include "chrome/browser/chromeos/drive/file_cache.h"
  13 #include "chrome/browser/chromeos/drive/file_system/download_operation.h"
  14 #include "chrome/browser/chromeos/drive/file_system/operation_observer.h"
  15 #include "chrome/browser/chromeos/drive/file_system/update_operation.h"
  16 #include "chrome/browser/chromeos/drive/file_system_util.h"
  17 #include "chrome/browser/chromeos/drive/sync/entry_update_performer.h"
  18 #include "content/public/browser/browser_thread.h"
  19 #include "google_apis/drive/task_util.h"
  20
  21 using content::BrowserThread;
  22
  23 namespace drive {
  24 namespace internal {
  25
  26 namespace {
  27
  28 // The delay constant is used to delay processing a sync task. We should not
  29 // process SyncTasks immediately for the following reasons:
  30 //
  31 // 1) For fetching, the user may accidentally click on "Make available
  32 //    offline" checkbox on a file, and immediately cancel it in a second.
  33 //    It's a waste to fetch the file in this case.
  34 //
  35 // 2) For uploading, file writing via HTML5 file system API is performed in
  36 //    two steps: 1) truncate a file to 0 bytes, 2) write contents. We
  37 //    shouldn't start uploading right after the step 1). Besides, the user
  38 //    may edit the same file repeatedly in a short period of time.
  39 //
  40 // TODO(satorux): We should find a way to handle the upload case more nicely,
  41 // and shorten the delay. crbug.com/134774
  42 const int kDelaySeconds = 5;
  43
  44 // The delay constant is used to delay retrying a sync task on server errors.
  45 const int kLongDelaySeconds = 600;
  46
  47 // Iterates entries and appends IDs to |to_fetch| if the file is pinned but not
  48 // fetched (not present locally), to |to_upload| if the file is dirty but not
  49 // uploaded, or to |to_remove| if the entry is in the trash.
  50 void CollectBacklog(ResourceMetadata* metadata,
  51                     std::vector<std::string>* to_fetch,
  52                     std::vector<std::string>* to_upload,
  53                     std::vector<std::string>* to_update) {
  54   DCHECK(to_fetch);
  55   DCHECK(to_upload);
  56   DCHECK(to_update);
  57
  58   scoped_ptr<ResourceMetadata::Iterator> it = metadata->GetIterator();
  59   for (; !it->IsAtEnd(); it->Advance()) {
  60     const std::string& local_id = it->GetID();
  61     const ResourceEntry& entry = it->GetValue();
  62     if (entry.parent_local_id() == util::kDriveTrashDirLocalId) {
  63       to_update->push_back(local_id);
  64       continue;
  65     }
  66
  67     switch (entry.metadata_edit_state()) {
  68       case ResourceEntry::CLEAN:
  69         break;
  70       case ResourceEntry::SYNCING:
  71       case ResourceEntry::DIRTY:
  72         to_update->push_back(local_id);
  73         break;
  74     }
  75
  76     FileCacheEntry cache_entry;
  77     if (it->GetCacheEntry(&cache_entry)) {
  78       if (cache_entry.is_pinned() && !cache_entry.is_present())
  79         to_fetch->push_back(local_id);
  80
  81       if (cache_entry.is_dirty())
  82         to_upload->push_back(local_id);
  83     }
  84   }
  85   DCHECK(!it->HasError());
  86 }
  87
  88 // Iterates cache entries and collects IDs of ones with obsolete cache files.
  89 void CheckExistingPinnedFiles(ResourceMetadata* metadata,
  90                               FileCache* cache,
  91                               std::vector<std::string>* local_ids) {
  92   scoped_ptr<FileCache::Iterator> it = cache->GetIterator();
  93   for (; !it->IsAtEnd(); it->Advance()) {
  94     const FileCacheEntry& cache_entry = it->GetValue();
  95     const std::string& local_id = it->GetID();
  96     if (!cache_entry.is_pinned() || !cache_entry.is_present())
  97       continue;
  98
  99     ResourceEntry entry;
 100     FileError error = metadata->GetResourceEntryById(local_id, &entry);
 101     if (error != FILE_ERROR_OK) {
 102       LOG(WARNING) << "Entry not found: " << local_id;
 103       continue;
 104     }
 105
 106     // If MD5s don't match, it indicates the local cache file is stale, unless
 107     // the file is dirty (the MD5 is "local"). We should never re-fetch the
 108     // file when we have a locally modified version.
 109     if (entry.file_specific_info().md5() == cache_entry.md5() ||
 110         cache_entry.is_dirty())
 111       continue;
 112
 113     error = cache->Remove(local_id);
 114     if (error != FILE_ERROR_OK) {
 115       LOG(WARNING) << "Failed to remove cache entry: " << local_id;
 116       continue;
 117     }
 118
 119     error = cache->Pin(local_id);
 120     if (error != FILE_ERROR_OK) {
 121       LOG(WARNING) << "Failed to pin cache entry: " << local_id;
 122       continue;
 123     }
 124
 125     local_ids->push_back(local_id);
 126   }
 127   DCHECK(!it->HasError());
 128 }
 129
 130 }  // namespace
 131
 132 SyncClient::SyncTask::SyncTask() : state(PENDING), should_run_again(false) {}
 133 SyncClient::SyncTask::~SyncTask() {}
 134
 135 SyncClient::SyncClient(base::SequencedTaskRunner* blocking_task_runner,
 136                        file_system::OperationObserver* observer,
 137                        JobScheduler* scheduler,
 138                        ResourceMetadata* metadata,
 139                        FileCache* cache,
 140                        const base::FilePath& temporary_file_directory)
 141     : blocking_task_runner_(blocking_task_runner),
 142       operation_observer_(observer),
 143       metadata_(metadata),
 144       cache_(cache),
 145       download_operation_(new file_system::DownloadOperation(
 146           blocking_task_runner,
 147           observer,
 148           scheduler,
 149           metadata,
 150           cache,
 151           temporary_file_directory)),
 152       update_operation_(new file_system::UpdateOperation(blocking_task_runner,
 153                                                          observer,
 154                                                          scheduler,
 155                                                          metadata,
 156                                                          cache)),
 157       entry_update_performer_(new EntryUpdatePerformer(blocking_task_runner,
 158                                                        observer,
 159                                                        scheduler,
 160                                                        metadata)),
 161       delay_(base::TimeDelta::FromSeconds(kDelaySeconds)),
 162       long_delay_(base::TimeDelta::FromSeconds(kLongDelaySeconds)),
 163       weak_ptr_factory_(this) {
 164   DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
 165 }
 166
 167 SyncClient::~SyncClient() {
 168   DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
 169 }
 170
 171 void SyncClient::StartProcessingBacklog() {
 172   DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
 173
 174   std::vector<std::string>* to_fetch = new std::vector<std::string>;
 175   std::vector<std::string>* to_upload = new std::vector<std::string>;
 176   std::vector<std::string>* to_remove = new std::vector<std::string>;
 177   blocking_task_runner_->PostTaskAndReply(
 178       FROM_HERE,
 179       base::Bind(&CollectBacklog, metadata_, to_fetch, to_upload, to_remove),
 180       base::Bind(&SyncClient::OnGetLocalIdsOfBacklog,
 181                  weak_ptr_factory_.GetWeakPtr(),
 182                  base::Owned(to_fetch),
 183                  base::Owned(to_upload),
 184                  base::Owned(to_remove)));
 185 }
 186
 187 void SyncClient::StartCheckingExistingPinnedFiles() {
 188   DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
 189
 190   std::vector<std::string>* local_ids = new std::vector<std::string>;
 191   blocking_task_runner_->PostTaskAndReply(
 192       FROM_HERE,
 193       base::Bind(&CheckExistingPinnedFiles,
 194                  metadata_,
 195                  cache_,
 196                  local_ids),
 197       base::Bind(&SyncClient::AddFetchTasks,
 198                  weak_ptr_factory_.GetWeakPtr(),
 199                  base::Owned(local_ids)));
 200 }
 201
 202 void SyncClient::AddFetchTask(const std::string& local_id) {
 203   DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
 204   AddFetchTaskInternal(local_id, delay_);
 205 }
 206
 207 void SyncClient::RemoveFetchTask(const std::string& local_id) {
 208   DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
 209
 210   SyncTasks::iterator it = tasks_.find(SyncTasks::key_type(FETCH, local_id));
 211   if (it == tasks_.end())
 212     return;
 213
 214   SyncTask* task = &it->second;
 215   switch (task->state) {
 216     case PENDING:
 217       tasks_.erase(it);
 218       break;
 219     case RUNNING:
 220       // TODO(kinaba): Cancel tasks in JobScheduler as well. crbug.com/248856
 221       break;
 222   }
 223 }
 224
 225 void SyncClient::AddUploadTask(const ClientContext& context,
 226                                const std::string& local_id) {
 227   DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
 228   AddUploadTaskInternal(context, local_id,
 229                         file_system::UpdateOperation::RUN_CONTENT_CHECK,
 230                         delay_);
 231 }
 232
 233 void SyncClient::AddUpdateTask(const std::string& local_id) {
 234   DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
 235   AddUpdateTaskInternal(local_id, base::TimeDelta::FromSeconds(0));
 236 }
 237
 238 void SyncClient::AddFetchTaskInternal(const std::string& local_id,
 239                                       const base::TimeDelta& delay) {
 240   DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
 241
 242   SyncTask task;
 243   task.task = base::Bind(
 244       &file_system::DownloadOperation::EnsureFileDownloadedByLocalId,
 245       base::Unretained(download_operation_.get()),
 246       local_id,
 247       ClientContext(BACKGROUND),
 248       GetFileContentInitializedCallback(),
 249       google_apis::GetContentCallback(),
 250       base::Bind(&SyncClient::OnFetchFileComplete,
 251                  weak_ptr_factory_.GetWeakPtr(),
 252                  local_id));
 253   AddTask(SyncTasks::key_type(FETCH, local_id), task, delay);
 254 }
 255
 256 void SyncClient::AddUploadTaskInternal(
 257     const ClientContext& context,
 258     const std::string& local_id,
 259     file_system::UpdateOperation::ContentCheckMode content_check_mode,
 260     const base::TimeDelta& delay) {
 261   DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
 262
 263   SyncTask task;
 264   task.task = base::Bind(
 265       &file_system::UpdateOperation::UpdateFileByLocalId,
 266       base::Unretained(update_operation_.get()),
 267       local_id,
 268       context,
 269       content_check_mode,
 270       base::Bind(&SyncClient::OnUploadFileComplete,
 271                  weak_ptr_factory_.GetWeakPtr(),
 272                  local_id));
 273   AddTask(SyncTasks::key_type(UPLOAD, local_id), task, delay);
 274 }
 275
 276 void SyncClient::AddUpdateTaskInternal(const std::string& local_id,
 277                                        const base::TimeDelta& delay) {
 278   SyncTask task;
 279   task.task = base::Bind(
 280       &EntryUpdatePerformer::UpdateEntry,
 281       base::Unretained(entry_update_performer_.get()),
 282       local_id,
 283       base::Bind(&SyncClient::OnUpdateComplete,
 284                  weak_ptr_factory_.GetWeakPtr(),
 285                  local_id));
 286   AddTask(SyncTasks::key_type(UPDATE, local_id), task, delay);
 287 }
 288
 289 void SyncClient::AddTask(const SyncTasks::key_type& key,
 290                          const SyncTask& task,
 291                          const base::TimeDelta& delay) {
 292   DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
 293
 294   SyncTasks::iterator it = tasks_.find(key);
 295   if (it != tasks_.end()) {
 296     switch (it->second.state) {
 297       case PENDING:
 298         // The same task will run, do nothing.
 299         break;
 300       case RUNNING:
 301         // Something has changed since the task started. Schedule rerun.
 302         it->second.should_run_again = true;
 303         break;
 304     }
 305     return;
 306   }
 307
 308   DCHECK_EQ(PENDING, task.state);
 309   tasks_[key] = task;
 310
 311   base::MessageLoopProxy::current()->PostDelayedTask(
 312       FROM_HERE,
 313       base::Bind(&SyncClient::StartTask, weak_ptr_factory_.GetWeakPtr(), key),
 314       delay);
 315 }
 316
 317 void SyncClient::StartTask(const SyncTasks::key_type& key) {
 318   SyncTasks::iterator it = tasks_.find(key);
 319   if (it == tasks_.end())
 320     return;
 321
 322   SyncTask* task = &it->second;
 323   switch (task->state) {
 324     case PENDING:
 325       task->state = RUNNING;
 326       task->task.Run();
 327       break;
 328     case RUNNING:  // Do nothing.
 329       break;
 330   }
 331 }
 332
 333 void SyncClient::OnGetLocalIdsOfBacklog(
 334     const std::vector<std::string>* to_fetch,
 335     const std::vector<std::string>* to_upload,
 336     const std::vector<std::string>* to_update) {
 337   DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
 338
 339   // Give priority to upload tasks over fetch tasks, so that dirty files are
 340   // uploaded as soon as possible.
 341   for (size_t i = 0; i < to_upload->size(); ++i) {
 342     const std::string& local_id = (*to_upload)[i];
 343     DVLOG(1) << "Queuing to upload: " << local_id;
 344     AddUploadTaskInternal(ClientContext(BACKGROUND), local_id,
 345                           file_system::UpdateOperation::NO_CONTENT_CHECK,
 346                           delay_);
 347   }
 348
 349   for (size_t i = 0; i < to_fetch->size(); ++i) {
 350     const std::string& local_id = (*to_fetch)[i];
 351     DVLOG(1) << "Queuing to fetch: " << local_id;
 352     AddFetchTaskInternal(local_id, delay_);
 353   }
 354
 355   for (size_t i = 0; i < to_update->size(); ++i) {
 356     const std::string& local_id = (*to_update)[i];
 357     DVLOG(1) << "Queuing to update: " << local_id;
 358     AddUpdateTask(local_id);
 359   }
 360 }
 361
 362 void SyncClient::AddFetchTasks(const std::vector<std::string>* local_ids) {
 363   DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
 364
 365   for (size_t i = 0; i < local_ids->size(); ++i)
 366     AddFetchTask((*local_ids)[i]);
 367 }
 368
 369 bool SyncClient::OnTaskComplete(SyncType type, const std::string& local_id) {
 370   DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
 371
 372   const SyncTasks::key_type key(type, local_id);
 373   SyncTasks::iterator it = tasks_.find(key);
 374   DCHECK(it != tasks_.end());
 375
 376   if (it->second.should_run_again) {
 377     DVLOG(1) << "Running again: type = " << type << ", id = " << local_id;
 378     it->second.should_run_again = false;
 379     it->second.task.Run();
 380     return false;
 381   }
 382
 383   tasks_.erase(it);
 384   return true;
 385 }
 386
 387 void SyncClient::OnFetchFileComplete(const std::string& local_id,
 388                                      FileError error,
 389                                      const base::FilePath& local_path,
 390                                      scoped_ptr<ResourceEntry> entry) {
 391   DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
 392
 393   if (!OnTaskComplete(FETCH, local_id))
 394     return;
 395
 396   if (error == FILE_ERROR_OK) {
 397     DVLOG(1) << "Fetched " << local_id << ": " << local_path.value();
 398   } else {
 399     switch (error) {
 400       case FILE_ERROR_ABORT:
 401         // If user cancels download, unpin the file so that we do not sync the
 402         // file again.
 403         base::PostTaskAndReplyWithResult(
 404             blocking_task_runner_,
 405             FROM_HERE,
 406             base::Bind(&FileCache::Unpin, base::Unretained(cache_), local_id),
 407             base::Bind(&util::EmptyFileOperationCallback));
 408         break;
 409       case FILE_ERROR_NO_CONNECTION:
 410         // Add the task again so that we'll retry once the connection is back.
 411         AddFetchTaskInternal(local_id, delay_);
 412         break;
 413       case FILE_ERROR_SERVICE_UNAVAILABLE:
 414         // Add the task again so that we'll retry once the service is back.
 415         AddFetchTaskInternal(local_id, long_delay_);
 416         operation_observer_->OnDriveSyncError(
 417             file_system::DRIVE_SYNC_ERROR_SERVICE_UNAVAILABLE,
 418             local_id);
 419         break;
 420       default:
 421         LOG(WARNING) << "Failed to fetch " << local_id
 422                      << ": " << FileErrorToString(error);
 423     }
 424   }
 425 }
 426
 427 void SyncClient::OnUploadFileComplete(const std::string& local_id,
 428                                       FileError error) {
 429   DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
 430
 431   if (!OnTaskComplete(UPLOAD, local_id))
 432     return;
 433
 434   if (error == FILE_ERROR_OK) {
 435     DVLOG(1) << "Uploaded " << local_id;
 436   } else {
 437     switch (error) {
 438       case FILE_ERROR_NO_CONNECTION:
 439         // Add the task again so that we'll retry once the connection is back.
 440         AddUploadTaskInternal(ClientContext(BACKGROUND), local_id,
 441                               file_system::UpdateOperation::NO_CONTENT_CHECK,
 442                               delay_);
 443         break;
 444       case FILE_ERROR_SERVICE_UNAVAILABLE:
 445         // Add the task again so that we'll retry once the service is back.
 446         AddUploadTaskInternal(ClientContext(BACKGROUND), local_id,
 447                               file_system::UpdateOperation::NO_CONTENT_CHECK,
 448                               long_delay_);
 449         operation_observer_->OnDriveSyncError(
 450             file_system::DRIVE_SYNC_ERROR_SERVICE_UNAVAILABLE,
 451             local_id);
 452         break;
 453       default:
 454         LOG(WARNING) << "Failed to upload " << local_id << ": "
 455                      << FileErrorToString(error);
 456     }
 457   }
 458 }
 459
 460 void SyncClient::OnUpdateComplete(const std::string& local_id,
 461                                   FileError error) {
 462   DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
 463
 464   if (!OnTaskComplete(UPDATE, local_id))
 465     return;
 466
 467   if (error == FILE_ERROR_OK) {
 468     DVLOG(1) << "Updated " << local_id;
 469   } else {
 470     switch (error) {
 471       case FILE_ERROR_NO_CONNECTION:
 472         // Add the task again so that we'll retry once the connection is back.
 473         AddUpdateTaskInternal(local_id, base::TimeDelta::FromSeconds(0));
 474         break;
 475       case FILE_ERROR_SERVICE_UNAVAILABLE:
 476         // Add the task again so that we'll retry once the service is back.
 477         AddUpdateTaskInternal(local_id, long_delay_);
 478         operation_observer_->OnDriveSyncError(
 479             file_system::DRIVE_SYNC_ERROR_SERVICE_UNAVAILABLE,
 480             local_id);
 481         break;
 482       default:
 483         LOG(WARNING) << "Failed to update " << local_id << ": "
 484                      << FileErrorToString(error);
 485     }
 486   }
 487 }
 488
 489 }  // namespace internal
 490 }  // namespace drive