From 4ae4dce5c5bb547c1ff54d07c8b70d287490cae9 Mon Sep 17 00:00:00 2001
From: Samuel Just <sam.just@inktank.com>
Date: Thu, 3 Jan 2013 09:59:45 -0800
Subject: [PATCH] OSD: for old osds, dispatch peering messages immediately

Normally, we batch up peering messages until the end of
process_peering_events to allow us to combine many notifies, etc
to the same osd into the same message.  However, old osds assume
that the actiavtion message (log or info) will be _dispatched
before the first sub_op_modify of the interval.  Thus, for those
peers, we need to send the peering messages before we drop the
pg lock, lest we issue a client repop from another thread before
activation message is sent.

Signed-off-by: Samuel Just <sam.just@inktank.com>
Reviewed-by: Sage Weil <sage@inktank.com>
---
 src/osd/OSD.cc | 24 +++++++++++++++++++++++-
 src/osd/OSD.h  |  1 +
 2 files changed, 24 insertions(+), 1 deletion(-)

diff --git a/src/osd/OSD.cc b/src/osd/OSD.cc
index 75f84ede5bb30..d0236e0641949 100644
--- a/src/osd/OSD.cc
+++ b/src/osd/OSD.cc
@@ -4890,6 +4890,23 @@ void OSD::dispatch_context_transaction(PG::RecoveryCtx &ctx, PG *pg)
   }
 }
 
+bool OSD::compat_must_dispatch_immediately(PG *pg)
+{
+  assert(pg->is_locked());
+  for (vector<int>::iterator i = pg->acting.begin();
+       i != pg->acting.end();
+       ++i) {
+    if (*i == whoami)
+      continue;
+    ConnectionRef conn =
+      service.get_con_osd_cluster(*i, pg->get_osdmap()->get_epoch());
+    if (conn && !(conn->features & CEPH_FEATURE_INDEP_PG_MAP)) {
+      return true;
+    }
+  }
+  return false;
+}
+
 void OSD::dispatch_context(PG::RecoveryCtx &ctx, PG *pg, OSDMapRef curmap)
 {
   do_notifies(*ctx.notify_list, curmap);
@@ -6160,7 +6177,12 @@ void OSD::process_peering_events(const list<PG*> &pgs)
       rctx.on_applied->add(new C_CompleteSplits(this, split_pgs));
       split_pgs.clear();
     }
-    dispatch_context_transaction(rctx, pg);
+    if (compat_must_dispatch_immediately(pg)) {
+      dispatch_context(rctx, pg, curmap);
+      rctx = create_context();
+    } else {
+      dispatch_context_transaction(rctx, pg);
+    }
     pg->unlock();
   }
   if (need_up_thru)
diff --git a/src/osd/OSD.h b/src/osd/OSD.h
index 6576d7fdbdaa7..8bab8a9905938 100644
--- a/src/osd/OSD.h
+++ b/src/osd/OSD.h
@@ -969,6 +969,7 @@ protected:
 
   // -- generic pg peering --
   PG::RecoveryCtx create_context();
+  bool compat_must_dispatch_immediately(PG *pg);
   void dispatch_context(PG::RecoveryCtx &ctx, PG *pg, OSDMapRef curmap);
   void dispatch_context_transaction(PG::RecoveryCtx &ctx, PG *pg);
   void do_notifies(map< int,vector<pair<pg_notify_t, pg_interval_map_t> > >& notify_list,
-- 
2.39.5