osd: make _share_map_outgoing() use a Connection

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

diff --git a/src/osd/OSD.cc b/src/osd/OSD.cc
index 541c5d00e02431f1d68eeeef31cbf2f3eaee2f24..efc94f399b9c00a1a3d3c4cbc594d16fe9102bca 100644 (file)
--- a/src/osd/OSD.cc
+++ b/src/osd/OSD.cc
@@ -1911,8 +1911,13 @@ void OSD::handle_osd_ping(MOSDPing *m)
 
       if (curmap->is_up(from)) {
        note_peer_epoch(from, m->map_epoch);
-       if (is_active())
-         _share_map_outgoing(curmap->get_cluster_inst(from));
+       if (is_active()) {
+         Connection *con = service.get_con_osd_cluster(from, curmap->get_epoch());
+         if (con) {
+           _share_map_outgoing(from, con);
+           con->put();
+         }
+       }
       }
     }
     break;
@@ -1932,8 +1937,13 @@ void OSD::handle_osd_ping(MOSDPing *m)
       if (m->map_epoch &&
          curmap->is_up(from)) {
        note_peer_epoch(from, m->map_epoch);
-       if (is_active())
-         _share_map_outgoing(curmap->get_cluster_inst(from));
+       if (is_active()) {
+         Connection *con = service.get_con_osd_cluster(from, curmap->get_epoch());
+         if (con) {
+           _share_map_outgoing(from, con);
+           con->put();
+         }
+       }
       }
 
       // Cancel false reports
@@ -3019,25 +3029,21 @@ bool OSD::_share_map_incoming(const entity_inst_t& inst, epoch_t epoch,
 }
 
 
-void OSD::_share_map_outgoing(const entity_inst_t& inst,
-                             OSDMapRef map)
+void OSD::_share_map_outgoing(int peer, Connection *con, OSDMapRef map)
 {
   if (!map)
     map = service.get_osdmap();
-  assert(inst.name.is_osd());
-
-  int peer = inst.name.num();
 
   // send map?
   epoch_t pe = get_peer_epoch(peer);
   if (pe) {
     if (pe < map->get_epoch()) {
-      send_incremental_map(pe, inst);
+      send_incremental_map(pe, con);
       note_peer_epoch(peer, map->get_epoch());
     } else
-      dout(20) << "_share_map_outgoing " << inst << " already has epoch " << pe << dendl;
+      dout(20) << "_share_map_outgoing " << con << " already has epoch " << pe << dendl;
   } else {
-    dout(20) << "_share_map_outgoing " << inst << " don't know epoch, doing nothing" << dendl;
+    dout(20) << "_share_map_outgoing " << con << " don't know epoch, doing nothing" << dendl;
     // no idea about peer's epoch.
     // ??? send recent ???
     // do nothing.
@@ -4682,7 +4688,7 @@ void OSD::do_notifies(
       continue;
     Connection *con =
       cluster_messenger->get_connection(curmap->get_cluster_inst(it->first));
-    _share_map_outgoing(curmap->get_cluster_inst(it->first), curmap);
+    _share_map_outgoing(it->first, con, curmap);
     if ((con->features & CEPH_FEATURE_INDEP_PG_MAP)) {
       dout(7) << "do_notify osd." << it->first
              << " on " << it->second.size() << " PGs" << dendl;
@@ -4722,7 +4728,7 @@ void OSD::do_queries(map< int, map<pg_t,pg_query_t> >& query_map,
     int who = pit->first;
     Connection *con =
       cluster_messenger->get_connection(curmap->get_cluster_inst(pit->first));
-    _share_map_outgoing(curmap->get_cluster_inst(who), curmap);
+    _share_map_outgoing(who, con, curmap);
     if ((con->features & CEPH_FEATURE_INDEP_PG_MAP)) {
       dout(7) << "do_queries querying osd." << who
              << " on " << pit->second.size() << " PGs" << dendl;
@@ -4761,7 +4767,7 @@ void OSD::do_infos(map<int,vector<pair<pg_notify_t, pg_interval_map_t> > >& info
     }
     Connection *con =
       cluster_messenger->get_connection(curmap->get_cluster_inst(p->first));
-    _share_map_outgoing(curmap->get_cluster_inst(p->first), curmap);
+    _share_map_outgoing(p->first, con, curmap);
     if ((con->features & CEPH_FEATURE_INDEP_PG_MAP)) {
       MOSDPGInfo *m = new MOSDPGInfo(curmap->get_epoch());
       m->pg_list = p->second;
@@ -5147,10 +5153,14 @@ void OSD::handle_pg_query(OpRequestRef op)
     pg_info_t empty(pgid);
     if (it->second.type == pg_query_t::LOG ||
        it->second.type == pg_query_t::FULLLOG) {
-      MOSDPGLog *mlog = new MOSDPGLog(osdmap->get_epoch(), empty,
-                                     it->second.epoch_sent);
-      _share_map_outgoing(osdmap->get_cluster_inst(from));
-      service.send_message_osd_cluster(from, mlog, osdmap->get_epoch());
+      Connection *con = service.get_con_osd_cluster(from, osdmap->get_epoch());
+      if (con) {
+       MOSDPGLog *mlog = new MOSDPGLog(osdmap->get_epoch(), empty,
+                                       it->second.epoch_sent);
+       _share_map_outgoing(from, con, osdmap);
+       cluster_messenger->send_message(mlog, con);
+       con->put();
+      }
     } else {
       notify_list[from].push_back(make_pair(pg_notify_t(it->second.epoch_sent,
                                                        osdmap->get_epoch(),

diff --git a/src/osd/OSD.h b/src/osd/OSD.h
index 8277e35c7dbbc85354edac51521a59c38fb8ac4c..e5258b0c99f023180254a16c660b227c5cb9c5af 100644 (file)
--- a/src/osd/OSD.h
+++ b/src/osd/OSD.h
@@ -710,7 +710,7 @@ private:
 
   bool _share_map_incoming(const entity_inst_t& inst, epoch_t epoch,
                           Session *session = 0);
-  void _share_map_outgoing(const entity_inst_t& inst,
+  void _share_map_outgoing(int peer, Connection *con,
                           OSDMapRef map = OSDMapRef());
 
   void wait_for_new_map(OpRequestRef op);

diff --git a/src/osd/PG.cc b/src/osd/PG.cc
index 21a6f514dbc2f2f4e7813f398b08384f673e95e9..82beaf6e29e875d45713ad24d0eb12f9621be309 100644 (file)
--- a/src/osd/PG.cc
+++ b/src/osd/PG.cc
@@ -4222,9 +4222,14 @@ void PG::fulfill_log(int from, const pg_query_t &query, epoch_t query_epoch)
 
   dout(10) << " sending " << mlog->log << " " << mlog->missing << dendl;
 
-  osd->osd->_share_map_outgoing(get_osdmap()->get_cluster_inst(from),
-                               get_osdmap());
-  osd->send_message_osd_cluster(from, mlog, get_osdmap()->get_epoch());
+  Connection *con = osd->get_con_osd_cluster(from, get_osdmap()->get_epoch());
+  if (con) {
+    osd->osd->_share_map_outgoing(from, con, get_osdmap());
+    osd->cluster_messenger->send_message(mlog, con);
+    con->put();
+  } else {
+    mlog->put();
+  }
 }