xio: alloc all portals and use for all connections

Allocate all xio_portal_threads and assign/spread both passive and
active connections to different portals of heavy messengers.

Signed-off-by: Vu Pham <vu@mellanox.com>

diff --git a/src/msg/xio/XioConnection.cc b/src/msg/xio/XioConnection.cc
index 40069fad0cf8d05ef5d680f36553d4d035bf4485..7f1cc0e913efe494b2fa65313ea3a07b829d32dd 100644 (file)
--- a/src/msg/xio/XioConnection.cc
+++ b/src/msg/xio/XioConnection.cc
@@ -81,7 +81,7 @@ XioConnection::XioConnection(XioMessenger *m, XioConnection::type _type,
                             const entity_inst_t& _peer) :
   Connection(m->cct, m),
   xio_conn_type(_type),
-  portal(m->default_portal()),
+  portal(m->get_portal()),
   connected(false),
   peer(_peer),
   session(NULL),

diff --git a/src/msg/xio/XioMessenger.cc b/src/msg/xio/XioMessenger.cc
index e6ff7dbf1352337fcdf2d66b9b5b62553ea70d19..4631aee707f034453a60056b8fff3ecb1a144b50 100644 (file)
--- a/src/msg/xio/XioMessenger.cc
+++ b/src/msg/xio/XioMessenger.cc
@@ -541,9 +541,9 @@ int XioMessenger::session_event(struct xio_session *session,
   case XIO_SESSION_CONNECTION_CLOSED_EVENT: /* orderly discon */
   case XIO_SESSION_CONNECTION_DISCONNECTED_EVENT: /* unexpected discon */
   case XIO_SESSION_CONNECTION_REFUSED_EVENT:
-    ldout(cct,2) << xio_session_event_types[event_data->event]
-      << " user_context " << event_data->conn_user_context << dendl;
     xcon = static_cast<XioConnection*>(event_data->conn_user_context);
+    ldout(cct,2) << xio_session_event_types[event_data->event]
+      << " xcon " << xcon << " session " << session  << dendl;
     if (likely(!!xcon)) {
       Spinlock::Locker lckr(conns_sp);
       XioConnection::EntitySet::iterator conn_iter =
@@ -565,9 +565,9 @@ int XioMessenger::session_event(struct xio_session *session,
     }
     break;
   case XIO_SESSION_CONNECTION_TEARDOWN_EVENT:
-    ldout(cct,2) << xio_session_event_types[event_data->event]
-      << " user_context " << event_data->conn_user_context << dendl;
     xcon = static_cast<XioConnection*>(event_data->conn_user_context);
+    ldout(cct,2) << xio_session_event_types[event_data->event]
+      << " xcon " << xcon << " session " << session << dendl;
     xcon->on_teardown_event();
     break;
   case XIO_SESSION_TEARDOWN_EVENT:
@@ -1004,7 +1004,7 @@ ConnectionRef XioMessenger::get_connection(const entity_inst_t& dest)
      * we can always set it explicitly */
     struct xio_connection_params xcp = {};
     xcp.session           = xcon->session;
-    xcp.ctx               = this->portals.get_portal0()->ctx;
+    xcp.ctx               = xcon->portal->ctx;
     xcp.conn_user_context = xcon;
 
     xcon->conn = xio_connect(&xcp);

diff --git a/src/msg/xio/XioMessenger.h b/src/msg/xio/XioMessenger.h
index 86f3e65959990f10aab12f81740d2009deed3357..a267962225795a943876760e690c5baaa2c9b3b7 100644 (file)
--- a/src/msg/xio/XioMessenger.h
+++ b/src/msg/xio/XioMessenger.h
@@ -68,7 +68,7 @@ public:
 
   virtual ~XioMessenger();
 
-  XioPortal* default_portal() { return portals.get_portal0(); }
+  XioPortal* get_portal() { return portals.get_next_portal(); }
 
   virtual void set_myaddr(const entity_addr_t& a) {
     Messenger::set_myaddr(a);

diff --git a/src/msg/xio/XioPortal.cc b/src/msg/xio/XioPortal.cc
index f6911850dec42044a51c2135cc10ea988f367a27..e2379fb331247a6b34a129035c42d7d73fcd4e8d 100644 (file)
--- a/src/msg/xio/XioPortal.cc
+++ b/src/msg/xio/XioPortal.cc
@@ -62,9 +62,6 @@ int XioPortals::bind(struct xio_session_ops *ops, const string& base_uri,
 
   /* bind the portals */
   for (size_t i = 0; i < portals.size(); i++) {
-    if (!portals[i])
-      portals[i] = new XioPortal(msgr, max_conns_per_ctx);
-
     uint16_t result_port;
     if (port != 0) {
       // bind directly to the given port

diff --git a/src/msg/xio/XioPortal.h b/src/msg/xio/XioPortal.h
index 42d580d2b3ef6f0cc65e704814820bb519d4a0e9..e43f95a0cdbff1d19274051195a115880e047f20 100644 (file)
--- a/src/msg/xio/XioPortal.h
+++ b/src/msg/xio/XioPortal.h
@@ -356,16 +356,19 @@ private:
   char **p_vec;
   int n;
   int last_unused;
-  int max_conns_per_ctx;
 
 public:
   XioPortals(Messenger *msgr, int _n, int nconns) : p_vec(NULL), last_unused(0)
   {
-    max_conns_per_ctx = nconns;
     n = max(_n, 1);
-    /* portal0 */
-    portals.push_back(new XioPortal(msgr, nconns));
-    /* additional portals allocated on bind() */
+
+    portals.resize(n);
+    for (int i = 0; i < n; i++) {
+      if (!portals[i]) {
+        portals[i] = new XioPortal(msgr, nconns);
+        assert(portals[i] != nullptr);
+      }
+    }
   }
 
   vector<XioPortal*>& get() { return portals; }
@@ -388,9 +391,10 @@ public:
     return pix;
   }
 
-  XioPortal* get_portal0()
+  XioPortal* get_next_portal()
   {
-    return portals[0];
+    int pix = get_last_unused();
+    return portals[pix];
   }
 
   int bind(struct xio_session_ops *ops, const string& base_uri,
@@ -420,7 +424,6 @@ public:
     p_vec = new char*[nportals];
     for (p_ix = 0; p_ix < nportals; ++p_ix) {
       portal = portals[p_ix];
-      /* shift left */
       p_vec[p_ix] = (char*) /* portal->xio_uri.c_str() */
                        portal->portal_id;
     }