sync: Refactor update application

sync/engine/update_applicator.cc

Index: sync/engine/update_applicator.cc
diff --git a/sync/engine/update_applicator.cc b/sync/engine/update_applicator.cc
index c5a80c6c6b5b3be12d8f0f33dfe58faa684ba503..081df22f012828ad2be83d0cc68ff95339435610 100644
--- a/sync/engine/update_applicator.cc
+++ b/sync/engine/update_applicator.cc
@@ -18,86 +18,79 @@ using std::vector;
 
 namespace syncer {
 
-UpdateApplicator::UpdateApplicator(ConflictResolver* resolver,
-                                   Cryptographer* cryptographer,
-                                   const UpdateIterator& begin,
-                                   const UpdateIterator& end,
+using syncable::ID;
+
+UpdateApplicator::UpdateApplicator(Cryptographer* cryptographer,
                                    const ModelSafeRoutingInfo& routes,
                                    ModelSafeGroup group_filter)
-    : resolver_(resolver),
-      cryptographer_(cryptographer),
-      begin_(begin),
-      end_(end),
-      pointer_(begin),
+    : cryptographer_(cryptographer),
       group_filter_(group_filter),
-      progress_(false),
       routing_info_(routes),
-      application_results_(end - begin) {
-  size_t item_count = end - begin;
-  DVLOG(1) << "UpdateApplicator created for " << item_count << " items.";
+      application_results_() {
 }
 
 UpdateApplicator::~UpdateApplicator() {
 }
 
-// Returns true if there's more to do.
-bool UpdateApplicator::AttemptOneApplication(
-    syncable::WriteTransaction* trans) {
-  // If there are no updates left to consider, we're done.
-  if (end_ == begin_)
-    return false;
-  if (pointer_ == end_) {
-    if (!progress_)
-      return false;
-
-    DVLOG(1) << "UpdateApplicator doing additional pass.";
-    pointer_ = begin_;
-    progress_ = false;
-
-    // Clear the tracked failures to avoid double-counting.
-    application_results_.ClearConflicts();
-  }
-
-  syncable::Entry read_only(trans, syncable::GET_BY_HANDLE, *pointer_);
-  if (SkipUpdate(read_only)) {
-    Advance();
-    return true;
-  }
-
-  syncable::MutableEntry entry(trans, syncable::GET_BY_HANDLE, *pointer_);
-  UpdateAttemptResponse updateResponse = AttemptToUpdateEntry(
-      trans, &entry, resolver_, cryptographer_);
-  switch (updateResponse) {
-    case SUCCESS:
-      Advance();
-      progress_ = true;
-      application_results_.AddSuccess(entry.Get(syncable::ID));
-      break;
-    case CONFLICT_SIMPLE:
-      pointer_++;
-      application_results_.AddSimpleConflict(entry.Get(syncable::ID));
-      break;
-    case CONFLICT_ENCRYPTION:
-      pointer_++;
-      application_results_.AddEncryptionConflict(entry.Get(syncable::ID));
-      break;
-    case CONFLICT_HIERARCHY:
-      pointer_++;
-      application_results_.AddHierarchyConflict(entry.Get(syncable::ID));
+// Attempt to apply all updates, using multiple passes if necessary.
+//
+// Some updates must be applied in order.  For example, children must be created
+// after their parent folder is created.  This function runs an O(n^2) algorithm
+// that will keep trying until there is nothing left to apply, or it stops
+// making progress, which would indicate that the hierarchy is invalid.
+//
+// The update applicator also has to deal with simple conflicts, which occur
+// when an item is modified on both the server and the local model, and
+// encryption conflicts.  There's not much we can do about them here, so we
+// don't bother re-processing them on subsequent passes.
+void UpdateApplicator::AttemptApplications(
+    syncable::WriteTransaction* trans,
+    std::vector<int64> to_apply) {
+  DVLOG(1) << "UpdateApplicator running over " << to_apply.size() << " items.";
+  while (!to_apply.empty()) {
+    std::vector<int64> to_reapply;
+
+    for (UpdateIterator i = to_apply.begin(); i != to_apply.end(); ++i) {
+      syncable::Entry read_entry(trans, syncable::GET_BY_HANDLE, *i);
+      if (SkipUpdate(read_entry)) {
+        continue;
+      }
+
+      syncable::MutableEntry entry(trans, syncable::GET_BY_HANDLE, *i);
+      UpdateAttemptResponse result = AttemptToUpdateEntry(
+          trans, &entry, cryptographer_);
+
+      switch (result) {
+        case SUCCESS:
+          application_results_.AddSuccess(entry.Get(ID));
+          break;
+        case CONFLICT_SIMPLE:
+          application_results_.AddSimpleConflict(entry.Get(ID));
+          break;
+        case CONFLICT_ENCRYPTION:
+          application_results_.AddEncryptionConflict(entry.Get(ID));
+          break;
+        case CONFLICT_HIERARCHY:
+          // Tentatively label these as hierarchy conflicts.  If we make any
+          // progress this round, we'll unlabel it and try to re-apply.
+          application_results_.AddHierarchyConflict(entry.Get(ID));
+          to_reapply.push_back(*i);
+          break;
+        default:
+          NOTREACHED();
+          break;
+      }
+    }
+
+    if (to_reapply.size() == to_apply.size()) {
+      // We made no progress.  Must be stubborn hierarchy conflicts.
       break;

[rlarocque, 2012/09/25 17:37:45]

I don't like this, but I think it's cleaner than a flag variable.  Suggestions
welcome.

[tim (not reviewing), 2012/09/26 21:13:14]

If we have unapplied updates for type T in the Directory, and then we try to run
a config job for a subset of (AllEnabledTypes - T), I believe the code in
ApplyUpdatesCommand will result in entries of type T being placed in
GROUP_PASSIVE and included as part of the sync cycle. In that case,
to_reapply.size() couldn't possibly == to_apply.size(), because we'll have
skipped some entries, but (I think) logically we'd be in the same case you're
targeting here with the 'break'?  

[rlarocque, 2012/09/26 21:32:29]

I don't think so, though I admit I had not anticipated the situation you
describe.

It's different in the next patch upload, but in this one it's not really obvious
that I'm modifying the value of to_apply with each loop iteration.  See the
comments on line 92.

If we skip items with SkipUpdate(), then they won't be in to_apply on the next
iteration.  Only the items we classified as hierarchy conflicts will attempt a
re-application on the following iteration.  Does that make sense?

-    default:
-      NOTREACHED();
-      break;
-  }
-  DVLOG(1) << "Apply Status for " << entry.Get(syncable::META_HANDLE)
-           << " is " << updateResponse;
-
-  return true;
-}
+    } else {
+      application_results_.ClearHierarchyConflicts();
+    }
 
-void UpdateApplicator::Advance() {
-  --end_;
-  *pointer_ = *end_;
+    to_apply = to_reapply;

[akalin, 2012/09/25 22:00:55]

not sure what this line does

[rlarocque, 2012/09/25 22:16:46]

I'm using the parameter as if it were a function-scoped variable here. 
Technically, it is.  However, one could argue this is clever and not very
C++-like.  

Maybe it would be more clear if I made the parameter a const-ref (as you
suggested earlier) and explicitly declared the local variable that I need.

[akalin, 2012/09/25 22:22:24]

oh, i missed that it was inside the while loop.

Yeah, I'd suggest an explicit local variable.  Then you can clear and swap.

+  }
 }
 
 bool UpdateApplicator::SkipUpdate(const syncable::Entry& entry) {
@@ -119,46 +112,38 @@ bool UpdateApplicator::SkipUpdate(const syncable::Entry& entry) {
   return false;
 }
 
-bool UpdateApplicator::AllUpdatesApplied() const {
-  return application_results_.no_conflicts() && begin_ == end_;
-}
-
 void UpdateApplicator::SaveProgressIntoSessionState(
     sessions::ConflictProgress* conflict_progress,
     sessions::UpdateProgress* update_progress) {
-  DCHECK(begin_ == end_ || ((pointer_ == end_) && !progress_))
-      << "SaveProgress called before updates exhausted.";
-
   application_results_.SaveProgress(conflict_progress, update_progress);
 }
 
-UpdateApplicator::ResultTracker::ResultTracker(size_t num_results) {
-  successful_ids_.reserve(num_results);
+UpdateApplicator::ResultTracker::ResultTracker() {
 }
 
 UpdateApplicator::ResultTracker::~ResultTracker() {
 }
 
 void UpdateApplicator::ResultTracker::AddSimpleConflict(syncable::Id id) {
-  conflicting_ids_.push_back(id);
+  conflicting_ids_.insert(id);
 }
 
 void UpdateApplicator::ResultTracker::AddEncryptionConflict(syncable::Id id) {
-  encryption_conflict_ids_.push_back(id);
+  encryption_conflict_ids_.insert(id);
 }
 
 void UpdateApplicator::ResultTracker::AddHierarchyConflict(syncable::Id id) {
-  hierarchy_conflict_ids_.push_back(id);
+  hierarchy_conflict_ids_.insert(id);
 }
 
 void UpdateApplicator::ResultTracker::AddSuccess(syncable::Id id) {
-  successful_ids_.push_back(id);
+  successful_ids_.insert(id);
 }
 
 void UpdateApplicator::ResultTracker::SaveProgress(
     sessions::ConflictProgress* conflict_progress,
     sessions::UpdateProgress* update_progress) {
-  vector<syncable::Id>::const_iterator i;
+  std::set<syncable::Id>::const_iterator i;
   for (i = conflicting_ids_.begin(); i != conflicting_ids_.end(); ++i) {
     conflict_progress->AddSimpleConflictingItemById(*i);
     update_progress->AddAppliedUpdate(CONFLICT_SIMPLE, *i);
@@ -179,9 +164,7 @@ void UpdateApplicator::ResultTracker::SaveProgress(
   }
 }
 
-void UpdateApplicator::ResultTracker::ClearConflicts() {
-  conflicting_ids_.clear();
-  encryption_conflict_ids_.clear();
+void UpdateApplicator::ResultTracker::ClearHierarchyConflicts() {
   hierarchy_conflict_ids_.clear();
 }