sync/engine/commit.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 "sync/engine/commit.h"
   6
   7 #include "base/debug/trace_event.h"
   8 #include "sync/engine/build_commit_command.h"
   9 #include "sync/engine/get_commit_ids_command.h"
  10 #include "sync/engine/process_commit_response_command.h"
  11 #include "sync/engine/syncer_proto_util.h"
  12 #include "sync/sessions/sync_session.h"
  13 #include "sync/syncable/mutable_entry.h"
  14 #include "sync/syncable/write_transaction.h"
  15
  16 namespace syncer {
  17
  18 using sessions::SyncSession;
  19 using sessions::StatusController;
  20 using syncable::SYNCER;
  21 using syncable::WriteTransaction;
  22
  23 namespace {
  24
  25 // Sets the SYNCING bits of all items in the commit set to value_to_set.
  26 void SetAllSyncingBitsToValue(WriteTransaction* trans,
  27                               const sessions::OrderedCommitSet& commit_set,
  28                               bool value_to_set) {
  29   const std::vector<syncable::Id>& commit_ids = commit_set.GetAllCommitIds();
  30   for (std::vector<syncable::Id>::const_iterator it = commit_ids.begin();
  31        it != commit_ids.end(); ++it) {
  32     syncable::MutableEntry entry(trans, syncable::GET_BY_ID, *it);
  33     if (entry.good()) {
  34       entry.Put(syncable::SYNCING, value_to_set);
  35     }
  36   }
  37 }
  38
  39 // Sets the SYNCING bits for all items in the OrderedCommitSet.
  40 void SetSyncingBits(WriteTransaction* trans,
  41                     const sessions::OrderedCommitSet& commit_set) {
  42   SetAllSyncingBitsToValue(trans, commit_set, true);
  43 }
  44
  45 // Clears the SYNCING bits for all items in the OrderedCommitSet.
  46 void ClearSyncingBits(syncable::Directory* dir,
  47                       const sessions::OrderedCommitSet& commit_set) {
  48   WriteTransaction trans(FROM_HERE, SYNCER, dir);
  49   SetAllSyncingBitsToValue(&trans, commit_set, false);
  50 }
  51
  52 // Helper function that finds sync items that are ready to be committed to the
  53 // server and serializes them into a commit message protobuf.  It will return
  54 // false iff there are no entries ready to be committed at this time.
  55 //
  56 // The OrderedCommitSet parameter is an output parameter which will contain
  57 // the set of all items which are to be committed.  The number of items in
  58 // the set shall not exceed the maximum batch size.  (The default batch size
  59 // is currently 25, though it can be overwritten by the server.)
  60 //
  61 // The ClientToServerMessage parameter is an output parameter which will contain
  62 // the commit message which should be sent to the server.  It is valid iff the
  63 // return value of this function is true.
  64 bool PrepareCommitMessage(sessions::SyncSession* session,
  65                           sessions::OrderedCommitSet* commit_set,
  66                           sync_pb::ClientToServerMessage* commit_message) {
  67   TRACE_EVENT0("sync", "PrepareCommitMessage");
  68
  69   commit_set->Clear();
  70   commit_message->Clear();
  71
  72   WriteTransaction trans(FROM_HERE, SYNCER, session->context()->directory());
  73   sessions::ScopedSetSessionWriteTransaction set_trans(session, &trans);
  74
  75   // Fetch the items to commit.
  76   const size_t batch_size = session->context()->max_commit_batch_size();
  77   GetCommitIdsCommand get_commit_ids_command(batch_size, commit_set);
  78   get_commit_ids_command.Execute(session);
  79
  80   DVLOG(1) << "Commit message will contain " << commit_set->Size() << " items.";
  81   if (commit_set->Empty()) {
  82     return false;
  83   }
  84
  85   // Serialize the message.
  86   BuildCommitCommand build_commit_command(*commit_set, commit_message);
  87   build_commit_command.Execute(session);
  88
  89   SetSyncingBits(session->write_transaction(), *commit_set);
  90   return true;
  91 }
  92
  93 SyncerError BuildAndPostCommitsImpl(Syncer* syncer,
  94                                     sessions::SyncSession* session,
  95                                     sessions::OrderedCommitSet* commit_set) {
  96   sync_pb::ClientToServerMessage commit_message;
  97   while (!syncer->ExitRequested() &&
  98          PrepareCommitMessage(session, commit_set, &commit_message)) {
  99     sync_pb::ClientToServerResponse commit_response;
 100
 101     DVLOG(1) << "Sending commit message.";
 102     TRACE_EVENT_BEGIN0("sync", "PostCommit");
 103     const SyncerError post_result = SyncerProtoUtil::PostClientToServerMessage(
 104         &commit_message, &commit_response, session);
 105     TRACE_EVENT_END0("sync", "PostCommit");
 106
 107     if (post_result != SYNCER_OK) {
 108       LOG(WARNING) << "Post commit failed";
 109       return post_result;
 110     }
 111
 112     if (!commit_response.has_commit()) {
 113       LOG(WARNING) << "Commit response has no commit body!";
 114       return SERVER_RESPONSE_VALIDATION_FAILED;
 115     }
 116
 117     const size_t num_responses = commit_response.commit().entryresponse_size();
 118     if (num_responses != commit_set->Size()) {
 119       LOG(ERROR)
 120          << "Commit response has wrong number of entries! "
 121          << "Expected: " << commit_set->Size() << ", "
 122          << "Got: " << num_responses;
 123       return SERVER_RESPONSE_VALIDATION_FAILED;
 124     }
 125
 126     TRACE_EVENT_BEGIN0("sync", "ProcessCommitResponse");
 127     ProcessCommitResponseCommand process_response_command(
 128         *commit_set, commit_message, commit_response);
 129     const SyncerError processing_result =
 130         process_response_command.Execute(session);
 131     TRACE_EVENT_END0("sync", "ProcessCommitResponse");
 132
 133     if (processing_result != SYNCER_OK) {
 134       return processing_result;
 135     }
 136     session->SendEventNotification(SyncEngineEvent::STATUS_CHANGED);
 137   }
 138
 139   return SYNCER_OK;
 140 }
 141
 142 }  // namespace
 143
 144
 145 SyncerError BuildAndPostCommits(Syncer* syncer,
 146                                 sessions::SyncSession* session) {
 147   sessions::OrderedCommitSet commit_set(session->routing_info());
 148   SyncerError result = BuildAndPostCommitsImpl(syncer, session, &commit_set);
 149   if (result != SYNCER_OK) {
 150     ClearSyncingBits(session->context()->directory(), commit_set);
 151   }
 152   return result;
 153 }
 154
 155 }  // namespace syncer