[SyncFS] Make the snapshot sync of local-to-remote sync parallelization enabled. (2/3)

chrome/browser/sync_file_system/local/local_file_change_tracker.cc

 // Copyright 2013 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/sync_file_system/local/local_file_change_tracker.h"
 
 #include <queue>
 
 #include "base/location.h"
 #include "base/logging.h"
@@ skipping 61 matching lines @@
 
 // LocalFileChangeTracker ------------------------------------------------------
 
 LocalFileChangeTracker::LocalFileChangeTracker(
     const base::FilePath& base_path,
     leveldb::Env* env_override,
     base::SequencedTaskRunner* file_task_runner)
     : initialized_(false),
       file_task_runner_(file_task_runner),
       tracker_db_(new TrackerDB(base_path, env_override)),
-      current_change_seq_(0),
+      current_change_seq_number_(0),
       num_changes_(0) {
 }
 
 LocalFileChangeTracker::~LocalFileChangeTracker() {
   DCHECK(file_task_runner_->RunsTasksOnCurrentThread());
   tracker_db_.reset();
 }
 
 void LocalFileChangeTracker::OnStartUpdate(const FileSystemURL& url) {
   DCHECK(file_task_runner_->RunsTasksOnCurrentThread());
-  if (ContainsKey(changes_, url))
+  if (ContainsKey(changes_, url) || ContainsKey(demoted_changes_, url))
     return;
   // TODO(nhiroki): propagate the error code (see http://crbug.com/152127).
   MarkDirtyOnDatabase(url);
 }
 
 void LocalFileChangeTracker::OnEndUpdate(const FileSystemURL& url) {}
 
 void LocalFileChangeTracker::OnCreateFile(const FileSystemURL& url) {
   RecordChange(url, FileChange(FileChange::FILE_CHANGE_ADD_OR_UPDATE,
                                SYNC_FILE_TYPE_FILE));
@@ skipping 75 matching lines @@
 }
 
 void LocalFileChangeTracker::RemoveMirrorAndCommitChangesForURL(
     const fileapi::FileSystemURL& url) {
   DCHECK(file_task_runner_->RunsTasksOnCurrentThread());
   FileChangeMap::iterator found = mirror_changes_.find(url);
   if (found == mirror_changes_.end())
     return;
   mirror_changes_.erase(found);
 
-  if (ContainsKey(changes_, url))
+  if (ContainsKey(changes_, url) || ContainsKey(demoted_changes_, url))
     MarkDirtyOnDatabase(url);
   else
     ClearDirtyOnDatabase(url);
   UpdateNumChanges();
 }
 
 void LocalFileChangeTracker::ResetToMirrorAndCommitChangesForURL(
     const fileapi::FileSystemURL& url) {
   DCHECK(file_task_runner_->RunsTasksOnCurrentThread());
   FileChangeMap::iterator found = mirror_changes_.find(url);
   if (found == mirror_changes_.end() || found->second.change_list.empty()) {
     ClearChangesForURL(url);
     return;
   }
   const ChangeInfo& info = found->second;
-  change_seqs_[info.change_seq] = url;
-  changes_[url] = info;
+  if (ContainsKey(demoted_changes_, url)) {
+    DCHECK(!ContainsKey(changes_, url));
+    demoted_changes_[url] = info;
+  } else {
+    DCHECK(!ContainsKey(demoted_changes_, url));
+    change_seqs_[info.change_seq] = url;
+    changes_[url] = info;
+  }
   RemoveMirrorAndCommitChangesForURL(url);
 }
 
 void LocalFileChangeTracker::DemoteChangesForURL(
     const fileapi::FileSystemURL& url) {
   DCHECK(file_task_runner_->RunsTasksOnCurrentThread());
+
   FileChangeMap::iterator found = changes_.find(url);
   if (found == changes_.end())
     return;
-  FileChangeList changes = found->second.change_list;
-
-  mirror_changes_.erase(url);
+  DCHECK(!ContainsKey(demoted_changes_, url));
   change_seqs_.erase(found->second.change_seq);
+  demoted_changes_.insert(*found);
   changes_.erase(found);
-
-  FileChangeList::List change_list = changes.list();
-  while (!change_list.empty()) {
-    RecordChangeToChangeMaps(url, change_list.front(), 0,
-                             &demoted_changes_, NULL);
-    change_list.pop_front();
-  }
   UpdateNumChanges();
 }
 
 void LocalFileChangeTracker::PromoteDemotedChangesForURL(
     const fileapi::FileSystemURL& url) {
   DCHECK(file_task_runner_->RunsTasksOnCurrentThread());
 
   FileChangeMap::iterator iter = demoted_changes_.find(url);
   if (iter == demoted_changes_.end())
     return;
 
   FileChangeList::List change_list = iter->second.change_list.list();
   // Make sure that this URL is in no queues.
+  DCHECK(!ContainsKey(change_seqs_, iter->second.change_seq));
   DCHECK(!ContainsKey(changes_, url));
-  DCHECK(!ContainsKey(mirror_changes_, url));
 
+  change_seqs_[iter->second.change_seq] = url;
+  changes_.insert(*iter);
   demoted_changes_.erase(iter);
-
-  while (!change_list.empty()) {
-    RecordChange(url, change_list.front());
-    change_list.pop_front();
-  }
 }
 
 bool LocalFileChangeTracker::PromoteDemotedChanges() {
   DCHECK(file_task_runner_->RunsTasksOnCurrentThread());
   if (demoted_changes_.empty())
     return false;
   while (!demoted_changes_.empty()) {
     fileapi::FileSystemURL url = demoted_changes_.begin()->first;
     PromoteDemotedChangesForURL(url);
   }
@@ skipping 14 matching lines @@
 }
 
 void LocalFileChangeTracker::ResetForFileSystem(
     const GURL& origin,
     fileapi::FileSystemType type) {
   DCHECK(file_task_runner_->RunsTasksOnCurrentThread());
   scoped_ptr<leveldb::WriteBatch> batch(new leveldb::WriteBatch);
   for (FileChangeMap::iterator iter = changes_.begin();
        iter != changes_.end();) {
     fileapi::FileSystemURL url = iter->first;
-    if (url.origin() != origin || url.type() != type) {
-      ++iter;
-      continue;
-    }
-    mirror_changes_.erase(url);
-    demoted_changes_.erase(url);
-    change_seqs_.erase(iter->second.change_seq);
-    changes_.erase(iter++);
+    int change_seq = iter->second.change_seq;
+    // Advance |iter| before calling ResetForURL to avoid the iterator
+    // invalidation in it.
+    ++iter;
+    if (url.origin() == origin && url.type() == type)
+      ResetForURL(url, change_seq, batch.get());
+  }
 
-    std::string serialized_url;
-    const bool should_success =
-        SerializeSyncableFileSystemURL(url, &serialized_url);
-    if (!should_success) {
-      NOTREACHED() << "Failed to serialize: " << url.DebugString();
-      continue;
-    }
-    batch->Delete(serialized_url);
+  for (FileChangeMap::iterator iter = demoted_changes_.begin();
+       iter != demoted_changes_.end();) {
+    fileapi::FileSystemURL url = iter->first;
+    int change_seq = iter->second.change_seq;
+    // Advance |iter| before calling ResetForURL to avoid the iterator
+    // invalidation in it.
+    ++iter;
+    if (url.origin() == origin && url.type() == type)
+      ResetForURL(url, change_seq, batch.get());
   }
+
   // Fail to apply batch to database wouldn't have critical effect, they'll be
   // just marked deleted on next relaunch.
   tracker_db_->WriteBatch(batch.Pass());
   UpdateNumChanges();
 }
 
 void LocalFileChangeTracker::UpdateNumChanges() {
   base::AutoLock lock(num_changes_lock_);
   num_changes_ = static_cast<int64>(change_seqs_.size());
 }
@@ skipping 96 matching lines @@
         LOG(WARNING) << "Failed to access local file.";
         break;
     }
   }
   return SYNC_STATUS_OK;
 }
 
 void LocalFileChangeTracker::RecordChange(
     const FileSystemURL& url, const FileChange& change) {
   DCHECK(file_task_runner_->RunsTasksOnCurrentThread());
+  int change_seq = current_change_seq_number_++;
   if (ContainsKey(demoted_changes_, url)) {
-    RecordChangeToChangeMaps(url, change, 0, &demoted_changes_, NULL);
-    return;
+    RecordChangeToChangeMaps(url, change, change_seq,
+                             &demoted_changes_, NULL);
+  } else {
+    RecordChangeToChangeMaps(url, change, change_seq, &changes_, &change_seqs_);
   }
-  int change_seq = current_change_seq_++;
-  RecordChangeToChangeMaps(url, change, change_seq, &changes_, &change_seqs_);
   if (ContainsKey(mirror_changes_, url))
     RecordChangeToChangeMaps(url, change, change_seq, &mirror_changes_, NULL);
   UpdateNumChanges();
 }
 
 // static
 void LocalFileChangeTracker::RecordChangeToChangeMaps(
     const FileSystemURL& url,
     const FileChange& change,
     int new_change_seq,
     FileChangeMap* changes,
     ChangeSeqMap* change_seqs) {
   ChangeInfo& info = (*changes)[url];
   if (info.change_seq >= 0 && change_seqs)
     change_seqs->erase(info.change_seq);
   info.change_list.Update(change);
   if (info.change_list.empty()) {
     changes->erase(url);
     return;
   }
   info.change_seq = new_change_seq;
   if (change_seqs)
     (*change_seqs)[info.change_seq] = url;
 }
 
+void LocalFileChangeTracker::ResetForURL(const fileapi::FileSystemURL& url,
+                                         int change_seq,
+                                         leveldb::WriteBatch* batch) {
+  mirror_changes_.erase(url);
+  demoted_changes_.erase(url);
+  change_seqs_.erase(change_seq);
+  changes_.erase(url);
+
+  std::string serialized_url;
+  if (!SerializeSyncableFileSystemURL(url, &serialized_url)) {
+    NOTREACHED() << "Failed to serialize: " << url.DebugString();
+    return;
+  }
+  batch->Delete(serialized_url);
+}
+
 // TrackerDB -------------------------------------------------------------------
 
 LocalFileChangeTracker::TrackerDB::TrackerDB(const base::FilePath& base_path,
                                              leveldb::Env* env_override)
   : base_path_(base_path),
     env_override_(env_override),
     db_status_(SYNC_STATUS_OK) {}
 
 SyncStatusCode LocalFileChangeTracker::TrackerDB::Init(
     RecoveryOption recovery_option) {
@@ skipping 134 matching lines @@
   if (!status.ok() && !status.IsNotFound()) {
     HandleError(FROM_HERE, status);
     db_status_ = LevelDBStatusToSyncStatusCode(status);
     db_.reset();
     return db_status_;
   }
   return SYNC_STATUS_OK;
 }
 
 }  // namespace sync_file_system