| 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() {
|
|
|