OSD: add handlers for MOSDPG(Push|PushReply|Pull)

MOSDPG(Push|PushReply|Pull|SubOp|SubOpReply) need the
same thing checked prior to queueing the op, so they
share a templated handler.

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

diff --git a/src/osd/OSD.cc b/src/osd/OSD.cc
index c15e93e2047520db16b1abed3efb015757c5dde5..585da0dcb4fe715d63556a0ca31ec8d172c4f41b 100644 (file)
--- a/src/osd/OSD.cc
+++ b/src/osd/OSD.cc
@@ -4410,10 +4410,19 @@ void OSD::dispatch_op(OpRequestRef op)
 
     // for replication etc.
   case MSG_OSD_SUBOP:
-    handle_sub_op(op);
+    handle_replica_op<MOSDSubOp, MSG_OSD_SUBOP>(op);
     break;
   case MSG_OSD_SUBOPREPLY:
-    handle_sub_op_reply(op);
+    handle_replica_op<MOSDSubOpReply, MSG_OSD_SUBOPREPLY>(op);
+    break;
+  case MSG_OSD_PG_PUSH:
+    handle_replica_op<MOSDPGPush, MSG_OSD_PG_PUSH>(op);
+    break;
+  case MSG_OSD_PG_PULL:
+    handle_replica_op<MOSDPGPull, MSG_OSD_PG_PULL>(op);
+    break;
+  case MSG_OSD_PG_PUSH_REPLY:
+    handle_replica_op<MOSDPGPushReply, MSG_OSD_PG_PUSH_REPLY>(op);
     break;
   }
 }
@@ -6767,12 +6776,13 @@ void OSD::handle_op(OpRequestRef op)
   enqueue_op(pg, op);
 }
 
-void OSD::handle_sub_op(OpRequestRef op)
+template<typename T, int MSGTYPE>
+void OSD::handle_replica_op(OpRequestRef op)
 {
-  MOSDSubOp *m = static_cast<MOSDSubOp*>(op->request);
-  assert(m->get_header().type == MSG_OSD_SUBOP);
+  T *m = static_cast<T *>(op->request);
+  assert(m->get_header().type == MSGTYPE);
 
-  dout(10) << "handle_sub_op " << *m << " epoch " << m->map_epoch << dendl;
+  dout(10) << __func__ << *m << " epoch " << m->map_epoch << dendl;
   if (m->map_epoch < up_epoch) {
     dout(3) << "replica op from before up" << dendl;
     return;
@@ -6784,9 +6794,6 @@ void OSD::handle_sub_op(OpRequestRef op)
   // must be a rep op.
   assert(m->get_source().is_osd());
   
-  // make sure we have the pg
-  const pg_t pgid = m->pgid;
-
   // require same or newer map
   if (!require_same_or_newer_map(op, m->map_epoch))
     return;
@@ -6795,6 +6802,8 @@ void OSD::handle_sub_op(OpRequestRef op)
   _share_map_incoming(m->get_source(), m->get_connection().get(), m->map_epoch,
                      static_cast<Session*>(m->get_connection()->get_priv()));
 
+  // make sure we have the pg
+  const pg_t pgid = m->pgid;
   if (service.splitting(pgid)) {
     waiting_for_pg[pgid].push_back(op);
     return;
@@ -6807,38 +6816,6 @@ void OSD::handle_sub_op(OpRequestRef op)
   enqueue_op(pg, op);
 }
 
-void OSD::handle_sub_op_reply(OpRequestRef op)
-{
-  MOSDSubOpReply *m = static_cast<MOSDSubOpReply*>(op->request);
-  assert(m->get_header().type == MSG_OSD_SUBOPREPLY);
-  if (m->get_map_epoch() < up_epoch) {
-    dout(3) << "replica op reply from before up" << dendl;
-    return;
-  }
-
-  if (!require_osd_peer(op))
-    return;
-
-  // must be a rep op.
-  assert(m->get_source().is_osd());
-  
-  // make sure we have the pg
-  const pg_t pgid = m->get_pg();
-
-  // require same or newer map
-  if (!require_same_or_newer_map(op, m->get_map_epoch())) return;
-
-  // share our map with sender, if they're old
-  _share_map_incoming(m->get_source(), m->get_connection().get(), m->get_map_epoch(),
-                     static_cast<Session*>(m->get_connection()->get_priv()));
-
-  PG *pg = _have_pg(pgid) ? _lookup_pg(pgid) : NULL;
-  if (!pg) {
-    return;
-  }
-  enqueue_op(pg, op);
-}
-
 bool OSD::op_is_discardable(MOSDOp *op)
 {
   // drop client request if they are not connected and can't get the

diff --git a/src/osd/OSD.h b/src/osd/OSD.h
index 3d8218e6aa15fcc2d2206a1c9707946db8681c7b..5114c99b66a39bd2d5ae8ec745055f994b7b9e6c 100644 (file)
--- a/src/osd/OSD.h
+++ b/src/osd/OSD.h
@@ -1668,8 +1668,9 @@ public:
   void handle_scrub(class MOSDScrub *m);
   void handle_osd_ping(class MOSDPing *m);
   void handle_op(OpRequestRef op);
-  void handle_sub_op(OpRequestRef op);
-  void handle_sub_op_reply(OpRequestRef op);
+
+  template <typename T, int MSGTYPE>
+  void handle_replica_op(OpRequestRef op);
 
   /// check if we can throw out op from a disconnected client
   static bool op_is_discardable(class MOSDOp *m);