osd: index pg (slots) by map epoch within each shard

This will replace the epoch tracking in OSDService shortly.

Signed-off-by: Sage Weil <sage@redhat.com>

src/osd/OSD.cc

@@ -9252,6 +9252,9 @@ void OSDShard::_attach_pg(OSDShardPGSlot *slot, PG *pg)
   pg->osd_shard = this;
   pg->pg_slot = slot;
   ++osd->num_pgs;
+
+  slot->epoch = pg->get_osdmap_epoch();
+  pg_slots_by_epoch.insert(*slot);
 }
 
 void OSDShard::_detach_pg(OSDShardPGSlot *slot)
@@ -9261,6 +9264,32 @@ void OSDShard::_detach_pg(OSDShardPGSlot *slot)
   slot->pg->pg_slot = nullptr;
   slot->pg = nullptr;
   --osd->num_pgs;
+
+  pg_slots_by_epoch.erase(pg_slots_by_epoch.iterator_to(*slot));
+  slot->epoch = 0;
+}
+
+void OSDShard::update_pg_epoch(OSDShardPGSlot *slot, epoch_t e)
+{
+  Mutex::Locker l(sdata_op_ordering_lock);
+  dout(20) << "min was " << pg_slots_by_epoch.begin()->epoch
+	   << " on " << pg_slots_by_epoch.begin()->pg->pg_id << dendl;
+  pg_slots_by_epoch.erase(pg_slots_by_epoch.iterator_to(*slot));
+  dout(20) << slot->pg->pg_id << " " << slot->epoch << " -> " << e << dendl;
+  slot->epoch = e;
+  pg_slots_by_epoch.insert(*slot);
+  dout(20) << "min is now " << pg_slots_by_epoch.begin()->epoch
+	   << " on " << pg_slots_by_epoch.begin()->pg->pg_id << dendl;
+}
+
+epoch_t OSDShard::get_min_pg_epoch()
+{
+  Mutex::Locker l(sdata_op_ordering_lock);
+  auto p = pg_slots_by_epoch.begin();
+  if (p == pg_slots_by_epoch.end()) {
+    return 0;
+  }
+  return p->epoch;
 }
 
 void OSDShard::consume_map(

src/osd/OSD.h

@@ -1104,6 +1104,9 @@ struct OSDShardPGSlot {
 
   /// waiting for split child to materialize
   bool waiting_for_split = false;
+
+  epoch_t epoch = 0;
+  boost::intrusive::set_member_hook<> pg_epoch_item;
 };
 
 struct OSDShard {
@@ -1122,6 +1125,21 @@ struct OSDShard {
   /// pg lock.  stale slots are removed by consume_map.
   unordered_map<spg_t,unique_ptr<OSDShardPGSlot>> pg_slots;
 
+  struct pg_slot_compare_by_epoch {
+    bool operator()(const OSDShardPGSlot& l, const OSDShardPGSlot& r) const {
+      return l.epoch < r.epoch;
+    }
+  };
+
+  /// maintain an ordering of pg slots by pg epoch
+  boost::intrusive::multiset<
+    OSDShardPGSlot,
+    boost::intrusive::member_hook<
+      OSDShardPGSlot,
+      boost::intrusive::set_member_hook<>,
+      &OSDShardPGSlot::pg_epoch_item>,
+    boost::intrusive::compare<pg_slot_compare_by_epoch>> pg_slots_by_epoch;
+
   /// priority queue
   std::unique_ptr<OpQueue<OpQueueItem, uint64_t>> pqueue;
 
@@ -1143,6 +1161,9 @@ struct OSDShard {
   void _attach_pg(OSDShardPGSlot *slot, PG *pg);
   void _detach_pg(OSDShardPGSlot *slot);
 
+  void update_pg_epoch(OSDShardPGSlot *slot, epoch_t epoch);
+  epoch_t get_min_pg_epoch();
+
   /// push osdmap into shard
   void consume_map(
     OSDMapRef& osdmap,

src/osd/PG.cc

@@ -6399,6 +6399,7 @@ void PG::handle_advance_map(
 	   << " -- " << up_primary << "/" << acting_primary
 	   << dendl;
   update_osdmap_ref(osdmap);
+  osd_shard->update_pg_epoch(pg_slot, osdmap->get_epoch());
 
   pool.update(cct, osdmap);