lnet/lnet/router.c

   1 /*
   2  * Copyright (c) 2007, 2010, Oracle and/or its affiliates. All rights reserved.
   3  *
   4  * Copyright (c) 2011, 2017, Intel Corporation.
   5  *
   6  *   This file is part of Lustre, https://wiki.whamcloud.com/
   7  *
   8  *   Portals is free software; you can redistribute it and/or
   9  *   modify it under the terms of version 2 of the GNU General Public
  10  *   License as published by the Free Software Foundation.
  11  *
  12  *   Portals is distributed in the hope that it will be useful,
  13  *   but WITHOUT ANY WARRANTY; without even the implied warranty of
  14  *   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  15  *   GNU General Public License for more details.
  16  *
  17  *   You should have received a copy of the GNU General Public License
  18  *   along with Portals; if not, write to the Free Software
  19  *   Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
  20  *
  21  */
  22
  23 #define DEBUG_SUBSYSTEM S_LNET
  24
  25 #include <linux/random.h>
  26 #include <lnet/lib-lnet.h>
  27
  28 #define LNET_NRB_TINY_MIN       512     /* min value for each CPT */
  29 #define LNET_NRB_TINY           (LNET_NRB_TINY_MIN * 4)
  30 #define LNET_NRB_SMALL_MIN      4096    /* min value for each CPT */
  31 #define LNET_NRB_SMALL          (LNET_NRB_SMALL_MIN * 4)
  32 #define LNET_NRB_SMALL_PAGES    1
  33 #define LNET_NRB_LARGE_MIN      256     /* min value for each CPT */
  34 #define LNET_NRB_LARGE          (LNET_NRB_LARGE_MIN * 4)
  35 #define LNET_NRB_LARGE_PAGES    ((LNET_MTU + PAGE_SIZE - 1) >> \
  36                                   PAGE_SHIFT)
  37
  38 static char *forwarding = "";
  39 module_param(forwarding, charp, 0444);
  40 MODULE_PARM_DESC(forwarding, "Explicitly enable/disable forwarding between networks");
  41
  42 static int tiny_router_buffers;
  43 module_param(tiny_router_buffers, int, 0444);
  44 MODULE_PARM_DESC(tiny_router_buffers, "# of 0 payload messages to buffer in the router");
  45 static int small_router_buffers;
  46 module_param(small_router_buffers, int, 0444);
  47 MODULE_PARM_DESC(small_router_buffers, "# of small (1 page) messages to buffer in the router");
  48 static int large_router_buffers;
  49 module_param(large_router_buffers, int, 0444);
  50 MODULE_PARM_DESC(large_router_buffers, "# of large messages to buffer in the router");
  51 static int peer_buffer_credits;
  52 module_param(peer_buffer_credits, int, 0444);
  53 MODULE_PARM_DESC(peer_buffer_credits, "# router buffer credits per peer");
  54
  55 static int auto_down = 1;
  56 module_param(auto_down, int, 0444);
  57 MODULE_PARM_DESC(auto_down, "Automatically mark peers down on comms error");
  58
  59 int
  60 lnet_peer_buffer_credits(struct lnet_net *net)
  61 {
  62         /* NI option overrides LNet default */
  63         if (net->net_tunables.lct_peer_rtr_credits > 0)
  64                 return net->net_tunables.lct_peer_rtr_credits;
  65         if (peer_buffer_credits > 0)
  66                 return peer_buffer_credits;
  67
  68         /* As an approximation, allow this peer the same number of router
  69          * buffers as it is allowed outstanding sends */
  70         return net->net_tunables.lct_peer_tx_credits;
  71 }
  72
  73 static int check_routers_before_use;
  74 module_param(check_routers_before_use, int, 0444);
  75 MODULE_PARM_DESC(check_routers_before_use, "Assume routers are down and ping them before use");
  76
  77 int avoid_asym_router_failure = 1;
  78 module_param(avoid_asym_router_failure, int, 0644);
  79 MODULE_PARM_DESC(avoid_asym_router_failure, "Avoid asymmetrical router failures (0 to disable)");
  80
  81 static int dead_router_check_interval = 60;
  82 module_param(dead_router_check_interval, int, 0644);
  83 MODULE_PARM_DESC(dead_router_check_interval, "Seconds between dead router health checks (<= 0 to disable)");
  84
  85 static int live_router_check_interval = 60;
  86 module_param(live_router_check_interval, int, 0644);
  87 MODULE_PARM_DESC(live_router_check_interval, "Seconds between live router health checks (<= 0 to disable)");
  88
  89 static int router_ping_timeout = 50;
  90 module_param(router_ping_timeout, int, 0644);
  91 MODULE_PARM_DESC(router_ping_timeout, "Seconds to wait for the reply to a router health query");
  92
  93 /*
  94  * A value between 0 and 100. 0 meaning that even if router's interfaces
  95  * have the worse health still consider the gateway usable.
  96  * 100 means that at least one interface on the route's remote net is 100%
  97  * healthy to consider the route alive.
  98  * The default is set to 100 to ensure we maintain the original behavior.
  99  */
 100 unsigned int router_sensitivity_percentage = 100;
 101 static int rtr_sensitivity_set(const char *val, cfs_kernel_param_arg_t *kp);
 102 static struct kernel_param_ops param_ops_rtr_sensitivity = {
 103         .set = rtr_sensitivity_set,
 104         .get = param_get_int,
 105 };
 106 #define param_check_rtr_sensitivity(name, p) \
 107                 __param_check(name, p, int)
 108 #ifdef HAVE_KERNEL_PARAM_OPS
 109 module_param(router_sensitivity_percentage, rtr_sensitivity, S_IRUGO|S_IWUSR);
 110 #else
 111 module_param_call(router_sensitivity_percentage, rtr_sensitivity_set, param_get_int,
 112                   &router_sensitivity_percentage, S_IRUGO|S_IWUSR);
 113 #endif
 114 MODULE_PARM_DESC(router_sensitivity_percentage,
 115                 "How healthy a gateway should be to be used in percent");
 116
 117 static int
 118 rtr_sensitivity_set(const char *val, cfs_kernel_param_arg_t *kp)
 119 {
 120         int rc;
 121         unsigned *sen = (unsigned *)kp->arg;
 122         unsigned long value;
 123
 124         rc = kstrtoul(val, 0, &value);
 125         if (rc) {
 126                 CERROR("Invalid module parameter value for 'router_sensitivity_percentage'\n");
 127                 return rc;
 128         }
 129
 130         if (value < 0 || value > 100) {
 131                 CERROR("Invalid value: %lu for 'router_sensitivity_percentage'\n", value);
 132                 return -EINVAL;
 133         }
 134
 135         /*
 136          * The purpose of locking the api_mutex here is to ensure that
 137          * the correct value ends up stored properly.
 138          */
 139         mutex_lock(&the_lnet.ln_api_mutex);
 140
 141         *sen = value;
 142
 143         mutex_unlock(&the_lnet.ln_api_mutex);
 144
 145         return 0;
 146 }
 147
 148 int
 149 lnet_peers_start_down(void)
 150 {
 151         return check_routers_before_use;
 152 }
 153
 154 /*
 155  * A net is alive if at least one gateway NI on the network is alive.
 156  */
 157 static bool
 158 lnet_is_gateway_net_alive(struct lnet_peer_net *lpn)
 159 {
 160         struct lnet_peer_ni *lpni;
 161
 162         list_for_each_entry(lpni, &lpn->lpn_peer_nis, lpni_peer_nis) {
 163                 if (lnet_is_peer_ni_alive(lpni))
 164                         return true;
 165         }
 166
 167         return false;
 168 }
 169
 170 /*
 171  * a gateway is alive only if all its nets are alive
 172  * called with cpt lock held
 173  */
 174 bool lnet_is_gateway_alive(struct lnet_peer *gw)
 175 {
 176         struct lnet_peer_net *lpn;
 177
 178         list_for_each_entry(lpn, &gw->lp_peer_nets, lpn_peer_nets) {
 179                 if (!lnet_is_gateway_net_alive(lpn))
 180                         return false;
 181         }
 182
 183         return true;
 184 }
 185
 186 /*
 187  * lnet_is_route_alive() needs to be called with cpt lock held
 188  * A route is alive if the gateway can route between the local network and
 189  * the remote network of the route.
 190  * This means at least one NI is alive on each of the local and remote
 191  * networks of the gateway.
 192  */
 193 bool lnet_is_route_alive(struct lnet_route *route)
 194 {
 195         struct lnet_peer *gw = route->lr_gateway;
 196         struct lnet_peer_net *llpn;
 197         struct lnet_peer_net *rlpn;
 198         bool route_alive;
 199
 200         /*
 201          * check the gateway's interfaces on the route rnet to make sure
 202          * that the gateway is viable.
 203          */
 204         llpn = lnet_peer_get_net_locked(gw, route->lr_lnet);
 205         if (!llpn)
 206                 return false;
 207
 208         route_alive = lnet_is_gateway_net_alive(llpn);
 209
 210         if (avoid_asym_router_failure) {
 211                 rlpn = lnet_peer_get_net_locked(gw, route->lr_net);
 212                 if (!rlpn)
 213                         return false;
 214                 route_alive = route_alive &&
 215                               lnet_is_gateway_net_alive(rlpn);
 216         }
 217
 218         if (!route_alive)
 219                 return route_alive;
 220
 221         spin_lock(&gw->lp_lock);
 222         if (!(gw->lp_state & LNET_PEER_ROUTER_ENABLED)) {
 223                 if (gw->lp_rtr_refcount > 0)
 224                         CERROR("peer %s is being used as a gateway but routing feature is not turned on\n",
 225                                libcfs_nid2str(gw->lp_primary_nid));
 226                 route_alive = false;
 227         }
 228         spin_unlock(&gw->lp_lock);
 229
 230         return route_alive;
 231 }
 232
 233 static void
 234 lnet_rtr_addref_locked(struct lnet_peer *lp)
 235 {
 236         LASSERT(lp->lp_rtr_refcount >= 0);
 237
 238         /* lnet_net_lock must be exclusively locked */
 239         lp->lp_rtr_refcount++;
 240         if (lp->lp_rtr_refcount == 1) {
 241                 list_add_tail(&lp->lp_rtr_list, &the_lnet.ln_routers);
 242                 /* addref for the_lnet.ln_routers */
 243                 lnet_peer_addref_locked(lp);
 244                 the_lnet.ln_routers_version++;
 245         }
 246 }
 247
 248 static void
 249 lnet_rtr_decref_locked(struct lnet_peer *lp)
 250 {
 251         LASSERT(atomic_read(&lp->lp_refcount) > 0);
 252         LASSERT(lp->lp_rtr_refcount > 0);
 253
 254         /* lnet_net_lock must be exclusively locked */
 255         lp->lp_rtr_refcount--;
 256         if (lp->lp_rtr_refcount == 0) {
 257                 LASSERT(list_empty(&lp->lp_routes));
 258
 259                 list_del(&lp->lp_rtr_list);
 260                 /* decref for the_lnet.ln_routers */
 261                 lnet_peer_decref_locked(lp);
 262                 the_lnet.ln_routers_version++;
 263         }
 264 }
 265
 266 struct lnet_remotenet *
 267 lnet_find_rnet_locked(__u32 net)
 268 {
 269         struct lnet_remotenet *rnet;
 270         struct list_head *tmp;
 271         struct list_head *rn_list;
 272
 273         LASSERT(the_lnet.ln_state == LNET_STATE_RUNNING);
 274
 275         rn_list = lnet_net2rnethash(net);
 276         list_for_each(tmp, rn_list) {
 277                 rnet = list_entry(tmp, struct lnet_remotenet, lrn_list);
 278
 279                 if (rnet->lrn_net == net)
 280                         return rnet;
 281         }
 282         return NULL;
 283 }
 284
 285 static void lnet_shuffle_seed(void)
 286 {
 287         static int seeded;
 288         struct lnet_ni *ni = NULL;
 289
 290         if (seeded)
 291                 return;
 292
 293         /* Nodes with small feet have little entropy
 294          * the NID for this node gives the most entropy in the low bits */
 295         while ((ni = lnet_get_next_ni_locked(NULL, ni)))
 296                 add_device_randomness(&ni->ni_nid, sizeof(ni->ni_nid));
 297
 298         seeded = 1;
 299         return;
 300 }
 301
 302 /* NB expects LNET_LOCK held */
 303 static void
 304 lnet_add_route_to_rnet(struct lnet_remotenet *rnet, struct lnet_route *route)
 305 {
 306         unsigned int len = 0;
 307         unsigned int offset = 0;
 308         struct list_head *e;
 309
 310         lnet_shuffle_seed();
 311
 312         list_for_each(e, &rnet->lrn_routes)
 313                 len++;
 314
 315         /*
 316          * Randomly adding routes to the list is done to ensure that when
 317          * different nodes are using the same list of routers, they end up
 318          * preferring different routers.
 319          */
 320         offset = cfs_rand() % (len + 1);
 321         list_for_each(e, &rnet->lrn_routes) {
 322                 if (offset == 0)
 323                         break;
 324                 offset--;
 325         }
 326         list_add(&route->lr_list, e);
 327         /*
 328          * force a router check on the gateway to make sure the route is
 329          * alive
 330          */
 331         route->lr_gateway->lp_rtrcheck_timestamp = 0;
 332
 333         the_lnet.ln_remote_nets_version++;
 334
 335         /* add the route on the gateway list */
 336         list_add(&route->lr_gwlist, &route->lr_gateway->lp_routes);
 337
 338         /* take a router reference count on the gateway */
 339         lnet_rtr_addref_locked(route->lr_gateway);
 340 }
 341
 342 int
 343 lnet_add_route(__u32 net, __u32 hops, lnet_nid_t gateway,
 344                unsigned int priority)
 345 {
 346         struct list_head *route_entry;
 347         struct lnet_remotenet *rnet;
 348         struct lnet_remotenet *rnet2;
 349         struct lnet_route *route;
 350         struct lnet_peer_ni *lpni;
 351         struct lnet_peer *gw;
 352         int add_route;
 353         int rc;
 354
 355         CDEBUG(D_NET, "Add route: remote net %s hops %d priority %u gw %s\n",
 356                libcfs_net2str(net), hops, priority, libcfs_nid2str(gateway));
 357
 358         if (gateway == LNET_NID_ANY ||
 359             LNET_NETTYP(LNET_NIDNET(gateway)) == LOLND ||
 360             net == LNET_NIDNET(LNET_NID_ANY) ||
 361             LNET_NETTYP(net) == LOLND ||
 362             LNET_NIDNET(gateway) == net ||
 363             (hops != LNET_UNDEFINED_HOPS && (hops < 1 || hops > 255)))
 364                 return -EINVAL;
 365
 366         /* it's a local network */
 367         if (lnet_islocalnet(net))
 368                 return -EEXIST;
 369
 370         /* Assume net, route, all new */
 371         LIBCFS_ALLOC(route, sizeof(*route));
 372         LIBCFS_ALLOC(rnet, sizeof(*rnet));
 373         if (route == NULL || rnet == NULL) {
 374                 CERROR("Out of memory creating route %s %d %s\n",
 375                        libcfs_net2str(net), hops, libcfs_nid2str(gateway));
 376                 if (route != NULL)
 377                         LIBCFS_FREE(route, sizeof(*route));
 378                 if (rnet != NULL)
 379                         LIBCFS_FREE(rnet, sizeof(*rnet));
 380                 return -ENOMEM;
 381         }
 382
 383         INIT_LIST_HEAD(&rnet->lrn_routes);
 384         rnet->lrn_net = net;
 385         /* store the local and remote net that the route represents */
 386         route->lr_lnet = LNET_NIDNET(gateway);
 387         route->lr_net = net;
 388         route->lr_priority = priority;
 389         route->lr_hops = hops;
 390
 391         lnet_net_lock(LNET_LOCK_EX);
 392
 393         /*
 394          * lnet_nid2peerni_ex() grabs a ref on the lpni. We will need to
 395          * lose that once we're done
 396          */
 397         lpni = lnet_nid2peerni_ex(gateway, LNET_LOCK_EX);
 398         if (IS_ERR(lpni)) {
 399                 lnet_net_unlock(LNET_LOCK_EX);
 400
 401                 LIBCFS_FREE(route, sizeof(*route));
 402                 LIBCFS_FREE(rnet, sizeof(*rnet));
 403
 404                 rc = PTR_ERR(lpni);
 405                 CERROR("Error %d creating route %s %d %s\n", rc,
 406                         libcfs_net2str(net), hops,
 407                         libcfs_nid2str(gateway));
 408                 return rc;
 409         }
 410
 411         LASSERT(lpni->lpni_peer_net && lpni->lpni_peer_net->lpn_peer);
 412         gw = lpni->lpni_peer_net->lpn_peer;
 413
 414         route->lr_gateway = gw;
 415
 416         rnet2 = lnet_find_rnet_locked(net);
 417         if (rnet2 == NULL) {
 418                 /* new network */
 419                 list_add_tail(&rnet->lrn_list, lnet_net2rnethash(net));
 420                 rnet2 = rnet;
 421         }
 422
 423         /* Search for a duplicate route (it's a NOOP if it is) */
 424         add_route = 1;
 425         list_for_each(route_entry, &rnet2->lrn_routes) {
 426                 struct lnet_route *route2;
 427
 428                 route2 = list_entry(route_entry, struct lnet_route, lr_list);
 429                 if (route2->lr_gateway == route->lr_gateway) {
 430                         add_route = 0;
 431                         break;
 432                 }
 433
 434                 /* our lookups must be true */
 435                 LASSERT(route2->lr_gateway->lp_primary_nid != gateway);
 436         }
 437
 438         /*
 439          * It is possible to add multiple routes through the same peer,
 440          * but it'll be using a different NID of that peer. When the
 441          * gateway is discovered, discovery will consolidate the different
 442          * peers into one peer. In this case the discovery code will have
 443          * to move the routes from the peer that's being deleted to the
 444          * consolidated peer lp_routes list
 445          */
 446         if (add_route)
 447                 lnet_add_route_to_rnet(rnet2, route);
 448
 449         /*
 450          * get rid of the reference on the lpni.
 451          */
 452         lnet_peer_ni_decref_locked(lpni);
 453         lnet_net_unlock(LNET_LOCK_EX);
 454
 455         rc = 0;
 456
 457         if (!add_route) {
 458                 rc = -EEXIST;
 459                 LIBCFS_FREE(route, sizeof(*route));
 460         }
 461
 462         if (rnet != rnet2)
 463                 LIBCFS_FREE(rnet, sizeof(*rnet));
 464
 465         /* kick start the monitor thread to handle the added route */
 466         wake_up(&the_lnet.ln_mt_waitq);
 467
 468         return rc;
 469 }
 470
 471 static void
 472 lnet_del_route_from_rnet(lnet_nid_t gw_nid, struct list_head *route_list,
 473                          struct list_head *zombies)
 474 {
 475         struct lnet_peer *gateway;
 476         struct lnet_route *route;
 477         struct lnet_route *tmp;
 478
 479         list_for_each_entry_safe(route, tmp, route_list, lr_list) {
 480                 gateway = route->lr_gateway;
 481                 if (gw_nid != LNET_NID_ANY &&
 482                     gw_nid != gateway->lp_primary_nid)
 483                         continue;
 484
 485                 /*
 486                  * move to zombie to delete outside the lock
 487                  * Note that this function is called with the
 488                  * ln_api_mutex held as well as the exclusive net
 489                  * lock. Adding to the remote net list happens
 490                  * under the same conditions. Same goes for the
 491                  * gateway router list
 492                  */
 493                 list_move(&route->lr_list, zombies);
 494                 the_lnet.ln_remote_nets_version++;
 495
 496                 list_del(&route->lr_gwlist);
 497                 lnet_rtr_decref_locked(gateway);
 498         }
 499 }
 500
 501 int
 502 lnet_del_route(__u32 net, lnet_nid_t gw_nid)
 503 {
 504         struct list_head rnet_zombies;
 505         struct lnet_remotenet *rnet;
 506         struct lnet_remotenet *tmp;
 507         struct list_head *rn_list;
 508         struct lnet_peer_ni *lpni;
 509         struct lnet_route *route;
 510         struct list_head zombies;
 511         struct lnet_peer *lp;
 512         int i = 0;
 513
 514         INIT_LIST_HEAD(&rnet_zombies);
 515         INIT_LIST_HEAD(&zombies);
 516
 517         CDEBUG(D_NET, "Del route: net %s : gw %s\n",
 518                libcfs_net2str(net), libcfs_nid2str(gw_nid));
 519
 520         /* NB Caller may specify either all routes via the given gateway
 521          * or a specific route entry actual NIDs) */
 522
 523         lnet_net_lock(LNET_LOCK_EX);
 524
 525         lpni = lnet_find_peer_ni_locked(gw_nid);
 526         if (lpni) {
 527                 lp = lpni->lpni_peer_net->lpn_peer;
 528                 LASSERT(lp);
 529                 gw_nid = lp->lp_primary_nid;
 530                 lnet_peer_ni_decref_locked(lpni);
 531         }
 532
 533         if (net != LNET_NIDNET(LNET_NID_ANY)) {
 534                 rnet = lnet_find_rnet_locked(net);
 535                 if (!rnet) {
 536                         lnet_net_unlock(LNET_LOCK_EX);
 537                         return -ENOENT;
 538                 }
 539                 lnet_del_route_from_rnet(gw_nid, &rnet->lrn_routes,
 540                                          &zombies);
 541                 if (list_empty(&rnet->lrn_routes))
 542                         list_move(&rnet->lrn_list, &rnet_zombies);
 543                 goto delete_zombies;
 544         }
 545
 546         for (i = 0; i < LNET_REMOTE_NETS_HASH_SIZE; i++) {
 547                 rn_list = &the_lnet.ln_remote_nets_hash[i];
 548
 549                 list_for_each_entry_safe(rnet, tmp, rn_list, lrn_list) {
 550                         lnet_del_route_from_rnet(gw_nid, &rnet->lrn_routes,
 551                                                  &zombies);
 552                         if (list_empty(&rnet->lrn_routes))
 553                                 list_move(&rnet->lrn_list, &rnet_zombies);
 554                 }
 555         }
 556
 557 delete_zombies:
 558         lnet_net_unlock(LNET_LOCK_EX);
 559
 560         while (!list_empty(&zombies)) {
 561                 route = list_first_entry(&zombies, struct lnet_route, lr_list);
 562                 list_del(&route->lr_list);
 563                 LIBCFS_FREE(route, sizeof(*route));
 564         }
 565
 566         while (!list_empty(&rnet_zombies)) {
 567                 rnet = list_first_entry(&rnet_zombies, struct lnet_remotenet,
 568                                         lrn_list);
 569                 list_del(&rnet->lrn_list);
 570                 LIBCFS_FREE(rnet, sizeof(*rnet));
 571         }
 572
 573         return 0;
 574 }
 575
 576 void
 577 lnet_destroy_routes (void)
 578 {
 579         lnet_del_route(LNET_NIDNET(LNET_NID_ANY), LNET_NID_ANY);
 580 }
 581
 582 int lnet_get_rtr_pool_cfg(int cpt, struct lnet_ioctl_pool_cfg *pool_cfg)
 583 {
 584         struct lnet_rtrbufpool *rbp;
 585         int i, rc = -ENOENT, j;
 586
 587         if (the_lnet.ln_rtrpools == NULL)
 588                 return rc;
 589
 590
 591         cfs_percpt_for_each(rbp, i, the_lnet.ln_rtrpools) {
 592                 if (i != cpt)
 593                         continue;
 594
 595                 lnet_net_lock(i);
 596                 for (j = 0; j < LNET_NRBPOOLS; j++) {
 597                         pool_cfg->pl_pools[j].pl_npages = rbp[j].rbp_npages;
 598                         pool_cfg->pl_pools[j].pl_nbuffers = rbp[j].rbp_nbuffers;
 599                         pool_cfg->pl_pools[j].pl_credits = rbp[j].rbp_credits;
 600                         pool_cfg->pl_pools[j].pl_mincredits = rbp[j].rbp_mincredits;
 601                 }
 602                 lnet_net_unlock(i);
 603                 rc = 0;
 604                 break;
 605         }
 606
 607         lnet_net_lock(LNET_LOCK_EX);
 608         pool_cfg->pl_routing = the_lnet.ln_routing;
 609         lnet_net_unlock(LNET_LOCK_EX);
 610
 611         return rc;
 612 }
 613
 614 int
 615 lnet_get_route(int idx, __u32 *net, __u32 *hops,
 616                lnet_nid_t *gateway, __u32 *alive, __u32 *priority)
 617 {
 618         struct list_head *e1;
 619         struct list_head *e2;
 620         struct lnet_remotenet *rnet;
 621         struct lnet_route        *route;
 622         int               cpt;
 623         int               i;
 624         struct list_head *rn_list;
 625
 626         cpt = lnet_net_lock_current();
 627
 628         for (i = 0; i < LNET_REMOTE_NETS_HASH_SIZE; i++) {
 629                 rn_list = &the_lnet.ln_remote_nets_hash[i];
 630                 list_for_each(e1, rn_list) {
 631                         rnet = list_entry(e1, struct lnet_remotenet, lrn_list);
 632
 633                         list_for_each(e2, &rnet->lrn_routes) {
 634                                 route = list_entry(e2, struct lnet_route,
 635                                                    lr_list);
 636
 637                                 if (idx-- == 0) {
 638                                         *net      = rnet->lrn_net;
 639                                         *hops     = route->lr_hops;
 640                                         *priority = route->lr_priority;
 641                                         *gateway  = route->lr_gateway->lp_primary_nid;
 642                                         *alive    = lnet_is_route_alive(route);
 643                                         lnet_net_unlock(cpt);
 644                                         return 0;
 645                                 }
 646                         }
 647                 }
 648         }
 649
 650         lnet_net_unlock(cpt);
 651         return -ENOENT;
 652 }
 653
 654 static void
 655 lnet_wait_known_routerstate(void)
 656 {
 657         struct lnet_peer *rtr;
 658         struct list_head *entry;
 659         int all_known;
 660
 661         LASSERT(the_lnet.ln_mt_state == LNET_MT_STATE_RUNNING);
 662
 663         for (;;) {
 664                 int cpt = lnet_net_lock_current();
 665
 666                 all_known = 1;
 667                 list_for_each(entry, &the_lnet.ln_routers) {
 668                         rtr = list_entry(entry, struct lnet_peer,
 669                                          lp_rtr_list);
 670
 671                         spin_lock(&rtr->lp_lock);
 672
 673                         if ((rtr->lp_state & LNET_PEER_DISCOVERED) == 0) {
 674                                 all_known = 0;
 675                                 spin_unlock(&rtr->lp_lock);
 676                                 break;
 677                         }
 678                         spin_unlock(&rtr->lp_lock);
 679                 }
 680
 681                 lnet_net_unlock(cpt);
 682
 683                 if (all_known)
 684                         return;
 685
 686                 set_current_state(TASK_UNINTERRUPTIBLE);
 687                 schedule_timeout(cfs_time_seconds(1));
 688         }
 689 }
 690
 691 static void
 692 lnet_update_ni_status_locked(void)
 693 {
 694         struct lnet_ni *ni = NULL;
 695         time64_t now;
 696         time64_t timeout;
 697
 698         LASSERT(the_lnet.ln_routing);
 699
 700         timeout = router_ping_timeout +
 701                   MAX(live_router_check_interval, dead_router_check_interval);
 702
 703         now = ktime_get_real_seconds();
 704         while ((ni = lnet_get_next_ni_locked(NULL, ni))) {
 705                 if (ni->ni_net->net_lnd->lnd_type == LOLND)
 706                         continue;
 707
 708                 if (now < ni->ni_last_alive + timeout)
 709                         continue;
 710
 711                 lnet_ni_lock(ni);
 712                 /* re-check with lock */
 713                 if (now < ni->ni_last_alive + timeout) {
 714                         lnet_ni_unlock(ni);
 715                         continue;
 716                 }
 717
 718                 LASSERT(ni->ni_status != NULL);
 719
 720                 if (ni->ni_status->ns_status != LNET_NI_STATUS_DOWN) {
 721                         CDEBUG(D_NET, "NI(%s:%lld) status changed to down\n",
 722                                libcfs_nid2str(ni->ni_nid), timeout);
 723                         /* NB: so far, this is the only place to set
 724                          * NI status to "down" */
 725                         ni->ni_status->ns_status = LNET_NI_STATUS_DOWN;
 726                 }
 727                 lnet_ni_unlock(ni);
 728         }
 729 }
 730
 731 void lnet_router_post_mt_start(void)
 732 {
 733         if (check_routers_before_use) {
 734                 /* Note that a helpful side-effect of pinging all known routers
 735                  * at startup is that it makes them drop stale connections they
 736                  * may have to a previous instance of me. */
 737                 lnet_wait_known_routerstate();
 738         }
 739 }
 740
 741 /*
 742  * This function is called from the monitor thread to check if there are
 743  * any active routers that need to be checked.
 744  */
 745 inline bool
 746 lnet_router_checker_active(void)
 747 {
 748         if (the_lnet.ln_mt_state != LNET_MT_STATE_RUNNING)
 749                 return true;
 750
 751         /* Router Checker thread needs to run when routing is enabled in
 752          * order to call lnet_update_ni_status_locked() */
 753         if (the_lnet.ln_routing)
 754                 return true;
 755
 756         return !list_empty(&the_lnet.ln_routers) &&
 757                 (live_router_check_interval > 0 ||
 758                  dead_router_check_interval > 0);
 759 }
 760
 761 void
 762 lnet_check_routers(void)
 763 {
 764         struct lnet_peer *rtr;
 765         struct list_head *entry;
 766         __u64   version;
 767         int     cpt;
 768
 769         cpt = lnet_net_lock_current();
 770 rescan:
 771         version = the_lnet.ln_routers_version;
 772
 773         list_for_each(entry, &the_lnet.ln_routers) {
 774                 rtr = list_entry(entry, struct lnet_peer,
 775                                  lp_rtr_list);
 776
 777                 /* TODO use discovery to determine if router is alive */
 778
 779                 /* NB dropped lock */
 780                 if (version != the_lnet.ln_routers_version) {
 781                         /* the routers list has changed */
 782                         goto rescan;
 783                 }
 784         }
 785
 786         if (the_lnet.ln_routing)
 787                 lnet_update_ni_status_locked();
 788
 789         lnet_net_unlock(cpt);
 790 }
 791
 792 void
 793 lnet_destroy_rtrbuf(struct lnet_rtrbuf *rb, int npages)
 794 {
 795         int sz = offsetof(struct lnet_rtrbuf, rb_kiov[npages]);
 796
 797         while (--npages >= 0)
 798                 __free_page(rb->rb_kiov[npages].kiov_page);
 799
 800         LIBCFS_FREE(rb, sz);
 801 }
 802
 803 static struct lnet_rtrbuf *
 804 lnet_new_rtrbuf(struct lnet_rtrbufpool *rbp, int cpt)
 805 {
 806         int            npages = rbp->rbp_npages;
 807         int            sz = offsetof(struct lnet_rtrbuf, rb_kiov[npages]);
 808         struct page   *page;
 809         struct lnet_rtrbuf *rb;
 810         int            i;
 811
 812         LIBCFS_CPT_ALLOC(rb, lnet_cpt_table(), cpt, sz);
 813         if (rb == NULL)
 814                 return NULL;
 815
 816         rb->rb_pool = rbp;
 817
 818         for (i = 0; i < npages; i++) {
 819                 page = cfs_page_cpt_alloc(lnet_cpt_table(), cpt,
 820                                           GFP_KERNEL | __GFP_ZERO);
 821                 if (page == NULL) {
 822                         while (--i >= 0)
 823                                 __free_page(rb->rb_kiov[i].kiov_page);
 824
 825                         LIBCFS_FREE(rb, sz);
 826                         return NULL;
 827                 }
 828
 829                 rb->rb_kiov[i].kiov_len = PAGE_SIZE;
 830                 rb->rb_kiov[i].kiov_offset = 0;
 831                 rb->rb_kiov[i].kiov_page = page;
 832         }
 833
 834         return rb;
 835 }
 836
 837 static void
 838 lnet_rtrpool_free_bufs(struct lnet_rtrbufpool *rbp, int cpt)
 839 {
 840         int npages = rbp->rbp_npages;
 841         struct lnet_rtrbuf *rb;
 842         struct list_head tmp;
 843
 844         if (rbp->rbp_nbuffers == 0) /* not initialized or already freed */
 845                 return;
 846
 847         INIT_LIST_HEAD(&tmp);
 848
 849         lnet_net_lock(cpt);
 850         list_splice_init(&rbp->rbp_msgs, &tmp);
 851         lnet_drop_routed_msgs_locked(&tmp, cpt);
 852         list_splice_init(&rbp->rbp_bufs, &tmp);
 853         rbp->rbp_req_nbuffers = 0;
 854         rbp->rbp_nbuffers = rbp->rbp_credits = 0;
 855         rbp->rbp_mincredits = 0;
 856         lnet_net_unlock(cpt);
 857
 858         /* Free buffers on the free list. */
 859         while (!list_empty(&tmp)) {
 860                 rb = list_entry(tmp.next, struct lnet_rtrbuf, rb_list);
 861                 list_del(&rb->rb_list);
 862                 lnet_destroy_rtrbuf(rb, npages);
 863         }
 864 }
 865
 866 static int
 867 lnet_rtrpool_adjust_bufs(struct lnet_rtrbufpool *rbp, int nbufs, int cpt)
 868 {
 869         struct list_head rb_list;
 870         struct lnet_rtrbuf *rb;
 871         int             num_rb;
 872         int             num_buffers = 0;
 873         int             old_req_nbufs;
 874         int             npages = rbp->rbp_npages;
 875
 876         lnet_net_lock(cpt);
 877         /* If we are called for less buffers than already in the pool, we
 878          * just lower the req_nbuffers number and excess buffers will be
 879          * thrown away as they are returned to the free list.  Credits
 880          * then get adjusted as well.
 881          * If we already have enough buffers allocated to serve the
 882          * increase requested, then we can treat that the same way as we
 883          * do the decrease. */
 884         num_rb = nbufs - rbp->rbp_nbuffers;
 885         if (nbufs <= rbp->rbp_req_nbuffers || num_rb <= 0) {
 886                 rbp->rbp_req_nbuffers = nbufs;
 887                 lnet_net_unlock(cpt);
 888                 return 0;
 889         }
 890         /* store the older value of rbp_req_nbuffers and then set it to
 891          * the new request to prevent lnet_return_rx_credits_locked() from
 892          * freeing buffers that we need to keep around */
 893         old_req_nbufs = rbp->rbp_req_nbuffers;
 894         rbp->rbp_req_nbuffers = nbufs;
 895         lnet_net_unlock(cpt);
 896
 897         INIT_LIST_HEAD(&rb_list);
 898
 899         /* allocate the buffers on a local list first.  If all buffers are
 900          * allocated successfully then join this list to the rbp buffer
 901          * list.  If not then free all allocated buffers. */
 902         while (num_rb-- > 0) {
 903                 rb = lnet_new_rtrbuf(rbp, cpt);
 904                 if (rb == NULL) {
 905                         CERROR("Failed to allocate %d route bufs of %d pages\n",
 906                                nbufs, npages);
 907
 908                         lnet_net_lock(cpt);
 909                         rbp->rbp_req_nbuffers = old_req_nbufs;
 910                         lnet_net_unlock(cpt);
 911
 912                         goto failed;
 913                 }
 914
 915                 list_add(&rb->rb_list, &rb_list);
 916                 num_buffers++;
 917         }
 918
 919         lnet_net_lock(cpt);
 920
 921         list_splice_tail(&rb_list, &rbp->rbp_bufs);
 922         rbp->rbp_nbuffers += num_buffers;
 923         rbp->rbp_credits += num_buffers;
 924         rbp->rbp_mincredits = rbp->rbp_credits;
 925         /* We need to schedule blocked msg using the newly
 926          * added buffers. */
 927         while (!list_empty(&rbp->rbp_bufs) &&
 928                !list_empty(&rbp->rbp_msgs))
 929                 lnet_schedule_blocked_locked(rbp);
 930
 931         lnet_net_unlock(cpt);
 932
 933         return 0;
 934
 935 failed:
 936         while (!list_empty(&rb_list)) {
 937                 rb = list_entry(rb_list.next, struct lnet_rtrbuf, rb_list);
 938                 list_del(&rb->rb_list);
 939                 lnet_destroy_rtrbuf(rb, npages);
 940         }
 941
 942         return -ENOMEM;
 943 }
 944
 945 static void
 946 lnet_rtrpool_init(struct lnet_rtrbufpool *rbp, int npages)
 947 {
 948         INIT_LIST_HEAD(&rbp->rbp_msgs);
 949         INIT_LIST_HEAD(&rbp->rbp_bufs);
 950
 951         rbp->rbp_npages = npages;
 952         rbp->rbp_credits = 0;
 953         rbp->rbp_mincredits = 0;
 954 }
 955
 956 void
 957 lnet_rtrpools_free(int keep_pools)
 958 {
 959         struct lnet_rtrbufpool *rtrp;
 960         int               i;
 961
 962         if (the_lnet.ln_rtrpools == NULL) /* uninitialized or freed */
 963                 return;
 964
 965         cfs_percpt_for_each(rtrp, i, the_lnet.ln_rtrpools) {
 966                 lnet_rtrpool_free_bufs(&rtrp[LNET_TINY_BUF_IDX], i);
 967                 lnet_rtrpool_free_bufs(&rtrp[LNET_SMALL_BUF_IDX], i);
 968                 lnet_rtrpool_free_bufs(&rtrp[LNET_LARGE_BUF_IDX], i);
 969         }
 970
 971         if (!keep_pools) {
 972                 cfs_percpt_free(the_lnet.ln_rtrpools);
 973                 the_lnet.ln_rtrpools = NULL;
 974         }
 975 }
 976
 977 static int
 978 lnet_nrb_tiny_calculate(void)
 979 {
 980         int     nrbs = LNET_NRB_TINY;
 981
 982         if (tiny_router_buffers < 0) {
 983                 LCONSOLE_ERROR_MSG(0x10c,
 984                                    "tiny_router_buffers=%d invalid when "
 985                                    "routing enabled\n", tiny_router_buffers);
 986                 return -EINVAL;
 987         }
 988
 989         if (tiny_router_buffers > 0)
 990                 nrbs = tiny_router_buffers;
 991
 992         nrbs /= LNET_CPT_NUMBER;
 993         return max(nrbs, LNET_NRB_TINY_MIN);
 994 }
 995
 996 static int
 997 lnet_nrb_small_calculate(void)
 998 {
 999         int     nrbs = LNET_NRB_SMALL;
1000
1001         if (small_router_buffers < 0) {
1002                 LCONSOLE_ERROR_MSG(0x10c,
1003                                    "small_router_buffers=%d invalid when "
1004                                    "routing enabled\n", small_router_buffers);
1005                 return -EINVAL;
1006         }
1007
1008         if (small_router_buffers > 0)
1009                 nrbs = small_router_buffers;
1010
1011         nrbs /= LNET_CPT_NUMBER;
1012         return max(nrbs, LNET_NRB_SMALL_MIN);
1013 }
1014
1015 static int
1016 lnet_nrb_large_calculate(void)
1017 {
1018         int     nrbs = LNET_NRB_LARGE;
1019
1020         if (large_router_buffers < 0) {
1021                 LCONSOLE_ERROR_MSG(0x10c,
1022                                    "large_router_buffers=%d invalid when "
1023                                    "routing enabled\n", large_router_buffers);
1024                 return -EINVAL;
1025         }
1026
1027         if (large_router_buffers > 0)
1028                 nrbs = large_router_buffers;
1029
1030         nrbs /= LNET_CPT_NUMBER;
1031         return max(nrbs, LNET_NRB_LARGE_MIN);
1032 }
1033
1034 int
1035 lnet_rtrpools_alloc(int im_a_router)
1036 {
1037         struct lnet_rtrbufpool *rtrp;
1038         int     nrb_tiny;
1039         int     nrb_small;
1040         int     nrb_large;
1041         int     rc;
1042         int     i;
1043
1044         if (!strcmp(forwarding, "")) {
1045                 /* not set either way */
1046                 if (!im_a_router)
1047                         return 0;
1048         } else if (!strcmp(forwarding, "disabled")) {
1049                 /* explicitly disabled */
1050                 return 0;
1051         } else if (!strcmp(forwarding, "enabled")) {
1052                 /* explicitly enabled */
1053         } else {
1054                 LCONSOLE_ERROR_MSG(0x10b, "'forwarding' not set to either "
1055                                    "'enabled' or 'disabled'\n");
1056                 return -EINVAL;
1057         }
1058
1059         nrb_tiny = lnet_nrb_tiny_calculate();
1060         if (nrb_tiny < 0)
1061                 return -EINVAL;
1062
1063         nrb_small = lnet_nrb_small_calculate();
1064         if (nrb_small < 0)
1065                 return -EINVAL;
1066
1067         nrb_large = lnet_nrb_large_calculate();
1068         if (nrb_large < 0)
1069                 return -EINVAL;
1070
1071         the_lnet.ln_rtrpools = cfs_percpt_alloc(lnet_cpt_table(),
1072                                                 LNET_NRBPOOLS *
1073                                                 sizeof(struct lnet_rtrbufpool));
1074         if (the_lnet.ln_rtrpools == NULL) {
1075                 LCONSOLE_ERROR_MSG(0x10c,
1076                                    "Failed to initialize router buffe pool\n");
1077                 return -ENOMEM;
1078         }
1079
1080         cfs_percpt_for_each(rtrp, i, the_lnet.ln_rtrpools) {
1081                 lnet_rtrpool_init(&rtrp[LNET_TINY_BUF_IDX], 0);
1082                 rc = lnet_rtrpool_adjust_bufs(&rtrp[LNET_TINY_BUF_IDX],
1083                                               nrb_tiny, i);
1084                 if (rc != 0)
1085                         goto failed;
1086
1087                 lnet_rtrpool_init(&rtrp[LNET_SMALL_BUF_IDX],
1088                                   LNET_NRB_SMALL_PAGES);
1089                 rc = lnet_rtrpool_adjust_bufs(&rtrp[LNET_SMALL_BUF_IDX],
1090                                               nrb_small, i);
1091                 if (rc != 0)
1092                         goto failed;
1093
1094                 lnet_rtrpool_init(&rtrp[LNET_LARGE_BUF_IDX],
1095                                   LNET_NRB_LARGE_PAGES);
1096                 rc = lnet_rtrpool_adjust_bufs(&rtrp[LNET_LARGE_BUF_IDX],
1097                                               nrb_large, i);
1098                 if (rc != 0)
1099                         goto failed;
1100         }
1101
1102         lnet_net_lock(LNET_LOCK_EX);
1103         the_lnet.ln_routing = 1;
1104         lnet_net_unlock(LNET_LOCK_EX);
1105         wake_up(&the_lnet.ln_mt_waitq);
1106         return 0;
1107
1108  failed:
1109         lnet_rtrpools_free(0);
1110         return rc;
1111 }
1112
1113 static int
1114 lnet_rtrpools_adjust_helper(int tiny, int small, int large)
1115 {
1116         int nrb = 0;
1117         int rc = 0;
1118         int i;
1119         struct lnet_rtrbufpool *rtrp;
1120
1121         /* If the provided values for each buffer pool are different than the
1122          * configured values, we need to take action. */
1123         if (tiny >= 0) {
1124                 tiny_router_buffers = tiny;
1125                 nrb = lnet_nrb_tiny_calculate();
1126                 cfs_percpt_for_each(rtrp, i, the_lnet.ln_rtrpools) {
1127                         rc = lnet_rtrpool_adjust_bufs(&rtrp[LNET_TINY_BUF_IDX],
1128                                                       nrb, i);
1129                         if (rc != 0)
1130                                 return rc;
1131                 }
1132         }
1133         if (small >= 0) {
1134                 small_router_buffers = small;
1135                 nrb = lnet_nrb_small_calculate();
1136                 cfs_percpt_for_each(rtrp, i, the_lnet.ln_rtrpools) {
1137                         rc = lnet_rtrpool_adjust_bufs(&rtrp[LNET_SMALL_BUF_IDX],
1138                                                       nrb, i);
1139                         if (rc != 0)
1140                                 return rc;
1141                 }
1142         }
1143         if (large >= 0) {
1144                 large_router_buffers = large;
1145                 nrb = lnet_nrb_large_calculate();
1146                 cfs_percpt_for_each(rtrp, i, the_lnet.ln_rtrpools) {
1147                         rc = lnet_rtrpool_adjust_bufs(&rtrp[LNET_LARGE_BUF_IDX],
1148                                                       nrb, i);
1149                         if (rc != 0)
1150                                 return rc;
1151                 }
1152         }
1153
1154         return 0;
1155 }
1156
1157 int
1158 lnet_rtrpools_adjust(int tiny, int small, int large)
1159 {
1160         /* this function doesn't revert the changes if adding new buffers
1161          * failed.  It's up to the user space caller to revert the
1162          * changes. */
1163
1164         if (!the_lnet.ln_routing)
1165                 return 0;
1166
1167         return lnet_rtrpools_adjust_helper(tiny, small, large);
1168 }
1169
1170 int
1171 lnet_rtrpools_enable(void)
1172 {
1173         int rc = 0;
1174
1175         if (the_lnet.ln_routing)
1176                 return 0;
1177
1178         if (the_lnet.ln_rtrpools == NULL)
1179                 /* If routing is turned off, and we have never
1180                  * initialized the pools before, just call the
1181                  * standard buffer pool allocation routine as
1182                  * if we are just configuring this for the first
1183                  * time. */
1184                 rc = lnet_rtrpools_alloc(1);
1185         else
1186                 rc = lnet_rtrpools_adjust_helper(0, 0, 0);
1187         if (rc != 0)
1188                 return rc;
1189
1190         lnet_net_lock(LNET_LOCK_EX);
1191         the_lnet.ln_routing = 1;
1192
1193         the_lnet.ln_ping_target->pb_info.pi_features &=
1194                 ~LNET_PING_FEAT_RTE_DISABLED;
1195         lnet_net_unlock(LNET_LOCK_EX);
1196
1197         return rc;
1198 }
1199
1200 void
1201 lnet_rtrpools_disable(void)
1202 {
1203         if (!the_lnet.ln_routing)
1204                 return;
1205
1206         lnet_net_lock(LNET_LOCK_EX);
1207         the_lnet.ln_routing = 0;
1208         the_lnet.ln_ping_target->pb_info.pi_features |=
1209                 LNET_PING_FEAT_RTE_DISABLED;
1210
1211         tiny_router_buffers = 0;
1212         small_router_buffers = 0;
1213         large_router_buffers = 0;
1214         lnet_net_unlock(LNET_LOCK_EX);
1215         lnet_rtrpools_free(1);
1216 }
1217
1218 int
1219 lnet_notify(struct lnet_ni *ni, lnet_nid_t nid, int alive, time64_t when)
1220 {
1221         struct lnet_peer_ni *lp = NULL;
1222         time64_t now = ktime_get_seconds();
1223         int cpt = lnet_cpt_of_nid(nid, ni);
1224
1225         LASSERT (!in_interrupt ());
1226
1227         CDEBUG (D_NET, "%s notifying %s: %s\n",
1228                 (ni == NULL) ? "userspace" : libcfs_nid2str(ni->ni_nid),
1229                 libcfs_nid2str(nid),
1230                 alive ? "up" : "down");
1231
1232         if (ni != NULL &&
1233             LNET_NIDNET(ni->ni_nid) != LNET_NIDNET(nid)) {
1234                 CWARN("Ignoring notification of %s %s by %s (different net)\n",
1235                       libcfs_nid2str(nid), alive ? "birth" : "death",
1236                       libcfs_nid2str(ni->ni_nid));
1237                 return -EINVAL;
1238         }
1239
1240         /* can't do predictions... */
1241         if (when > now) {
1242                 CWARN("Ignoring prediction from %s of %s %s "
1243                       "%lld seconds in the future\n",
1244                       (ni == NULL) ? "userspace" : libcfs_nid2str(ni->ni_nid),
1245                       libcfs_nid2str(nid), alive ? "up" : "down", when - now);
1246                 return -EINVAL;
1247         }
1248
1249         if (ni != NULL && !alive &&             /* LND telling me she's down */
1250             !auto_down) {                       /* auto-down disabled */
1251                 CDEBUG(D_NET, "Auto-down disabled\n");
1252                 return 0;
1253         }
1254
1255         lnet_net_lock(cpt);
1256
1257         if (the_lnet.ln_state != LNET_STATE_RUNNING) {
1258                 lnet_net_unlock(cpt);
1259                 return -ESHUTDOWN;
1260         }
1261
1262         lp = lnet_find_peer_ni_locked(nid);
1263         if (lp == NULL) {
1264                 /* nid not found */
1265                 lnet_net_unlock(cpt);
1266                 CDEBUG(D_NET, "%s not found\n", libcfs_nid2str(nid));
1267                 return 0;
1268         }
1269
1270         /*
1271          * It is possible for this function to be called for the same peer
1272          * but with different NIs. We want to synchronize the notification
1273          * between the different calls. So we will use the lpni_cpt to
1274          * grab the net lock.
1275          */
1276         if (lp->lpni_cpt != cpt) {
1277                 lnet_net_unlock(cpt);
1278                 cpt = lp->lpni_cpt;
1279                 lnet_net_lock(cpt);
1280         }
1281
1282         lnet_peer_ni_decref_locked(lp);
1283
1284         lnet_net_unlock(cpt);
1285         return 0;
1286 }
1287 EXPORT_SYMBOL(lnet_notify);