| OLD | NEW |
|---|
| (Empty) |
| 1 // Copyright (c) 2011 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 "chrome/browser/sync/engine/syncer_thread2.h" | |
| 6 | |
| 7 #include <algorithm> | |
| 8 | |
| 9 #include "base/rand_util.h" | |
| 10 #include "chrome/browser/sync/engine/syncer.h" | |
| 11 | |
| 12 using base::TimeDelta; | |
| 13 using base::TimeTicks; | |
| 14 | |
| 15 namespace browser_sync { | |
| 16 | |
| 17 using sessions::SyncSession; | |
| 18 using sessions::SyncSessionSnapshot; | |
| 19 using sessions::SyncSourceInfo; | |
| 20 using syncable::ModelTypePayloadMap; | |
| 21 using syncable::ModelTypeBitSet; | |
| 22 using sync_pb::GetUpdatesCallerInfo; | |
| 23 | |
| 24 namespace s3 { | |
| 25 | |
| 26 SyncerThread::DelayProvider::DelayProvider() {} | |
| 27 SyncerThread::DelayProvider::~DelayProvider() {} | |
| 28 | |
| 29 SyncerThread::WaitInterval::WaitInterval() {} | |
| 30 SyncerThread::WaitInterval::~WaitInterval() {} | |
| 31 | |
| 32 SyncerThread::SyncSessionJob::SyncSessionJob() {} | |
| 33 SyncerThread::SyncSessionJob::~SyncSessionJob() {} | |
| 34 | |
| 35 SyncerThread::SyncSessionJob::SyncSessionJob(SyncSessionJobPurpose purpose, | |
| 36 base::TimeTicks start, | |
| 37 linked_ptr<sessions::SyncSession> session, bool is_canary_job, | |
| 38 const tracked_objects::Location& nudge_location) : purpose(purpose), | |
| 39 scheduled_start(start), | |
| 40 session(session), | |
| 41 is_canary_job(is_canary_job), | |
| 42 nudge_location(nudge_location) { | |
| 43 } | |
| 44 | |
| 45 TimeDelta SyncerThread::DelayProvider::GetDelay( | |
| 46 const base::TimeDelta& last_delay) { | |
| 47 return SyncerThread::GetRecommendedDelay(last_delay); | |
| 48 } | |
| 49 | |
| 50 GetUpdatesCallerInfo::GetUpdatesSource GetUpdatesFromNudgeSource( | |
| 51 NudgeSource source) { | |
| 52 switch (source) { | |
| 53 case NUDGE_SOURCE_NOTIFICATION: | |
| 54 return GetUpdatesCallerInfo::NOTIFICATION; | |
| 55 case NUDGE_SOURCE_LOCAL: | |
| 56 return GetUpdatesCallerInfo::LOCAL; | |
| 57 case NUDGE_SOURCE_CONTINUATION: | |
| 58 return GetUpdatesCallerInfo::SYNC_CYCLE_CONTINUATION; | |
| 59 case NUDGE_SOURCE_UNKNOWN: | |
| 60 return GetUpdatesCallerInfo::UNKNOWN; | |
| 61 default: | |
| 62 NOTREACHED(); | |
| 63 return GetUpdatesCallerInfo::UNKNOWN; | |
| 64 } | |
| 65 } | |
| 66 | |
| 67 SyncerThread::WaitInterval::WaitInterval(Mode mode, TimeDelta length) | |
| 68 : mode(mode), had_nudge(false), length(length) { } | |
| 69 | |
| 70 SyncerThread::SyncerThread(sessions::SyncSessionContext* context, | |
| 71 Syncer* syncer) | |
| 72 : thread_("SyncEngine_SyncerThread"), | |
| 73 syncer_short_poll_interval_seconds_( | |
| 74 TimeDelta::FromSeconds(kDefaultShortPollIntervalSeconds)), | |
| 75 syncer_long_poll_interval_seconds_( | |
| 76 TimeDelta::FromSeconds(kDefaultLongPollIntervalSeconds)), | |
| 77 mode_(NORMAL_MODE), | |
| 78 server_connection_ok_(false), | |
| 79 delay_provider_(new DelayProvider()), | |
| 80 syncer_(syncer), | |
| 81 session_context_(context) { | |
| 82 } | |
| 83 | |
| 84 SyncerThread::~SyncerThread() { | |
| 85 DCHECK(!thread_.IsRunning()); | |
| 86 } | |
| 87 | |
| 88 void SyncerThread::CheckServerConnectionManagerStatus( | |
| 89 HttpResponse::ServerConnectionCode code) { | |
| 90 | |
| 91 VLOG(2) << "SyncerThread(" << this << ")" << " Server connection changed." | |
| 92 << "Old mode: " << server_connection_ok_ << " Code: " << code; | |
| 93 // Note, be careful when adding cases here because if the SyncerThread | |
| 94 // thinks there is no valid connection as determined by this method, it | |
| 95 // will drop out of *all* forward progress sync loops (it won't poll and it | |
| 96 // will queue up Talk notifications but not actually call SyncShare) until | |
| 97 // some external action causes a ServerConnectionManager to broadcast that | |
| 98 // a valid connection has been re-established. | |
| 99 if (HttpResponse::CONNECTION_UNAVAILABLE == code || | |
| 100 HttpResponse::SYNC_AUTH_ERROR == code) { | |
| 101 server_connection_ok_ = false; | |
| 102 VLOG(2) << "SyncerThread(" << this << ")" << " Server connection changed." | |
| 103 << " new mode:" << server_connection_ok_; | |
| 104 } else if (HttpResponse::SERVER_CONNECTION_OK == code) { | |
| 105 server_connection_ok_ = true; | |
| 106 VLOG(2) << "SyncerThread(" << this << ")" << " Server connection changed." | |
| 107 << " new mode:" << server_connection_ok_; | |
| 108 DoCanaryJob(); | |
| 109 } | |
| 110 } | |
| 111 | |
| 112 void SyncerThread::Start(Mode mode, ModeChangeCallback* callback) { | |
| 113 VLOG(2) << "SyncerThread(" << this << ")" << " Start called from thread " | |
| 114 << MessageLoop::current()->thread_name(); | |
| 115 if (!thread_.IsRunning()) { | |
| 116 VLOG(2) << "SyncerThread(" << this << ")" << " Starting thread with mode " | |
| 117 << mode; | |
| 118 if (!thread_.Start()) { | |
| 119 NOTREACHED() << "Unable to start SyncerThread."; | |
| 120 return; | |
| 121 } | |
| 122 WatchConnectionManager(); | |
| 123 thread_.message_loop()->PostTask(FROM_HERE, NewRunnableMethod( | |
| 124 this, &SyncerThread::SendInitialSnapshot)); | |
| 125 } | |
| 126 | |
| 127 VLOG(2) << "SyncerThread(" << this << ")" << " Entering start with mode = " | |
| 128 << mode; | |
| 129 | |
| 130 thread_.message_loop()->PostTask(FROM_HERE, NewRunnableMethod( | |
| 131 this, &SyncerThread::StartImpl, mode, make_linked_ptr(callback))); | |
| 132 } | |
| 133 | |
| 134 void SyncerThread::SendInitialSnapshot() { | |
| 135 DCHECK_EQ(MessageLoop::current(), thread_.message_loop()); | |
| 136 scoped_ptr<SyncSession> dummy(new SyncSession(session_context_.get(), this, | |
| 137 SyncSourceInfo(), ModelSafeRoutingInfo(), | |
| 138 std::vector<ModelSafeWorker*>())); | |
| 139 SyncEngineEvent event(SyncEngineEvent::STATUS_CHANGED); | |
| 140 sessions::SyncSessionSnapshot snapshot(dummy->TakeSnapshot()); | |
| 141 event.snapshot = &snapshot; | |
| 142 session_context_->NotifyListeners(event); | |
| 143 } | |
| 144 | |
| 145 void SyncerThread::WatchConnectionManager() { | |
| 146 ServerConnectionManager* scm = session_context_->connection_manager(); | |
| 147 CheckServerConnectionManagerStatus(scm->server_status()); | |
| 148 scm->AddListener(this); | |
| 149 } | |
| 150 | |
| 151 void SyncerThread::StartImpl(Mode mode, | |
| 152 linked_ptr<ModeChangeCallback> callback) { | |
| 153 VLOG(2) << "SyncerThread(" << this << ")" << " Doing StartImpl with mode " | |
| 154 << mode; | |
| 155 DCHECK_EQ(MessageLoop::current(), thread_.message_loop()); | |
| 156 DCHECK(!session_context_->account_name().empty()); | |
| 157 DCHECK(syncer_.get()); | |
| 158 mode_ = mode; | |
| 159 AdjustPolling(NULL); // Will kick start poll timer if needed. | |
| 160 if (callback.get()) | |
| 161 callback->Run(); | |
| 162 | |
| 163 // We just changed our mode. See if there are any pending jobs that we could | |
| 164 // execute in the new mode. | |
| 165 DoPendingJobIfPossible(false); | |
| 166 } | |
| 167 | |
| 168 SyncerThread::JobProcessDecision SyncerThread::DecideWhileInWaitInterval( | |
| 169 const SyncSessionJob& job) { | |
| 170 | |
| 171 DCHECK(wait_interval_.get()); | |
| 172 DCHECK_NE(job.purpose, SyncSessionJob::CLEAR_USER_DATA); | |
| 173 | |
| 174 VLOG(2) << "SyncerThread(" << this << ")" << " Wait interval mode : " | |
| 175 << wait_interval_->mode << "Wait interval had nudge : " | |
| 176 << wait_interval_->had_nudge << "is canary job : " | |
| 177 << job.is_canary_job; | |
| 178 | |
| 179 if (job.purpose == SyncSessionJob::POLL) | |
| 180 return DROP; | |
| 181 | |
| 182 DCHECK(job.purpose == SyncSessionJob::NUDGE || | |
| 183 job.purpose == SyncSessionJob::CONFIGURATION); | |
| 184 if (wait_interval_->mode == WaitInterval::THROTTLED) | |
| 185 return SAVE; | |
| 186 | |
| 187 DCHECK_EQ(wait_interval_->mode, WaitInterval::EXPONENTIAL_BACKOFF); | |
| 188 if (job.purpose == SyncSessionJob::NUDGE) { | |
| 189 if (mode_ == CONFIGURATION_MODE) | |
| 190 return SAVE; | |
| 191 | |
| 192 // If we already had one nudge then just drop this nudge. We will retry | |
| 193 // later when the timer runs out. | |
| 194 return wait_interval_->had_nudge ? DROP : CONTINUE; | |
| 195 } | |
| 196 // This is a config job. | |
| 197 return job.is_canary_job ? CONTINUE : SAVE; | |
| 198 } | |
| 199 | |
| 200 SyncerThread::JobProcessDecision SyncerThread::DecideOnJob( | |
| 201 const SyncSessionJob& job) { | |
| 202 if (job.purpose == SyncSessionJob::CLEAR_USER_DATA) | |
| 203 return CONTINUE; | |
| 204 | |
| 205 if (wait_interval_.get()) | |
| 206 return DecideWhileInWaitInterval(job); | |
| 207 | |
| 208 if (mode_ == CONFIGURATION_MODE) { | |
| 209 if (job.purpose == SyncSessionJob::NUDGE) | |
| 210 return SAVE; | |
| 211 else if (job.purpose == SyncSessionJob::CONFIGURATION) | |
| 212 return CONTINUE; | |
| 213 else | |
| 214 return DROP; | |
| 215 } | |
| 216 | |
| 217 // We are in normal mode. | |
| 218 DCHECK_EQ(mode_, NORMAL_MODE); | |
| 219 DCHECK_NE(job.purpose, SyncSessionJob::CONFIGURATION); | |
| 220 | |
| 221 // Freshness condition | |
| 222 if (job.scheduled_start < last_sync_session_end_time_) { | |
| 223 VLOG(2) << "SyncerThread(" << this << ")" | |
| 224 << " Dropping job because of freshness"; | |
| 225 return DROP; | |
| 226 } | |
| 227 | |
| 228 if (server_connection_ok_) | |
| 229 return CONTINUE; | |
| 230 | |
| 231 VLOG(2) << "SyncerThread(" << this << ")" | |
| 232 << " Bad server connection. Using that to decide on job."; | |
| 233 return job.purpose == SyncSessionJob::NUDGE ? SAVE : DROP; | |
| 234 } | |
| 235 | |
| 236 void SyncerThread::InitOrCoalescePendingJob(const SyncSessionJob& job) { | |
| 237 DCHECK(job.purpose != SyncSessionJob::CONFIGURATION); | |
| 238 if (pending_nudge_.get() == NULL) { | |
| 239 VLOG(2) << "SyncerThread(" << this << ")" | |
| 240 << " Creating a pending nudge job"; | |
| 241 SyncSession* s = job.session.get(); | |
| 242 scoped_ptr<SyncSession> session(new SyncSession(s->context(), | |
| 243 s->delegate(), s->source(), s->routing_info(), s->workers())); | |
| 244 | |
| 245 SyncSessionJob new_job(SyncSessionJob::NUDGE, job.scheduled_start, | |
| 246 make_linked_ptr(session.release()), false, job.nudge_location); | |
| 247 pending_nudge_.reset(new SyncSessionJob(new_job)); | |
| 248 | |
| 249 return; | |
| 250 } | |
| 251 | |
| 252 VLOG(2) << "SyncerThread(" << this << ")" << " Coalescing a pending nudge"; | |
| 253 pending_nudge_->session->Coalesce(*(job.session.get())); | |
| 254 pending_nudge_->scheduled_start = job.scheduled_start; | |
| 255 | |
| 256 // Unfortunately the nudge location cannot be modified. So it stores the | |
| 257 // location of the first caller. | |
| 258 } | |
| 259 | |
| 260 bool SyncerThread::ShouldRunJob(const SyncSessionJob& job) { | |
| 261 JobProcessDecision decision = DecideOnJob(job); | |
| 262 VLOG(2) << "SyncerThread(" << this << ")" << " Should run job, decision: " | |
| 263 << decision << " Job purpose " << job.purpose << "mode " << mode_; | |
| 264 if (decision != SAVE) | |
| 265 return decision == CONTINUE; | |
| 266 | |
| 267 DCHECK(job.purpose == SyncSessionJob::NUDGE || job.purpose == | |
| 268 SyncSessionJob::CONFIGURATION); | |
| 269 | |
| 270 SaveJob(job); | |
| 271 return false; | |
| 272 } | |
| 273 | |
| 274 void SyncerThread::SaveJob(const SyncSessionJob& job) { | |
| 275 DCHECK(job.purpose != SyncSessionJob::CLEAR_USER_DATA); | |
| 276 if (job.purpose == SyncSessionJob::NUDGE) { | |
| 277 VLOG(2) << "SyncerThread(" << this << ")" << " Saving a nudge job"; | |
| 278 InitOrCoalescePendingJob(job); | |
| 279 } else if (job.purpose == SyncSessionJob::CONFIGURATION){ | |
| 280 VLOG(2) << "SyncerThread(" << this << ")" << " Saving a configuration job"; | |
| 281 DCHECK(wait_interval_.get()); | |
| 282 DCHECK(mode_ == CONFIGURATION_MODE); | |
| 283 | |
| 284 SyncSession* old = job.session.get(); | |
| 285 SyncSession* s(new SyncSession(session_context_.get(), this, | |
| 286 old->source(), old->routing_info(), old->workers())); | |
| 287 SyncSessionJob new_job(job.purpose, TimeTicks::Now(), | |
| 288 make_linked_ptr(s), false, job.nudge_location); | |
| 289 wait_interval_->pending_configure_job.reset(new SyncSessionJob(new_job)); | |
| 290 } // drop the rest. | |
| 291 } | |
| 292 | |
| 293 // Functor for std::find_if to search by ModelSafeGroup. | |
| 294 struct ModelSafeWorkerGroupIs { | |
| 295 explicit ModelSafeWorkerGroupIs(ModelSafeGroup group) : group(group) {} | |
| 296 bool operator()(ModelSafeWorker* w) { | |
| 297 return group == w->GetModelSafeGroup(); | |
| 298 } | |
| 299 ModelSafeGroup group; | |
| 300 }; | |
| 301 | |
| 302 void SyncerThread::ScheduleClearUserData() { | |
| 303 if (!thread_.IsRunning()) { | |
| 304 NOTREACHED(); | |
| 305 return; | |
| 306 } | |
| 307 thread_.message_loop()->PostTask(FROM_HERE, NewRunnableMethod( | |
| 308 this, &SyncerThread::ScheduleClearUserDataImpl)); | |
| 309 } | |
| 310 | |
| 311 void SyncerThread::ScheduleNudge(const TimeDelta& delay, | |
| 312 NudgeSource source, const ModelTypeBitSet& types, | |
| 313 const tracked_objects::Location& nudge_location) { | |
| 314 if (!thread_.IsRunning()) { | |
| 315 NOTREACHED(); | |
| 316 return; | |
| 317 } | |
| 318 | |
| 319 VLOG(2) << "SyncerThread(" << this << ")" << " Nudge scheduled"; | |
| 320 | |
| 321 ModelTypePayloadMap types_with_payloads = | |
| 322 syncable::ModelTypePayloadMapFromBitSet(types, std::string()); | |
| 323 thread_.message_loop()->PostTask(FROM_HERE, NewRunnableMethod( | |
| 324 this, &SyncerThread::ScheduleNudgeImpl, delay, | |
| 325 GetUpdatesFromNudgeSource(source), types_with_payloads, false, | |
| 326 nudge_location)); | |
| 327 } | |
| 328 | |
| 329 void SyncerThread::ScheduleNudgeWithPayloads(const TimeDelta& delay, | |
| 330 NudgeSource source, const ModelTypePayloadMap& types_with_payloads, | |
| 331 const tracked_objects::Location& nudge_location) { | |
| 332 if (!thread_.IsRunning()) { | |
| 333 NOTREACHED(); | |
| 334 return; | |
| 335 } | |
| 336 | |
| 337 VLOG(2) << "SyncerThread(" << this << ")" << " Nudge scheduled with payloads"; | |
| 338 | |
| 339 thread_.message_loop()->PostTask(FROM_HERE, NewRunnableMethod( | |
| 340 this, &SyncerThread::ScheduleNudgeImpl, delay, | |
| 341 GetUpdatesFromNudgeSource(source), types_with_payloads, false, | |
| 342 nudge_location)); | |
| 343 } | |
| 344 | |
| 345 void SyncerThread::ScheduleClearUserDataImpl() { | |
| 346 DCHECK_EQ(MessageLoop::current(), thread_.message_loop()); | |
| 347 SyncSession* session = new SyncSession(session_context_.get(), this, | |
| 348 SyncSourceInfo(), ModelSafeRoutingInfo(), | |
| 349 std::vector<ModelSafeWorker*>()); | |
| 350 ScheduleSyncSessionJob(TimeDelta::FromSeconds(0), | |
| 351 SyncSessionJob::CLEAR_USER_DATA, session, FROM_HERE); | |
| 352 } | |
| 353 | |
| 354 void SyncerThread::ScheduleNudgeImpl(const TimeDelta& delay, | |
| 355 GetUpdatesCallerInfo::GetUpdatesSource source, | |
| 356 const ModelTypePayloadMap& types_with_payloads, | |
| 357 bool is_canary_job, const tracked_objects::Location& nudge_location) { | |
| 358 DCHECK_EQ(MessageLoop::current(), thread_.message_loop()); | |
| 359 | |
| 360 VLOG(2) << "SyncerThread(" << this << ")" << " Running Schedule nudge impl"; | |
| 361 // Note we currently nudge for all types regardless of the ones incurring | |
| 362 // the nudge. Doing different would throw off some syncer commands like | |
| 363 // CleanupDisabledTypes. We may want to change this in the future. | |
| 364 SyncSourceInfo info(source, types_with_payloads); | |
| 365 | |
| 366 SyncSession* session(CreateSyncSession(info)); | |
| 367 SyncSessionJob job(SyncSessionJob::NUDGE, TimeTicks::Now() + delay, | |
| 368 make_linked_ptr(session), is_canary_job, | |
| 369 nudge_location); | |
| 370 | |
| 371 session = NULL; | |
| 372 if (!ShouldRunJob(job)) | |
| 373 return; | |
| 374 | |
| 375 if (pending_nudge_.get()) { | |
| 376 if (IsBackingOff() && delay > TimeDelta::FromSeconds(1)) { | |
| 377 VLOG(2) << "SyncerThread(" << this << ")" << " Dropping the nudge because" | |
| 378 << "we are in backoff"; | |
| 379 return; | |
| 380 } | |
| 381 | |
| 382 VLOG(2) << "SyncerThread(" << this << ")" << " Coalescing pending nudge"; | |
| 383 pending_nudge_->session->Coalesce(*(job.session.get())); | |
| 384 | |
| 385 if (!IsBackingOff()) { | |
| 386 VLOG(2) << "SyncerThread(" << this << ")" << " Dropping a nudge because" | |
| 387 << " we are not in backoff and the job was coalesced"; | |
| 388 return; | |
| 389 } else { | |
| 390 VLOG(2) << "SyncerThread(" << this << ")" | |
| 391 << " Rescheduling pending nudge"; | |
| 392 SyncSession* s = pending_nudge_->session.get(); | |
| 393 job.session.reset(new SyncSession(s->context(), s->delegate(), | |
| 394 s->source(), s->routing_info(), s->workers())); | |
| 395 pending_nudge_.reset(); | |
| 396 } | |
| 397 } | |
| 398 | |
| 399 // TODO(lipalani) - pass the job itself to ScheduleSyncSessionJob. | |
| 400 ScheduleSyncSessionJob(delay, SyncSessionJob::NUDGE, job.session.release(), | |
| 401 nudge_location); | |
| 402 } | |
| 403 | |
| 404 // Helper to extract the routing info and workers corresponding to types in | |
| 405 // |types| from |registrar|. | |
| 406 void GetModelSafeParamsForTypes(const ModelTypeBitSet& types, | |
| 407 ModelSafeWorkerRegistrar* registrar, ModelSafeRoutingInfo* routes, | |
| 408 std::vector<ModelSafeWorker*>* workers) { | |
| 409 ModelSafeRoutingInfo r_tmp; | |
| 410 std::vector<ModelSafeWorker*> w_tmp; | |
| 411 registrar->GetModelSafeRoutingInfo(&r_tmp); | |
| 412 registrar->GetWorkers(&w_tmp); | |
| 413 | |
| 414 typedef std::vector<ModelSafeWorker*>::const_iterator iter; | |
| 415 for (size_t i = syncable::FIRST_REAL_MODEL_TYPE; i < types.size(); ++i) { | |
| 416 if (!types.test(i)) | |
| 417 continue; | |
| 418 syncable::ModelType t = syncable::ModelTypeFromInt(i); | |
| 419 DCHECK_EQ(1U, r_tmp.count(t)); | |
| 420 (*routes)[t] = r_tmp[t]; | |
| 421 iter it = std::find_if(w_tmp.begin(), w_tmp.end(), | |
| 422 ModelSafeWorkerGroupIs(r_tmp[t])); | |
| 423 if (it != w_tmp.end()) | |
| 424 workers->push_back(*it); | |
| 425 else | |
| 426 NOTREACHED(); | |
| 427 } | |
| 428 | |
| 429 iter it = std::find_if(w_tmp.begin(), w_tmp.end(), | |
| 430 ModelSafeWorkerGroupIs(GROUP_PASSIVE)); | |
| 431 if (it != w_tmp.end()) | |
| 432 workers->push_back(*it); | |
| 433 else | |
| 434 NOTREACHED(); | |
| 435 } | |
| 436 | |
| 437 void SyncerThread::ScheduleConfig(const ModelTypeBitSet& types) { | |
| 438 if (!thread_.IsRunning()) { | |
| 439 NOTREACHED(); | |
| 440 return; | |
| 441 } | |
| 442 | |
| 443 VLOG(2) << "SyncerThread(" << this << ")" << " Scheduling a config"; | |
| 444 ModelSafeRoutingInfo routes; | |
| 445 std::vector<ModelSafeWorker*> workers; | |
| 446 GetModelSafeParamsForTypes(types, session_context_->registrar(), | |
| 447 &routes, &workers); | |
| 448 | |
| 449 thread_.message_loop()->PostTask(FROM_HERE, NewRunnableMethod( | |
| 450 this, &SyncerThread::ScheduleConfigImpl, routes, workers, | |
| 451 GetUpdatesCallerInfo::FIRST_UPDATE)); | |
| 452 } | |
| 453 | |
| 454 void SyncerThread::ScheduleConfigImpl(const ModelSafeRoutingInfo& routing_info, | |
| 455 const std::vector<ModelSafeWorker*>& workers, | |
| 456 const sync_pb::GetUpdatesCallerInfo::GetUpdatesSource source) { | |
| 457 DCHECK_EQ(MessageLoop::current(), thread_.message_loop()); | |
| 458 | |
| 459 VLOG(2) << "SyncerThread(" << this << ")" << " ScheduleConfigImpl..."; | |
| 460 // TODO(tim): config-specific GetUpdatesCallerInfo value? | |
| 461 SyncSession* session = new SyncSession(session_context_.get(), this, | |
| 462 SyncSourceInfo(source, | |
| 463 syncable::ModelTypePayloadMapFromRoutingInfo( | |
| 464 routing_info, std::string())), | |
| 465 routing_info, workers); | |
| 466 ScheduleSyncSessionJob(TimeDelta::FromSeconds(0), | |
| 467 SyncSessionJob::CONFIGURATION, session, FROM_HERE); | |
| 468 } | |
| 469 | |
| 470 void SyncerThread::ScheduleSyncSessionJob(const base::TimeDelta& delay, | |
| 471 SyncSessionJob::SyncSessionJobPurpose purpose, | |
| 472 sessions::SyncSession* session, | |
| 473 const tracked_objects::Location& nudge_location) { | |
| 474 DCHECK_EQ(MessageLoop::current(), thread_.message_loop()); | |
| 475 | |
| 476 SyncSessionJob job(purpose, TimeTicks::Now() + delay, | |
| 477 make_linked_ptr(session), false, nudge_location); | |
| 478 if (purpose == SyncSessionJob::NUDGE) { | |
| 479 VLOG(2) << "SyncerThread(" << this << ")" << " Resetting pending_nudge in" | |
| 480 << " ScheduleSyncSessionJob"; | |
| 481 DCHECK(!pending_nudge_.get() || pending_nudge_->session.get() == session); | |
| 482 pending_nudge_.reset(new SyncSessionJob(job)); | |
| 483 } | |
| 484 VLOG(2) << "SyncerThread(" << this << ")" | |
| 485 << " Posting job to execute in DoSyncSessionJob. Job purpose " | |
| 486 << job.purpose; | |
| 487 MessageLoop::current()->PostDelayedTask(FROM_HERE, NewRunnableMethod(this, | |
| 488 &SyncerThread::DoSyncSessionJob, job), | |
| 489 delay.InMilliseconds()); | |
| 490 } | |
| 491 | |
| 492 void SyncerThread::SetSyncerStepsForPurpose( | |
| 493 SyncSessionJob::SyncSessionJobPurpose purpose, | |
| 494 SyncerStep* start, SyncerStep* end) { | |
| 495 *end = SYNCER_END; | |
| 496 switch (purpose) { | |
| 497 case SyncSessionJob::CONFIGURATION: | |
| 498 *start = DOWNLOAD_UPDATES; | |
| 499 *end = APPLY_UPDATES; | |
| 500 return; | |
| 501 case SyncSessionJob::CLEAR_USER_DATA: | |
| 502 *start = CLEAR_PRIVATE_DATA; | |
| 503 return; | |
| 504 case SyncSessionJob::NUDGE: | |
| 505 case SyncSessionJob::POLL: | |
| 506 *start = SYNCER_BEGIN; | |
| 507 return; | |
| 508 default: | |
| 509 NOTREACHED(); | |
| 510 } | |
| 511 } | |
| 512 | |
| 513 void SyncerThread::DoSyncSessionJob(const SyncSessionJob& job) { | |
| 514 DCHECK_EQ(MessageLoop::current(), thread_.message_loop()); | |
| 515 if (!ShouldRunJob(job)) | |
| 516 return; | |
| 517 | |
| 518 if (job.purpose == SyncSessionJob::NUDGE) { | |
| 519 DCHECK(pending_nudge_.get()); | |
| 520 if (pending_nudge_->session != job.session) | |
| 521 return; // Another nudge must have been scheduled in in the meantime. | |
| 522 pending_nudge_.reset(); | |
| 523 } | |
| 524 VLOG(2) << "SyncerThread(" << this << ")" << " DoSyncSessionJob. job purpose " | |
| 525 << job.purpose; | |
| 526 | |
| 527 SyncerStep begin(SYNCER_BEGIN); | |
| 528 SyncerStep end(SYNCER_END); | |
| 529 SetSyncerStepsForPurpose(job.purpose, &begin, &end); | |
| 530 | |
| 531 bool has_more_to_sync = true; | |
| 532 while (ShouldRunJob(job) && has_more_to_sync) { | |
| 533 VLOG(2) << "SyncerThread(" << this << ")" | |
| 534 << " SyncerThread: Calling SyncShare."; | |
| 535 // Synchronously perform the sync session from this thread. | |
| 536 syncer_->SyncShare(job.session.get(), begin, end); | |
| 537 has_more_to_sync = job.session->HasMoreToSync(); | |
| 538 if (has_more_to_sync) | |
| 539 job.session->ResetTransientState(); | |
| 540 } | |
| 541 VLOG(2) << "SyncerThread(" << this << ")" | |
| 542 << " SyncerThread: Done SyncShare looping."; | |
| 543 FinishSyncSessionJob(job); | |
| 544 } | |
| 545 | |
| 546 void SyncerThread::UpdateCarryoverSessionState(const SyncSessionJob& old_job) { | |
| 547 if (old_job.purpose == SyncSessionJob::CONFIGURATION) { | |
| 548 // Whatever types were part of a configuration task will have had updates | |
| 549 // downloaded. For that reason, we make sure they get recorded in the | |
| 550 // event that they get disabled at a later time. | |
| 551 ModelSafeRoutingInfo r(session_context_->previous_session_routing_info()); | |
| 552 if (!r.empty()) { | |
| 553 ModelSafeRoutingInfo temp_r; | |
| 554 ModelSafeRoutingInfo old_info(old_job.session->routing_info()); | |
| 555 std::set_union(r.begin(), r.end(), old_info.begin(), old_info.end(), | |
| 556 std::insert_iterator<ModelSafeRoutingInfo>(temp_r, temp_r.begin())); | |
| 557 session_context_->set_previous_session_routing_info(temp_r); | |
| 558 } | |
| 559 } else { | |
| 560 session_context_->set_previous_session_routing_info( | |
| 561 old_job.session->routing_info()); | |
| 562 } | |
| 563 } | |
| 564 | |
| 565 void SyncerThread::FinishSyncSessionJob(const SyncSessionJob& job) { | |
| 566 DCHECK_EQ(MessageLoop::current(), thread_.message_loop()); | |
| 567 // Update timing information for how often datatypes are triggering nudges. | |
| 568 base::TimeTicks now = TimeTicks::Now(); | |
| 569 if (!last_sync_session_end_time_.is_null()) { | |
| 570 ModelTypePayloadMap::const_iterator iter; | |
| 571 for (iter = job.session->source().types.begin(); | |
| 572 iter != job.session->source().types.end(); | |
| 573 ++iter) { | |
| 574 syncable::PostTimeToTypeHistogram(iter->first, | |
| 575 now - last_sync_session_end_time_); | |
| 576 } | |
| 577 } | |
| 578 last_sync_session_end_time_ = now; | |
| 579 UpdateCarryoverSessionState(job); | |
| 580 if (IsSyncingCurrentlySilenced()) { | |
| 581 VLOG(2) << "SyncerThread(" << this << ")" | |
| 582 << " We are currently throttled. So not scheduling the next sync."; | |
| 583 SaveJob(job); | |
| 584 return; // Nothing to do. | |
| 585 } | |
| 586 | |
| 587 VLOG(2) << "SyncerThread(" << this << ")" | |
| 588 << " Updating the next polling time after SyncMain"; | |
| 589 ScheduleNextSync(job); | |
| 590 } | |
| 591 | |
| 592 void SyncerThread::ScheduleNextSync(const SyncSessionJob& old_job) { | |
| 593 DCHECK_EQ(MessageLoop::current(), thread_.message_loop()); | |
| 594 DCHECK(!old_job.session->HasMoreToSync()); | |
| 595 // Note: |num_server_changes_remaining| > 0 here implies that we received a | |
| 596 // broken response while trying to download all updates, because the Syncer | |
| 597 // will loop until this value is exhausted. Also, if unsynced_handles exist | |
| 598 // but HasMoreToSync is false, this implies that the Syncer determined no | |
| 599 // forward progress was possible at this time (an error, such as an HTTP | |
| 600 // 500, is likely to have occurred during commit). | |
| 601 const bool work_to_do = | |
| 602 old_job.session->status_controller()->num_server_changes_remaining() > 0 | |
| 603 || old_job.session->status_controller()->unsynced_handles().size() > 0; | |
| 604 VLOG(2) << "SyncerThread(" << this << ")" << " syncer has work to do: " | |
| 605 << work_to_do; | |
| 606 | |
| 607 AdjustPolling(&old_job); | |
| 608 | |
| 609 // TODO(tim): Old impl had special code if notifications disabled. Needed? | |
| 610 if (!work_to_do) { | |
| 611 // Success implies backoff relief. Note that if this was a "one-off" job | |
| 612 // (i.e. purpose == SyncSessionJob::CLEAR_USER_DATA), if there was | |
| 613 // work_to_do before it ran this wont have changed, as jobs like this don't | |
| 614 // run a full sync cycle. So we don't need special code here. | |
| 615 wait_interval_.reset(); | |
| 616 VLOG(2) << "SyncerThread(" << this << ")" | |
| 617 << " Job suceeded so not scheduling more jobs"; | |
| 618 return; | |
| 619 } | |
| 620 | |
| 621 if (old_job.session->source().updates_source == | |
| 622 GetUpdatesCallerInfo::SYNC_CYCLE_CONTINUATION) { | |
| 623 VLOG(2) << "SyncerThread(" << this << ")" | |
| 624 << " Job failed with source continuation"; | |
| 625 // We don't seem to have made forward progress. Start or extend backoff. | |
| 626 HandleConsecutiveContinuationError(old_job); | |
| 627 } else if (IsBackingOff()) { | |
| 628 VLOG(2) << "SyncerThread(" << this << ")" | |
| 629 << " A nudge during backoff failed"; | |
| 630 // We weren't continuing but we're in backoff; must have been a nudge. | |
| 631 DCHECK_EQ(SyncSessionJob::NUDGE, old_job.purpose); | |
| 632 DCHECK(!wait_interval_->had_nudge); | |
| 633 wait_interval_->had_nudge = true; | |
| 634 wait_interval_->timer.Reset(); | |
| 635 } else { | |
| 636 VLOG(2) << "SyncerThread(" << this << ")" | |
| 637 << " Failed. Schedule a job with continuation as source"; | |
| 638 // We weren't continuing and we aren't in backoff. Schedule a normal | |
| 639 // continuation. | |
| 640 if (old_job.purpose == SyncSessionJob::CONFIGURATION) { | |
| 641 ScheduleConfigImpl(old_job.session->routing_info(), | |
| 642 old_job.session->workers(), | |
| 643 GetUpdatesFromNudgeSource(NUDGE_SOURCE_CONTINUATION)); | |
| 644 } else { | |
| 645 // For all other purposes(nudge and poll) we schedule a retry nudge. | |
| 646 ScheduleNudgeImpl(TimeDelta::FromSeconds(0), | |
| 647 GetUpdatesFromNudgeSource(NUDGE_SOURCE_CONTINUATION), | |
| 648 old_job.session->source().types, false, FROM_HERE); | |
| 649 } | |
| 650 } | |
| 651 } | |
| 652 | |
| 653 void SyncerThread::AdjustPolling(const SyncSessionJob* old_job) { | |
| 654 DCHECK(thread_.IsRunning()); | |
| 655 DCHECK_EQ(MessageLoop::current(), thread_.message_loop()); | |
| 656 | |
| 657 TimeDelta poll = (!session_context_->notifications_enabled()) ? | |
| 658 syncer_short_poll_interval_seconds_ : | |
| 659 syncer_long_poll_interval_seconds_; | |
| 660 bool rate_changed = !poll_timer_.IsRunning() || | |
| 661 poll != poll_timer_.GetCurrentDelay(); | |
| 662 | |
| 663 if (old_job && old_job->purpose != SyncSessionJob::POLL && !rate_changed) | |
| 664 poll_timer_.Reset(); | |
| 665 | |
| 666 if (!rate_changed) | |
| 667 return; | |
| 668 | |
| 669 // Adjust poll rate. | |
| 670 poll_timer_.Stop(); | |
| 671 poll_timer_.Start(poll, this, &SyncerThread::PollTimerCallback); | |
| 672 } | |
| 673 | |
| 674 void SyncerThread::HandleConsecutiveContinuationError( | |
| 675 const SyncSessionJob& old_job) { | |
| 676 DCHECK_EQ(MessageLoop::current(), thread_.message_loop()); | |
| 677 // This if conditions should be compiled out in retail builds. | |
| 678 if (IsBackingOff()) { | |
| 679 DCHECK(wait_interval_->timer.IsRunning() || old_job.is_canary_job); | |
| 680 } | |
| 681 SyncSession* old = old_job.session.get(); | |
| 682 SyncSession* s(new SyncSession(session_context_.get(), this, | |
| 683 old->source(), old->routing_info(), old->workers())); | |
| 684 TimeDelta length = delay_provider_->GetDelay( | |
| 685 IsBackingOff() ? wait_interval_->length : TimeDelta::FromSeconds(1)); | |
| 686 | |
| 687 VLOG(2) << "SyncerThread(" << this << ")" | |
| 688 << " In handle continuation error. Old job purpose is " | |
| 689 << old_job.purpose; | |
| 690 VLOG(2) << "SyncerThread(" << this << ")" | |
| 691 << " In Handle continuation error. The time delta(ms) is: " | |
| 692 << length.InMilliseconds(); | |
| 693 | |
| 694 // This will reset the had_nudge variable as well. | |
| 695 wait_interval_.reset(new WaitInterval(WaitInterval::EXPONENTIAL_BACKOFF, | |
| 696 length)); | |
| 697 if (old_job.purpose == SyncSessionJob::CONFIGURATION) { | |
| 698 SyncSessionJob job(old_job.purpose, TimeTicks::Now() + length, | |
| 699 make_linked_ptr(s), false, FROM_HERE); | |
| 700 wait_interval_->pending_configure_job.reset(new SyncSessionJob(job)); | |
| 701 } else { | |
| 702 // We are not in configuration mode. So wait_interval's pending job | |
| 703 // should be null. | |
| 704 DCHECK(wait_interval_->pending_configure_job.get() == NULL); | |
| 705 | |
| 706 // TODO(lipalani) - handle clear user data. | |
| 707 InitOrCoalescePendingJob(old_job); | |
| 708 } | |
| 709 wait_interval_->timer.Start(length, this, &SyncerThread::DoCanaryJob); | |
| 710 } | |
| 711 | |
| 712 // static | |
| 713 TimeDelta SyncerThread::GetRecommendedDelay(const TimeDelta& last_delay) { | |
| 714 if (last_delay.InSeconds() >= kMaxBackoffSeconds) | |
| 715 return TimeDelta::FromSeconds(kMaxBackoffSeconds); | |
| 716 | |
| 717 // This calculates approx. base_delay_seconds * 2 +/- base_delay_seconds / 2 | |
| 718 int64 backoff_s = | |
| 719 std::max(static_cast<int64>(1), | |
| 720 last_delay.InSeconds() * kBackoffRandomizationFactor); | |
| 721 | |
| 722 // Flip a coin to randomize backoff interval by +/- 50%. | |
| 723 int rand_sign = base::RandInt(0, 1) * 2 - 1; | |
| 724 | |
| 725 // Truncation is adequate for rounding here. | |
| 726 backoff_s = backoff_s + | |
| 727 (rand_sign * (last_delay.InSeconds() / kBackoffRandomizationFactor)); | |
| 728 | |
| 729 // Cap the backoff interval. | |
| 730 backoff_s = std::max(static_cast<int64>(1), | |
| 731 std::min(backoff_s, kMaxBackoffSeconds)); | |
| 732 | |
| 733 return TimeDelta::FromSeconds(backoff_s); | |
| 734 } | |
| 735 | |
| 736 void SyncerThread::Stop() { | |
| 737 VLOG(2) << "SyncerThread(" << this << ")" << " stop called"; | |
| 738 syncer_->RequestEarlyExit(); // Safe to call from any thread. | |
| 739 session_context_->connection_manager()->RemoveListener(this); | |
| 740 thread_.Stop(); | |
| 741 } | |
| 742 | |
| 743 void SyncerThread::DoCanaryJob() { | |
| 744 VLOG(2) << "SyncerThread(" << this << ")" << " Do canary job"; | |
| 745 DoPendingJobIfPossible(true); | |
| 746 } | |
| 747 | |
| 748 void SyncerThread::DoPendingJobIfPossible(bool is_canary_job) { | |
| 749 SyncSessionJob* job_to_execute = NULL; | |
| 750 if (mode_ == CONFIGURATION_MODE && wait_interval_.get() | |
| 751 && wait_interval_->pending_configure_job.get()) { | |
| 752 VLOG(2) << "SyncerThread(" << this << ")" << " Found pending configure job"; | |
| 753 job_to_execute = wait_interval_->pending_configure_job.get(); | |
| 754 } else if (mode_ == NORMAL_MODE && pending_nudge_.get()) { | |
| 755 VLOG(2) << "SyncerThread(" << this << ")" << " Found pending nudge job"; | |
| 756 // Pending jobs mostly have time from the past. Reset it so this job | |
| 757 // will get executed. | |
| 758 if (pending_nudge_->scheduled_start < TimeTicks::Now()) | |
| 759 pending_nudge_->scheduled_start = TimeTicks::Now(); | |
| 760 | |
| 761 scoped_ptr<SyncSession> session(CreateSyncSession( | |
| 762 pending_nudge_->session->source())); | |
| 763 | |
| 764 // Also the routing info might have been changed since we cached the | |
| 765 // pending nudge. Update it by coalescing to the latest. | |
| 766 pending_nudge_->session->Coalesce(*(session.get())); | |
| 767 // The pending nudge would be cleared in the DoSyncSessionJob function. | |
| 768 job_to_execute = pending_nudge_.get(); | |
| 769 } | |
| 770 | |
| 771 if (job_to_execute != NULL) { | |
| 772 VLOG(2) << "SyncerThread(" << this << ")" << " Executing pending job"; | |
| 773 SyncSessionJob copy = *job_to_execute; | |
| 774 copy.is_canary_job = is_canary_job; | |
| 775 DoSyncSessionJob(copy); | |
| 776 } | |
| 777 } | |
| 778 | |
| 779 SyncSession* SyncerThread::CreateSyncSession(const SyncSourceInfo& source) { | |
| 780 ModelSafeRoutingInfo routes; | |
| 781 std::vector<ModelSafeWorker*> workers; | |
| 782 session_context_->registrar()->GetModelSafeRoutingInfo(&routes); | |
| 783 session_context_->registrar()->GetWorkers(&workers); | |
| 784 SyncSourceInfo info(source); | |
| 785 | |
| 786 SyncSession* session(new SyncSession(session_context_.get(), this, info, | |
| 787 routes, workers)); | |
| 788 | |
| 789 return session; | |
| 790 } | |
| 791 | |
| 792 void SyncerThread::PollTimerCallback() { | |
| 793 DCHECK_EQ(MessageLoop::current(), thread_.message_loop()); | |
| 794 ModelSafeRoutingInfo r; | |
| 795 ModelTypePayloadMap types_with_payloads = | |
| 796 syncable::ModelTypePayloadMapFromRoutingInfo(r, std::string()); | |
| 797 SyncSourceInfo info(GetUpdatesCallerInfo::PERIODIC, types_with_payloads); | |
| 798 SyncSession* s = CreateSyncSession(info); | |
| 799 ScheduleSyncSessionJob(TimeDelta::FromSeconds(0), SyncSessionJob::POLL, s, | |
| 800 FROM_HERE); | |
| 801 } | |
| 802 | |
| 803 void SyncerThread::Unthrottle() { | |
| 804 DCHECK_EQ(WaitInterval::THROTTLED, wait_interval_->mode); | |
| 805 VLOG(2) << "SyncerThread(" << this << ")" << " Unthrottled.."; | |
| 806 DoCanaryJob(); | |
| 807 wait_interval_.reset(); | |
| 808 } | |
| 809 | |
| 810 void SyncerThread::Notify(SyncEngineEvent::EventCause cause) { | |
| 811 DCHECK_EQ(MessageLoop::current(), thread_.message_loop()); | |
| 812 session_context_->NotifyListeners(SyncEngineEvent(cause)); | |
| 813 } | |
| 814 | |
| 815 bool SyncerThread::IsBackingOff() const { | |
| 816 return wait_interval_.get() && wait_interval_->mode == | |
| 817 WaitInterval::EXPONENTIAL_BACKOFF; | |
| 818 } | |
| 819 | |
| 820 void SyncerThread::OnSilencedUntil(const base::TimeTicks& silenced_until) { | |
| 821 wait_interval_.reset(new WaitInterval(WaitInterval::THROTTLED, | |
| 822 silenced_until - TimeTicks::Now())); | |
| 823 wait_interval_->timer.Start(wait_interval_->length, this, | |
| 824 &SyncerThread::Unthrottle); | |
| 825 } | |
| 826 | |
| 827 bool SyncerThread::IsSyncingCurrentlySilenced() { | |
| 828 return wait_interval_.get() && wait_interval_->mode == | |
| 829 WaitInterval::THROTTLED; | |
| 830 } | |
| 831 | |
| 832 void SyncerThread::OnReceivedShortPollIntervalUpdate( | |
| 833 const base::TimeDelta& new_interval) { | |
| 834 DCHECK_EQ(MessageLoop::current(), thread_.message_loop()); | |
| 835 syncer_short_poll_interval_seconds_ = new_interval; | |
| 836 } | |
| 837 | |
| 838 void SyncerThread::OnReceivedLongPollIntervalUpdate( | |
| 839 const base::TimeDelta& new_interval) { | |
| 840 DCHECK_EQ(MessageLoop::current(), thread_.message_loop()); | |
| 841 syncer_long_poll_interval_seconds_ = new_interval; | |
| 842 } | |
| 843 | |
| 844 void SyncerThread::OnShouldStopSyncingPermanently() { | |
| 845 VLOG(2) << "SyncerThread(" << this << ")" | |
| 846 << " OnShouldStopSyncingPermanently"; | |
| 847 syncer_->RequestEarlyExit(); // Thread-safe. | |
| 848 Notify(SyncEngineEvent::STOP_SYNCING_PERMANENTLY); | |
| 849 } | |
| 850 | |
| 851 void SyncerThread::OnServerConnectionEvent( | |
| 852 const ServerConnectionEvent2& event) { | |
| 853 thread_.message_loop()->PostTask(FROM_HERE, NewRunnableMethod(this, | |
| 854 &SyncerThread::CheckServerConnectionManagerStatus, | |
| 855 event.connection_code)); | |
| 856 } | |
| 857 | |
| 858 void SyncerThread::set_notifications_enabled(bool notifications_enabled) { | |
| 859 session_context_->set_notifications_enabled(notifications_enabled); | |
| 860 } | |
| 861 | |
| 862 } // s3 | |
| 863 } // browser_sync | |
| OLD | NEW |
|---|
Chromium Code Reviews
Issue 6874018:
make new syncer thread the default. (Closed)
Base URL: svn://svn.chromium.org/chrome/trunk/src