lnet/lnet/peer.c

   1 /*
   2  * GPL HEADER START
   3  *
   4  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
   5  *
   6  * This program is free software; you can redistribute it and/or modify
   7  * it under the terms of the GNU General Public License version 2 only,
   8  * as published by the Free Software Foundation.
   9  *
  10  * This program is distributed in the hope that it will be useful, but
  11  * WITHOUT ANY WARRANTY; without even the implied warranty of
  12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  13  * General Public License version 2 for more details (a copy is included
  14  * in the LICENSE file that accompanied this code).
  15  *
  16  * You should have received a copy of the GNU General Public License
  17  * version 2 along with this program; If not, see
  18  * http://www.gnu.org/licenses/gpl-2.0.html
  19  *
  20  * GPL HEADER END
  21  */
  22 /*
  23  * Copyright (c) 2007, 2010, Oracle and/or its affiliates. All rights reserved.
  24  * Use is subject to license terms.
  25  *
  26  * Copyright (c) 2012, 2014, Intel Corporation.
  27  */
  28 /*
  29  * This file is part of Lustre, http://www.lustre.org/
  30  * Lustre is a trademark of Sun Microsystems, Inc.
  31  *
  32  * lnet/lnet/peer.c
  33  */
  34
  35 #define DEBUG_SUBSYSTEM S_LNET
  36
  37 #include <lnet/lib-lnet.h>
  38 #include <lnet/lib-dlc.h>
  39
  40 static void
  41 lnet_peer_remove_from_remote_list(struct lnet_peer_ni *lpni)
  42 {
  43         if (!list_empty(&lpni->lpni_on_remote_peer_ni_list)) {
  44                 list_del_init(&lpni->lpni_on_remote_peer_ni_list);
  45                 lnet_peer_ni_decref_locked(lpni);
  46         }
  47 }
  48
  49 void
  50 lnet_peer_net_added(struct lnet_net *net)
  51 {
  52         struct lnet_peer_ni *lpni, *tmp;
  53
  54         list_for_each_entry_safe(lpni, tmp, &the_lnet.ln_remote_peer_ni_list,
  55                                  lpni_on_remote_peer_ni_list) {
  56
  57                 if (LNET_NIDNET(lpni->lpni_nid) == net->net_id) {
  58                         lpni->lpni_net = net;
  59                         lpni->lpni_txcredits =
  60                         lpni->lpni_mintxcredits =
  61                                 lpni->lpni_net->net_tunables.lct_peer_tx_credits;
  62                         lpni->lpni_rtrcredits =
  63                         lpni->lpni_minrtrcredits =
  64                                 lnet_peer_buffer_credits(lpni->lpni_net);
  65
  66                         lnet_peer_remove_from_remote_list(lpni);
  67                 }
  68         }
  69 }
  70
  71 static void
  72 lnet_peer_tables_destroy(void)
  73 {
  74         struct lnet_peer_table  *ptable;
  75         struct list_head        *hash;
  76         int                     i;
  77         int                     j;
  78
  79         if (!the_lnet.ln_peer_tables)
  80                 return;
  81
  82         cfs_percpt_for_each(ptable, i, the_lnet.ln_peer_tables) {
  83                 hash = ptable->pt_hash;
  84                 if (!hash) /* not intialized */
  85                         break;
  86
  87                 ptable->pt_hash = NULL;
  88                 for (j = 0; j < LNET_PEER_HASH_SIZE; j++)
  89                         LASSERT(list_empty(&hash[j]));
  90
  91                 LIBCFS_FREE(hash, LNET_PEER_HASH_SIZE * sizeof(*hash));
  92         }
  93
  94         cfs_percpt_free(the_lnet.ln_peer_tables);
  95         the_lnet.ln_peer_tables = NULL;
  96 }
  97
  98 int
  99 lnet_peer_tables_create(void)
 100 {
 101         struct lnet_peer_table  *ptable;
 102         struct list_head        *hash;
 103         int                     i;
 104         int                     j;
 105
 106         the_lnet.ln_peer_tables = cfs_percpt_alloc(lnet_cpt_table(),
 107                                                    sizeof(*ptable));
 108         if (the_lnet.ln_peer_tables == NULL) {
 109                 CERROR("Failed to allocate cpu-partition peer tables\n");
 110                 return -ENOMEM;
 111         }
 112
 113         cfs_percpt_for_each(ptable, i, the_lnet.ln_peer_tables) {
 114                 LIBCFS_CPT_ALLOC(hash, lnet_cpt_table(), i,
 115                                  LNET_PEER_HASH_SIZE * sizeof(*hash));
 116                 if (hash == NULL) {
 117                         CERROR("Failed to create peer hash table\n");
 118                         lnet_peer_tables_destroy();
 119                         return -ENOMEM;
 120                 }
 121
 122                 for (j = 0; j < LNET_PEER_HASH_SIZE; j++)
 123                         INIT_LIST_HEAD(&hash[j]);
 124                 ptable->pt_hash = hash; /* sign of initialization */
 125         }
 126
 127         return 0;
 128 }
 129
 130 void lnet_peer_uninit()
 131 {
 132         int cpt;
 133         struct lnet_peer_ni *lpni, *tmp;
 134         struct lnet_peer_table *ptable = NULL;
 135
 136         /* remove all peer_nis from the remote peer and he hash list */
 137         list_for_each_entry_safe(lpni, tmp, &the_lnet.ln_remote_peer_ni_list,
 138                                  lpni_on_remote_peer_ni_list) {
 139                 list_del_init(&lpni->lpni_on_remote_peer_ni_list);
 140                 lnet_peer_ni_decref_locked(lpni);
 141
 142                 cpt = lnet_cpt_of_nid_locked(lpni->lpni_nid, NULL);
 143                 ptable = the_lnet.ln_peer_tables[cpt];
 144                 ptable->pt_zombies++;
 145
 146                 list_del_init(&lpni->lpni_hashlist);
 147                 lnet_peer_ni_decref_locked(lpni);
 148         }
 149
 150         lnet_peer_tables_destroy();
 151 }
 152
 153 static void
 154 lnet_peer_table_cleanup_locked(lnet_ni_t *ni, struct lnet_peer_table *ptable)
 155 {
 156         int                      i;
 157         struct lnet_peer_ni     *lp;
 158         struct lnet_peer_ni     *tmp;
 159
 160         for (i = 0; i < LNET_PEER_HASH_SIZE; i++) {
 161                 list_for_each_entry_safe(lp, tmp, &ptable->pt_hash[i],
 162                                          lpni_hashlist) {
 163                         if (ni != NULL && ni->ni_net != lp->lpni_net)
 164                                 continue;
 165                         list_del_init(&lp->lpni_hashlist);
 166                         /* Lose hash table's ref */
 167                         ptable->pt_zombies++;
 168                         lnet_peer_ni_decref_locked(lp);
 169                 }
 170         }
 171 }
 172
 173 static void
 174 lnet_peer_table_finalize_wait_locked(struct lnet_peer_table *ptable,
 175                                      int cpt_locked)
 176 {
 177         int     i;
 178
 179         for (i = 3; ptable->pt_zombies != 0; i++) {
 180                 lnet_net_unlock(cpt_locked);
 181
 182                 if (IS_PO2(i)) {
 183                         CDEBUG(D_WARNING,
 184                                "Waiting for %d zombies on peer table\n",
 185                                ptable->pt_zombies);
 186                 }
 187                 set_current_state(TASK_UNINTERRUPTIBLE);
 188                 schedule_timeout(cfs_time_seconds(1) >> 1);
 189                 lnet_net_lock(cpt_locked);
 190         }
 191 }
 192
 193 static void
 194 lnet_peer_table_del_rtrs_locked(lnet_ni_t *ni, struct lnet_peer_table *ptable,
 195                                 int cpt_locked)
 196 {
 197         struct lnet_peer_ni     *lp;
 198         struct lnet_peer_ni     *tmp;
 199         lnet_nid_t              lpni_nid;
 200         int                     i;
 201
 202         for (i = 0; i < LNET_PEER_HASH_SIZE; i++) {
 203                 list_for_each_entry_safe(lp, tmp, &ptable->pt_hash[i],
 204                                          lpni_hashlist) {
 205                         if (ni->ni_net != lp->lpni_net)
 206                                 continue;
 207
 208                         if (lp->lpni_rtr_refcount == 0)
 209                                 continue;
 210
 211                         lpni_nid = lp->lpni_nid;
 212
 213                         lnet_net_unlock(cpt_locked);
 214                         lnet_del_route(LNET_NIDNET(LNET_NID_ANY), lpni_nid);
 215                         lnet_net_lock(cpt_locked);
 216                 }
 217         }
 218 }
 219
 220 void
 221 lnet_peer_tables_cleanup(lnet_ni_t *ni)
 222 {
 223         int                             i;
 224         struct lnet_peer_table          *ptable;
 225
 226         LASSERT(the_lnet.ln_shutdown || ni != NULL);
 227         /* If just deleting the peers for a NI, get rid of any routes these
 228          * peers are gateways for. */
 229         cfs_percpt_for_each(ptable, i, the_lnet.ln_peer_tables) {
 230                 lnet_net_lock(LNET_LOCK_EX);
 231                 lnet_peer_table_del_rtrs_locked(ni, ptable, i);
 232                 lnet_net_unlock(LNET_LOCK_EX);
 233         }
 234
 235         /* Start the cleanup process */
 236         cfs_percpt_for_each(ptable, i, the_lnet.ln_peer_tables) {
 237                 lnet_net_lock(LNET_LOCK_EX);
 238                 lnet_peer_table_cleanup_locked(ni, ptable);
 239                 lnet_net_unlock(LNET_LOCK_EX);
 240         }
 241
 242         /* Wait until all peers have been destroyed. */
 243         cfs_percpt_for_each(ptable, i, the_lnet.ln_peer_tables) {
 244                 lnet_net_lock(LNET_LOCK_EX);
 245                 lnet_peer_table_finalize_wait_locked(ptable, i);
 246                 lnet_net_unlock(LNET_LOCK_EX);
 247         }
 248 }
 249
 250 static struct lnet_peer_ni *
 251 lnet_get_peer_ni_locked(struct lnet_peer_table *ptable, lnet_nid_t nid)
 252 {
 253         struct list_head        *peers;
 254         struct lnet_peer_ni     *lp;
 255
 256         LASSERT(!the_lnet.ln_shutdown);
 257
 258         peers = &ptable->pt_hash[lnet_nid2peerhash(nid)];
 259         list_for_each_entry(lp, peers, lpni_hashlist) {
 260                 if (lp->lpni_nid == nid) {
 261                         lnet_peer_ni_addref_locked(lp);
 262                         return lp;
 263                 }
 264         }
 265
 266         return NULL;
 267 }
 268
 269 struct lnet_peer_ni *
 270 lnet_find_peer_ni_locked(lnet_nid_t nid)
 271 {
 272         struct lnet_peer_ni *lpni;
 273         struct lnet_peer_table *ptable;
 274         int cpt;
 275
 276         cpt = lnet_nid_cpt_hash(nid, LNET_CPT_NUMBER);
 277
 278         ptable = the_lnet.ln_peer_tables[cpt];
 279         lpni = lnet_get_peer_ni_locked(ptable, nid);
 280
 281         return lpni;
 282 }
 283
 284 int
 285 lnet_find_or_create_peer_locked(lnet_nid_t dst_nid, int cpt, struct lnet_peer **peer)
 286 {
 287         struct lnet_peer_ni *lpni;
 288
 289         lpni = lnet_find_peer_ni_locked(dst_nid);
 290         if (!lpni) {
 291                 int rc;
 292                 rc = lnet_nid2peerni_locked(&lpni, dst_nid, cpt);
 293                 if (rc != 0)
 294                         return rc;
 295         }
 296
 297         *peer = lpni->lpni_peer_net->lpn_peer;
 298         lnet_peer_ni_decref_locked(lpni);
 299
 300         return 0;
 301 }
 302
 303 struct lnet_peer_ni *
 304 lnet_get_peer_ni_idx_locked(int idx, struct lnet_peer_net **lpn,
 305                             struct lnet_peer **lp)
 306 {
 307         struct lnet_peer_ni     *lpni;
 308
 309         list_for_each_entry((*lp), &the_lnet.ln_peers, lp_on_lnet_peer_list) {
 310                 list_for_each_entry((*lpn), &((*lp)->lp_peer_nets), lpn_on_peer_list) {
 311                         list_for_each_entry(lpni, &((*lpn)->lpn_peer_nis),
 312                                             lpni_on_peer_net_list)
 313                                 if (idx-- == 0)
 314                                         return lpni;
 315                 }
 316         }
 317
 318         return NULL;
 319 }
 320
 321 struct lnet_peer_ni *
 322 lnet_get_next_peer_ni_locked(struct lnet_peer *peer,
 323                              struct lnet_peer_net *peer_net,
 324                              struct lnet_peer_ni *prev)
 325 {
 326         struct lnet_peer_ni *lpni;
 327         struct lnet_peer_net *net = peer_net;
 328
 329         if (!prev) {
 330                 if (!net)
 331                         net = list_entry(peer->lp_peer_nets.next,
 332                                          struct lnet_peer_net,
 333                                          lpn_on_peer_list);
 334                 lpni = list_entry(net->lpn_peer_nis.next, struct lnet_peer_ni,
 335                                   lpni_on_peer_net_list);
 336
 337                 return lpni;
 338         }
 339
 340         if (prev->lpni_on_peer_net_list.next ==
 341             &prev->lpni_peer_net->lpn_peer_nis) {
 342                 /*
 343                  * if you reached the end of the peer ni list and the peer
 344                  * net is specified then there are no more peer nis in that
 345                  * net.
 346                  */
 347                 if (net)
 348                         return NULL;
 349
 350                 /*
 351                  * we reached the end of this net ni list. move to the
 352                  * next net
 353                  */
 354                 if (prev->lpni_peer_net->lpn_on_peer_list.next ==
 355                     &peer->lp_peer_nets)
 356                         /* no more nets and no more NIs. */
 357                         return NULL;
 358
 359                 /* get the next net */
 360                 net = list_entry(prev->lpni_peer_net->lpn_on_peer_list.next,
 361                                  struct lnet_peer_net,
 362                                  lpn_on_peer_list);
 363                 /* get the ni on it */
 364                 lpni = list_entry(net->lpn_peer_nis.next, struct lnet_peer_ni,
 365                                   lpni_on_peer_net_list);
 366
 367                 return lpni;
 368         }
 369
 370         /* there are more nis left */
 371         lpni = list_entry(prev->lpni_on_peer_net_list.next,
 372                           struct lnet_peer_ni, lpni_on_peer_net_list);
 373
 374         return lpni;
 375 }
 376
 377 bool
 378 lnet_peer_is_ni_pref_locked(struct lnet_peer_ni *lpni, struct lnet_ni *ni)
 379 {
 380         int i;
 381
 382         for (i = 0; i < lpni->lpni_pref_nnids; i++) {
 383                 if (lpni->lpni_pref_nids[i] == ni->ni_nid)
 384                         return true;
 385         }
 386         return false;
 387 }
 388
 389 static void
 390 lnet_try_destroy_peer_hierarchy_locked(struct lnet_peer_ni *lpni)
 391 {
 392         struct lnet_peer_net *peer_net;
 393         struct lnet_peer *peer;
 394
 395         /* TODO: could the below situation happen? accessing an already
 396          * destroyed peer? */
 397         if (lpni->lpni_peer_net == NULL ||
 398             lpni->lpni_peer_net->lpn_peer == NULL)
 399                 return;
 400
 401         peer_net = lpni->lpni_peer_net;
 402         peer = lpni->lpni_peer_net->lpn_peer;
 403
 404         list_del_init(&lpni->lpni_on_peer_net_list);
 405         lpni->lpni_peer_net = NULL;
 406
 407         /* if peer_net is empty, then remove it from the peer */
 408         if (list_empty(&peer_net->lpn_peer_nis)) {
 409                 list_del_init(&peer_net->lpn_on_peer_list);
 410                 peer_net->lpn_peer = NULL;
 411                 LIBCFS_FREE(peer_net, sizeof(*peer_net));
 412
 413                 /* if the peer is empty then remove it from the
 414                  * the_lnet.ln_peers */
 415                 if (list_empty(&peer->lp_peer_nets)) {
 416                         list_del_init(&peer->lp_on_lnet_peer_list);
 417                         LIBCFS_FREE(peer, sizeof(*peer));
 418                 }
 419         }
 420 }
 421
 422 static int
 423 lnet_build_peer_hierarchy(struct lnet_peer_ni *lpni)
 424 {
 425         struct lnet_peer *peer;
 426         struct lnet_peer_net *peer_net;
 427         __u32 lpni_net = LNET_NIDNET(lpni->lpni_nid);
 428
 429         peer = NULL;
 430         peer_net = NULL;
 431
 432         LIBCFS_ALLOC(peer, sizeof(*peer));
 433         if (peer == NULL)
 434                 return -ENOMEM;
 435
 436         LIBCFS_ALLOC(peer_net, sizeof(*peer_net));
 437         if (peer_net == NULL) {
 438                 LIBCFS_FREE(peer, sizeof(*peer));
 439                 return -ENOMEM;
 440         }
 441
 442         INIT_LIST_HEAD(&peer->lp_on_lnet_peer_list);
 443         INIT_LIST_HEAD(&peer->lp_peer_nets);
 444         INIT_LIST_HEAD(&peer_net->lpn_on_peer_list);
 445         INIT_LIST_HEAD(&peer_net->lpn_peer_nis);
 446
 447         /* build the hierarchy */
 448         peer_net->lpn_net_id = lpni_net;
 449         peer_net->lpn_peer = peer;
 450         lpni->lpni_peer_net = peer_net;
 451         peer->lp_primary_nid = lpni->lpni_nid;
 452         list_add_tail(&peer_net->lpn_on_peer_list, &peer->lp_peer_nets);
 453         list_add_tail(&lpni->lpni_on_peer_net_list, &peer_net->lpn_peer_nis);
 454         list_add_tail(&peer->lp_on_lnet_peer_list, &the_lnet.ln_peers);
 455
 456         return 0;
 457 }
 458
 459 struct lnet_peer_net *
 460 lnet_peer_get_net_locked(struct lnet_peer *peer, __u32 net_id)
 461 {
 462         struct lnet_peer_net *peer_net;
 463         list_for_each_entry(peer_net, &peer->lp_peer_nets, lpn_on_peer_list) {
 464                 if (peer_net->lpn_net_id == net_id)
 465                         return peer_net;
 466         }
 467         return NULL;
 468 }
 469
 470 /*
 471  * given the key nid find the peer to add the new peer NID to. If the key
 472  * nid is NULL, then create a new peer, but first make sure that the NID
 473  * is unique
 474  */
 475 int
 476 lnet_add_peer_ni_to_peer(lnet_nid_t key_nid, lnet_nid_t nid)
 477 {
 478         struct lnet_peer_ni *lpni, *lpni2;
 479         struct lnet_peer *peer;
 480         struct lnet_peer_net *peer_net, *pn;
 481         int cpt, cpt2, rc;
 482         struct lnet_peer_table *ptable = NULL;
 483         __u32 net_id = LNET_NIDNET(nid);
 484
 485         if (nid == LNET_NID_ANY)
 486                 return -EINVAL;
 487
 488         /* check that nid is unique */
 489         cpt = lnet_nid_cpt_hash(nid, LNET_CPT_NUMBER);
 490         lnet_net_lock(cpt);
 491         lpni = lnet_find_peer_ni_locked(nid);
 492         if (lpni != NULL) {
 493                 lnet_peer_ni_decref_locked(lpni);
 494                 lnet_net_unlock(cpt);
 495                 return -EEXIST;
 496         }
 497         lnet_net_unlock(cpt);
 498
 499         if (key_nid != LNET_NID_ANY) {
 500                 cpt2 = lnet_nid_cpt_hash(key_nid, LNET_CPT_NUMBER);
 501                 lnet_net_lock(cpt2);
 502                 lpni = lnet_find_peer_ni_locked(key_nid);
 503                 if (lpni == NULL) {
 504                         lnet_net_unlock(cpt2);
 505                         /* key_nid refers to a non-existant peer_ni.*/
 506                         return -EINVAL;
 507                 }
 508                 peer = lpni->lpni_peer_net->lpn_peer;
 509                 peer->lp_multi_rail = true;
 510                 lnet_peer_ni_decref_locked(lpni);
 511                 lnet_net_unlock(cpt2);
 512         } else {
 513                 lnet_net_lock(LNET_LOCK_EX);
 514                 rc = lnet_nid2peerni_locked(&lpni, nid, LNET_LOCK_EX);
 515                 if (rc == 0) {
 516                         lpni->lpni_peer_net->lpn_peer->lp_multi_rail = true;
 517                         lnet_peer_ni_decref_locked(lpni);
 518                 }
 519                 lnet_net_unlock(LNET_LOCK_EX);
 520                 return rc;
 521         }
 522
 523         lpni = NULL;
 524
 525         LIBCFS_CPT_ALLOC(lpni, lnet_cpt_table(), cpt, sizeof(*lpni));
 526         if (lpni == NULL)
 527                 return -ENOMEM;
 528
 529         INIT_LIST_HEAD(&lpni->lpni_txq);
 530         INIT_LIST_HEAD(&lpni->lpni_rtrq);
 531         INIT_LIST_HEAD(&lpni->lpni_routes);
 532         INIT_LIST_HEAD(&lpni->lpni_hashlist);
 533         INIT_LIST_HEAD(&lpni->lpni_on_peer_net_list);
 534         INIT_LIST_HEAD(&lpni->lpni_on_remote_peer_ni_list);
 535
 536         lpni->lpni_alive = !lnet_peers_start_down(); /* 1 bit!! */
 537         lpni->lpni_last_alive = cfs_time_current(); /* assumes alive */
 538         lpni->lpni_ping_feats = LNET_PING_FEAT_INVAL;
 539         lpni->lpni_nid = nid;
 540         lpni->lpni_cpt = cpt;
 541         lnet_set_peer_ni_health_locked(lpni, true);
 542
 543         /* allocate here in case we need to add a new peer_net */
 544         peer_net = NULL;
 545         LIBCFS_ALLOC(peer_net, sizeof(*peer_net));
 546         if (peer_net == NULL) {
 547                 rc = -ENOMEM;
 548                 if (lpni != NULL)
 549                         LIBCFS_FREE(lpni, sizeof(*lpni));
 550                 return rc;
 551         }
 552
 553         lnet_net_lock(LNET_LOCK_EX);
 554
 555         ptable = the_lnet.ln_peer_tables[cpt];
 556         ptable->pt_number++;
 557
 558         lpni2 = lnet_find_peer_ni_locked(nid);
 559         if (lpni2 != NULL) {
 560                 lnet_peer_ni_decref_locked(lpni2);
 561                 /* sanity check that lpni2's peer is what we expect */
 562                 if (lpni2->lpni_peer_net->lpn_peer != peer)
 563                         rc = -EEXIST;
 564                 else
 565                         rc = -EINVAL;
 566
 567                 ptable->pt_number--;
 568                 /* another thread has already added it */
 569                 lnet_net_unlock(LNET_LOCK_EX);
 570                 LIBCFS_FREE(peer_net, sizeof(*peer_net));
 571                 return rc;
 572         }
 573
 574         lpni->lpni_net = lnet_get_net_locked(LNET_NIDNET(lpni->lpni_nid));
 575         if (lpni->lpni_net != NULL) {
 576                 lpni->lpni_txcredits    =
 577                 lpni->lpni_mintxcredits =
 578                         lpni->lpni_net->net_tunables.lct_peer_tx_credits;
 579                 lpni->lpni_rtrcredits =
 580                 lpni->lpni_minrtrcredits = lnet_peer_buffer_credits(lpni->lpni_net);
 581         } else {
 582                 /*
 583                  * if you're adding a peer which is not on a local network
 584                  * then we can't assign any of the credits. It won't be
 585                  * picked for sending anyway. Eventually a network can be
 586                  * added, in this case we need to revisit this peer and
 587                  * update its credits.
 588                  */
 589
 590                 /* increment refcount for remote peer list */
 591                 atomic_inc(&lpni->lpni_refcount);
 592                 list_add_tail(&lpni->lpni_on_remote_peer_ni_list,
 593                               &the_lnet.ln_remote_peer_ni_list);
 594         }
 595
 596         /* increment refcount for peer on hash list */
 597         atomic_inc(&lpni->lpni_refcount);
 598
 599         list_add_tail(&lpni->lpni_hashlist,
 600                       &ptable->pt_hash[lnet_nid2peerhash(nid)]);
 601         ptable->pt_version++;
 602
 603         /* add the lpni to a net */
 604         list_for_each_entry(pn, &peer->lp_peer_nets, lpn_on_peer_list) {
 605                 if (pn->lpn_net_id == net_id) {
 606                         list_add_tail(&lpni->lpni_on_peer_net_list,
 607                                       &pn->lpn_peer_nis);
 608                         lpni->lpni_peer_net = pn;
 609                         lnet_net_unlock(LNET_LOCK_EX);
 610                         LIBCFS_FREE(peer_net, sizeof(*peer_net));
 611                         return 0;
 612                 }
 613         }
 614
 615         INIT_LIST_HEAD(&peer_net->lpn_on_peer_list);
 616         INIT_LIST_HEAD(&peer_net->lpn_peer_nis);
 617
 618         /* build the hierarchy */
 619         peer_net->lpn_net_id = net_id;
 620         peer_net->lpn_peer = peer;
 621         lpni->lpni_peer_net = peer_net;
 622         list_add_tail(&lpni->lpni_on_peer_net_list, &peer_net->lpn_peer_nis);
 623         list_add_tail(&peer_net->lpn_on_peer_list, &peer->lp_peer_nets);
 624
 625         lnet_net_unlock(LNET_LOCK_EX);
 626         return 0;
 627 }
 628
 629 int
 630 lnet_del_peer_ni_from_peer(lnet_nid_t key_nid, lnet_nid_t nid)
 631 {
 632         int cpt;
 633         lnet_nid_t local_nid;
 634         struct lnet_peer *peer;
 635         struct lnet_peer_ni *lpni, *lpni2;
 636         struct lnet_peer_table *ptable = NULL;
 637
 638         if (key_nid == LNET_NID_ANY)
 639                 return -EINVAL;
 640
 641         local_nid = (nid != LNET_NID_ANY) ? nid : key_nid;
 642         cpt = lnet_nid_cpt_hash(local_nid, LNET_CPT_NUMBER);
 643         lnet_net_lock(LNET_LOCK_EX);
 644
 645         lpni = lnet_find_peer_ni_locked(local_nid);
 646         if (lpni == NULL) {
 647                 lnet_net_unlock(cpt);
 648                 return -EINVAL;
 649         }
 650         lnet_peer_ni_decref_locked(lpni);
 651
 652         peer = lpni->lpni_peer_net->lpn_peer;
 653         LASSERT(peer != NULL);
 654
 655         if (peer->lp_primary_nid == lpni->lpni_nid) {
 656                 /*
 657                  * deleting the primary ni is equivalent to deleting the
 658                  * entire peer
 659                  */
 660                 lpni = NULL;
 661                 lpni = lnet_get_next_peer_ni_locked(peer, NULL, lpni);
 662                 while (lpni != NULL) {
 663                         lpni2 = lnet_get_next_peer_ni_locked(peer, NULL, lpni);
 664                         cpt = lnet_nid_cpt_hash(lpni->lpni_nid,
 665                                                 LNET_CPT_NUMBER);
 666                         lnet_peer_remove_from_remote_list(lpni);
 667                         ptable = the_lnet.ln_peer_tables[cpt];
 668                         ptable->pt_zombies++;
 669                         list_del_init(&lpni->lpni_hashlist);
 670                         lnet_peer_ni_decref_locked(lpni);
 671                         lpni = lpni2;
 672                 }
 673                 lnet_net_unlock(LNET_LOCK_EX);
 674
 675                 return 0;
 676         }
 677
 678         lnet_peer_remove_from_remote_list(lpni);
 679         cpt = lnet_nid_cpt_hash(lpni->lpni_nid, LNET_CPT_NUMBER);
 680         ptable = the_lnet.ln_peer_tables[cpt];
 681         ptable->pt_zombies++;
 682         list_del_init(&lpni->lpni_hashlist);
 683         lnet_peer_ni_decref_locked(lpni);
 684         lnet_net_unlock(LNET_LOCK_EX);
 685
 686         return 0;
 687 }
 688
 689 void
 690 lnet_destroy_peer_ni_locked(struct lnet_peer_ni *lpni)
 691 {
 692         struct lnet_peer_table *ptable;
 693
 694         LASSERT(atomic_read(&lpni->lpni_refcount) == 0);
 695         LASSERT(lpni->lpni_rtr_refcount == 0);
 696         LASSERT(list_empty(&lpni->lpni_txq));
 697         LASSERT(list_empty(&lpni->lpni_hashlist));
 698         LASSERT(lpni->lpni_txqnob == 0);
 699         LASSERT(lpni->lpni_peer_net != NULL);
 700         LASSERT(lpni->lpni_peer_net->lpn_peer != NULL);
 701
 702         ptable = the_lnet.ln_peer_tables[lpni->lpni_cpt];
 703         LASSERT(ptable->pt_number > 0);
 704         ptable->pt_number--;
 705
 706         lpni->lpni_net = NULL;
 707
 708         lnet_try_destroy_peer_hierarchy_locked(lpni);
 709
 710         LIBCFS_FREE(lpni, sizeof(*lpni));
 711
 712         LASSERT(ptable->pt_zombies > 0);
 713         ptable->pt_zombies--;
 714 }
 715
 716 int
 717 lnet_nid2peerni_locked(struct lnet_peer_ni **lpnip, lnet_nid_t nid, int cpt)
 718 {
 719         struct lnet_peer_table  *ptable;
 720         struct lnet_peer_ni     *lpni = NULL;
 721         struct lnet_peer_ni     *lpni2;
 722         int                     cpt2;
 723         int                     rc = 0;
 724
 725         *lpnip = NULL;
 726         if (the_lnet.ln_shutdown) /* it's shutting down */
 727                 return -ESHUTDOWN;
 728
 729         /*
 730          * calculate cpt2 with the standard hash function
 731          * This cpt2 becomes the slot where we'll find or create the peer.
 732          */
 733         cpt2 = lnet_nid_cpt_hash(nid, LNET_CPT_NUMBER);
 734
 735         /*
 736          * Any changes to the peer tables happen under exclusive write
 737          * lock. Any reads to the peer tables can be done via a standard
 738          * CPT read lock.
 739          */
 740         if (cpt != LNET_LOCK_EX) {
 741                 lnet_net_unlock(cpt);
 742                 lnet_net_lock(LNET_LOCK_EX);
 743         }
 744
 745         ptable = the_lnet.ln_peer_tables[cpt2];
 746         lpni = lnet_get_peer_ni_locked(ptable, nid);
 747         if (lpni != NULL) {
 748                 *lpnip = lpni;
 749                 if (cpt != LNET_LOCK_EX) {
 750                         lnet_net_unlock(LNET_LOCK_EX);
 751                         lnet_net_lock(cpt);
 752                 }
 753                 return 0;
 754         }
 755
 756         /*
 757          * take extra refcount in case another thread has shutdown LNet
 758          * and destroyed locks and peer-table before I finish the allocation
 759          */
 760         ptable->pt_number++;
 761         lnet_net_unlock(LNET_LOCK_EX);
 762
 763         LIBCFS_CPT_ALLOC(lpni, lnet_cpt_table(), cpt2, sizeof(*lpni));
 764
 765         if (lpni == NULL) {
 766                 rc = -ENOMEM;
 767                 lnet_net_lock(cpt);
 768                 goto out;
 769         }
 770
 771         INIT_LIST_HEAD(&lpni->lpni_txq);
 772         INIT_LIST_HEAD(&lpni->lpni_rtrq);
 773         INIT_LIST_HEAD(&lpni->lpni_routes);
 774         INIT_LIST_HEAD(&lpni->lpni_hashlist);
 775         INIT_LIST_HEAD(&lpni->lpni_on_peer_net_list);
 776         INIT_LIST_HEAD(&lpni->lpni_on_remote_peer_ni_list);
 777
 778         lpni->lpni_alive = !lnet_peers_start_down(); /* 1 bit!! */
 779         lpni->lpni_last_alive = cfs_time_current(); /* assumes alive */
 780         lpni->lpni_ping_feats = LNET_PING_FEAT_INVAL;
 781         lpni->lpni_nid = nid;
 782         lpni->lpni_cpt = cpt2;
 783         atomic_set(&lpni->lpni_refcount, 2);    /* 1 for caller; 1 for hash */
 784
 785         rc = lnet_build_peer_hierarchy(lpni);
 786         if (rc != 0)
 787                 goto out;
 788
 789         lnet_net_lock(LNET_LOCK_EX);
 790
 791         if (the_lnet.ln_shutdown) {
 792                 rc = -ESHUTDOWN;
 793                 goto out;
 794         }
 795
 796         lpni2 = lnet_get_peer_ni_locked(ptable, nid);
 797         if (lpni2 != NULL) {
 798                 *lpnip = lpni2;
 799                 goto out;
 800         }
 801
 802         lpni->lpni_net = lnet_get_net_locked(LNET_NIDNET(lpni->lpni_nid));
 803         if (lpni->lpni_net) {
 804                 lpni->lpni_txcredits    =
 805                 lpni->lpni_mintxcredits =
 806                         lpni->lpni_net->net_tunables.lct_peer_tx_credits;
 807                 lpni->lpni_rtrcredits    =
 808                 lpni->lpni_minrtrcredits =
 809                         lnet_peer_buffer_credits(lpni->lpni_net);
 810         } else {
 811                 /*
 812                  * if you're adding a peer which is not on a local network
 813                  * then we can't assign any of the credits. It won't be
 814                  * picked for sending anyway. Eventually a network can be
 815                  * added, in this case we need to revisit this peer and
 816                  * update its credits.
 817                  */
 818
 819                 CDEBUG(D_NET, "peer_ni %s is not directly connected\n",
 820                        libcfs_nid2str(nid));
 821                 /* increment refcount for remote peer list */
 822                 atomic_inc(&lpni->lpni_refcount);
 823                 list_add_tail(&lpni->lpni_on_remote_peer_ni_list,
 824                               &the_lnet.ln_remote_peer_ni_list);
 825         }
 826
 827         lnet_set_peer_ni_health_locked(lpni, true);
 828
 829         list_add_tail(&lpni->lpni_hashlist,
 830                         &ptable->pt_hash[lnet_nid2peerhash(nid)]);
 831         ptable->pt_version++;
 832         *lpnip = lpni;
 833
 834         if (cpt != LNET_LOCK_EX) {
 835                 lnet_net_unlock(LNET_LOCK_EX);
 836                 lnet_net_lock(cpt);
 837         }
 838
 839         return 0;
 840 out:
 841         if (lpni != NULL) {
 842                 lnet_try_destroy_peer_hierarchy_locked(lpni);
 843                 LIBCFS_FREE(lpni, sizeof(*lpni));
 844         }
 845         ptable->pt_number--;
 846         if (cpt != LNET_LOCK_EX) {
 847                 lnet_net_unlock(LNET_LOCK_EX);
 848                 lnet_net_lock(cpt);
 849         }
 850         return rc;
 851 }
 852
 853 void
 854 lnet_debug_peer(lnet_nid_t nid)
 855 {
 856         char                    *aliveness = "NA";
 857         struct lnet_peer_ni     *lp;
 858         int                     rc;
 859         int                     cpt;
 860
 861         cpt = lnet_cpt_of_nid(nid, NULL);
 862         lnet_net_lock(cpt);
 863
 864         rc = lnet_nid2peerni_locked(&lp, nid, cpt);
 865         if (rc != 0) {
 866                 lnet_net_unlock(cpt);
 867                 CDEBUG(D_WARNING, "No peer %s\n", libcfs_nid2str(nid));
 868                 return;
 869         }
 870
 871         if (lnet_isrouter(lp) || lnet_peer_aliveness_enabled(lp))
 872                 aliveness = lp->lpni_alive ? "up" : "down";
 873
 874         CDEBUG(D_WARNING, "%-24s %4d %5s %5d %5d %5d %5d %5d %ld\n",
 875                libcfs_nid2str(lp->lpni_nid), atomic_read(&lp->lpni_refcount),
 876                aliveness, lp->lpni_net->net_tunables.lct_peer_tx_credits,
 877                lp->lpni_rtrcredits, lp->lpni_minrtrcredits,
 878                lp->lpni_txcredits, lp->lpni_mintxcredits, lp->lpni_txqnob);
 879
 880         lnet_peer_ni_decref_locked(lp);
 881
 882         lnet_net_unlock(cpt);
 883 }
 884
 885 int lnet_get_peer_ni_info(__u32 peer_index, __u64 *nid,
 886                           char aliveness[LNET_MAX_STR_LEN],
 887                           __u32 *cpt_iter, __u32 *refcount,
 888                           __u32 *ni_peer_tx_credits, __u32 *peer_tx_credits,
 889                           __u32 *peer_rtr_credits, __u32 *peer_min_rtr_credits,
 890                           __u32 *peer_tx_qnob)
 891 {
 892         struct lnet_peer_table          *peer_table;
 893         struct lnet_peer_ni             *lp;
 894         int                             j;
 895         int                             lncpt;
 896         bool                            found = false;
 897
 898         /* get the number of CPTs */
 899         lncpt = cfs_percpt_number(the_lnet.ln_peer_tables);
 900
 901         /* if the cpt number to be examined is >= the number of cpts in
 902          * the system then indicate that there are no more cpts to examin
 903          */
 904         if (*cpt_iter >= lncpt)
 905                 return -ENOENT;
 906
 907         /* get the current table */
 908         peer_table = the_lnet.ln_peer_tables[*cpt_iter];
 909         /* if the ptable is NULL then there are no more cpts to examine */
 910         if (peer_table == NULL)
 911                 return -ENOENT;
 912
 913         lnet_net_lock(*cpt_iter);
 914
 915         for (j = 0; j < LNET_PEER_HASH_SIZE && !found; j++) {
 916                 struct list_head *peers = &peer_table->pt_hash[j];
 917
 918                 list_for_each_entry(lp, peers, lpni_hashlist) {
 919                         if (peer_index-- > 0)
 920                                 continue;
 921
 922                         snprintf(aliveness, LNET_MAX_STR_LEN, "NA");
 923                         if (lnet_isrouter(lp) ||
 924                                 lnet_peer_aliveness_enabled(lp))
 925                                 snprintf(aliveness, LNET_MAX_STR_LEN,
 926                                          lp->lpni_alive ? "up" : "down");
 927
 928                         *nid = lp->lpni_nid;
 929                         *refcount = atomic_read(&lp->lpni_refcount);
 930                         *ni_peer_tx_credits =
 931                                 lp->lpni_net->net_tunables.lct_peer_tx_credits;
 932                         *peer_tx_credits = lp->lpni_txcredits;
 933                         *peer_rtr_credits = lp->lpni_rtrcredits;
 934                         *peer_min_rtr_credits = lp->lpni_mintxcredits;
 935                         *peer_tx_qnob = lp->lpni_txqnob;
 936
 937                         found = true;
 938                 }
 939
 940         }
 941         lnet_net_unlock(*cpt_iter);
 942
 943         *cpt_iter = lncpt;
 944
 945         return found ? 0 : -ENOENT;
 946 }
 947
 948 int lnet_get_peer_info(__u32 idx, lnet_nid_t *primary_nid, lnet_nid_t *nid,
 949                        struct lnet_peer_ni_credit_info *peer_ni_info)
 950 {
 951         struct lnet_peer_ni *lpni = NULL;
 952         struct lnet_peer_net *lpn = NULL;
 953         struct lnet_peer *lp = NULL;
 954
 955         lpni = lnet_get_peer_ni_idx_locked(idx, &lpn, &lp);
 956
 957         if (!lpni)
 958                 return -ENOENT;
 959
 960         *primary_nid = lp->lp_primary_nid;
 961         *nid = lpni->lpni_nid;
 962         snprintf(peer_ni_info->cr_aliveness, LNET_MAX_STR_LEN, "NA");
 963         if (lnet_isrouter(lpni) ||
 964                 lnet_peer_aliveness_enabled(lpni))
 965                 snprintf(peer_ni_info->cr_aliveness, LNET_MAX_STR_LEN,
 966                          lpni->lpni_alive ? "up" : "down");
 967
 968         peer_ni_info->cr_refcount = atomic_read(&lpni->lpni_refcount);
 969         peer_ni_info->cr_ni_peer_tx_credits = (lpni->lpni_net != NULL) ?
 970                 lpni->lpni_net->net_tunables.lct_peer_tx_credits : 0;
 971         peer_ni_info->cr_peer_tx_credits = lpni->lpni_txcredits;
 972         peer_ni_info->cr_peer_rtr_credits = lpni->lpni_rtrcredits;
 973         peer_ni_info->cr_peer_min_rtr_credits = lpni->lpni_mintxcredits;
 974         peer_ni_info->cr_peer_tx_qnob = lpni->lpni_txqnob;
 975
 976         return 0;
 977 }