stamp = inc.stamp;
pool_stat_t pg_sum_old = pg_sum;
+ hash_map<uint64_t, pool_stat_t> pg_pool_sum_old;
bool ratios_changed = false;
if (inc.full_ratio != full_ratio && inc.full_ratio != -1) {
const pg_t &update_pg(p->first);
const pg_stat_t &update_stat(p->second);
+ if (pg_pool_sum_old.count(update_pg.pool()) == 0)
+ pg_pool_sum_old[update_pg.pool()] = pg_pool_sum[update_pg.pool()];
+
hash_map<pg_t,pg_stat_t>::iterator t = pg_stat.find(update_pg);
if (t == pg_stat.end()) {
hash_map<pg_t,pg_stat_t>::value_type v(update_pg, update_stat);
++p) {
int osd = p->first;
const osd_stat_t &new_stats(p->second);
-
+
hash_map<int32_t,osd_stat_t>::iterator t = osd_stat.find(osd);
if (t == osd_stat.end()) {
hash_map<int32_t,osd_stat_t>::value_type v(osd, new_stats);
osd_epochs.insert(*(inc.get_osd_epochs().find(osd)));
stat_osd_add(new_stats);
-
+
// adjust [near]full status
register_nearfull_status(osd, new_stats);
}
pg_stat.erase(s);
}
}
-
+
for (set<int>::iterator p = inc.get_osd_stat_rm().begin();
p != inc.get_osd_stat_rm().end();
++p) {
stamp_delta -= pg_sum_deltas.front().second;
pg_sum_deltas.pop_front();
}
-
+
+ update_pool_deltas(cct, inc.stamp, pg_pool_sum_old);
+
if (inc.osdmap_epoch)
last_osdmap_epoch = inc.osdmap_epoch;
if (inc.pg_scan)
}
}
-void PGMap::update_delta(CephContext *cct, utime_t inc_stamp, pool_stat_t& pg_sum_old)
+
+/**
+ * update aggregated delta
+ *
+ * @param cct ceph context
+ * @param ts Timestamp for the stats being delta'ed
+ * @param old_pool_sum Previous stats sum
+ * @param last_ts Last timestamp for pool
+ * @param result_pool_sum Resulting stats
+ * @param result_ts_delta Resulting timestamp delta
+ * @param delta_avg_list List of last N computed deltas, used to average
+ */
+void PGMap::update_delta(CephContext *cct,
+ const utime_t ts,
+ const pool_stat_t& old_pool_sum,
+ utime_t *last_ts,
+ const pool_stat_t& current_pool_sum,
+ pool_stat_t *result_pool_delta,
+ utime_t *result_ts_delta,
+ list<pair<pool_stat_t,utime_t> > *delta_avg_list)
{
+ /* @p ts is the timestamp we want to associate with the data
+ * in @p old_pool_sum, and on which we will base ourselves to
+ * calculate the delta, stored in 'delta_t'.
+ */
utime_t delta_t;
- delta_t = inc_stamp;
- delta_t -= stamp;
- stamp = inc_stamp;
+ delta_t = ts; // start with the provided timestamp
+ delta_t -= *last_ts; // take the last timestamp we saw
+ *last_ts = ts; // @p ts becomes the last timestamp we saw
// calculate a delta, and average over the last 2 deltas.
- pool_stat_t d = pg_sum;
- d.stats.sub(pg_sum_old.stats);
- pg_sum_deltas.push_back(make_pair(d, delta_t));
- stamp_delta += delta_t;
+ /* start by taking a copy of our current @p result_pool_sum, and by
+ * taking out the stats from @p old_pool_sum. This generates a stats
+ * delta. Stash this stats delta in @p delta_avg_list, along with the
+ * timestamp delta for these results.
+ */
+ pool_stat_t d = current_pool_sum;
+ d.stats.sub(old_pool_sum.stats);
+ delta_avg_list->push_back(make_pair(d,delta_t));
+ *result_ts_delta += delta_t;
+
+ /* Aggregate current delta, and take out the last seen delta (if any) to
+ * average it out.
+ */
+ result_pool_delta->stats.add(d.stats);
+ size_t s = MAX(1, cct ? cct->_conf->mon_stat_smooth_intervals : 1);
+ if (delta_avg_list->size() > s) {
+ result_pool_delta->stats.sub(delta_avg_list->front().first.stats);
+ *result_ts_delta -= delta_avg_list->front().second;
+ delta_avg_list->pop_front();
+ }
+}
- pg_sum_delta.stats.add(d.stats);
- if (pg_sum_deltas.size() > (std::list< pair<pool_stat_t, utime_t> >::size_type)MAX(1, cct ? cct->_conf->mon_stat_smooth_intervals : 1)) {
- pg_sum_delta.stats.sub(pg_sum_deltas.front().first.stats);
- stamp_delta -= pg_sum_deltas.front().second;
- pg_sum_deltas.pop_front();
+/**
+ * update aggregated delta
+ *
+ * @param cct ceph context
+ * @param ts Timestamp
+ * @param pg_sum_old Old pg_sum
+ */
+void PGMap::update_global_delta(CephContext *cct,
+ const utime_t ts, const pool_stat_t& pg_sum_old)
+{
+ update_delta(cct, ts, pg_sum_old, &stamp, pg_sum, &pg_sum_delta,
+ &stamp_delta, &pg_sum_deltas);
+}
+
+/**
+ * Update a given pool's deltas
+ *
+ * @param cct Ceph Context
+ * @param ts Timestamp for the stats being delta'ed
+ * @param pool Pool's id
+ * @param old_pool_sum Previous stats sum
+ */
+void PGMap::update_one_pool_delta(CephContext *cct,
+ const utime_t ts,
+ const uint64_t pool,
+ const pool_stat_t& old_pool_sum)
+{
+ if (per_pool_sum_deltas.count(pool) == 0) {
+ assert(per_pool_sum_deltas_stamps.count(pool) == 0);
+ assert(per_pool_sum_delta.count(pool) == 0);
+ }
+
+ pair<pool_stat_t,utime_t>& sum_delta = per_pool_sum_delta[pool];
+
+ update_delta(cct, ts, old_pool_sum, &sum_delta.second, pg_pool_sum[pool],
+ &sum_delta.first, &per_pool_sum_deltas_stamps[pool],
+ &per_pool_sum_deltas[pool]);
+}
+
+/**
+ * Update pools' deltas
+ *
+ * @param cct CephContext
+ * @param ts Timestamp for the stats being delta'ed
+ * @param pg_pool_sum_old Map of pool stats for delta calcs.
+ */
+void PGMap::update_pool_deltas(CephContext *cct, const utime_t ts,
+ const hash_map<uint64_t,pool_stat_t>& pg_pool_sum_old)
+{
+ for (hash_map<uint64_t,pool_stat_t>::const_iterator it = pg_pool_sum_old.begin();
+ it != pg_pool_sum_old.end(); ++it) {
+ update_one_pool_delta(cct, ts, it->first, it->second);
}
}
utime_t stamp;
// recent deltas, and summation
+ /**
+ * keep track of last deltas for each pool, calculated using
+ * @p pg_pool_sum as baseline.
+ */
+ hash_map<uint64_t, list< pair<pool_stat_t, utime_t> > > per_pool_sum_deltas;
+ /**
+ * keep track of per-pool timestamp deltas, according to last update on
+ * each pool.
+ */
+ hash_map<uint64_t, utime_t> per_pool_sum_deltas_stamps;
+ /**
+ * keep track of sum deltas, per-pool, taking into account any previous
+ * deltas existing in @p per_pool_sum_deltas. The utime_t as second member
+ * of the pair is the timestamp refering to the last update (i.e., the first
+ * member of the pair) for a given pool.
+ */
+ hash_map<uint64_t, pair<pool_stat_t,utime_t> > per_pool_sum_delta;
+
list< pair<pool_stat_t, utime_t> > pg_sum_deltas;
pool_stat_t pg_sum_delta;
utime_t stamp_delta;
- void update_delta(CephContext *cct, utime_t inc_stamp, pool_stat_t& pg_sum_old);
+ void update_global_delta(CephContext *cct,
+ const utime_t ts, const pool_stat_t& pg_sum_old);
+ void update_pool_deltas(CephContext *cct,
+ const utime_t ts,
+ const hash_map<uint64_t, pool_stat_t>& pg_pool_sum_old);
void clear_delta();
+ private:
+ void update_delta(CephContext *cct,
+ const utime_t ts,
+ const pool_stat_t& old_pool_sum,
+ utime_t *last_ts,
+ const pool_stat_t& current_pool_sum,
+ pool_stat_t *result_pool_delta,
+ utime_t *result_ts_delta,
+ list<pair<pool_stat_t,utime_t> > *delta_avg_list);
+
+ void update_one_pool_delta(CephContext *cct,
+ const utime_t ts,
+ const uint64_t pool,
+ const pool_stat_t& old_pool_sum);
+ public:
+
set<pg_t> creating_pgs; // lru: front = new additions, back = recently pinged
map<int,set<pg_t> > creating_pgs_by_osd;
}
}
+ /* If we have deltas for pools, run through pgmap's 'per_pool_sum_delta' and
+ * clear any deltas that are old enough.
+ *
+ * Note that 'per_pool_sum_delta' keeps a pool id as key, and a pair containing
+ * the calc'ed stats delta and an absolute timestamp from when those stats were
+ * obtained -- the timestamp IS NOT a delta itself.
+ */
+ if (!pg_map.per_pool_sum_deltas.empty()) {
+ hash_map<uint64_t,pair<pool_stat_t,utime_t> >::iterator it;
+ for (it = pg_map.per_pool_sum_delta.begin();
+ it != pg_map.per_pool_sum_delta.end(); ) {
+ utime_t age = ceph_clock_now(g_ceph_context) - it->second.second;
+ if (age > 2*g_conf->mon_delta_reset_interval) {
+ dout(10) << " clearing pg_map delta for pool " << it->first
+ << " (" << age << " > " << g_conf->mon_delta_reset_interval
+ << " seconds old)" << dendl;
+ pg_map.per_pool_sum_deltas.erase(it->first);
+ pg_map.per_pool_sum_deltas_stamps.erase(it->first);
+ pg_map.per_pool_sum_delta.erase((it++)->first);
+ } else {
+ ++it;
+ }
+ }
+ }
+
dout(10) << pg_map << dendl;
}
}
pool_stat_t pg_sum_old = pg_map.pg_sum;
+ hash_map<uint64_t, pool_stat_t> pg_pool_sum_old;
// pgs
bufferlist::iterator p = dirty_pgs.begin();
dout(20) << " refreshing pg " << pgid << dendl;
bufferlist bl;
int r = mon->store->get(pgmap_pg_prefix, stringify(pgid), bl);
+
+ if (pg_pool_sum_old.count(pgid.pool()) == 0)
+ pg_pool_sum_old[pgid.pool()] = pg_map.pg_pool_sum[pgid.pool()];
+
if (r >= 0) {
pg_map.update_pg(pgid, bl);
} else {
}
}
- pg_map.update_delta(g_ceph_context, inc_stamp, pg_sum_old);
+ pg_map.update_global_delta(g_ceph_context, inc_stamp, pg_sum_old);
+ pg_map.update_pool_deltas(g_ceph_context, inc_stamp, pg_pool_sum_old);
// ok, we're now on the new version
pg_map.version = v;
projects / ceph.git / commitdiff