osd: only trim pg log if pg contains complete set of osds

Eventually we may want to also impose some maximum pg log size.  At
some point the cost of the long log will approach the cost of
building a backlog...

2 files changed, 4 insertions, 1 deletions

diff --git a/src/osd/PG.h b/src/osd/PG.h
index 638ac8cbbbb..24e7b86e7f2 100644
--- a/src/osd/PG.h
+++ b/src/osd/PG.h
@@ -724,6 +724,8 @@ public:
 
   bool  is_empty() const { return info.last_update == eversion_t(0,0); }
 
+  bool is_complete_pg() { return acting.size() == info.pgid.size(); }
+
   // pg on-disk state
   void write_info(ObjectStore::Transaction& t);
   void write_log(ObjectStore::Transaction& t);
diff --git a/src/osd/ReplicatedPG.cc b/src/osd/ReplicatedPG.cc
index 986623b93fe..27def3f7322 100644
--- a/src/osd/ReplicatedPG.cc
+++ b/src/osd/ReplicatedPG.cc
@@ -1275,7 +1275,8 @@ void ReplicatedPG::issue_repop(RepGather *repop, int dest, utime_t now)
   wr->snapset = repop->snapset;
   wr->snapc = repop->snapc;
   wr->get_data() = repop->op->get_data();   // _copy_ bufferlist
-  wr->pg_trim_to = peers_complete_thru;
+  if (is_complete_pg())
+    wr->pg_trim_to = peers_complete_thru;
   wr->peer_stat = osd->get_my_stat_for(now, dest);
   osd->messenger->send_message(wr, osd->osdmap->get_inst(dest));
 }