osd: make get_pg_or_queue_for_pg() method safer

On succeeding in getting a PG, put a reference on it before
we drop the pg_map_lock and give it back to caller.
This is necessary because we may race with _remove_pg(),
who will remove the last reference and destruct PG in most case.

Signed-off-by: xie xingguo <xie.xingguo@zte.com.cn>

src/osd/OSD.cc

@@ -3050,6 +3050,11 @@ PG *OSD::get_pg_or_queue_for_pg(const spg_t& pgid, OpRequestRef& op)
   PG *out = NULL;
   if (wlistiter == session->waiting_for_pg.end()) {
     out = i->second;
+    // On succeeding in getting a PG, put a reference on it before
+    // we drop the pg_map_lock and give it back to caller.
+    // This is necessary because we may race with _remove_pg(),
+    // who will remove the last reference and destruct PG in most case.
+    out->get("get_pg_or_queue_for_pg");
   } else {
     wlistiter->second.push_back(op);
     register_session_waiting_on_pg(session, pgid);
@@ -8470,6 +8475,7 @@ void OSD::handle_op(OpRequestRef& op, OSDMapRef& osdmap)
     op->send_map_update = share_map.should_send;
     op->sent_epoch = m->get_map_epoch();
     enqueue_op(pg, op);
+    pg->put("get_pg_or_queue_for_pg");
     share_map.should_send = false;
     return;
   }
@@ -8558,6 +8564,7 @@ void OSD::handle_replica_op(OpRequestRef& op, OSDMapRef& osdmap)
     op->send_map_update = should_share_map;
     op->sent_epoch = m->map_epoch;
     enqueue_op(pg, op);
+    pg->put("get_pg_or_queue_for_pg");
   } else if (should_share_map && m->get_connection()->is_connected()) {
     C_SendMap *send_map = new C_SendMap(this, m->get_source(),
 					m->get_connection(),