osd/: move proc_* and log handling methods

Signed-off-by: Samuel Just <sjust@redhat.com>

diff --git a/src/osd/PG.cc b/src/osd/PG.cc
index dcb6331470f043ff48acb249da8aad32d03e373b..02de6bcbd01bafef4aef08a7b045bd4f4ffc564e 100644 (file)
--- a/src/osd/PG.cc
+++ b/src/osd/PG.cc
@@ -309,64 +309,6 @@ PerfCounters &PG::get_peering_perf() {
   
 /********* PG **********/
 
-void PG::proc_master_log(
-  ObjectStore::Transaction& t, pg_info_t &oinfo,
-  pg_log_t &olog, pg_missing_t& omissing, pg_shard_t from)
-{
-  dout(10) << "proc_master_log for osd." << from << ": "
-          << olog << " " << omissing << dendl;
-  ceph_assert(!is_peered() && is_primary());
-
-  // merge log into our own log to build master log.  no need to
-  // make any adjustments to their missing map; we are taking their
-  // log to be authoritative (i.e., their entries are by definitely
-  // non-divergent).
-  merge_log(t, oinfo, olog, from);
-  peer_info[from] = oinfo;
-  dout(10) << " peer osd." << from << " now " << oinfo << " " << omissing << dendl;
-  might_have_unfound.insert(from);
-
-  // See doc/dev/osd_internals/last_epoch_started
-  if (oinfo.last_epoch_started > info.last_epoch_started) {
-    info.last_epoch_started = oinfo.last_epoch_started;
-    dirty_info = true;
-  }
-  if (oinfo.last_interval_started > info.last_interval_started) {
-    info.last_interval_started = oinfo.last_interval_started;
-    dirty_info = true;
-  }
-  update_history(oinfo.history);
-  ceph_assert(cct->_conf->osd_find_best_info_ignore_history_les ||
-        info.last_epoch_started >= info.history.last_epoch_started);
-
-  peer_missing[from].claim(omissing);
-}
-    
-void PG::proc_replica_log(
-  pg_info_t &oinfo,
-  const pg_log_t &olog,
-  pg_missing_t& omissing,
-  pg_shard_t from)
-{
-  dout(10) << "proc_replica_log for osd." << from << ": "
-          << oinfo << " " << olog << " " << omissing << dendl;
-
-  pg_log.proc_replica_log(oinfo, olog, omissing, from);
-
-  peer_info[from] = oinfo;
-  dout(10) << " peer osd." << from << " now " << oinfo << " " << omissing << dendl;
-  might_have_unfound.insert(from);
-
-  for (map<hobject_t, pg_missing_item>::const_iterator i =
-        omissing.get_items().begin();
-       i != omissing.get_items().end();
-       ++i) {
-    dout(20) << " after missing " << i->first << " need " << i->second.need
-            << " have " << i->second.have << dendl;
-  }
-  peer_missing[from].claim(omissing);
-}
-
 void PG::remove_snap_mapped_object(
   ObjectStore::Transaction &t, const hobject_t &soid)
 {
@@ -409,21 +351,6 @@ void PG::update_object_snap_mapping(
     &_t);
 }
 
-void PG::merge_log(
-  ObjectStore::Transaction& t, pg_info_t &oinfo, pg_log_t &olog, pg_shard_t from)
-{
-  PGLogEntryHandler rollbacker{this, &t};
-  pg_log.merge_log(
-    oinfo, olog, from, info, &rollbacker, dirty_info, dirty_big_info);
-}
-
-void PG::rewind_divergent_log(ObjectStore::Transaction& t, eversion_t newhead)
-{
-  PGLogEntryHandler rollbacker{this, &t};
-  pg_log.rewind_divergent_log(
-    newhead, info, &rollbacker, dirty_info, dirty_big_info);
-}
-
 /******* PG ***********/
 void PG::clear_primary_state()
 {
@@ -4401,27 +4328,6 @@ bool PG::try_flush_or_schedule_async()
   }
 }
 
-void PG::proc_primary_info(ObjectStore::Transaction &t, const pg_info_t &oinfo)
-{
-  ceph_assert(!is_primary());
-
-  update_history(oinfo.history);
-  if (!info.stats.stats_invalid && info.stats.stats.sum.num_scrub_errors) {
-    info.stats.stats.sum.num_scrub_errors = 0;
-    info.stats.stats.sum.num_shallow_scrub_errors = 0;
-    info.stats.stats.sum.num_deep_scrub_errors = 0;
-    dirty_info = true;
-  }
-
-  if (!(info.purged_snaps == oinfo.purged_snaps)) {
-    dout(10) << __func__ << " updating purged_snaps to " << oinfo.purged_snaps
-            << dendl;
-    info.purged_snaps = oinfo.purged_snaps;
-    dirty_info = true;
-    dirty_big_info = true;
-  }
-}
-
 ostream& operator<<(ostream& out, const PG& pg)
 {
   out << "pg[" << pg.info

diff --git a/src/osd/PG.h b/src/osd/PG.h
index 2c7a0f986cc7b345edc257ed7593b6460930a2a8..f1fe3f4b3d6b7faf1874b2053f936b64042de1d4 100644 (file)
--- a/src/osd/PG.h
+++ b/src/osd/PG.h
@@ -480,6 +480,11 @@ public:
     return recovery_state.is_forced_recovery_or_backfill();
   }
 
+  PGLog::LogEntryHandlerRef get_log_handler(
+    ObjectStore::Transaction *t) override {
+    return std::make_unique<PG::PGLogEntryHandler>(this, t);
+  }
+
   void queue_peering_event(PGPeeringEventRef evt);
   void do_peering_event(PGPeeringEventRef evt, PeeringCtx *rcx);
   void queue_null(epoch_t msg_epoch, epoch_t query_epoch);
@@ -1018,10 +1023,6 @@ protected:
 
   virtual void calc_trim_to_aggressive() = 0;
 
-  void proc_replica_log(pg_info_t &oinfo, const pg_log_t &olog,
-                       pg_missing_t& omissing, pg_shard_t from);
-  void proc_master_log(ObjectStore::Transaction& t, pg_info_t &oinfo, pg_log_t &olog,
-                      pg_missing_t& omissing, pg_shard_t from);
   struct PGLogEntryHandler : public PGLog::LogEntryHandler {
     PG *pg;
     ObjectStore::Transaction *t;
@@ -1053,12 +1054,6 @@ protected:
     ObjectStore::Transaction *t, const hobject_t &soid);
   void remove_snap_mapped_object(
     ObjectStore::Transaction& t, const hobject_t& soid);
-  void merge_log(
-    ObjectStore::Transaction& t, pg_info_t &oinfo,
-    pg_log_t &olog, pg_shard_t from);
-  void rewind_divergent_log(ObjectStore::Transaction& t, eversion_t newhead);
-
-  void proc_primary_info(ObjectStore::Transaction &t, const pg_info_t &info);
 
   bool have_unfound() const { 
     return recovery_state.have_unfound();

diff --git a/src/osd/PGLog.h b/src/osd/PGLog.h
index f02b4c4c23d199dcdd02ac1049ac290762d9d34c..7ac66eda870f8c36b4043375342e7e9ea19806c8 100644 (file)
--- a/src/osd/PGLog.h
+++ b/src/osd/PGLog.h
@@ -59,6 +59,7 @@ struct PGLog : DoutPrefixProvider {
       version_t v) = 0;
     virtual ~LogEntryHandler() {}
   };
+  using LogEntryHandlerRef = unique_ptr<LogEntryHandler>;
 
 public:
   /**

diff --git a/src/osd/PeeringState.cc b/src/osd/PeeringState.cc
index 0166912c77e8e1749b81cc74de32e774824f1e78..8009032553d3bec5e206d6cc1f12bb7f8bdcb989 100644 (file)
--- a/src/osd/PeeringState.cc
+++ b/src/osd/PeeringState.cc
@@ -2329,8 +2329,8 @@ void PeeringState::activate(
     pl->on_activate(std::move(to_trim));
   }
   if (acting.size() >= pool.info.min_size) {
-    PG::PGLogEntryHandler handler{pg, &t};
-    pg_log.roll_forward(&handler);
+    PGLog::LogEntryHandlerRef rollbacker{pl->get_log_handler(&t)};
+    pg_log.roll_forward(rollbacker.get());
   }
 }
 
@@ -2363,6 +2363,109 @@ void PeeringState::share_pg_info()
   }
 }
 
+void PeeringState::merge_log(
+  ObjectStore::Transaction& t, pg_info_t &oinfo, pg_log_t &olog,
+  pg_shard_t from)
+{
+  PGLog::LogEntryHandlerRef rollbacker{pl->get_log_handler(&t)};
+  pg_log.merge_log(
+    oinfo, olog, from, info, rollbacker.get(), dirty_info, dirty_big_info);
+}
+
+void PeeringState::rewind_divergent_log(
+  ObjectStore::Transaction& t, eversion_t newhead)
+{
+  PGLog::LogEntryHandlerRef rollbacker{pl->get_log_handler(&t)};
+  pg_log.rewind_divergent_log(
+    newhead, info, rollbacker.get(), dirty_info, dirty_big_info);
+}
+
+
+void PeeringState::proc_primary_info(
+  ObjectStore::Transaction &t, const pg_info_t &oinfo)
+{
+  ceph_assert(!is_primary());
+
+  update_history(oinfo.history);
+  if (!info.stats.stats_invalid && info.stats.stats.sum.num_scrub_errors) {
+    info.stats.stats.sum.num_scrub_errors = 0;
+    info.stats.stats.sum.num_shallow_scrub_errors = 0;
+    info.stats.stats.sum.num_deep_scrub_errors = 0;
+    dirty_info = true;
+  }
+
+  if (!(info.purged_snaps == oinfo.purged_snaps)) {
+    psdout(10) << __func__ << " updating purged_snaps to "
+              << oinfo.purged_snaps
+              << dendl;
+    info.purged_snaps = oinfo.purged_snaps;
+    dirty_info = true;
+    dirty_big_info = true;
+  }
+}
+
+void PeeringState::proc_master_log(
+  ObjectStore::Transaction& t, pg_info_t &oinfo,
+  pg_log_t &olog, pg_missing_t& omissing, pg_shard_t from)
+{
+  psdout(10) << "proc_master_log for osd." << from << ": "
+            << olog << " " << omissing << dendl;
+  ceph_assert(!is_peered() && is_primary());
+
+  // merge log into our own log to build master log.  no need to
+  // make any adjustments to their missing map; we are taking their
+  // log to be authoritative (i.e., their entries are by definitely
+  // non-divergent).
+  merge_log(t, oinfo, olog, from);
+  peer_info[from] = oinfo;
+  psdout(10) << " peer osd." << from << " now " << oinfo
+            << " " << omissing << dendl;
+  might_have_unfound.insert(from);
+
+  // See doc/dev/osd_internals/last_epoch_started
+  if (oinfo.last_epoch_started > info.last_epoch_started) {
+    info.last_epoch_started = oinfo.last_epoch_started;
+    dirty_info = true;
+  }
+  if (oinfo.last_interval_started > info.last_interval_started) {
+    info.last_interval_started = oinfo.last_interval_started;
+    dirty_info = true;
+  }
+  update_history(oinfo.history);
+  ceph_assert(cct->_conf->osd_find_best_info_ignore_history_les ||
+        info.last_epoch_started >= info.history.last_epoch_started);
+
+  peer_missing[from].claim(omissing);
+}
+
+void PeeringState::proc_replica_log(
+  pg_info_t &oinfo,
+  const pg_log_t &olog,
+  pg_missing_t& omissing,
+  pg_shard_t from)
+{
+  psdout(10) << "proc_replica_log for osd." << from << ": "
+            << oinfo << " " << olog << " " << omissing << dendl;
+
+  pg_log.proc_replica_log(oinfo, olog, omissing, from);
+
+  peer_info[from] = oinfo;
+  psdout(10) << " peer osd." << from << " now "
+            << oinfo << " " << omissing << dendl;
+  might_have_unfound.insert(from);
+
+  for (map<hobject_t, pg_missing_item>::const_iterator i =
+        omissing.get_items().begin();
+       i != omissing.get_items().end();
+       ++i) {
+    psdout(20) << " after missing " << i->first
+              << " need " << i->second.need
+              << " have " << i->second.have << dendl;
+  }
+  peer_missing[from].claim(omissing);
+}
+
+
 /*------------ Peering State Machine----------------*/
 #undef dout_prefix
 #define dout_prefix (context< PeeringMachine >().dpp->gen_prefix(*_dout) \
@@ -3853,7 +3956,7 @@ boost::statechart::result PeeringState::Active::react(const MLogRec& logevt)
   DECLARE_LOCALS
   psdout(10) << "searching osd." << logevt.from
                     << " log for unfound items" << dendl;
-  pg->proc_replica_log(
+  ps->proc_replica_log(
     logevt.msg->info, logevt.msg->log, logevt.msg->missing, logevt.from);
   bool got_missing = ps->search_for_missing(
     ps->peer_info[logevt.from],
@@ -4081,7 +4184,7 @@ boost::statechart::result PeeringState::ReplicaActive::react(
 boost::statechart::result PeeringState::ReplicaActive::react(const MInfoRec& infoevt)
 {
   DECLARE_LOCALS
-  pg->proc_primary_info(*context<PeeringMachine>().get_cur_transaction(),
+  ps->proc_primary_info(*context<PeeringMachine>().get_cur_transaction(),
                        infoevt.info);
   return discard_event();
 }
@@ -4091,7 +4194,7 @@ boost::statechart::result PeeringState::ReplicaActive::react(const MLogRec& loge
   DECLARE_LOCALS
   psdout(10) << "received log from " << logevt.from << dendl;
   ObjectStore::Transaction* t = context<PeeringMachine>().get_cur_transaction();
-  pg->merge_log(*t, logevt.msg->info, logevt.msg->log, logevt.from);
+  ps->merge_log(*t, logevt.msg->info, logevt.msg->log, logevt.from);
   ceph_assert(ps->pg_log.get_head() == ps->info.last_update);
 
   return discard_event();
@@ -4185,12 +4288,12 @@ boost::statechart::result PeeringState::Stray::react(const MLogRec& logevt)
     ps->dirty_info = true;
     ps->dirty_big_info = true;  // maybe.
 
-    PG::PGLogEntryHandler rollbacker{pg, t};
-    ps->pg_log.reset_backfill_claim_log(msg->log, &rollbacker);
+    PGLog::LogEntryHandlerRef rollbacker{pl->get_log_handler(t)};
+    ps->pg_log.reset_backfill_claim_log(msg->log, rollbacker.get());
 
     ps->pg_log.reset_backfill();
   } else {
-    pg->merge_log(*t, msg->info, msg->log, logevt.from);
+    ps->merge_log(*t, msg->info, msg->log, logevt.from);
   }
 
   ceph_assert(ps->pg_log.get_head() == ps->info.last_update);
@@ -4207,7 +4310,7 @@ boost::statechart::result PeeringState::Stray::react(const MInfoRec& infoevt)
   if (ps->info.last_update > infoevt.info.last_update) {
     // rewind divergent log entries
     ObjectStore::Transaction* t = context<PeeringMachine>().get_cur_transaction();
-    pg->rewind_divergent_log(*t, infoevt.info.last_update);
+    ps->rewind_divergent_log(*t, infoevt.info.last_update);
     ps->info.stats = infoevt.info.stats;
     ps->info.hit_set = infoevt.info.hit_set;
   }
@@ -4592,7 +4695,7 @@ boost::statechart::result PeeringState::GetLog::react(const GotLog&)
   psdout(10) << "leaving GetLog" << dendl;
   if (msg) {
     psdout(10) << "processing master log" << dendl;
-    pg->proc_master_log(*context<PeeringMachine>().get_cur_transaction(),
+    ps->proc_master_log(*context<PeeringMachine>().get_cur_transaction(),
                        msg->info, msg->log, msg->missing,
                        auth_log_shard);
   }
@@ -4904,7 +5007,7 @@ boost::statechart::result PeeringState::GetMissing::react(const MLogRec& logevt)
   DECLARE_LOCALS
 
   peer_missing_requested.erase(logevt.from);
-  pg->proc_replica_log(logevt.msg->info, logevt.msg->log, logevt.msg->missing, logevt.from);
+  ps->proc_replica_log(logevt.msg->info, logevt.msg->log, logevt.msg->missing, logevt.from);
 
   if (peer_missing_requested.empty()) {
     if (ps->need_up_thru) {

diff --git a/src/osd/PeeringState.h b/src/osd/PeeringState.h
index 1549628b6427e8f1744300317d72c6d65d0b4a8e..33778a3af2cbbd4b5f7a344a4782fc845cec20c0 100644 (file)
--- a/src/osd/PeeringState.h
+++ b/src/osd/PeeringState.h
@@ -143,6 +143,10 @@ public:
       int64_t primary_num_bytes, int64_t local_num_bytes) = 0;
     virtual void unreserve_recovery_space() = 0;
 
+    // Peering log events
+    virtual PGLog::LogEntryHandlerRef get_log_handler(
+      ObjectStore::Transaction *t) = 0;
+
     virtual epoch_t oldest_stored_osdmap() = 0;
     virtual LogChannel &get_clog() = 0;
 
@@ -1348,6 +1352,18 @@ public:
     PeeringCtx *ctx);
   void share_pg_info();
 
+  void rewind_divergent_log(ObjectStore::Transaction& t, eversion_t newhead);
+  void merge_log(
+    ObjectStore::Transaction& t, pg_info_t &oinfo,
+    pg_log_t &olog, pg_shard_t from);
+
+  void proc_primary_info(ObjectStore::Transaction &t, const pg_info_t &info);
+  void proc_master_log(ObjectStore::Transaction& t, pg_info_t &oinfo,
+                      pg_log_t &olog, pg_missing_t& omissing,
+                      pg_shard_t from);
+  void proc_replica_log(pg_info_t &oinfo, const pg_log_t &olog,
+                       pg_missing_t& omissing, pg_shard_t from);
+
 public:
   PeeringState(
     CephContext *cct,