PG: add split_into to populate child members

Signed-off-by: Samuel Just <sam.just@inktank.com>

4 files changed, 134 insertions, 0 deletions

diff --git a/src/osd/PG.cc b/src/osd/PG.cc
index 89ae7d06d79..dfac75c5ca5 100644
--- a/src/osd/PG.cc
+++ b/src/osd/PG.cc
@@ -1907,6 +1907,85 @@ void PG::finish_recovery_op(const hobject_t& soid, bool dequeue)
   osd->osd->finish_recovery_op(this, soid, dequeue);
 }
 
+void PG::IndexedLog::split_into(
+  pg_t child_pgid,
+  unsigned split_bits,
+  PG::IndexedLog *olog)
+{
+  list<pg_log_entry_t> oldlog;
+  oldlog.swap(log);
+
+  eversion_t old_tail;
+  olog->head = head;
+  olog->tail = tail;
+  unsigned mask = ~((~0)<<split_bits);
+  for (list<pg_log_entry_t>::iterator i = oldlog.begin();
+       i != oldlog.end();
+       ) {
+    if ((i->soid.hash & mask) == child_pgid.m_seed) {
+      olog->log.push_back(*i);
+      if (log.empty())
+	tail = i->version;
+    } else {
+      log.push_back(*i);
+      if (olog->empty())
+	olog->tail = i->version;
+    }
+    oldlog.erase(i++);
+  }
+
+  if (log.empty())
+    tail = head;
+  else
+    head = log.rbegin()->version;
+
+  if (olog->empty())
+    olog->tail = olog->head;
+  else
+    olog->head = olog->log.rbegin()->version;
+
+  olog->index();
+  index();
+}
+
+void PG::split_into(pg_t child_pgid, PG *child, unsigned split_bits)
+{
+  child->osdmap_ref = osdmap_ref;
+
+  child->pool = pool;
+
+  // Log
+  log.split_into(child_pgid, split_bits, &(child->log));
+  child->info.last_complete = info.last_complete;
+
+  info.last_update = log.head;
+  child->info.last_update = child->log.head;
+
+  info.log_tail = log.tail;
+  child->info.log_tail = child->log.tail;
+
+  if (info.last_complete < log.tail)
+    info.last_complete = log.tail;
+  if (child->info.last_complete < child->log.tail)
+    child->info.last_complete = child->log.tail;
+
+  // Missing
+  missing.split_into(child_pgid, split_bits, &(child->missing));
+
+  // Info
+  child->info.history = info.history;
+  child->info.purged_snaps = info.purged_snaps;
+  child->info.last_backfill = info.last_backfill;
+
+  child->info.stats = info.stats;
+  info.stats.stats_invalid = true;
+  child->info.stats.stats_invalid = true;
+
+  child->snap_trimq = snap_trimq;
+
+  // History
+  child->past_intervals = past_intervals;
+}
 
 void PG::defer_recovery()
 {
diff --git a/src/osd/PG.h b/src/osd/PG.h
index 536f04dda64..cc2e112f48d 100644
--- a/src/osd/PG.h
+++ b/src/osd/PG.h
@@ -180,6 +180,11 @@ public:
       index();
     }
 
+    void split_into(
+      pg_t child_pgid,
+      unsigned split_bits,
+      IndexedLog *olog);
+
     void zero() {
       unindex();
       pg_log_t::clear();
@@ -789,6 +794,8 @@ public:
   void start_recovery_op(const hobject_t& soid);
   void finish_recovery_op(const hobject_t& soid, bool dequeue=false);
 
+  void split_into(pg_t child_pgid, PG *child, unsigned split_bits);
+
   loff_t get_log_write_pos() {
     return 0;
   }
diff --git a/src/osd/osd_types.cc b/src/osd/osd_types.cc
index c65797b9988..c52cf20e28d 100644
--- a/src/osd/osd_types.cc
+++ b/src/osd/osd_types.cc
@@ -221,6 +221,27 @@ bool pg_t::is_split(unsigned old_pg_num, unsigned new_pg_num, set<pg_t> *childre
   return split;
 }
 
+unsigned pg_t::get_split_bits(unsigned pg_num) const {
+  assert(pg_num > 1);
+
+  // Find unique p such that pg_num \in [2^(p-1), 2^p)
+  unsigned p = pg_pool_t::calc_bits_of(pg_num);
+
+  if ((m_seed % (1<<(p-1))) < (pg_num % (1<<(p-1))))
+    return p;
+  else
+    return p - 1;
+}
+
+pg_t pg_t::get_parent() const
+{
+  unsigned bits = pg_pool_t::calc_bits_of(m_seed);
+  assert(bits);
+  pg_t retval = *this;
+  retval.m_seed &= ~((~0)<<(bits - 1));
+  return retval;
+}
+
 void pg_t::dump(Formatter *f) const
 {
   f->dump_unsigned("pool", m_pool);
@@ -2023,6 +2044,24 @@ void pg_missing_t::got(const std::map<hobject_t, pg_missing_t::item>::iterator &
   missing.erase(m);
 }
 
+void pg_missing_t::split_into(
+  pg_t child_pgid,
+  unsigned split_bits,
+  pg_missing_t *omissing)
+{
+  unsigned mask = ~((~0)<<split_bits);
+  for (map<hobject_t, item>::iterator i = missing.begin();
+       i != missing.end();
+       ) {
+    if ((i->first.hash & mask) == child_pgid.m_seed) {
+      omissing->add(i->first, i->second.need, i->second.have);
+      rm(i++);
+    } else {
+      ++i;
+    }
+  }
+}
+
 // -- pg_create_t --
 
 void pg_create_t::encode(bufferlist &bl) const
diff --git a/src/osd/osd_types.h b/src/osd/osd_types.h
index 0cc961965d7..a1aa20226a6 100644
--- a/src/osd/osd_types.h
+++ b/src/osd/osd_types.h
@@ -235,11 +235,19 @@ struct pg_t {
     m_preferred = osd;
   }
 
+  pg_t get_parent() const;
+
   int print(char *o, int maxlen) const;
   bool parse(const char *s);
 
   bool is_split(unsigned old_pg_num, unsigned new_pg_num, set<pg_t> *pchildren) const;
 
+  /**
+   * Returns b such that for all object o:
+   *   ~((~0)<<b) & o.hash) == 0 iff o is in the pg for *this
+   */
+  unsigned get_split_bits(unsigned pg_num) const;
+
   void encode(bufferlist& bl) const {
     __u8 v = 1;
     ::encode(v, bl);
@@ -1450,6 +1458,7 @@ struct pg_missing_t {
   void rm(const std::map<hobject_t, pg_missing_t::item>::iterator &m);
   void got(const hobject_t& oid, eversion_t v);
   void got(const std::map<hobject_t, pg_missing_t::item>::iterator &m);
+  void split_into(pg_t child_pgid, unsigned split_bits, pg_missing_t *omissing);
 
   void clear() {
     missing.clear();