sync: Implement per-type GetCommitIds

sync/engine/get_commit_ids.cc

 // Copyright 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 "sync/engine/get_commit_ids_command.h"
+#include "sync/engine/get_commit_ids.h"
 
 #include <set>
-#include <utility>
 #include <vector>
 
+#include "base/basictypes.h"
 #include "sync/engine/syncer_util.h"
-#include "sync/sessions/nudge_tracker.h"
+#include "sync/syncable/directory.h"
 #include "sync/syncable/entry.h"
 #include "sync/syncable/nigori_handler.h"
 #include "sync/syncable/nigori_util.h"
 #include "sync/syncable/syncable_base_transaction.h"
 #include "sync/syncable/syncable_util.h"
 #include "sync/util/cryptographer.h"
 
 using std::set;
 using std::vector;
 
 namespace syncer {
 
-using sessions::OrderedCommitSet;
-using sessions::SyncSession;
-using sessions::StatusController;
+void GetCommitIdsForType(
+    syncable::BaseTransaction* trans,
+    ModelType type,
+    const size_t max_entries,
+    std::vector<int64>* out) {
+  syncable::Directory* dir = trans->directory();
 
-GetCommitIdsCommand::GetCommitIdsCommand(
-    syncable::BaseTransaction* trans,
-    ModelTypeSet requested_types,
-    const size_t commit_batch_size,
-    sessions::OrderedCommitSet* commit_set)
-    : trans_(trans),
-      requested_types_(requested_types),
-      requested_commit_batch_size_(commit_batch_size),
-      commit_set_(commit_set) {
-}
-
-GetCommitIdsCommand::~GetCommitIdsCommand() {}
-
-SyncerError GetCommitIdsCommand::ExecuteImpl(SyncSession* session) {
   // Gather the full set of unsynced items and store it in the session. They
   // are not in the correct order for commit.
   std::set<int64> ready_unsynced_set;
   syncable::Directory::Metahandles all_unsynced_handles;
-  GetUnsyncedEntries(trans_,
-                     &all_unsynced_handles);
+  GetUnsyncedEntries(trans, &all_unsynced_handles);
 
   ModelTypeSet encrypted_types;
   bool passphrase_missing = false;
-  Cryptographer* cryptographer =
-      session->context()->
-      directory()->GetCryptographer(trans_);
+  Cryptographer* cryptographer = dir->GetCryptographer(trans);
   if (cryptographer) {
-    encrypted_types = session->context()->directory()->GetNigoriHandler()->
-        GetEncryptedTypes(trans_);
+    encrypted_types = dir->GetNigoriHandler()->GetEncryptedTypes(trans);
     passphrase_missing = cryptographer->has_pending_keys();
   };
 
   // We filter out all unready entries from the set of unsynced handles. This
   // new set of ready and unsynced items is then what we use to determine what
   // is a candidate for commit.  The caller of this SyncerCommand is responsible
   // for ensuring that no throttled types are included among the
   // requested_types.
-  FilterUnreadyEntries(trans_,
-                       requested_types_,
+  FilterUnreadyEntries(trans,
+                       ModelTypeSet(type),
                        encrypted_types,
                        passphrase_missing,
                        all_unsynced_handles,
                        &ready_unsynced_set);
 
-  BuildCommitIds(trans_,
-                 session->context()->routing_info(),
-                 ready_unsynced_set);
+  OrderCommitIds(trans, max_entries, ready_unsynced_set, out);
 
-  return SYNCER_OK;
+  for (size_t i = 0; i < out->size(); i++)
+    DVLOG(1) << "Debug commit batch result:" << (*out)[i];
 }
 
-namespace {
-
 bool IsEntryInConflict(const syncable::Entry& entry) {
   if (entry.Get(syncable::IS_UNSYNCED) &&
       entry.Get(syncable::SERVER_VERSION) > 0 &&
       (entry.Get(syncable::SERVER_VERSION) >
        entry.Get(syncable::BASE_VERSION))) {
     // The local and server versions don't match. The item must be in
     // conflict, so there's no point in attempting to commit.
     DCHECK(entry.Get(syncable::IS_UNAPPLIED_UPDATE));
     DVLOG(1) << "Excluding entry from commit due to version mismatch "
              << entry;
@@ skipping 55 matching lines @@
 
   if (entry.IsRoot()) {
     NOTREACHED() << "Permanent item became unsynced " << entry;
     return false;
   }
 
   DVLOG(2) << "Entry is ready for commit: " << entry;
   return true;
 }
 
-}  // namespace
-
-void GetCommitIdsCommand::FilterUnreadyEntries(
+void FilterUnreadyEntries(
     syncable::BaseTransaction* trans,
     ModelTypeSet requested_types,
     ModelTypeSet encrypted_types,
     bool passphrase_missing,
     const syncable::Directory::Metahandles& unsynced_handles,
     std::set<int64>* ready_unsynced_set) {
   for (syncable::Directory::Metahandles::const_iterator iter =
        unsynced_handles.begin(); iter != unsynced_handles.end(); ++iter) {
     syncable::Entry entry(trans, syncable::GET_BY_HANDLE, *iter);
     if (IsEntryReadyForCommit(requested_types,
                               encrypted_types,
                               passphrase_missing,
                               entry)) {
       ready_unsynced_set->insert(*iter);
     }
   }
 }
 
-bool GetCommitIdsCommand::AddUncommittedParentsAndTheirPredecessors(
+namespace {
+
+// This class helps to implement OrderCommitIds().  Its members track the
+// progress of a traversal while its methods extend it.  It can return early if
+// the traversal reaches the desired size before the full traversal is complete.
+class Traversal {
+ public:
+  Traversal(
     syncable::BaseTransaction* trans,
-    const ModelSafeRoutingInfo& routes,
+    int64 max_entries,
+    std::vector<int64>* out);
+  ~Traversal();
+
+  // The following functions do not modify the traversal directly.  They return
+  // their results in the |result| vector instead.
+
+  bool AddUncommittedParentsAndTheirPredecessors(

[Nicolas Zea, 2013/09/05 23:55:37]

make these methods private?

[rlarocque, 2013/09/06 18:58:26]

Done.

+      const std::set<int64>& ready_unsynced_set,
+      const syncable::Entry& item,
+      std::vector<int64>* result) const;
+
+  void TryAddItem(const std::set<int64>& ready_unsynced_set,
+                  const syncable::Entry& item,
+                  std::vector<int64>* result) const;
+
+  void AddItemThenPredecessors(
+      const std::set<int64>& ready_unsynced_set,
+      const syncable::Entry& item,
+      std::vector<int64>* result) const;
+
+  void AddPredecessorsThenItem(
+      const std::set<int64>& ready_unsynced_set,
+      const syncable::Entry& item,
+      std::vector<int64>* result) const;
+
+  // First step of traversal building.  Adds non-deleted items in order.
+  void AddCreatesAndMoves(const std::set<int64>& ready_unsynced_set);
+
+  // Second step of traverals building.  Appends deleted items.
+  void AddDeletes(const std::set<int64>& ready_unsynced_set);
+
+ private:
+  // Returns true if we've collected enough items.
+  bool IsFull() const;
+
+  // Returns true if the specified handle is already in the traversal.
+  bool HaveItem(int64 handle) const;
+
+  // Adds the specified handles to the traversal.
+  void AppendManyToTraversal(std::vector<int64>);

[Nicolas Zea, 2013/09/05 23:55:37]

variable name

[rlarocque, 2013/09/06 18:58:26]

Done.

+
+  // Adds the specifed handle to the traversal.
+  void AppendToTraversal(int64 handle);
+
+  std::vector<int64>* out_;

[Nicolas Zea, 2013/09/05 23:55:37]

perhaps use Metahandles, which I believe is just the same type? (here and
elsewhere you have vectors of handles)

[rlarocque, 2013/09/06 18:58:26]

Done.

+  std::set<int64> added_handles_;

[rlarocque, 2013/09/05 18:33:48]

This member is not really necessary, but I felt bad about implementing
HaveItem() with an O(n) search where n ~= 90.  The old code's equivalent used an
OrderedCommitSet here, so its search was O(lg(n)).

I haven't profiled the change, so this may be premature optimization, but I
figured it would be safest to try to mimic the old code's algorithms.

+  const size_t max_entries_;
+  syncable::BaseTransaction* trans_;

[Nicolas Zea, 2013/09/05 23:55:37]

DISALLOW_COPY_AND_ASSIGN?

[rlarocque, 2013/09/06 18:58:26]

Done.

+};
+
+Traversal::Traversal(
+    syncable::BaseTransaction* trans,
+    int64 max_entries,
+    std::vector<int64>* out)
+  : out_(out),
+    max_entries_(max_entries),
+    trans_(trans) { }
+
+Traversal::~Traversal() {}
+
+bool Traversal::AddUncommittedParentsAndTheirPredecessors(
     const std::set<int64>& ready_unsynced_set,
     const syncable::Entry& item,
-    sessions::OrderedCommitSet* result) const {
-  OrderedCommitSet item_dependencies(routes);
+    std::vector<int64>* result) const {
+  std::vector<int64> dependencies;
   syncable::Id parent_id = item.Get(syncable::PARENT_ID);
 
   // Climb the tree adding entries leaf -> root.
   while (!parent_id.ServerKnows()) {
-    syncable::Entry parent(trans, syncable::GET_BY_ID, parent_id);
+    syncable::Entry parent(trans_, syncable::GET_BY_ID, parent_id);
     CHECK(parent.good()) << "Bad user-only parent in item path.";
     int64 handle = parent.Get(syncable::META_HANDLE);
-    if (commit_set_->HaveCommitItem(handle)) {
+    if (HaveItem(handle)) {
       // We've already added this parent (and therefore all of its parents).
       // We can return early.
       break;
     }
     if (IsEntryInConflict(parent)) {
       // We ignore all entries that are children of a conflicing item.  Return
       // false immediately to forget the traversal we've built up so far.
       DVLOG(1) << "Parent was in conflict, omitting " << item;
       return false;
     }
-    AddItemThenPredecessors(trans,
-                            ready_unsynced_set,
+    AddItemThenPredecessors(ready_unsynced_set,
                             parent,
-                            &item_dependencies);
+                            &dependencies);
     parent_id = parent.Get(syncable::PARENT_ID);
   }
 
   // Reverse what we added to get the correct order.
-  result->AppendReverse(item_dependencies);
+  result->insert(result->end(), dependencies.begin(), dependencies.end());

[Nicolas Zea, 2013/09/05 23:55:37]

this is no longer reversing right?

[rlarocque, 2013/09/06 18:58:26]

Good catch!  Somehow the unit tests missed this.

I've fixed the bug and added a test that will catch similar fixes in the future.

   return true;
 }
 
 // Adds the given item to the list if it is unsynced and ready for commit.
-void GetCommitIdsCommand::TryAddItem(const std::set<int64>& ready_unsynced_set,
-                                     const syncable::Entry& item,
-                                     OrderedCommitSet* result) const {
+void Traversal::TryAddItem(const std::set<int64>& ready_unsynced_set,
+                           const syncable::Entry& item,
+                           std::vector<int64>* result) const {
   DCHECK(item.Get(syncable::IS_UNSYNCED));
   int64 item_handle = item.Get(syncable::META_HANDLE);
   if (ready_unsynced_set.count(item_handle) != 0) {
-    result->AddCommitItem(item_handle, item.GetModelType());
+    result->push_back(item_handle);
   }
 }
 
 // Adds the given item, and all its unsynced predecessors.  The traversal will
 // be cut short if any item along the traversal is not IS_UNSYNCED, or if we
 // detect that this area of the tree has already been traversed.  Items that are
 // not 'ready' for commit (see IsEntryReadyForCommit()) will not be added to the
 // list, though they will not stop the traversal.
-void GetCommitIdsCommand::AddItemThenPredecessors(
-    syncable::BaseTransaction* trans,
+void Traversal::AddItemThenPredecessors(
     const std::set<int64>& ready_unsynced_set,
     const syncable::Entry& item,
-    OrderedCommitSet* result) const {
+    std::vector<int64>* result) const {
   int64 item_handle = item.Get(syncable::META_HANDLE);
-  if (commit_set_->HaveCommitItem(item_handle)) {
+  if (HaveItem(item_handle)) {
     // We've already added this item to the commit set, and so must have
     // already added the predecessors as well.
     return;
   }
   TryAddItem(ready_unsynced_set, item, result);
   if (item.Get(syncable::IS_DEL))
     return;  // Deleted items have no predecessors.
 
   syncable::Id prev_id = item.GetPredecessorId();
   while (!prev_id.IsRoot()) {
-    syncable::Entry prev(trans, syncable::GET_BY_ID, prev_id);
+    syncable::Entry prev(trans_, syncable::GET_BY_ID, prev_id);
     CHECK(prev.good()) << "Bad id when walking predecessors.";
     if (!prev.Get(syncable::IS_UNSYNCED)) {
       // We're interested in "runs" of unsynced items.  This item breaks
       // the streak, so we stop traversing.
       return;
     }
     int64 handle = prev.Get(syncable::META_HANDLE);
-    if (commit_set_->HaveCommitItem(handle)) {
+    if (HaveItem(handle)) {
       // We've already added this item to the commit set, and so must have
       // already added the predecessors as well.
       return;
     }
     TryAddItem(ready_unsynced_set, prev, result);
     prev_id = prev.GetPredecessorId();
   }
 }
 
 // Same as AddItemThenPredecessor, but the traversal order will be reversed.
-void GetCommitIdsCommand::AddPredecessorsThenItem(
-    syncable::BaseTransaction* trans,
-    const ModelSafeRoutingInfo& routes,
+void Traversal::AddPredecessorsThenItem(
     const std::set<int64>& ready_unsynced_set,
     const syncable::Entry& item,
-    OrderedCommitSet* result) const {
-  OrderedCommitSet item_dependencies(routes);
-  AddItemThenPredecessors(trans, ready_unsynced_set, item, &item_dependencies);
+    std::vector<int64>* result) const {
+  std::vector<int64> dependencies;
+  AddItemThenPredecessors(ready_unsynced_set, item, &dependencies);
 
   // Reverse what we added to get the correct order.
-  result->AppendReverse(item_dependencies);
+  result->insert(result->end(), dependencies.rbegin(), dependencies.rend());
 }
 
-bool GetCommitIdsCommand::IsCommitBatchFull() const {
-  return commit_set_->Size() >= requested_commit_batch_size_;
+bool Traversal::IsFull() const {
+  return out_->size() >= max_entries_;
 }
 
-void GetCommitIdsCommand::AddCreatesAndMoves(
-    syncable::BaseTransaction* trans,
-    const ModelSafeRoutingInfo& routes,
+bool Traversal::HaveItem(int64 handle) const {
+  return added_handles_.find(handle) != added_handles_.end();
+}
+
+void Traversal::AppendManyToTraversal(std::vector<int64> handles) {
+  out_->insert(out_->end(), handles.begin(), handles.end());
+  added_handles_.insert(handles.begin(), handles.end());
+}
+
+void Traversal::AppendToTraversal(int64 metahandle) {
+  out_->push_back(metahandle);
+  added_handles_.insert(metahandle);
+}
+
+void Traversal::AddCreatesAndMoves(
     const std::set<int64>& ready_unsynced_set) {
   // Add moves and creates, and prepend their uncommitted parents.
   for (std::set<int64>::const_iterator iter = ready_unsynced_set.begin();
-       !IsCommitBatchFull() && iter != ready_unsynced_set.end(); ++iter) {
+       !IsFull() && iter != ready_unsynced_set.end(); ++iter) {
     int64 metahandle = *iter;
-    if (commit_set_->HaveCommitItem(metahandle))
+    if (HaveItem(metahandle))
       continue;
 
-    syncable::Entry entry(trans,
+    syncable::Entry entry(trans_,
                           syncable::GET_BY_HANDLE,
                           metahandle);
     if (!entry.Get(syncable::IS_DEL)) {
       // We only commit an item + its dependencies if it and all its
       // dependencies are not in conflict.
-      OrderedCommitSet item_dependencies(routes);
+      std::vector<int64> item_dependencies;
       if (AddUncommittedParentsAndTheirPredecessors(
-              trans,
-              routes,
               ready_unsynced_set,
               entry,
               &item_dependencies)) {
-        AddPredecessorsThenItem(trans,
-                                routes,
-                                ready_unsynced_set,
+        AddPredecessorsThenItem(ready_unsynced_set,
                                 entry,
                                 &item_dependencies);
-        commit_set_->Append(item_dependencies);
+        AppendManyToTraversal(item_dependencies);
       }
     }
   }
 
   // It's possible that we overcommitted while trying to expand dependent
   // items.  If so, truncate the set down to the allowed size.
-  commit_set_->Truncate(requested_commit_batch_size_);
+  if (out_->size() > max_entries_) {
+    out_->resize(max_entries_);
+  }
 }
 
-void GetCommitIdsCommand::AddDeletes(
-    syncable::BaseTransaction* trans,
+void Traversal::AddDeletes(
     const std::set<int64>& ready_unsynced_set) {
   set<syncable::Id> legal_delete_parents;
 
   for (std::set<int64>::const_iterator iter = ready_unsynced_set.begin();
-       !IsCommitBatchFull() && iter != ready_unsynced_set.end(); ++iter) {
+       !IsFull() && iter != ready_unsynced_set.end(); ++iter) {
     int64 metahandle = *iter;
-    if (commit_set_->HaveCommitItem(metahandle))
+    if (HaveItem(metahandle))
       continue;
 
-    syncable::Entry entry(trans, syncable::GET_BY_HANDLE,
+    syncable::Entry entry(trans_, syncable::GET_BY_HANDLE,
                           metahandle);
 
     if (entry.Get(syncable::IS_DEL)) {
-      syncable::Entry parent(trans, syncable::GET_BY_ID,
+      syncable::Entry parent(trans_, syncable::GET_BY_ID,
                              entry.Get(syncable::PARENT_ID));
       // If the parent is deleted and unsynced, then any children of that
       // parent don't need to be added to the delete queue.
       //
       // Note: the parent could be synced if there was an update deleting a
       // folder when we had a deleted all items in it.
       // We may get more updates, or we may want to delete the entry.
       if (parent.good() &&
           parent.Get(syncable::IS_DEL) &&
           parent.Get(syncable::IS_UNSYNCED)) {
         // However, if an entry is moved, these rules can apply differently.
         //
         // If the entry was moved, then the destination parent was deleted,
         // then we'll miss it in the roll up. We have to add it in manually.
         // TODO(chron): Unit test for move / delete cases:
         // Case 1: Locally moved, then parent deleted
         // Case 2: Server moved, then locally issue recursive delete.
         if (entry.Get(syncable::ID).ServerKnows() &&
             entry.Get(syncable::PARENT_ID) !=
                 entry.Get(syncable::SERVER_PARENT_ID)) {
           DVLOG(1) << "Inserting moved and deleted entry, will be missed by "
                    << "delete roll." << entry.Get(syncable::ID);
 
-          commit_set_->AddCommitItem(metahandle, entry.GetModelType());
+          AppendToTraversal(metahandle);
         }
 
         // Skip this entry since it's a child of a parent that will be
         // deleted. The server will unroll the delete and delete the
         // child as well.
         continue;
       }
 
       legal_delete_parents.insert(entry.Get(syncable::PARENT_ID));
     }
   }
 
   // We could store all the potential entries with a particular parent during
   // the above scan, but instead we rescan here. This is less efficient, but
   // we're dropping memory alloc/dealloc in favor of linear scans of recently
   // examined entries.
   //
   // Scan through the UnsyncedMetaHandles again. If we have a deleted
   // entry, then check if the parent is in legal_delete_parents.
   //
   // Parent being in legal_delete_parents means for the child:
   //   a recursive delete is not currently happening (no recent deletes in same
   //     folder)
   //   parent did expect at least one old deleted child
   //   parent was not deleted
   for (std::set<int64>::const_iterator iter = ready_unsynced_set.begin();
-       !IsCommitBatchFull() && iter != ready_unsynced_set.end(); ++iter) {
+       !IsFull() && iter != ready_unsynced_set.end(); ++iter) {
     int64 metahandle = *iter;
-    if (commit_set_->HaveCommitItem(metahandle))
+    if (HaveItem(metahandle))
       continue;
-    syncable::Entry entry(trans, syncable::GET_BY_HANDLE,
+    syncable::Entry entry(trans_, syncable::GET_BY_HANDLE,
                           metahandle);
     if (entry.Get(syncable::IS_DEL)) {
       syncable::Id parent_id = entry.Get(syncable::PARENT_ID);
       if (legal_delete_parents.count(parent_id)) {
-        commit_set_->AddCommitItem(metahandle, entry.GetModelType());
+        AppendToTraversal(metahandle);
       }
     }
   }
 }
 
-void GetCommitIdsCommand::BuildCommitIds(
+}  // namespace
+
+void OrderCommitIds(
     syncable::BaseTransaction* trans,
-    const ModelSafeRoutingInfo& routes,
-    const std::set<int64>& ready_unsynced_set) {
+    size_t max_entries,
+    const std::set<int64>& ready_unsynced_set,
+    std::vector<int64>* out) {
   // Commits follow these rules:
   // 1. Moves or creates are preceded by needed folder creates, from
   //    root to leaf.  For folders whose contents are ordered, moves
   //    and creates appear in order.
   // 2. Moves/Creates before deletes.
   // 3. Deletes, collapsed.
   // We commit deleted moves under deleted items as moves when collapsing
   // delete trees.
 
+  Traversal traversal(trans, max_entries, out);
+
   // Add moves and creates, and prepend their uncommitted parents.
-  AddCreatesAndMoves(trans, routes, ready_unsynced_set);
+  traversal.AddCreatesAndMoves(ready_unsynced_set);
 
   // Add all deletes.
-  AddDeletes(trans, ready_unsynced_set);
+  traversal.AddDeletes(ready_unsynced_set);
+}
+
+void GetCommitIds(
+    syncable::BaseTransaction* trans,
+    ModelTypeSet requested_types,
+    const size_t commit_batch_size,
+    sessions::OrderedCommitSet* ordered_commit_set) {
+  for (ModelTypeSet::Iterator it = requested_types.First();
+       it.Good(); it.Inc()) {
+    size_t space_remaining = commit_batch_size - ordered_commit_set->Size();

[Nicolas Zea, 2013/09/05 23:55:37]

size_t is unsigned right? The <= 0 condition doesn't really make sense in that
case

[rlarocque, 2013/09/06 18:58:26]

Good point.

This should still work as written, since all we really need to do here is break
when space_remaining == 0.  space_remaining < 0 should never happen.  

You're right that this doesn't make much sense, though.  I've rewritten this to
make it more explicit and avoid < 0 comparisons on unsigned ints.

+    if (space_remaining <= 0)
+      break;
+    std::vector<int64> out;
+    GetCommitIdsForType(
+        trans,
+        it.Get(),
+        space_remaining,
+        &out);
+    ordered_commit_set->AddCommitItems(out, it.Get());
+  }
 }
 
 }  // namespace syncer