move ldlm namespace creation in setup phase to avoid grab

[fs/lustre-release.git] / lnet / lnet / router.c

diff --git a/lnet/lnet/router.c b/lnet/lnet/router.c
index 1be457b..42df3a2 100644 (file)
--- a/lnet/lnet/router.c
+++ b/lnet/lnet/router.c
@@ -39,6 +39,9 @@ CFS_MODULE_PARM(small_router_buffers, "i", int, 0444,
 static int large_router_buffers = 512;
 CFS_MODULE_PARM(large_router_buffers, "i", int, 0444,
                 "# of large messages to buffer in the router");
+static int peer_buffer_credits = 0;
+CFS_MODULE_PARM(peer_buffer_credits, "i", int, 0444,
+                "# router buffer credits per peer");
 
 static int auto_down = 1;
 CFS_MODULE_PARM(auto_down, "i", int, 0444,
@@ -66,6 +69,20 @@ lnet_peers_start_down(void)
         return check_routers_before_use;
 }
 
+int
+lnet_peer_buffer_credits(lnet_ni_t *ni)
+{
+        /* NI option overrides LNet default */
+        if (ni->ni_peerrtrcredits > 0)
+                return ni->ni_peerrtrcredits;
+        if (peer_buffer_credits > 0)
+                return peer_buffer_credits;
+
+        /* As an approximation, allow this peer the same number of router
+         * buffers as it is allowed outstanding sends */
+        return ni->ni_peertxcredits;
+}
+
 void
 lnet_notify_locked(lnet_peer_t *lp, int notifylnd, int alive, time_t when)
 {
@@ -184,6 +201,13 @@ lnet_notify (lnet_ni_t *ni, lnet_nid_t nid, int alive, time_t when)
                 return 0;
         }
 
+        /* We can't fully trust LND on reporting exact peer last_alive
+         * if he notifies us about dead peer. For example ksocklnd can
+         * call us with when == _time_when_the_node_was_booted_ if
+         * no connections were successfully established */
+        if (ni != NULL && !alive && when < lp->lp_last_alive)
+                when = lp->lp_last_alive;
+
         lnet_notify_locked(lp, ni == NULL, alive, when);
 
         LNET_UNLOCK();
@@ -207,6 +231,12 @@ lnet_notify (lnet_ni_t *ni, lnet_nid_t nid, int alive, time_t when)
         return -EOPNOTSUPP;
 }
 
+void
+lnet_notify_locked (lnet_peer_t *lp, int notifylnd, int alive, time_t when)
+{
+        return;
+}
+
 #endif
 
 static void
@@ -606,6 +636,7 @@ lnet_router_checker(void *arg)
         int                  rc;
         lnet_handle_md_t     mdh;
         lnet_peer_t         *rtr;
+        lnet_md_t            md = {0};
         struct list_head    *entry;
         time_t               now;
         lnet_process_id_t    rtr_id;
@@ -618,13 +649,16 @@ lnet_router_checker(void *arg)
 
         LASSERT (the_lnet.ln_rc_state == LNET_RC_STATE_SHUTDOWN);
 
-        rc = LNetMDBind((lnet_md_t){.start     = &pinginfo,
-                                    .length    = sizeof(pinginfo),
-                                    .threshold = LNET_MD_THRESH_INF,
-                                    .options   = LNET_MD_TRUNCATE,
-                                    .eq_handle = the_lnet.ln_rc_eqh},
-                        LNET_UNLINK,
-                        &mdh);
+        /* initialize md content */
+        md.start     = &pinginfo;
+        md.length    = sizeof(pinginfo);
+        md.threshold = LNET_MD_THRESH_INF;
+        md.max_size  = 0;
+        md.options   = LNET_MD_TRUNCATE,
+        md.user_ptr  = NULL;
+        md.eq_handle = the_lnet.ln_rc_eqh;
+
+        rc = LNetMDBind(md, LNET_UNLINK, &mdh);
 
         if (rc < 0) {
                 CERROR("Can't bind MD: %d\n", rc);
@@ -1043,6 +1077,12 @@ lnet_peers_start_down(void)
         return 0;
 }
 
+int
+lnet_peer_buffer_credits(lnet_ni_t *ni)
+{
+        return 0;
+}
+
 void
 lnet_router_checker_stop(void)
 {