OSD: change Session handling around _share_map_incoming

Move responsibility for the reference up to _share_map_incoming's caller,
and start using the Session::sent_epoch_lock. This looks a little silly
now, but we're going to split up the decide-to-send-maps and send-maps steps
and don't want to block in the decide-to-send step, so we need some
pretty flexible locking up at this level.

Signed-off-by: Greg Farnum <greg@inktank.com>

diff --git a/src/osd/OSD.cc b/src/osd/OSD.cc
index 104f7c298470507efd2d988641a1233490a2c949..6e3f3a96f0db3b00593d5310343cb7377171bf11 100644 (file)
--- a/src/osd/OSD.cc
+++ b/src/osd/OSD.cc
@@ -4710,8 +4710,6 @@ void OSD::_share_map_incoming(
       send_incremental_map(epoch, con, osdmap);
     }
   }
-  if (session)
-    session->put();
 }
 
 
@@ -7429,12 +7427,21 @@ void OSD::handle_op(OpRequestRef op, OSDMapRef osdmap)
   }
 
   // share our map with sender, if they're old
+  Session *client_session = static_cast<Session *>(m->get_connection()->get_priv());
+  if (client_session) {
+    client_session->sent_epoch_lock.Lock();
+  }
   _share_map_incoming(
-    m->get_source(),
-    m->get_connection().get(),
-    m->get_map_epoch(),
-    osdmap,
-    static_cast<Session *>(m->get_connection()->get_priv()));
+      m->get_source(),
+      m->get_connection().get(),
+      m->get_map_epoch(),
+      osdmap,
+      client_session
+  );
+  if (client_session) {
+    client_session->sent_epoch_lock.Unlock();
+    client_session->put();
+  }
 
   if (op->rmw_flags == 0) {
     int r = init_op_flags(op);
@@ -7567,10 +7574,19 @@ void OSD::handle_replica_op(OpRequestRef op, OSDMapRef osdmap)
     return;
 
   // share our map with sender, if they're old
+  Session *peer_session =
+      static_cast<Session*>(m->get_connection()->get_priv());
+  if (peer_session) {
+    peer_session->sent_epoch_lock.Lock();
+  }
   _share_map_incoming(
     m->get_source(), m->get_connection().get(), m->map_epoch,
     osdmap,
-    static_cast<Session*>(m->get_connection()->get_priv()));
+    peer_session);
+  if (peer_session) {
+    peer_session->sent_epoch_lock.Unlock();
+    peer_session->put();
+  }
 
   // make sure we have the pg
   const spg_t pgid = m->pgid;