sync/notifier/ack_tracker.cc

   1 // Copyright (c) 2012 The Chromium Authors. All rights reserved.
   2 // Use of this source code is governed by a BSD-style license that can be
   3 // found in the LICENSE file.
   4
   5 #include "sync/notifier/ack_tracker.h"
   6
   7 #include <algorithm>
   8 #include <iterator>
   9 #include <utility>
  10
  11 #include "base/callback.h"
  12 #include "base/stl_util.h"
  13 #include "base/time/tick_clock.h"
  14 #include "google/cacheinvalidation/include/types.h"
  15
  16 namespace syncer {
  17
  18 namespace {
  19
  20 // All times are in milliseconds.
  21 const net::BackoffEntry::Policy kDefaultBackoffPolicy = {
  22   // Number of initial errors (in sequence) to ignore before applying
  23   // exponential back-off rules.
  24   // Note this value is set to 1 to work in conjunction with a hack in
  25   // AckTracker::Track.
  26   1,
  27
  28   // Initial delay.  The interpretation of this value depends on
  29   // always_use_initial_delay.  It's either how long we wait between
  30   // requests before backoff starts, or how much we delay the first request
  31   // after backoff starts.
  32   60 * 1000,
  33
  34   // Factor by which the waiting time will be multiplied.
  35   2,
  36
  37   // Fuzzing percentage. ex: 10% will spread requests randomly
  38   // between 90%-100% of the calculated time.
  39   0,
  40
  41   // Maximum amount of time we are willing to delay our request, -1
  42   // for no maximum.
  43   60 * 10 * 1000,
  44
  45   // Time to keep an entry from being discarded even when it
  46   // has no significant state, -1 to never discard.
  47   -1,
  48
  49   // If true, we always use a delay of initial_delay_ms, even before
  50   // we've seen num_errors_to_ignore errors.  Otherwise, initial_delay_ms
  51   // is the first delay once we start exponential backoff.
  52   //
  53   // So if we're ignoring 1 error, we'll see (N, N, Nm, Nm^2, ...) if true,
  54   // and (0, 0, N, Nm, ...) when false, where N is initial_backoff_ms and
  55   // m is multiply_factor, assuming we've already seen one success.
  56   true,
  57 };
  58
  59 scoped_ptr<net::BackoffEntry> CreateDefaultBackoffEntry(
  60     const net::BackoffEntry::Policy* const policy) {
  61   return scoped_ptr<net::BackoffEntry>(new net::BackoffEntry(policy));
  62 }
  63
  64 }  // namespace
  65
  66 AckTracker::Delegate::~Delegate() {
  67 }
  68
  69 AckTracker::Entry::Entry(scoped_ptr<net::BackoffEntry> backoff,
  70                          const ObjectIdSet& ids)
  71     : backoff(backoff.Pass()), ids(ids) {
  72 }
  73
  74 AckTracker::Entry::~Entry() {
  75 }
  76
  77 AckTracker::AckTracker(base::TickClock* tick_clock, Delegate* delegate)
  78     : create_backoff_entry_callback_(base::Bind(&CreateDefaultBackoffEntry)),
  79       tick_clock_(tick_clock),
  80       delegate_(delegate) {
  81   DCHECK(tick_clock_);
  82   DCHECK(delegate_);
  83 }
  84
  85 AckTracker::~AckTracker() {
  86   DCHECK(thread_checker_.CalledOnValidThread());
  87
  88   Clear();
  89 }
  90
  91 void AckTracker::Clear() {
  92   DCHECK(thread_checker_.CalledOnValidThread());
  93
  94   timer_.Stop();
  95   STLDeleteValues(&queue_);
  96 }
  97
  98 void AckTracker::Track(const ObjectIdSet& ids) {
  99   DCHECK(thread_checker_.CalledOnValidThread());
 100   DCHECK(!ids.empty());
 101
 102   scoped_ptr<Entry> entry(new Entry(
 103       create_backoff_entry_callback_.Run(&kDefaultBackoffPolicy), ids));
 104   // This is a small hack. When net::BackoffRequest is first created,
 105   // GetReleaseTime() always returns the default base::TimeTicks value: 0.
 106   // In order to work around that, we mark it as failed right away.
 107   entry->backoff->InformOfRequest(false /* succeeded */);
 108   const base::TimeTicks release_time = entry->backoff->GetReleaseTime();
 109   queue_.insert(std::make_pair(release_time, entry.release()));
 110   NudgeTimer();
 111 }
 112
 113 void AckTracker::Ack(const ObjectIdSet& ids) {
 114   DCHECK(thread_checker_.CalledOnValidThread());
 115
 116   // We could be clever and maintain a mapping of object IDs to their position
 117   // in the multimap, but that makes things a lot more complicated.
 118   for (std::multimap<base::TimeTicks, Entry*>::iterator it = queue_.begin();
 119        it != queue_.end(); ) {
 120     ObjectIdSet remaining_ids;
 121     std::set_difference(it->second->ids.begin(), it->second->ids.end(),
 122                         ids.begin(), ids.end(),
 123                         std::inserter(remaining_ids, remaining_ids.begin()),
 124                         ids.value_comp());
 125     it->second->ids.swap(remaining_ids);
 126     if (it->second->ids.empty()) {
 127       std::multimap<base::TimeTicks, Entry*>::iterator erase_it = it;
 128       ++it;
 129       delete erase_it->second;
 130       queue_.erase(erase_it);
 131     } else {
 132       ++it;
 133     }
 134   }
 135   NudgeTimer();
 136 }
 137
 138 void AckTracker::NudgeTimer() {
 139   DCHECK(thread_checker_.CalledOnValidThread());
 140
 141   if (queue_.empty()) {
 142     return;
 143   }
 144
 145   const base::TimeTicks now = tick_clock_->NowTicks();
 146   // There are two cases when the timer needs to be started:
 147   // 1. |desired_run_time_| is in the past. By definition, the timer has already
 148   //    fired at this point. Since the queue is non-empty, we need to set the
 149   //    timer to fire again.
 150   // 2. The timer is already running but we need it to fire sooner if the first
 151   //    entry's timeout occurs before |desired_run_time_|.
 152   if (desired_run_time_ <= now || queue_.begin()->first < desired_run_time_) {
 153     base::TimeDelta delay = queue_.begin()->first - now;
 154     if (delay < base::TimeDelta()) {
 155       delay = base::TimeDelta();
 156     }
 157     timer_.Start(FROM_HERE, delay, this, &AckTracker::OnTimeout);
 158     desired_run_time_ = queue_.begin()->first;
 159   }
 160 }
 161
 162 void AckTracker::OnTimeout() {
 163   DCHECK(thread_checker_.CalledOnValidThread());
 164
 165   OnTimeoutAt(tick_clock_->NowTicks());
 166 }
 167
 168 void AckTracker::OnTimeoutAt(base::TimeTicks now) {
 169   DCHECK(thread_checker_.CalledOnValidThread());
 170
 171   if (queue_.empty())
 172     return;
 173
 174   ObjectIdSet expired_ids;
 175   std::multimap<base::TimeTicks, Entry*>::iterator end =
 176       queue_.upper_bound(now);
 177   std::vector<Entry*> expired_entries;
 178   for (std::multimap<base::TimeTicks, Entry*>::iterator it = queue_.begin();
 179        it != end; ++it) {
 180     expired_ids.insert(it->second->ids.begin(), it->second->ids.end());
 181     it->second->backoff->InformOfRequest(false /* succeeded */);
 182     expired_entries.push_back(it->second);
 183   }
 184   queue_.erase(queue_.begin(), end);
 185   for (std::vector<Entry*>::const_iterator it = expired_entries.begin();
 186        it != expired_entries.end(); ++it) {
 187     queue_.insert(std::make_pair((*it)->backoff->GetReleaseTime(), *it));
 188   }
 189   delegate_->OnTimeout(expired_ids);
 190   NudgeTimer();
 191 }
 192
 193 // Testing helpers.
 194 void AckTracker::SetCreateBackoffEntryCallbackForTest(
 195     const CreateBackoffEntryCallback& create_backoff_entry_callback) {
 196   DCHECK(thread_checker_.CalledOnValidThread());
 197
 198   create_backoff_entry_callback_ = create_backoff_entry_callback;
 199 }
 200
 201 bool AckTracker::TriggerTimeoutAtForTest(base::TimeTicks now) {
 202   DCHECK(thread_checker_.CalledOnValidThread());
 203
 204   bool no_timeouts_before_now = (queue_.lower_bound(now) == queue_.begin());
 205   OnTimeoutAt(now);
 206   return no_timeouts_before_now;
 207 }
 208
 209 bool AckTracker::IsQueueEmptyForTest() const {
 210   DCHECK(thread_checker_.CalledOnValidThread());
 211
 212   return queue_.empty();
 213 }
 214
 215 const base::Timer& AckTracker::GetTimerForTest() const {
 216   DCHECK(thread_checker_.CalledOnValidThread());
 217
 218   return timer_;
 219 }
 220
 221 }  // namespace syncer