ReplicatedPG: remove pushing/pulling

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

2 files changed, 19 insertions, 73 deletions

diff --git a/src/osd/ReplicatedPG.cc b/src/osd/ReplicatedPG.cc
index 242943f14c6..b0f31251a31 100644
--- a/src/osd/ReplicatedPG.cc
+++ b/src/osd/ReplicatedPG.cc
@@ -202,8 +202,9 @@ void ReplicatedPG::on_global_recover(
   const hobject_t &soid)
 {
   publish_stats_to_osd();
-  pushing.erase(soid);
   dout(10) << "pushed " << soid << " to all replicas" << dendl;
+  assert(recovering.count(soid));
+  recovering.erase(soid);
   finish_recovery_op(soid);
   if (waiting_for_degraded_object.count(soid)) {
     requeue_ops(waiting_for_degraded_object[soid]);
@@ -268,8 +269,8 @@ void ReplicatedPG::wait_for_missing_object(const hobject_t& soid, OpRequestRef o
   assert(g != missing.missing.end());
   const eversion_t &v(g->second.need);
 
-  map<hobject_t, PullInfo>::const_iterator p = pulling.find(soid);
-  if (p != pulling.end()) {
+  set<hobject_t>::const_iterator p = recovering.find(soid);
+  if (p != recovering.end()) {
     dout(7) << "missing " << soid << " v " << v << ", already pulling." << dendl;
   }
   else if (missing_loc.find(soid) == missing_loc.end()) {
@@ -316,15 +317,15 @@ void ReplicatedPG::wait_for_degraded_object(const hobject_t& soid, OpRequestRef
   assert(is_degraded_object(soid));
 
   // we don't have it (yet).
-  if (pushing.count(soid)) {
+  if (recovering.count(soid)) {
     dout(7) << "degraded "
 	    << soid 
-	    << ", already pushing"
+	    << ", already recovering"
 	    << dendl;
   } else {
     dout(7) << "degraded " 
 	    << soid 
-	    << ", pushing"
+	    << ", recovering"
 	    << dendl;
     eversion_t v;
     for (unsigned i = 1; i < acting.size(); i++) {
@@ -6002,6 +6003,8 @@ int ReplicatedPG::recover_missing(
     assert(head_obc);
   }
   start_recovery_op(soid);
+  assert(!recovering.count(soid));
+  recovering.insert(soid);
   pgbackend->recover_object(
     soid,
     head_obc,
@@ -6963,7 +6966,8 @@ void ReplicatedBackend::sub_op_push(OpRequestRef op)
 
 void ReplicatedPG::failed_push(int from, const hobject_t &soid)
 {
-  // TODOSAM: this will need to update recovering
+  assert(recovering.count(soid));
+  recovering.erase(soid);
   map<hobject_t,set<int> >::iterator p = missing_loc.find(soid);
   if (p != missing_loc.end()) {
     dout(0) << "_failed_push " << soid << " from osd." << from
@@ -7343,10 +7347,6 @@ void ReplicatedPG::on_change(ObjectStore::Transaction *t)
   apply_and_flush_repops(is_primary());
 
   pgbackend->on_change(t);
-  // clear pushing/pulling maps
-  pushing.clear();
-  pulling.clear();
-  pull_from_peer.clear();
 
   // clear snap_trimmer state
   snap_trimmer_machine.process_event(Reset());
@@ -7372,10 +7372,8 @@ void ReplicatedPG::_clear_recovery_state()
   backfill_pos = hobject_t();
   backfills_in_flight.clear();
   pending_backfill_updates.clear();
+  recovering.clear();
   pgbackend->clear_state();
-  pulling.clear();
-  pushing.clear();
-  pull_from_peer.clear();
 }
 
 void ReplicatedPG::cancel_pull(const hobject_t &soid)
@@ -7586,7 +7584,8 @@ int ReplicatedPG::recover_primary(int max, ThreadPool::TPHandle &handle)
 
   const pg_missing_t &missing = pg_log.get_missing();
 
-  dout(10) << "recover_primary pulling " << pulling.size() << " in pg" << dendl;
+  dout(10) << "recover_primary recovering " << recovering.size()
+	   << " in pg" << dendl;
   dout(10) << "recover_primary " << missing << dendl;
   dout(25) << "recover_primary " << missing.missing << dendl;
 
@@ -7626,8 +7625,8 @@ int ReplicatedPG::recover_primary(int max, ThreadPool::TPHandle &handle)
 	     << (unfound ? " (unfound)":"")
 	     << (missing.is_missing(soid) ? " (missing)":"")
 	     << (missing.is_missing(head) ? " (missing head)":"")
-             << (pulling.count(soid) ? " (pulling)":"")
-	     << (pulling.count(head) ? " (pulling head)":"")
+             << (recovering.count(soid) ? " (recovering)":"")
+	     << (recovering.count(head) ? " (recovering head)":"")
              << dendl;
 
     if (latest) {
@@ -7816,8 +7815,8 @@ int ReplicatedPG::recover_replicas(int max, ThreadPool::TPHandle &handle)
       handle.reset_tp_timeout();
       const hobject_t soid(p->second);
 
-      if (pushing.count(soid)) {
-	dout(10) << __func__ << ": already pushing " << soid << dendl;
+      if (recovering.count(soid)) {
+	dout(10) << __func__ << ": already recovering" << soid << dendl;
 	continue;
       }
 
@@ -8063,7 +8062,7 @@ void ReplicatedPG::prep_backfill_object_push(
 
   backfills_in_flight.insert(oid);
 
-  if (!pushing.count(oid))
+  if (!recovering.count(oid))
     start_recovery_op(oid);
   ObjectContextRef obc = get_object_context(oid, false);
   obc->ondisk_read_lock();
diff --git a/src/osd/ReplicatedPG.h b/src/osd/ReplicatedPG.h
index 0b438163908..94514a56e25 100644
--- a/src/osd/ReplicatedPG.h
+++ b/src/osd/ReplicatedPG.h
@@ -499,51 +499,6 @@ protected:
   }
   void put_snapset_context(SnapSetContext *ssc);
 
-  // push
-  struct PushInfo {
-    ObjectRecoveryProgress recovery_progress;
-    ObjectRecoveryInfo recovery_info;
-    int priority;
-
-    void dump(Formatter *f) const {
-      {
-	f->open_object_section("recovery_progress");
-	recovery_progress.dump(f);
-	f->close_section();
-      }
-      {
-	f->open_object_section("recovery_info");
-	recovery_info.dump(f);
-	f->close_section();
-      }
-    }
-  };
-  map<hobject_t, map<int, PushInfo> > pushing;
-
-  // pull
-  struct PullInfo {
-    ObjectRecoveryProgress recovery_progress;
-    ObjectRecoveryInfo recovery_info;
-    int priority;
-
-    void dump(Formatter *f) const {
-      {
-	f->open_object_section("recovery_progress");
-	recovery_progress.dump(f);
-	f->close_section();
-      }
-      {
-	f->open_object_section("recovery_info");
-	recovery_info.dump(f);
-	f->close_section();
-      }
-    }
-
-    bool is_complete() const {
-      return recovery_progress.is_complete(recovery_info);
-    }
-  };
-  map<hobject_t, PullInfo> pulling;
   set<hobject_t> recovering;
 
   ObjectRecoveryInfo recalc_subsets(const ObjectRecoveryInfo& recovery_info);
@@ -622,9 +577,6 @@ protected:
   /// leading edge of backfill
   hobject_t backfill_pos;
 
-  // Reverse mapping from osd peer to objects beging pulled from that peer
-  map<int, set<hobject_t> > pull_from_peer;
-
   int prep_object_replica_pushes(const hobject_t& soid, eversion_t v,
 				 int priority,
 				 map<int, vector<PushOp> > *pushes);
@@ -795,11 +747,6 @@ protected:
       const hobject_t &hoid,
       epoch_t epoch) : pg(pg), hoid(hoid), epoch(epoch) {}
     void finish(int) {
-      pg->lock();
-      if (!pg->pg_has_reset_since(epoch)) {
-	pg->finish_recovery_op(hoid);
-      }
-      pg->unlock();
     }
   };
   friend struct C_OSD_CompletedPull;