msg/async: use v1 for v1 <-> [v2,v1] peers

If *peers* are communicating, i.e. there may be bidirectional connection
attempts, we must use the same protocol version from both ends or else
we will get very confused.

Fix this by forcing the use of v1 when we

- are bound to a v1 endpoint only (people can't connect to us via v2)
- we are connecting to a *peer*

If it is a non-peer, then connections are uni-directional.  If we both
have v2, we will both use v2.

If we ever switch to [v2,v1], it will be as part of a restart.

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

diff --git a/src/msg/async/AsyncMessenger.cc b/src/msg/async/AsyncMessenger.cc
index 8d8ed05107c706639eacb73c505f1d62b2867414..7f34f886b83bf4d09564f0d72aac75684b4585b6 100644 (file)
--- a/src/msg/async/AsyncMessenger.cc
+++ b/src/msg/async/AsyncMessenger.cc
@@ -601,32 +601,38 @@ AsyncConnectionRef AsyncMessenger::create_connect(
   return conn;
 }
 
-ConnectionRef AsyncMessenger::connect_to(int type, const entity_addrvec_t& addrs)
-{
-  Mutex::Locker l(lock);
-  if (*my_addrs == addrs ||
-      (addrs.v.size() == 1 &&
-       my_addrs->contains(addrs.front()))) {
-    // local
-    return local_connection;
-  }
-
-  AsyncConnectionRef conn = _lookup_conn(addrs);
-  if (conn) {
-    ldout(cct, 10) << __func__ << " " << addrs << " existing " << conn << dendl;
-  } else {
-    conn = create_connect(addrs, type);
-    ldout(cct, 10) << __func__ << " " << addrs << " new " << conn << dendl;
-  }
-
-  return conn;
-}
 
 ConnectionRef AsyncMessenger::get_loopback_connection()
 {
   return local_connection;
 }
 
+entity_addrvec_t AsyncMessenger::_filter_addrs(int type,
+                                              const entity_addrvec_t& addrs)
+{
+  if (did_bind &&
+      !get_myaddrs().has_msgr2() &&
+      get_mytype() == type) {
+    // if we are bound to v1 only, and we are connecting to a peer, we cannot
+    // use the peer's v2 address (yet). otherwise the connection is assymetrical,
+    // because they would have to use v1 to connect to us, and we would use v2,
+    // and connection race detection etc would totally break down (among other
+    // things).
+    ldout(cct, 10) << __func__ << " " << addrs << " type " << type
+                  << " limiting to v1 ()" << dendl;
+    entity_addrvec_t r;
+    for (auto& i : addrs.v) {
+      if (i.is_msgr2()) {
+       continue;
+      }
+      r.v.push_back(i);
+    }
+    return r;
+  } else {
+    return addrs;
+  }
+}
+
 int AsyncMessenger::send_to(Message *m, int type, const entity_addrvec_t& addrs)
 {
   Mutex::Locker l(lock);
@@ -650,11 +656,35 @@ int AsyncMessenger::send_to(Message *m, int type, const entity_addrvec_t& addrs)
     return -EINVAL;
   }
 
-  AsyncConnectionRef conn = _lookup_conn(addrs);
-  submit_message(m, conn, addrs, type);
+  auto av = _filter_addrs(type, addrs);
+  AsyncConnectionRef conn = _lookup_conn(av);
+  submit_message(m, conn, av, type);
   return 0;
 }
 
+ConnectionRef AsyncMessenger::connect_to(int type, const entity_addrvec_t& addrs)
+{
+  Mutex::Locker l(lock);
+  if (*my_addrs == addrs ||
+      (addrs.v.size() == 1 &&
+       my_addrs->contains(addrs.front()))) {
+    // local
+    return local_connection;
+  }
+
+  auto av = _filter_addrs(type, addrs);
+
+  AsyncConnectionRef conn = _lookup_conn(av);
+  if (conn) {
+    ldout(cct, 10) << __func__ << " " << av << " existing " << conn << dendl;
+  } else {
+    conn = create_connect(av, type);
+    ldout(cct, 10) << __func__ << " " << av << " new " << conn << dendl;
+  }
+
+  return conn;
+}
+
 void AsyncMessenger::submit_message(Message *m, AsyncConnectionRef con,
                                     const entity_addrvec_t& dest_addrs,
                                    int dest_type)

diff --git a/src/msg/async/AsyncMessenger.h b/src/msg/async/AsyncMessenger.h
index 2a91e0cf4019d70fa20b2ec75dc3c54b18053022..666187a720deebaaf85d466fbe03e1dd6c77429a 100644 (file)
--- a/src/msg/async/AsyncMessenger.h
+++ b/src/msg/async/AsyncMessenger.h
@@ -215,6 +215,9 @@ private:
   void _finish_bind(const entity_addrvec_t& bind_addrs,
                    const entity_addrvec_t& listen_addrs);
 
+  entity_addrvec_t _filter_addrs(int type,
+                                const entity_addrvec_t& addrs);
+
  private:
   static const uint64_t ReapDeadConnectionThreshold = 5;