Chromium Code Reviews
chromiumcodereview-hr@appspot.gserviceaccount.com (chromiumcodereview-hr) | Please choose your nickname with Settings | Help | Chromium Project | Gerrit Changes | Sign out
(659)

Unified Diff: sync/engine/sync_scheduler_impl.cc

Issue 11342008: Revert 164565 - sync: make scheduling logic and job ownership more obvious. (Closed) Base URL: svn://svn.chromium.org/chrome/branches/1311/src/
Patch Set: Created 8 years, 2 months ago
Use n/p to move between diff chunks; N/P to move between comments. Draft comments are only viewable by you.
Jump to:
View side-by-side diff with in-line comments
Download patch
« no previous file with comments | « sync/engine/sync_scheduler_impl.h ('k') | sync/engine/sync_scheduler_unittest.cc » ('j') | no next file with comments »
Expand Comments ('e') | Collapse Comments ('c') | Show Comments Hide Comments ('s')
Index: sync/engine/sync_scheduler_impl.cc
===================================================================
--- sync/engine/sync_scheduler_impl.cc (revision 164641)
+++ sync/engine/sync_scheduler_impl.cc (working copy)
@@ -9,7 +9,6 @@
#include "base/auto_reset.h"
#include "base/bind.h"
-#include "base/bind_helpers.h"
#include "base/compiler_specific.h"
#include "base/location.h"
#include "base/logging.h"
@@ -84,13 +83,9 @@
SyncSchedulerImpl::WaitInterval::WaitInterval()
: mode(UNKNOWN),
- had_nudge(false),
- pending_configure_job(NULL) {}
+ had_nudge(false) {
+}
-SyncSchedulerImpl::WaitInterval::WaitInterval(Mode mode, TimeDelta length)
- : mode(mode), had_nudge(false), length(length),
- pending_configure_job(NULL) {}
-
SyncSchedulerImpl::WaitInterval::~WaitInterval() {}
#define ENUM_CASE(x) case x: return #x; break;
@@ -105,6 +100,39 @@
return "";
}
+SyncSchedulerImpl::SyncSessionJob::SyncSessionJob()
+ : purpose(UNKNOWN),
+ is_canary_job(false) {
+}
+
+SyncSchedulerImpl::SyncSessionJob::~SyncSessionJob() {}
+
+SyncSchedulerImpl::SyncSessionJob::SyncSessionJob(SyncSessionJobPurpose purpose,
+ base::TimeTicks start,
+ linked_ptr<sessions::SyncSession> session,
+ bool is_canary_job,
+ const ConfigurationParams& config_params,
+ const tracked_objects::Location& from_here)
+ : purpose(purpose),
+ scheduled_start(start),
+ session(session),
+ is_canary_job(is_canary_job),
+ config_params(config_params),
+ from_here(from_here) {
+}
+
+const char* SyncSchedulerImpl::SyncSessionJob::GetPurposeString(
+ SyncSchedulerImpl::SyncSessionJob::SyncSessionJobPurpose purpose) {
+ switch (purpose) {
+ ENUM_CASE(UNKNOWN);
+ ENUM_CASE(POLL);
+ ENUM_CASE(NUDGE);
+ ENUM_CASE(CONFIGURATION);
+ }
+ NOTREACHED();
+ return "";
+}
+
GetUpdatesCallerInfo::GetUpdatesSource GetUpdatesFromNudgeSource(
NudgeSource source) {
switch (source) {
@@ -124,6 +152,9 @@
}
}
+SyncSchedulerImpl::WaitInterval::WaitInterval(Mode mode, TimeDelta length)
+ : mode(mode), had_nudge(false), length(length) { }
+
// Helper macros to log with the syncer thread name; useful when there
// are multiple syncer threads involved.
@@ -174,7 +205,6 @@
// Start with assuming everything is fine with the connection.
// At the end of the sync cycle we would have the correct status.
connection_code_(HttpResponse::SERVER_CONNECTION_OK),
- pending_nudge_(NULL),
delay_provider_(delay_provider),
syncer_(syncer),
session_context_(context),
@@ -211,25 +241,10 @@
void SyncSchedulerImpl::OnServerConnectionErrorFixed() {
connection_code_ = HttpResponse::SERVER_CONNECTION_OK;
- // There could be a pending nudge or configuration job in several cases:
- //
- // 1. We're in exponential backoff.
- // 2. We're silenced / throttled.
- // 3. A nudge was saved previously due to not having a valid auth token.
- // 4. A nudge was scheduled + saved while in configuration mode.
- //
- // In all cases except (2), we want to retry contacting the server. We
- // call DoCanaryJob to achieve this, and note that nothing -- not even a
- // canary job -- can bypass a THROTTLED WaitInterval. The only thing that
- // has the authority to do that is the Unthrottle timer.
- scoped_ptr<SyncSessionJob> pending(TakePendingJobForCurrentMode());
- if (!pending.get())
- return;
-
PostTask(FROM_HERE, "DoCanaryJob",
base::Bind(&SyncSchedulerImpl::DoCanaryJob,
- weak_ptr_factory_.GetWeakPtr(),
- base::Passed(&pending)));
+ weak_ptr_factory_.GetWeakPtr()));
+
}
void SyncSchedulerImpl::UpdateServerConnectionManagerStatus(
@@ -265,20 +280,11 @@
if (mode_ == NORMAL_MODE) {
// It is illegal to switch to NORMAL_MODE if a previous CONFIGURATION job
// has not yet completed.
- DCHECK(!wait_interval_.get() || !wait_interval_->pending_configure_job);
+ DCHECK(!wait_interval_.get() ||
+ !wait_interval_->pending_configure_job.get());
}
- scoped_ptr<SyncSessionJob> pending(TakePendingJobForCurrentMode());
- if (pending.get()) {
- // TODO(tim): We should be able to remove this...
- scoped_ptr<SyncSession> session(CreateSyncSession(
- pending->session()->source()));
- // Also the routing info might have been changed since we cached the
- // pending nudge. Update it by coalescing to the latest.
- pending->mutable_session()->Coalesce(*session);
- SDVLOG(2) << "Executing pending job. Good luck!";
- DoSyncSessionJob(pending.Pass());
- }
+ DoPendingJobIfPossible(false);
}
}
@@ -322,7 +328,7 @@
// Only one configuration is allowed at a time. Verify we're not waiting
// for a pending configure job.
- DCHECK(!wait_interval_.get() || !wait_interval_->pending_configure_job);
+ DCHECK(!wait_interval_.get() || !wait_interval_->pending_configure_job.get());
ModelSafeRoutingInfo restricted_routes;
BuildModelSafeParams(params.types_to_download,
@@ -333,7 +339,7 @@
// Only reconfigure if we have types to download.
if (!params.types_to_download.Empty()) {
DCHECK(!restricted_routes.empty());
- scoped_ptr<SyncSession> session(new SyncSession(
+ linked_ptr<SyncSession> session(new SyncSession(
session_context_,
this,
SyncSourceInfo(params.source,
@@ -342,18 +348,19 @@
std::string())),
restricted_routes,
session_context_->workers()));
- scoped_ptr<SyncSessionJob> job(new SyncSessionJob(
- SyncSessionJob::CONFIGURATION,
- TimeTicks::Now(),
- session.Pass(),
- params,
- FROM_HERE));
- bool succeeded = DoSyncSessionJob(job.Pass());
+ SyncSessionJob job(SyncSessionJob::CONFIGURATION,
+ TimeTicks::Now(),
+ session,
+ false,
+ params,
+ FROM_HERE);
+ DoSyncSessionJob(job);
// If we failed, the job would have been saved as the pending configure
// job and a wait interval would have been set.
- if (!succeeded) {
- DCHECK(wait_interval_.get() && wait_interval_->pending_configure_job);
+ if (!session->Succeeded()) {
+ DCHECK(wait_interval_.get() &&
+ wait_interval_->pending_configure_job.get());
return false;
}
} else {
@@ -365,42 +372,37 @@
}
SyncSchedulerImpl::JobProcessDecision
-SyncSchedulerImpl::DecideWhileInWaitInterval(const SyncSessionJob& job) {
+SyncSchedulerImpl::DecideWhileInWaitInterval(
+ const SyncSessionJob& job) {
DCHECK_EQ(MessageLoop::current(), sync_loop_);
DCHECK(wait_interval_.get());
SDVLOG(2) << "DecideWhileInWaitInterval with WaitInterval mode "
<< WaitInterval::GetModeString(wait_interval_->mode)
<< (wait_interval_->had_nudge ? " (had nudge)" : "")
- << (job.is_canary() ? " (canary)" : "");
+ << (job.is_canary_job ? " (canary)" : "");
- if (job.purpose() == SyncSessionJob::POLL)
+ if (job.purpose == SyncSessionJob::POLL)
return DROP;
- // If we save a job while in a WaitInterval, there is a well-defined moment
- // in time in the future when it makes sense for that SAVE-worthy job to try
- // running again -- the end of the WaitInterval.
- DCHECK(job.purpose() == SyncSessionJob::NUDGE ||
- job.purpose() == SyncSessionJob::CONFIGURATION);
-
- // If throttled, there's a clock ticking to unthrottle. We want to get
- // on the same train.
+ DCHECK(job.purpose == SyncSessionJob::NUDGE ||
+ job.purpose == SyncSessionJob::CONFIGURATION);
if (wait_interval_->mode == WaitInterval::THROTTLED)
return SAVE;
DCHECK_EQ(wait_interval_->mode, WaitInterval::EXPONENTIAL_BACKOFF);
- if (job.purpose() == SyncSessionJob::NUDGE) {
+ if (job.purpose == SyncSessionJob::NUDGE) {
if (mode_ == CONFIGURATION_MODE)
return SAVE;
// If we already had one nudge then just drop this nudge. We will retry
// later when the timer runs out.
- if (!job.is_canary())
+ if (!job.is_canary_job)
return wait_interval_->had_nudge ? DROP : CONTINUE;
else // We are here because timer ran out. So retry.
return CONTINUE;
}
- return job.is_canary() ? CONTINUE : SAVE;
+ return job.is_canary_job ? CONTINUE : SAVE;
}
SyncSchedulerImpl::JobProcessDecision SyncSchedulerImpl::DecideOnJob(
@@ -410,22 +412,16 @@
// See if our type is throttled.
ModelTypeSet throttled_types =
session_context_->throttled_data_type_tracker()->GetThrottledTypes();
- if (job.purpose() == SyncSessionJob::NUDGE &&
- job.session()->source().updates_source == GetUpdatesCallerInfo::LOCAL) {
+ if (job.purpose == SyncSessionJob::NUDGE &&
+ job.session->source().updates_source == GetUpdatesCallerInfo::LOCAL) {
ModelTypeSet requested_types;
for (ModelTypeInvalidationMap::const_iterator i =
- job.session()->source().types.begin();
- i != job.session()->source().types.end();
+ job.session->source().types.begin();
+ i != job.session->source().types.end();
++i) {
requested_types.Put(i->first);
}
- // If all types are throttled, do not CONTINUE. Today, we don't treat
- // a per-datatype "unthrottle" event as something that should force a
- // canary job. For this reason, there's no good time to reschedule this job
- // to run -- we'll lazily wait for an independent event to trigger a sync.
- // Note that there may already be such an event if we're in a WaitInterval,
- // so we can retry it then.
if (!requested_types.Empty() && throttled_types.HasAll(requested_types))
return SAVE;
}
@@ -434,9 +430,9 @@
return DecideWhileInWaitInterval(job);
if (mode_ == CONFIGURATION_MODE) {
- if (job.purpose() == SyncSessionJob::NUDGE)
- return SAVE; // Running requires a mode switch.
- else if (job.purpose() == SyncSessionJob::CONFIGURATION)
+ if (job.purpose == SyncSessionJob::NUDGE)
+ return SAVE;
+ else if (job.purpose == SyncSessionJob::CONFIGURATION)
return CONTINUE;
else
return DROP;
@@ -444,7 +440,7 @@
// We are in normal mode.
DCHECK_EQ(mode_, NORMAL_MODE);
- DCHECK_NE(job.purpose(), SyncSessionJob::CONFIGURATION);
+ DCHECK_NE(job.purpose, SyncSessionJob::CONFIGURATION);
// Note about some subtle scheduling semantics.
//
@@ -487,61 +483,83 @@
return CONTINUE;
SDVLOG(2) << "No valid auth token. Using that to decide on job.";
- // Running the job would require updated auth, so we can't honour
- // job.scheduled_start().
- return job.purpose() == SyncSessionJob::NUDGE ? SAVE : DROP;
+ return job.purpose == SyncSessionJob::NUDGE ? SAVE : DROP;
}
-void SyncSchedulerImpl::HandleSaveJobDecision(scoped_ptr<SyncSessionJob> job) {
- DCHECK_EQ(DecideOnJob(*job), SAVE);
- const bool is_nudge = job->purpose() == SyncSessionJob::NUDGE;
- if (is_nudge && pending_nudge_) {
- SDVLOG(2) << "Coalescing a pending nudge";
- // TODO(tim): This basically means we never use the more-careful coalescing
- // logic in ScheduleNudgeImpl that takes the min of the two nudge start
- // times, because we're calling this function first. Pull this out
- // into a function to coalesce + set start times and reuse.
- pending_nudge_->mutable_session()->Coalesce(*(job->session()));
- return;
- }
+void SyncSchedulerImpl::InitOrCoalescePendingJob(const SyncSessionJob& job) {
+ DCHECK_EQ(MessageLoop::current(), sync_loop_);
+ DCHECK(job.purpose != SyncSessionJob::CONFIGURATION);
+ if (pending_nudge_.get() == NULL) {
+ SDVLOG(2) << "Creating a pending nudge job";
+ SyncSession* s = job.session.get();
- scoped_ptr<SyncSessionJob> job_to_save = job->CloneAndAbandon();
- if (wait_interval_.get() && !wait_interval_->pending_configure_job) {
- // This job should be made the new canary.
- if (is_nudge) {
- pending_nudge_ = job_to_save.get();
- } else {
- SDVLOG(2) << "Saving a configuration job";
- DCHECK_EQ(job->purpose(), SyncSessionJob::CONFIGURATION);
- DCHECK(!wait_interval_->pending_configure_job);
- DCHECK_EQ(mode_, CONFIGURATION_MODE);
- DCHECK(!job->config_params().ready_task.is_null());
- // The only nudge that could exist is a scheduled canary nudge.
- DCHECK(!unscheduled_nudge_storage_.get());
- if (pending_nudge_) {
- // Pre-empt the nudge canary and abandon the old nudge (owned by task).
- unscheduled_nudge_storage_ = pending_nudge_->CloneAndAbandon();
- pending_nudge_ = unscheduled_nudge_storage_.get();
- }
- wait_interval_->pending_configure_job = job_to_save.get();
- }
- TimeDelta length =
- wait_interval_->timer.desired_run_time() - TimeTicks::Now();
- wait_interval_->length = length < TimeDelta::FromSeconds(0) ?
- TimeDelta::FromSeconds(0) : length;
- RestartWaiting(job_to_save.Pass());
+ // Get a fresh session with similar configuration as before (resets
+ // StatusController).
+ scoped_ptr<SyncSession> session(new SyncSession(s->context(),
+ s->delegate(), s->source(), s->routing_info(), s->workers()));
+
+ SyncSessionJob new_job(SyncSessionJob::NUDGE, job.scheduled_start,
+ make_linked_ptr(session.release()), false,
+ ConfigurationParams(), job.from_here);
+ pending_nudge_.reset(new SyncSessionJob(new_job));
return;
}
- // Note that today there are no cases where we SAVE a CONFIGURATION job
- // when we're not in a WaitInterval. See bug 147736.
- DCHECK(is_nudge);
- // There may or may not be a pending_configure_job. Either way this nudge
- // is unschedulable.
- pending_nudge_ = job_to_save.get();
- unscheduled_nudge_storage_ = job_to_save.Pass();
+ SDVLOG(2) << "Coalescing a pending nudge";
+ pending_nudge_->session->Coalesce(*(job.session.get()));
+ pending_nudge_->scheduled_start = job.scheduled_start;
+
+ // Unfortunately the nudge location cannot be modified. So it stores the
+ // location of the first caller.
}
+bool SyncSchedulerImpl::ShouldRunJob(const SyncSessionJob& job) {
+ DCHECK_EQ(MessageLoop::current(), sync_loop_);
+ DCHECK(started_);
+
+ JobProcessDecision decision = DecideOnJob(job);
+ SDVLOG(2) << "Should run "
+ << SyncSessionJob::GetPurposeString(job.purpose)
+ << " job in mode " << GetModeString(mode_)
+ << ": " << GetDecisionString(decision);
+ if (decision != SAVE)
+ return decision == CONTINUE;
+
+ DCHECK(job.purpose == SyncSessionJob::NUDGE || job.purpose ==
+ SyncSessionJob::CONFIGURATION);
+
+ SaveJob(job);
+ return false;
+}
+
+void SyncSchedulerImpl::SaveJob(const SyncSessionJob& job) {
+ DCHECK_EQ(MessageLoop::current(), sync_loop_);
+ if (job.purpose == SyncSessionJob::NUDGE) {
+ SDVLOG(2) << "Saving a nudge job";
+ InitOrCoalescePendingJob(job);
+ } else if (job.purpose == SyncSessionJob::CONFIGURATION){
+ SDVLOG(2) << "Saving a configuration job";
+ DCHECK(wait_interval_.get());
+ DCHECK(mode_ == CONFIGURATION_MODE);
+
+ // Config params should always get set.
+ DCHECK(!job.config_params.ready_task.is_null());
+ SyncSession* old = job.session.get();
+ SyncSession* s(new SyncSession(session_context_, this, old->source(),
+ old->routing_info(), old->workers()));
+ SyncSessionJob new_job(job.purpose,
+ TimeTicks::Now(),
+ make_linked_ptr(s),
+ false,
+ job.config_params,
+ job.from_here);
+ wait_interval_->pending_configure_job.reset(new SyncSessionJob(new_job));
+ } // drop the rest.
+ // TODO(sync): Is it okay to drop the rest? It's weird that
+ // SaveJob() only does what it says sometimes. (See
+ // http://crbug.com/90868.)
+}
+
// Functor for std::find_if to search by ModelSafeGroup.
struct ModelSafeWorkerGroupIs {
explicit ModelSafeWorkerGroupIs(ModelSafeGroup group) : group(group) {}
@@ -552,39 +570,39 @@
};
void SyncSchedulerImpl::ScheduleNudgeAsync(
- const TimeDelta& desired_delay,
+ const TimeDelta& delay,
NudgeSource source, ModelTypeSet types,
const tracked_objects::Location& nudge_location) {
DCHECK_EQ(MessageLoop::current(), sync_loop_);
SDVLOG_LOC(nudge_location, 2)
- << "Nudge scheduled with delay "
- << desired_delay.InMilliseconds() << " ms, "
+ << "Nudge scheduled with delay " << delay.InMilliseconds() << " ms, "
<< "source " << GetNudgeSourceString(source) << ", "
<< "types " << ModelTypeSetToString(types);
ModelTypeInvalidationMap invalidation_map =
ModelTypeSetToInvalidationMap(types, std::string());
- SyncSchedulerImpl::ScheduleNudgeImpl(desired_delay,
+ SyncSchedulerImpl::ScheduleNudgeImpl(delay,
GetUpdatesFromNudgeSource(source),
invalidation_map,
+ false,
nudge_location);
}
void SyncSchedulerImpl::ScheduleNudgeWithStatesAsync(
- const TimeDelta& desired_delay,
+ const TimeDelta& delay,
NudgeSource source, const ModelTypeInvalidationMap& invalidation_map,
const tracked_objects::Location& nudge_location) {
DCHECK_EQ(MessageLoop::current(), sync_loop_);
SDVLOG_LOC(nudge_location, 2)
- << "Nudge scheduled with delay "
- << desired_delay.InMilliseconds() << " ms, "
+ << "Nudge scheduled with delay " << delay.InMilliseconds() << " ms, "
<< "source " << GetNudgeSourceString(source) << ", "
<< "payloads "
<< ModelTypeInvalidationMapToString(invalidation_map);
- SyncSchedulerImpl::ScheduleNudgeImpl(desired_delay,
+ SyncSchedulerImpl::ScheduleNudgeImpl(delay,
GetUpdatesFromNudgeSource(source),
invalidation_map,
+ false,
nudge_location);
}
@@ -592,7 +610,7 @@
const TimeDelta& delay,
GetUpdatesCallerInfo::GetUpdatesSource source,
const ModelTypeInvalidationMap& invalidation_map,
- const tracked_objects::Location& nudge_location) {
+ bool is_canary_job, const tracked_objects::Location& nudge_location) {
DCHECK_EQ(MessageLoop::current(), sync_loop_);
DCHECK(!invalidation_map.empty()) << "Nudge scheduled for no types!";
@@ -601,55 +619,44 @@
<< delay.InMilliseconds() << " ms, "
<< "source " << GetUpdatesSourceString(source) << ", "
<< "payloads "
- << ModelTypeInvalidationMapToString(invalidation_map);
+ << ModelTypeInvalidationMapToString(invalidation_map)
+ << (is_canary_job ? " (canary)" : "");
SyncSourceInfo info(source, invalidation_map);
UpdateNudgeTimeRecords(info);
- scoped_ptr<SyncSessionJob> job(new SyncSessionJob(
- SyncSessionJob::NUDGE,
- TimeTicks::Now() + delay,
- CreateSyncSession(info).Pass(),
- ConfigurationParams(),
- nudge_location));
+ SyncSession* session(CreateSyncSession(info));
+ SyncSessionJob job(SyncSessionJob::NUDGE, TimeTicks::Now() + delay,
+ make_linked_ptr(session), is_canary_job,
+ ConfigurationParams(), nudge_location);
- JobProcessDecision decision = DecideOnJob(*job);
- SDVLOG(2) << "Should run "
- << SyncSessionJob::GetPurposeString(job->purpose())
- << " job " << job->session()
- << " in mode " << GetModeString(mode_)
- << ": " << GetDecisionString(decision);
- if (decision != CONTINUE) {
- // End of the line, though we may save the job for later.
- if (decision == SAVE) {
- HandleSaveJobDecision(job.Pass());
- } else {
- DCHECK_EQ(decision, DROP);
- }
+ session = NULL;
+ if (!ShouldRunJob(job))
return;
- }
- if (pending_nudge_) {
+ if (pending_nudge_.get()) {
+ if (IsBackingOff() && delay > TimeDelta::FromSeconds(1)) {
+ SDVLOG(2) << "Dropping the nudge because we are in backoff";
+ return;
+ }
+
+ SDVLOG(2) << "Coalescing pending nudge";
+ pending_nudge_->session->Coalesce(*(job.session.get()));
+
SDVLOG(2) << "Rescheduling pending nudge";
- pending_nudge_->mutable_session()->Coalesce(*(job->session()));
- // Choose the start time as the earliest of the 2. Note that this means
- // if a nudge arrives with delay (e.g. kDefaultSessionsCommitDelaySeconds)
- // but a nudge is already scheduled to go out, we'll send the (tab) commit
- // without waiting.
- pending_nudge_->set_scheduled_start(
- std::min(job->scheduled_start(), pending_nudge_->scheduled_start()));
- // Abandon the old task by cloning and replacing the session.
- // It's possible that by "rescheduling" we're actually taking a job that
- // was previously unscheduled and giving it wings, so take care to reset
- // unscheduled nudge storage.
- job = pending_nudge_->CloneAndAbandon();
- unscheduled_nudge_storage_.reset();
- pending_nudge_ = NULL;
+ SyncSession* s = pending_nudge_->session.get();
+ job.session.reset(new SyncSession(s->context(), s->delegate(),
+ s->source(), s->routing_info(), s->workers()));
+
+ // Choose the start time as the earliest of the 2.
+ job.scheduled_start = std::min(job.scheduled_start,
+ pending_nudge_->scheduled_start);
+ pending_nudge_.reset();
}
// TODO(zea): Consider adding separate throttling/backoff for datatype
// refresh requests.
- ScheduleSyncSessionJob(job.Pass());
+ ScheduleSyncSessionJob(job);
}
const char* SyncSchedulerImpl::GetModeString(SyncScheduler::Mode mode) {
@@ -670,6 +677,29 @@
return "";
}
+// static
+void SyncSchedulerImpl::SetSyncerStepsForPurpose(
+ SyncSessionJob::SyncSessionJobPurpose purpose,
+ SyncerStep* start,
+ SyncerStep* end) {
+ switch (purpose) {
+ case SyncSessionJob::CONFIGURATION:
+ *start = DOWNLOAD_UPDATES;
+ *end = APPLY_UPDATES;
+ return;
+ case SyncSessionJob::NUDGE:
+ case SyncSessionJob::POLL:
+ *start = SYNCER_BEGIN;
+ *end = SYNCER_END;
+ return;
+ default:
+ NOTREACHED();
+ *start = SYNCER_END;
+ *end = SYNCER_END;
+ return;
+ }
+}
+
void SyncSchedulerImpl::PostTask(
const tracked_objects::Location& from_here,
const char* name, const base::Closure& task) {
@@ -695,93 +725,82 @@
sync_loop_->PostDelayedTask(from_here, task, delay);
}
-void SyncSchedulerImpl::ScheduleSyncSessionJob(
- scoped_ptr<SyncSessionJob> job) {
+void SyncSchedulerImpl::ScheduleSyncSessionJob(const SyncSessionJob& job) {
DCHECK_EQ(MessageLoop::current(), sync_loop_);
if (no_scheduling_allowed_) {
NOTREACHED() << "Illegal to schedule job while session in progress.";
return;
}
- TimeDelta delay = job->scheduled_start() - TimeTicks::Now();
- tracked_objects::Location loc(job->from_location());
+ TimeDelta delay = job.scheduled_start - TimeTicks::Now();
if (delay < TimeDelta::FromMilliseconds(0))
delay = TimeDelta::FromMilliseconds(0);
- SDVLOG_LOC(loc, 2)
+ SDVLOG_LOC(job.from_here, 2)
<< "In ScheduleSyncSessionJob with "
- << SyncSessionJob::GetPurposeString(job->purpose())
+ << SyncSessionJob::GetPurposeString(job.purpose)
<< " job and " << delay.InMilliseconds() << " ms delay";
- DCHECK(job->purpose() == SyncSessionJob::NUDGE ||
- job->purpose() == SyncSessionJob::POLL);
- if (job->purpose() == SyncSessionJob::NUDGE) {
- SDVLOG_LOC(loc, 2) << "Resetting pending_nudge to ";
- DCHECK(!pending_nudge_ || pending_nudge_->session() ==
- job->session());
- pending_nudge_ = job.get();
+ DCHECK(job.purpose == SyncSessionJob::NUDGE ||
+ job.purpose == SyncSessionJob::POLL);
+ if (job.purpose == SyncSessionJob::NUDGE) {
+ SDVLOG_LOC(job.from_here, 2) << "Resetting pending_nudge";
+ DCHECK(!pending_nudge_.get() || pending_nudge_->session.get() ==
+ job.session);
+ pending_nudge_.reset(new SyncSessionJob(job));
}
-
- PostDelayedTask(loc, "DoSyncSessionJob",
- base::Bind(base::IgnoreResult(&SyncSchedulerImpl::DoSyncSessionJob),
- weak_ptr_factory_.GetWeakPtr(),
- base::Passed(&job)),
- delay);
+ PostDelayedTask(job.from_here, "DoSyncSessionJob",
+ base::Bind(&SyncSchedulerImpl::DoSyncSessionJob,
+ weak_ptr_factory_.GetWeakPtr(),
+ job),
+ delay);
}
-bool SyncSchedulerImpl::DoSyncSessionJob(scoped_ptr<SyncSessionJob> job) {
+void SyncSchedulerImpl::DoSyncSessionJob(const SyncSessionJob& job) {
DCHECK_EQ(MessageLoop::current(), sync_loop_);
- if (job->purpose() == SyncSessionJob::NUDGE) {
- if (pending_nudge_ == NULL ||
- pending_nudge_->session() != job->session()) {
- // |job| is abandoned.
+
+ AutoReset<bool> protector(&no_scheduling_allowed_, true);
+ if (!ShouldRunJob(job)) {
+ SLOG(WARNING)
+ << "Not executing "
+ << SyncSessionJob::GetPurposeString(job.purpose) << " job from "
+ << GetUpdatesSourceString(job.session->source().updates_source);
+ return;
+ }
+
+ if (job.purpose == SyncSessionJob::NUDGE) {
+ if (pending_nudge_.get() == NULL ||
+ pending_nudge_->session != job.session) {
SDVLOG(2) << "Dropping a nudge in "
<< "DoSyncSessionJob because another nudge was scheduled";
- return false;
+ return; // Another nudge must have been scheduled in in the meantime.
}
- pending_nudge_ = NULL;
+ pending_nudge_.reset();
- // Rebase the session with the latest model safe table and use it to purge
+ // Create the session with the latest model safe table and use it to purge
// and update any disabled or modified entries in the job.
- job->mutable_session()->RebaseRoutingInfoWithLatest(
- session_context_->routing_info(), session_context_->workers());
- }
+ scoped_ptr<SyncSession> session(CreateSyncSession(job.session->source()));
- AutoReset<bool> protector(&no_scheduling_allowed_, true);
- JobProcessDecision decision = DecideOnJob(*job);
- SDVLOG(2) << "Should run "
- << SyncSessionJob::GetPurposeString(job->purpose())
- << " job " << job->session()
- << " in mode " << GetModeString(mode_)
- << " with source " << job->session()->source().updates_source
- << ": " << GetDecisionString(decision);
- if (decision != CONTINUE) {
- if (decision == SAVE) {
- HandleSaveJobDecision(job.Pass());
- } else {
- DCHECK_EQ(decision, DROP);
- }
- return false;
+ job.session->RebaseRoutingInfoWithLatest(*session);
}
-
SDVLOG(2) << "DoSyncSessionJob with "
- << SyncSessionJob::GetPurposeString(job->purpose()) << " job";
+ << SyncSessionJob::GetPurposeString(job.purpose) << " job";
+ SyncerStep begin(SYNCER_END);
+ SyncerStep end(SYNCER_END);
+ SetSyncerStepsForPurpose(job.purpose, &begin, &end);
+
bool has_more_to_sync = true;
- bool premature_exit = false;
- while (DecideOnJob(*job) == CONTINUE && has_more_to_sync) {
+ while (ShouldRunJob(job) && has_more_to_sync) {
SDVLOG(2) << "Calling SyncShare.";
// Synchronously perform the sync session from this thread.
- premature_exit = !syncer_->SyncShare(job->mutable_session(),
- job->start_step(),
- job->end_step());
-
- has_more_to_sync = job->session()->HasMoreToSync();
+ syncer_->SyncShare(job.session.get(), begin, end);
+ has_more_to_sync = job.session->HasMoreToSync();
if (has_more_to_sync)
- job->mutable_session()->PrepareForAnotherSyncCycle();
+ job.session->PrepareForAnotherSyncCycle();
}
SDVLOG(2) << "Done SyncShare looping.";
- return FinishSyncSessionJob(job.Pass(), premature_exit);
+ FinishSyncSessionJob(job);
}
void SyncSchedulerImpl::UpdateNudgeTimeRecords(const SyncSourceInfo& info) {
@@ -804,82 +823,64 @@
#define PER_DATA_TYPE_MACRO(type_str) \
SYNC_FREQ_HISTOGRAM("Sync.Freq" type_str, now - previous);
- SYNC_DATA_TYPE_HISTOGRAM(iter->first);
+ SYNC_DATA_TYPE_HISTOGRAM(iter->first);
#undef PER_DATA_TYPE_MACRO
}
}
-bool SyncSchedulerImpl::FinishSyncSessionJob(scoped_ptr<SyncSessionJob> job,
- bool exited_prematurely) {
+void SyncSchedulerImpl::FinishSyncSessionJob(const SyncSessionJob& job) {
DCHECK_EQ(MessageLoop::current(), sync_loop_);
// Now update the status of the connection from SCM. We need this to decide
- // whether we need to save/run future jobs. The notifications from SCM are
- // not reliable.
+ // whether we need to save/run future jobs. The notifications from SCM are not
+ // reliable.
//
// TODO(rlarocque): crbug.com/110954
// We should get rid of the notifications and it is probably not needed to
- // maintain this status variable in 2 places. We should query it directly
- // from SCM when needed.
+ // maintain this status variable in 2 places. We should query it directly from
+ // SCM when needed.
ServerConnectionManager* scm = session_context_->connection_manager();
UpdateServerConnectionManagerStatus(scm->server_status());
- // Let job know that we're through syncing (calling SyncShare) at this point.
- bool succeeded = false;
- {
+ if (IsSyncingCurrentlySilenced()) {
+ SDVLOG(2) << "We are currently throttled; not scheduling the next sync.";
+ // TODO(sync): Investigate whether we need to check job.purpose
+ // here; see DCHECKs in SaveJob(). (See http://crbug.com/90868.)
+ SaveJob(job);
+ return; // Nothing to do.
+ } else if (job.session->Succeeded() &&
+ !job.config_params.ready_task.is_null()) {
+ // If this was a configuration job with a ready task, invoke it now that
+ // we finished successfully.
AutoReset<bool> protector(&no_scheduling_allowed_, true);
- succeeded = job->Finish(exited_prematurely);
+ job.config_params.ready_task.Run();
}
SDVLOG(2) << "Updating the next polling time after SyncMain";
- ScheduleNextSync(job.Pass(), succeeded);
- return succeeded;
+ ScheduleNextSync(job);
}
-void SyncSchedulerImpl::ScheduleNextSync(
- scoped_ptr<SyncSessionJob> finished_job, bool succeeded) {
+void SyncSchedulerImpl::ScheduleNextSync(const SyncSessionJob& old_job) {
DCHECK_EQ(MessageLoop::current(), sync_loop_);
- DCHECK(!finished_job->session()->HasMoreToSync());
+ DCHECK(!old_job.session->HasMoreToSync());
- AdjustPolling(finished_job.get());
+ AdjustPolling(&old_job);
- if (succeeded) {
+ if (old_job.session->Succeeded()) {
// Only reset backoff if we actually reached the server.
- // It's possible that we reached the server on one attempt, then had an
- // error on the next (or didn't perform some of the server-communicating
- // commands). We want to verify that, for all commands attempted, we
- // successfully spoke with the server. Therefore, we verify no errors
- // and at least one SYNCER_OK.
- if (finished_job->session()->DidReachServer())
+ if (old_job.session->SuccessfullyReachedServer())
wait_interval_.reset();
SDVLOG(2) << "Job succeeded so not scheduling more jobs";
return;
}
- if (IsSyncingCurrentlySilenced()) {
- SDVLOG(2) << "We are currently throttled; scheduling Unthrottle.";
- // If we're here, it's because |job| was silenced until a server specified
- // time. (Note, it had to be |job|, because DecideOnJob would not permit
- // any job through while in WaitInterval::THROTTLED).
- scoped_ptr<SyncSessionJob> clone = finished_job->Clone();
- if (clone->purpose() == SyncSessionJob::NUDGE)
- pending_nudge_ = clone.get();
- else if (clone->purpose() == SyncSessionJob::CONFIGURATION)
- wait_interval_->pending_configure_job = clone.get();
- else
- clone.reset(); // Unthrottling is enough, no need to force a canary.
-
- RestartWaiting(clone.Pass());
- return;
- }
-
- if (finished_job->purpose() == SyncSessionJob::POLL) {
+ if (old_job.purpose == SyncSessionJob::POLL) {
return; // We don't retry POLL jobs.
}
// TODO(rlarocque): There's no reason why we should blindly backoff and retry
// if we don't succeed. Some types of errors are not likely to disappear on
- // their own. With the return values now available in the old_job.session,
- // we should be able to detect such errors and only retry when we detect
+ // their own. With the return values now available in the old_job.session, we
+ // should be able to detect such errors and only retry when we detect
// transient errors.
if (IsBackingOff() && wait_interval_->timer.IsRunning() &&
@@ -887,24 +888,22 @@
// When in normal mode, we allow up to one nudge per backoff interval. It
// appears that this was our nudge for this interval, and it failed.
//
- // Note: This does not prevent us from running canary jobs. For example,
- // an IP address change might still result in another nudge being executed
+ // Note: This does not prevent us from running canary jobs. For example, an
+ // IP address change might still result in another nudge being executed
// during this backoff interval.
- SDVLOG(2) << "A nudge during backoff failed, creating new pending nudge.";
- DCHECK_EQ(SyncSessionJob::NUDGE, finished_job->purpose());
+ SDVLOG(2) << "A nudge during backoff failed";
+
+ DCHECK_EQ(SyncSessionJob::NUDGE, old_job.purpose);
DCHECK(!wait_interval_->had_nudge);
wait_interval_->had_nudge = true;
- DCHECK(!pending_nudge_);
-
- scoped_ptr<SyncSessionJob> new_job = finished_job->Clone();
- pending_nudge_ = new_job.get();
- RestartWaiting(new_job.Pass());
+ InitOrCoalescePendingJob(old_job);
+ RestartWaiting();
} else {
// Either this is the first failure or a consecutive failure after our
// backoff timer expired. We handle it the same way in either case.
SDVLOG(2) << "Non-'backoff nudge' SyncShare job failed";
- HandleContinuationError(finished_job.Pass());
+ HandleContinuationError(old_job);
}
}
@@ -917,7 +916,7 @@
bool rate_changed = !poll_timer_.IsRunning() ||
poll != poll_timer_.GetCurrentDelay();
- if (old_job && old_job->purpose() != SyncSessionJob::POLL && !rate_changed)
+ if (old_job && old_job->purpose != SyncSessionJob::POLL && !rate_changed)
poll_timer_.Reset();
if (!rate_changed)
@@ -929,61 +928,55 @@
&SyncSchedulerImpl::PollTimerCallback);
}
-void SyncSchedulerImpl::RestartWaiting(scoped_ptr<SyncSessionJob> job) {
+void SyncSchedulerImpl::RestartWaiting() {
CHECK(wait_interval_.get());
wait_interval_->timer.Stop();
- DCHECK(wait_interval_->length >= TimeDelta::FromSeconds(0));
- if (wait_interval_->mode == WaitInterval::THROTTLED) {
- wait_interval_->timer.Start(FROM_HERE, wait_interval_->length,
- base::Bind(&SyncSchedulerImpl::Unthrottle,
- weak_ptr_factory_.GetWeakPtr(),
- base::Passed(&job)));
- } else {
- wait_interval_->timer.Start(FROM_HERE, wait_interval_->length,
- base::Bind(&SyncSchedulerImpl::DoCanaryJob,
- weak_ptr_factory_.GetWeakPtr(),
- base::Passed(&job)));
- }
+ wait_interval_->timer.Start(FROM_HERE, wait_interval_->length,
+ this, &SyncSchedulerImpl::DoCanaryJob);
}
void SyncSchedulerImpl::HandleContinuationError(
- scoped_ptr<SyncSessionJob> old_job) {
+ const SyncSessionJob& old_job) {
DCHECK_EQ(MessageLoop::current(), sync_loop_);
if (DCHECK_IS_ON()) {
if (IsBackingOff()) {
- DCHECK(wait_interval_->timer.IsRunning() || old_job->is_canary());
+ DCHECK(wait_interval_->timer.IsRunning() || old_job.is_canary_job);
}
}
TimeDelta length = delay_provider_->GetDelay(
IsBackingOff() ? wait_interval_->length :
delay_provider_->GetInitialDelay(
- old_job->session()->status_controller().model_neutral_state()));
+ old_job.session->status_controller().model_neutral_state()));
SDVLOG(2) << "In handle continuation error with "
- << SyncSessionJob::GetPurposeString(old_job->purpose())
+ << SyncSessionJob::GetPurposeString(old_job.purpose)
<< " job. The time delta(ms) is "
<< length.InMilliseconds();
// This will reset the had_nudge variable as well.
wait_interval_.reset(new WaitInterval(WaitInterval::EXPONENTIAL_BACKOFF,
length));
- scoped_ptr<SyncSessionJob> new_job(old_job->CloneFromLocation(FROM_HERE));
- new_job->set_scheduled_start(TimeTicks::Now() + length);
- if (old_job->purpose() == SyncSessionJob::CONFIGURATION) {
+ if (old_job.purpose == SyncSessionJob::CONFIGURATION) {
SDVLOG(2) << "Configuration did not succeed, scheduling retry.";
// Config params should always get set.
- DCHECK(!old_job->config_params().ready_task.is_null());
- wait_interval_->pending_configure_job = new_job.get();
+ DCHECK(!old_job.config_params.ready_task.is_null());
+ SyncSession* old = old_job.session.get();
+ SyncSession* s(new SyncSession(session_context_, this,
+ old->source(), old->routing_info(), old->workers()));
+ SyncSessionJob job(old_job.purpose, TimeTicks::Now() + length,
+ make_linked_ptr(s), false, old_job.config_params,
+ FROM_HERE);
+ wait_interval_->pending_configure_job.reset(new SyncSessionJob(job));
} else {
// We are not in configuration mode. So wait_interval's pending job
// should be null.
- DCHECK(wait_interval_->pending_configure_job == NULL);
- DCHECK(!pending_nudge_);
- pending_nudge_ = new_job.get();
- }
+ DCHECK(wait_interval_->pending_configure_job.get() == NULL);
- RestartWaiting(new_job.Pass());
+ // TODO(lipalani) - handle clear user data.
+ InitOrCoalescePendingJob(old_job);
+ }
+ RestartWaiting();
}
void SyncSchedulerImpl::RequestStop(const base::Closure& callback) {
@@ -1010,56 +1003,51 @@
callback.Run();
}
-void SyncSchedulerImpl::DoCanaryJob(scoped_ptr<SyncSessionJob> to_be_canary) {
+void SyncSchedulerImpl::DoCanaryJob() {
DCHECK_EQ(MessageLoop::current(), sync_loop_);
SDVLOG(2) << "Do canary job";
-
- // Only set canary privileges here, when we are about to run the job. This
- // avoids confusion in managing canary bits during scheduling, when you
- // consider that mode switches (e.g., to config) can "pre-empt" a NUDGE that
- // was scheduled as canary, and send it to an "unscheduled" state.
- to_be_canary->GrantCanaryPrivilege();
-
- if (to_be_canary->purpose() == SyncSessionJob::NUDGE) {
- // TODO(tim): We should be able to remove this...
- scoped_ptr<SyncSession> temp = CreateSyncSession(
- to_be_canary->session()->source()).Pass();
- // The routing info might have been changed since we cached the
- // pending nudge. Update it by coalescing to the latest.
- to_be_canary->mutable_session()->Coalesce(*(temp));
- }
- DoSyncSessionJob(to_be_canary.Pass());
+ DoPendingJobIfPossible(true);
}
-scoped_ptr<SyncSessionJob> SyncSchedulerImpl::TakePendingJobForCurrentMode() {
+void SyncSchedulerImpl::DoPendingJobIfPossible(bool is_canary_job) {
DCHECK_EQ(MessageLoop::current(), sync_loop_);
- // If we find a scheduled pending_ job, abandon the old one and return a
- // a clone. If unscheduled, just hand over ownership.
- scoped_ptr<SyncSessionJob> candidate;
+ SyncSessionJob* job_to_execute = NULL;
if (mode_ == CONFIGURATION_MODE && wait_interval_.get()
- && wait_interval_->pending_configure_job) {
+ && wait_interval_->pending_configure_job.get()) {
SDVLOG(2) << "Found pending configure job";
- candidate =
- wait_interval_->pending_configure_job->CloneAndAbandon().Pass();
- wait_interval_->pending_configure_job = candidate.get();
- } else if (mode_ == NORMAL_MODE && pending_nudge_) {
+ job_to_execute = wait_interval_->pending_configure_job.get();
+ } else if (mode_ == NORMAL_MODE && pending_nudge_.get()) {
SDVLOG(2) << "Found pending nudge job";
- candidate = pending_nudge_->CloneAndAbandon();
- pending_nudge_ = candidate.get();
- unscheduled_nudge_storage_.reset();
+
+ scoped_ptr<SyncSession> session(CreateSyncSession(
+ pending_nudge_->session->source()));
+
+ // Also the routing info might have been changed since we cached the
+ // pending nudge. Update it by coalescing to the latest.
+ pending_nudge_->session->Coalesce(*(session.get()));
+ // The pending nudge would be cleared in the DoSyncSessionJob function.
+ job_to_execute = pending_nudge_.get();
}
- return candidate.Pass();
+
+ if (job_to_execute != NULL) {
+ SDVLOG(2) << "Executing pending job";
+ SyncSessionJob copy = *job_to_execute;
+ copy.is_canary_job = is_canary_job;
+ DoSyncSessionJob(copy);
+ }
}
-scoped_ptr<SyncSession> SyncSchedulerImpl::CreateSyncSession(
+SyncSession* SyncSchedulerImpl::CreateSyncSession(
const SyncSourceInfo& source) {
DCHECK_EQ(MessageLoop::current(), sync_loop_);
DVLOG(2) << "Creating sync session with routes "
<< ModelSafeRoutingInfoToString(session_context_->routing_info());
SyncSourceInfo info(source);
- return scoped_ptr<SyncSession>(new SyncSession(session_context_, this, info,
+ SyncSession* session(new SyncSession(session_context_, this, info,
session_context_->routing_info(), session_context_->workers()));
+
+ return session;
}
void SyncSchedulerImpl::PollTimerCallback() {
@@ -1068,27 +1056,22 @@
ModelTypeInvalidationMap invalidation_map =
ModelSafeRoutingInfoToInvalidationMap(r, std::string());
SyncSourceInfo info(GetUpdatesCallerInfo::PERIODIC, invalidation_map);
- scoped_ptr<SyncSession> s(CreateSyncSession(info));
- scoped_ptr<SyncSessionJob> job(new SyncSessionJob(SyncSessionJob::POLL,
- TimeTicks::Now(),
- s.Pass(),
- ConfigurationParams(),
- FROM_HERE));
- ScheduleSyncSessionJob(job.Pass());
+ SyncSession* s = CreateSyncSession(info);
+
+ SyncSessionJob job(SyncSessionJob::POLL, TimeTicks::Now(),
+ make_linked_ptr(s),
+ false,
+ ConfigurationParams(),
+ FROM_HERE);
+
+ ScheduleSyncSessionJob(job);
}
-void SyncSchedulerImpl::Unthrottle(scoped_ptr<SyncSessionJob> to_be_canary) {
+void SyncSchedulerImpl::Unthrottle() {
DCHECK_EQ(MessageLoop::current(), sync_loop_);
DCHECK_EQ(WaitInterval::THROTTLED, wait_interval_->mode);
- SDVLOG(2) << "Unthrottled " << (to_be_canary.get() ? "with " : "without ")
- << "canary.";
- if (to_be_canary.get())
- DoCanaryJob(to_be_canary.Pass());
-
- // TODO(tim): The way DecideOnJob works today, canary privileges aren't
- // enough to bypass a THROTTLED wait interval, which would suggest we need
- // to reset before DoCanaryJob (though trusting canary in DecideOnJob is
- // probably the "right" thing to do). Bug 154216.
+ SDVLOG(2) << "Unthrottled.";
+ DoCanaryJob();
wait_interval_.reset();
}
@@ -1108,6 +1091,8 @@
DCHECK_EQ(MessageLoop::current(), sync_loop_);
wait_interval_.reset(new WaitInterval(WaitInterval::THROTTLED,
silenced_until - TimeTicks::Now()));
+ wait_interval_->timer.Start(FROM_HERE, wait_interval_->length, this,
+ &SyncSchedulerImpl::Unthrottle);
}
bool SyncSchedulerImpl::IsSyncingCurrentlySilenced() {
« no previous file with comments | « sync/engine/sync_scheduler_impl.h ('k') | sync/engine/sync_scheduler_unittest.cc » ('j') | no next file with comments »

Powered by Google App Engine
This is Rietveld 408576698