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..9e7e690424818cda46c763e3f85da6e103e7c83c 100644
--- a/sync/engine/update_applicator.cc
+++ b/sync/engine/update_applicator.cc
@@ -18,86 +18,81 @@ 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,
+    const std::vector<int64>& handles) {
+  std::vector<int64> to_apply = handles;
+  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) {

[tim (not reviewing), 2012/09/27 15:04:31]

As far as I can tell, the only logical differences between this and the previous
code are 1) the previous code wound up processing items in reverse order, which
I don't think was optimizing for anything but instead making it easier to 2)
avoid re-considering successfully applied items *in-place* (rather than using
extra storage as you do here).  Assuming I have that right, this new code is way
easier to understand.  Nice job.

The old code was old -- dates back to very early days.  I suspect at that time,
space may have been more of a concern (perhaps there were no batch size
limitations), I'm not entirely sure.  Another hint to that effect is it used to
take care to check early_exit_requested_ between each call to
AttemptOneApplication (that's why it was singled-out rather than "do all" as
you've done here).  I think this is also fine given todays system.

For archaeological purposes (if you're interested), using critique, look up

11682892 (see update_applicator.cc)
4493866-p10 (see syncer.cc)
2841323-p10 (see syncer.cc) This is the initial checkin of the update
application code, which remained mostly intact (although we stopped 'swapping'
iterator values at some point).

+      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);

[tim (not reviewing), 2012/09/26 21:56:25]

Add a comment somewhere explaining what constitutes a re_apply-able item and
what doesn't. 

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

Oh, I see you did this above.  I guess I expected to see it next to / near
to_reapply.  Well, food for thought.  No big deal I guess.

+          break;
+        default:
+          NOTREACHED();
+          break;
+      }
+    }
+
+    if (to_reapply.size() == to_apply.size()) {
+      // We made no progress.  Must be stubborn hierarchy conflicts.

[tim (not reviewing), 2012/09/26 21:56:25]

Would it be clearer to add a size-getter on application_results_'s hierarchy
conflicts and check that against to_apply?  Or as a dcheck?

[rlarocque, 2012/09/26 22:24:32]

I was thinking of the application_results_'s contents as an output of this
process.  I'd rather not feed the results back in as an input if it can be
avoided.

On a more practical note, I hope to replace application_result_'s set of
hierarchy conflicts with a counter.  The only consumer of that data is the
about:sync page, and it only cares about the size of the set.  That's another
(probably better) reason to keep a local to_reapply_ set.

       break;
-    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.swap(to_reapply);
+    to_reapply.clear();
+  }
 }
 
 bool UpdateApplicator::SkipUpdate(const syncable::Entry& entry) {
@@ -119,46 +114,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 +166,7 @@ void UpdateApplicator::ResultTracker::SaveProgress(
   }
 }
 
-void UpdateApplicator::ResultTracker::ClearConflicts() {
-  conflicting_ids_.clear();
-  encryption_conflict_ids_.clear();
+void UpdateApplicator::ResultTracker::ClearHierarchyConflicts() {
   hierarchy_conflict_ids_.clear();
 }