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
  60                         spin_lock(&lpni->lpni_lock);
  61                         lpni->lpni_txcredits =
  62                                 lpni->lpni_net->net_tunables.lct_peer_tx_credits;
  63                         lpni->lpni_mintxcredits = lpni->lpni_txcredits;
  64                         lpni->lpni_rtrcredits =
  65                                 lnet_peer_buffer_credits(lpni->lpni_net);
  66                         lpni->lpni_minrtrcredits = lpni->lpni_rtrcredits;
  67                         spin_unlock(&lpni->lpni_lock);
  68
  69                         lnet_peer_remove_from_remote_list(lpni);
  70                 }
  71         }
  72 }
  73
  74 static void
  75 lnet_peer_tables_destroy(void)
  76 {
  77         struct lnet_peer_table  *ptable;
  78         struct list_head        *hash;
  79         int                     i;
  80         int                     j;
  81
  82         if (!the_lnet.ln_peer_tables)
  83                 return;
  84
  85         cfs_percpt_for_each(ptable, i, the_lnet.ln_peer_tables) {
  86                 hash = ptable->pt_hash;
  87                 if (!hash) /* not intialized */
  88                         break;
  89
  90                 LASSERT(list_empty(&ptable->pt_zombie_list));
  91
  92                 ptable->pt_hash = NULL;
  93                 for (j = 0; j < LNET_PEER_HASH_SIZE; j++)
  94                         LASSERT(list_empty(&hash[j]));
  95
  96                 LIBCFS_FREE(hash, LNET_PEER_HASH_SIZE * sizeof(*hash));
  97         }
  98
  99         cfs_percpt_free(the_lnet.ln_peer_tables);
 100         the_lnet.ln_peer_tables = NULL;
 101 }
 102
 103 int
 104 lnet_peer_tables_create(void)
 105 {
 106         struct lnet_peer_table  *ptable;
 107         struct list_head        *hash;
 108         int                     i;
 109         int                     j;
 110
 111         the_lnet.ln_peer_tables = cfs_percpt_alloc(lnet_cpt_table(),
 112                                                    sizeof(*ptable));
 113         if (the_lnet.ln_peer_tables == NULL) {
 114                 CERROR("Failed to allocate cpu-partition peer tables\n");
 115                 return -ENOMEM;
 116         }
 117
 118         cfs_percpt_for_each(ptable, i, the_lnet.ln_peer_tables) {
 119                 LIBCFS_CPT_ALLOC(hash, lnet_cpt_table(), i,
 120                                  LNET_PEER_HASH_SIZE * sizeof(*hash));
 121                 if (hash == NULL) {
 122                         CERROR("Failed to create peer hash table\n");
 123                         lnet_peer_tables_destroy();
 124                         return -ENOMEM;
 125                 }
 126
 127                 spin_lock_init(&ptable->pt_zombie_lock);
 128                 INIT_LIST_HEAD(&ptable->pt_zombie_list);
 129
 130                 INIT_LIST_HEAD(&ptable->pt_peer_list);
 131
 132                 for (j = 0; j < LNET_PEER_HASH_SIZE; j++)
 133                         INIT_LIST_HEAD(&hash[j]);
 134                 ptable->pt_hash = hash; /* sign of initialization */
 135         }
 136
 137         return 0;
 138 }
 139
 140 static struct lnet_peer_ni *
 141 lnet_peer_ni_alloc(lnet_nid_t nid)
 142 {
 143         struct lnet_peer_ni *lpni;
 144         struct lnet_net *net;
 145         int cpt;
 146
 147         cpt = lnet_nid_cpt_hash(nid, LNET_CPT_NUMBER);
 148
 149         LIBCFS_CPT_ALLOC(lpni, lnet_cpt_table(), cpt, sizeof(*lpni));
 150         if (!lpni)
 151                 return NULL;
 152
 153         INIT_LIST_HEAD(&lpni->lpni_txq);
 154         INIT_LIST_HEAD(&lpni->lpni_rtrq);
 155         INIT_LIST_HEAD(&lpni->lpni_routes);
 156         INIT_LIST_HEAD(&lpni->lpni_hashlist);
 157         INIT_LIST_HEAD(&lpni->lpni_peer_nis);
 158         INIT_LIST_HEAD(&lpni->lpni_on_remote_peer_ni_list);
 159
 160         spin_lock_init(&lpni->lpni_lock);
 161
 162         lpni->lpni_alive = !lnet_peers_start_down(); /* 1 bit!! */
 163         lpni->lpni_last_alive = cfs_time_current(); /* assumes alive */
 164         lpni->lpni_ping_feats = LNET_PING_FEAT_INVAL;
 165         lpni->lpni_nid = nid;
 166         lpni->lpni_cpt = cpt;
 167         lnet_set_peer_ni_health_locked(lpni, true);
 168
 169         net = lnet_get_net_locked(LNET_NIDNET(nid));
 170         lpni->lpni_net = net;
 171         if (net) {
 172                 lpni->lpni_txcredits = net->net_tunables.lct_peer_tx_credits;
 173                 lpni->lpni_mintxcredits = lpni->lpni_txcredits;
 174                 lpni->lpni_rtrcredits = lnet_peer_buffer_credits(net);
 175                 lpni->lpni_minrtrcredits = lpni->lpni_rtrcredits;
 176         } else {
 177                 /*
 178                  * This peer_ni is not on a local network, so we
 179                  * cannot add the credits here. In case the net is
 180                  * added later, add the peer_ni to the remote peer ni
 181                  * list so it can be easily found and revisited.
 182                  */
 183                 /* FIXME: per-net implementation instead? */
 184                 atomic_inc(&lpni->lpni_refcount);
 185                 list_add_tail(&lpni->lpni_on_remote_peer_ni_list,
 186                               &the_lnet.ln_remote_peer_ni_list);
 187         }
 188
 189         CDEBUG(D_NET, "%p nid %s\n", lpni, libcfs_nid2str(lpni->lpni_nid));
 190
 191         return lpni;
 192 }
 193
 194 static struct lnet_peer_net *
 195 lnet_peer_net_alloc(__u32 net_id)
 196 {
 197         struct lnet_peer_net *lpn;
 198
 199         LIBCFS_CPT_ALLOC(lpn, lnet_cpt_table(), CFS_CPT_ANY, sizeof(*lpn));
 200         if (!lpn)
 201                 return NULL;
 202
 203         INIT_LIST_HEAD(&lpn->lpn_peer_nets);
 204         INIT_LIST_HEAD(&lpn->lpn_peer_nis);
 205         lpn->lpn_net_id = net_id;
 206
 207         CDEBUG(D_NET, "%p net %s\n", lpn, libcfs_net2str(lpn->lpn_net_id));
 208
 209         return lpn;
 210 }
 211
 212 void
 213 lnet_destroy_peer_net_locked(struct lnet_peer_net *lpn)
 214 {
 215         struct lnet_peer *lp;
 216
 217         CDEBUG(D_NET, "%p net %s\n", lpn, libcfs_net2str(lpn->lpn_net_id));
 218
 219         LASSERT(atomic_read(&lpn->lpn_refcount) == 0);
 220         LASSERT(list_empty(&lpn->lpn_peer_nis));
 221         LASSERT(list_empty(&lpn->lpn_peer_nets));
 222         lp = lpn->lpn_peer;
 223         lpn->lpn_peer = NULL;
 224         LIBCFS_FREE(lpn, sizeof(*lpn));
 225
 226         lnet_peer_decref_locked(lp);
 227 }
 228
 229 static struct lnet_peer *
 230 lnet_peer_alloc(lnet_nid_t nid)
 231 {
 232         struct lnet_peer *lp;
 233
 234         LIBCFS_CPT_ALLOC(lp, lnet_cpt_table(), CFS_CPT_ANY, sizeof(*lp));
 235         if (!lp)
 236                 return NULL;
 237
 238         INIT_LIST_HEAD(&lp->lp_peer_list);
 239         INIT_LIST_HEAD(&lp->lp_peer_nets);
 240         spin_lock_init(&lp->lp_lock);
 241         lp->lp_primary_nid = nid;
 242         lp->lp_cpt = lnet_nid_cpt_hash(nid, LNET_CPT_NUMBER);
 243
 244         CDEBUG(D_NET, "%p nid %s\n", lp, libcfs_nid2str(lp->lp_primary_nid));
 245
 246         return lp;
 247 }
 248
 249 void
 250 lnet_destroy_peer_locked(struct lnet_peer *lp)
 251 {
 252         CDEBUG(D_NET, "%p nid %s\n", lp, libcfs_nid2str(lp->lp_primary_nid));
 253
 254         LASSERT(atomic_read(&lp->lp_refcount) == 0);
 255         LASSERT(list_empty(&lp->lp_peer_nets));
 256         LASSERT(list_empty(&lp->lp_peer_list));
 257
 258         LIBCFS_FREE(lp, sizeof(*lp));
 259 }
 260
 261 /*
 262  * Detach a peer_ni from its peer_net. If this was the last peer_ni on
 263  * that peer_net, detach the peer_net from the peer.
 264  *
 265  * Call with lnet_net_lock/EX held
 266  */
 267 static void
 268 lnet_peer_detach_peer_ni_locked(struct lnet_peer_ni *lpni)
 269 {
 270         struct lnet_peer_table *ptable;
 271         struct lnet_peer_net *lpn;
 272         struct lnet_peer *lp;
 273
 274         /*
 275          * Belts and suspenders: gracefully handle teardown of a
 276          * partially connected peer_ni.
 277          */
 278         lpn = lpni->lpni_peer_net;
 279
 280         list_del_init(&lpni->lpni_peer_nis);
 281         /*
 282          * If there are no lpni's left, we detach lpn from
 283          * lp_peer_nets, so it cannot be found anymore.
 284          */
 285         if (list_empty(&lpn->lpn_peer_nis))
 286                 list_del_init(&lpn->lpn_peer_nets);
 287
 288         /* Update peer NID count. */
 289         lp = lpn->lpn_peer;
 290         ptable = the_lnet.ln_peer_tables[lp->lp_cpt];
 291         lp->lp_nnis--;
 292         ptable->pt_peer_nnids--;
 293
 294         /*
 295          * If there are no more peer nets, make the peer unfindable
 296          * via the peer_tables.
 297          */
 298         if (list_empty(&lp->lp_peer_nets)) {
 299                 list_del_init(&lp->lp_peer_list);
 300                 ptable->pt_peers--;
 301         }
 302         CDEBUG(D_NET, "peer %s NID %s\n",
 303                 libcfs_nid2str(lp->lp_primary_nid),
 304                 libcfs_nid2str(lpni->lpni_nid));
 305 }
 306
 307 /* called with lnet_net_lock LNET_LOCK_EX held */
 308 static int
 309 lnet_peer_ni_del_locked(struct lnet_peer_ni *lpni)
 310 {
 311         struct lnet_peer_table *ptable = NULL;
 312
 313         /* don't remove a peer_ni if it's also a gateway */
 314         if (lpni->lpni_rtr_refcount > 0) {
 315                 CERROR("Peer NI %s is a gateway. Can not delete it\n",
 316                        libcfs_nid2str(lpni->lpni_nid));
 317                 return -EBUSY;
 318         }
 319
 320         lnet_peer_remove_from_remote_list(lpni);
 321
 322         /* remove peer ni from the hash list. */
 323         list_del_init(&lpni->lpni_hashlist);
 324
 325         /* decrement the ref count on the peer table */
 326         ptable = the_lnet.ln_peer_tables[lpni->lpni_cpt];
 327         LASSERT(ptable->pt_number > 0);
 328         ptable->pt_number--;
 329
 330         /*
 331          * The peer_ni can no longer be found with a lookup. But there
 332          * can be current users, so keep track of it on the zombie
 333          * list until the reference count has gone to zero.
 334          *
 335          * The last reference may be lost in a place where the
 336          * lnet_net_lock locks only a single cpt, and that cpt may not
 337          * be lpni->lpni_cpt. So the zombie list of lnet_peer_table
 338          * has its own lock.
 339          */
 340         spin_lock(&ptable->pt_zombie_lock);
 341         list_add(&lpni->lpni_hashlist, &ptable->pt_zombie_list);
 342         ptable->pt_zombies++;
 343         spin_unlock(&ptable->pt_zombie_lock);
 344
 345         /* no need to keep this peer_ni on the hierarchy anymore */
 346         lnet_peer_detach_peer_ni_locked(lpni);
 347
 348         /* remove hashlist reference on peer_ni */
 349         lnet_peer_ni_decref_locked(lpni);
 350
 351         return 0;
 352 }
 353
 354 void lnet_peer_uninit(void)
 355 {
 356         struct lnet_peer_ni *lpni, *tmp;
 357
 358         lnet_net_lock(LNET_LOCK_EX);
 359
 360         /* remove all peer_nis from the remote peer and the hash list */
 361         list_for_each_entry_safe(lpni, tmp, &the_lnet.ln_remote_peer_ni_list,
 362                                  lpni_on_remote_peer_ni_list)
 363                 lnet_peer_ni_del_locked(lpni);
 364
 365         lnet_peer_tables_destroy();
 366
 367         lnet_net_unlock(LNET_LOCK_EX);
 368 }
 369
 370 static int
 371 lnet_peer_del_locked(struct lnet_peer *peer)
 372 {
 373         struct lnet_peer_ni *lpni = NULL, *lpni2;
 374         int rc = 0, rc2 = 0;
 375
 376         CDEBUG(D_NET, "peer %s\n", libcfs_nid2str(peer->lp_primary_nid));
 377
 378         lpni = lnet_get_next_peer_ni_locked(peer, NULL, lpni);
 379         while (lpni != NULL) {
 380                 lpni2 = lnet_get_next_peer_ni_locked(peer, NULL, lpni);
 381                 rc = lnet_peer_ni_del_locked(lpni);
 382                 if (rc != 0)
 383                         rc2 = rc;
 384                 lpni = lpni2;
 385         }
 386
 387         return rc2;
 388 }
 389
 390 static int
 391 lnet_peer_del(struct lnet_peer *peer)
 392 {
 393         lnet_net_lock(LNET_LOCK_EX);
 394         lnet_peer_del_locked(peer);
 395         lnet_net_unlock(LNET_LOCK_EX);
 396
 397         return 0;
 398 }
 399
 400 /*
 401  * Delete a NID from a peer. Call with ln_api_mutex held.
 402  *
 403  * Error codes:
 404  *  -EPERM:  Non-DLC deletion from DLC-configured peer.
 405  *  -ENOENT: No lnet_peer_ni corresponding to the nid.
 406  *  -ECHILD: The lnet_peer_ni isn't connected to the peer.
 407  *  -EBUSY:  The lnet_peer_ni is the primary, and not the only peer_ni.
 408  */
 409 static int
 410 lnet_peer_del_nid(struct lnet_peer *lp, lnet_nid_t nid, unsigned flags)
 411 {
 412         struct lnet_peer_ni *lpni;
 413         lnet_nid_t primary_nid = lp->lp_primary_nid;
 414         int rc = 0;
 415
 416         if (!(flags & LNET_PEER_CONFIGURED)) {
 417                 if (lp->lp_state & LNET_PEER_CONFIGURED) {
 418                         rc = -EPERM;
 419                         goto out;
 420                 }
 421         }
 422         lpni = lnet_find_peer_ni_locked(nid);
 423         if (!lpni) {
 424                 rc = -ENOENT;
 425                 goto out;
 426         }
 427         lnet_peer_ni_decref_locked(lpni);
 428         if (lp != lpni->lpni_peer_net->lpn_peer) {
 429                 rc = -ECHILD;
 430                 goto out;
 431         }
 432
 433         /*
 434          * This function only allows deletion of the primary NID if it
 435          * is the only NID.
 436          */
 437         if (nid == lp->lp_primary_nid && lp->lp_nnis != 1) {
 438                 rc = -EBUSY;
 439                 goto out;
 440         }
 441
 442         lnet_net_lock(LNET_LOCK_EX);
 443         lnet_peer_ni_del_locked(lpni);
 444         lnet_net_unlock(LNET_LOCK_EX);
 445
 446 out:
 447         CDEBUG(D_NET, "peer %s NID %s flags %#x: %d\n",
 448                libcfs_nid2str(primary_nid), libcfs_nid2str(nid), flags, rc);
 449
 450         return rc;
 451 }
 452
 453 static void
 454 lnet_peer_table_cleanup_locked(struct lnet_net *net,
 455                                struct lnet_peer_table *ptable)
 456 {
 457         int                      i;
 458         struct lnet_peer_ni     *next;
 459         struct lnet_peer_ni     *lpni;
 460         struct lnet_peer        *peer;
 461
 462         for (i = 0; i < LNET_PEER_HASH_SIZE; i++) {
 463                 list_for_each_entry_safe(lpni, next, &ptable->pt_hash[i],
 464                                          lpni_hashlist) {
 465                         if (net != NULL && net != lpni->lpni_net)
 466                                 continue;
 467
 468                         peer = lpni->lpni_peer_net->lpn_peer;
 469                         if (peer->lp_primary_nid != lpni->lpni_nid) {
 470                                 lnet_peer_ni_del_locked(lpni);
 471                                 continue;
 472                         }
 473                         /*
 474                          * Removing the primary NID implies removing
 475                          * the entire peer. Advance next beyond any
 476                          * peer_ni that belongs to the same peer.
 477                          */
 478                         list_for_each_entry_from(next, &ptable->pt_hash[i],
 479                                                  lpni_hashlist) {
 480                                 if (next->lpni_peer_net->lpn_peer != peer)
 481                                         break;
 482                         }
 483                         lnet_peer_del_locked(peer);
 484                 }
 485         }
 486 }
 487
 488 static void
 489 lnet_peer_ni_finalize_wait(struct lnet_peer_table *ptable)
 490 {
 491         int     i = 3;
 492
 493         spin_lock(&ptable->pt_zombie_lock);
 494         while (ptable->pt_zombies) {
 495                 spin_unlock(&ptable->pt_zombie_lock);
 496
 497                 if (IS_PO2(i)) {
 498                         CDEBUG(D_WARNING,
 499                                "Waiting for %d zombies on peer table\n",
 500                                ptable->pt_zombies);
 501                 }
 502                 set_current_state(TASK_UNINTERRUPTIBLE);
 503                 schedule_timeout(cfs_time_seconds(1) >> 1);
 504                 spin_lock(&ptable->pt_zombie_lock);
 505         }
 506         spin_unlock(&ptable->pt_zombie_lock);
 507 }
 508
 509 static void
 510 lnet_peer_table_del_rtrs_locked(struct lnet_net *net,
 511                                 struct lnet_peer_table *ptable)
 512 {
 513         struct lnet_peer_ni     *lp;
 514         struct lnet_peer_ni     *tmp;
 515         lnet_nid_t              lpni_nid;
 516         int                     i;
 517
 518         for (i = 0; i < LNET_PEER_HASH_SIZE; i++) {
 519                 list_for_each_entry_safe(lp, tmp, &ptable->pt_hash[i],
 520                                          lpni_hashlist) {
 521                         if (net != lp->lpni_net)
 522                                 continue;
 523
 524                         if (lp->lpni_rtr_refcount == 0)
 525                                 continue;
 526
 527                         lpni_nid = lp->lpni_nid;
 528
 529                         lnet_net_unlock(LNET_LOCK_EX);
 530                         lnet_del_route(LNET_NIDNET(LNET_NID_ANY), lpni_nid);
 531                         lnet_net_lock(LNET_LOCK_EX);
 532                 }
 533         }
 534 }
 535
 536 void
 537 lnet_peer_tables_cleanup(struct lnet_net *net)
 538 {
 539         int                             i;
 540         struct lnet_peer_table          *ptable;
 541
 542         LASSERT(the_lnet.ln_state != LNET_STATE_SHUTDOWN || net != NULL);
 543         /* If just deleting the peers for a NI, get rid of any routes these
 544          * peers are gateways for. */
 545         cfs_percpt_for_each(ptable, i, the_lnet.ln_peer_tables) {
 546                 lnet_net_lock(LNET_LOCK_EX);
 547                 lnet_peer_table_del_rtrs_locked(net, ptable);
 548                 lnet_net_unlock(LNET_LOCK_EX);
 549         }
 550
 551         /* Start the cleanup process */
 552         cfs_percpt_for_each(ptable, i, the_lnet.ln_peer_tables) {
 553                 lnet_net_lock(LNET_LOCK_EX);
 554                 lnet_peer_table_cleanup_locked(net, ptable);
 555                 lnet_net_unlock(LNET_LOCK_EX);
 556         }
 557
 558         cfs_percpt_for_each(ptable, i, the_lnet.ln_peer_tables)
 559                 lnet_peer_ni_finalize_wait(ptable);
 560 }
 561
 562 static struct lnet_peer_ni *
 563 lnet_get_peer_ni_locked(struct lnet_peer_table *ptable, lnet_nid_t nid)
 564 {
 565         struct list_head        *peers;
 566         struct lnet_peer_ni     *lp;
 567
 568         LASSERT(the_lnet.ln_state == LNET_STATE_RUNNING);
 569
 570         peers = &ptable->pt_hash[lnet_nid2peerhash(nid)];
 571         list_for_each_entry(lp, peers, lpni_hashlist) {
 572                 if (lp->lpni_nid == nid) {
 573                         lnet_peer_ni_addref_locked(lp);
 574                         return lp;
 575                 }
 576         }
 577
 578         return NULL;
 579 }
 580
 581 struct lnet_peer_ni *
 582 lnet_find_peer_ni_locked(lnet_nid_t nid)
 583 {
 584         struct lnet_peer_ni *lpni;
 585         struct lnet_peer_table *ptable;
 586         int cpt;
 587
 588         cpt = lnet_nid_cpt_hash(nid, LNET_CPT_NUMBER);
 589
 590         ptable = the_lnet.ln_peer_tables[cpt];
 591         lpni = lnet_get_peer_ni_locked(ptable, nid);
 592
 593         return lpni;
 594 }
 595
 596 struct lnet_peer_ni *
 597 lnet_get_peer_ni_idx_locked(int idx, struct lnet_peer_net **lpn,
 598                             struct lnet_peer **lp)
 599 {
 600         struct lnet_peer_table  *ptable;
 601         struct lnet_peer_ni     *lpni;
 602         int                     lncpt;
 603         int                     cpt;
 604
 605         lncpt = cfs_percpt_number(the_lnet.ln_peer_tables);
 606
 607         for (cpt = 0; cpt < lncpt; cpt++) {
 608                 ptable = the_lnet.ln_peer_tables[cpt];
 609                 if (ptable->pt_peer_nnids > idx)
 610                         break;
 611                 idx -= ptable->pt_peer_nnids;
 612         }
 613         if (cpt >= lncpt)
 614                 return NULL;
 615
 616         list_for_each_entry((*lp), &ptable->pt_peer_list, lp_peer_list) {
 617                 if ((*lp)->lp_nnis <= idx) {
 618                         idx -= (*lp)->lp_nnis;
 619                         continue;
 620                 }
 621                 list_for_each_entry((*lpn), &((*lp)->lp_peer_nets),
 622                                     lpn_peer_nets) {
 623                         list_for_each_entry(lpni, &((*lpn)->lpn_peer_nis),
 624                                             lpni_peer_nis) {
 625                                 if (idx-- == 0)
 626                                         return lpni;
 627                         }
 628                 }
 629         }
 630
 631         return NULL;
 632 }
 633
 634 struct lnet_peer_ni *
 635 lnet_get_next_peer_ni_locked(struct lnet_peer *peer,
 636                              struct lnet_peer_net *peer_net,
 637                              struct lnet_peer_ni *prev)
 638 {
 639         struct lnet_peer_ni *lpni;
 640         struct lnet_peer_net *net = peer_net;
 641
 642         if (!prev) {
 643                 if (!net) {
 644                         if (list_empty(&peer->lp_peer_nets))
 645                                 return NULL;
 646
 647                         net = list_entry(peer->lp_peer_nets.next,
 648                                          struct lnet_peer_net,
 649                                          lpn_peer_nets);
 650                 }
 651                 lpni = list_entry(net->lpn_peer_nis.next, struct lnet_peer_ni,
 652                                   lpni_peer_nis);
 653
 654                 return lpni;
 655         }
 656
 657         if (prev->lpni_peer_nis.next == &prev->lpni_peer_net->lpn_peer_nis) {
 658                 /*
 659                  * if you reached the end of the peer ni list and the peer
 660                  * net is specified then there are no more peer nis in that
 661                  * net.
 662                  */
 663                 if (net)
 664                         return NULL;
 665
 666                 /*
 667                  * we reached the end of this net ni list. move to the
 668                  * next net
 669                  */
 670                 if (prev->lpni_peer_net->lpn_peer_nets.next ==
 671                     &peer->lp_peer_nets)
 672                         /* no more nets and no more NIs. */
 673                         return NULL;
 674
 675                 /* get the next net */
 676                 net = list_entry(prev->lpni_peer_net->lpn_peer_nets.next,
 677                                  struct lnet_peer_net,
 678                                  lpn_peer_nets);
 679                 /* get the ni on it */
 680                 lpni = list_entry(net->lpn_peer_nis.next, struct lnet_peer_ni,
 681                                   lpni_peer_nis);
 682
 683                 return lpni;
 684         }
 685
 686         /* there are more nis left */
 687         lpni = list_entry(prev->lpni_peer_nis.next,
 688                           struct lnet_peer_ni, lpni_peer_nis);
 689
 690         return lpni;
 691 }
 692
 693 /*
 694  * Test whether a ni is a preferred ni for this peer_ni, e.g, whether
 695  * this is a preferred point-to-point path. Call with lnet_net_lock in
 696  * shared mmode.
 697  */
 698 bool
 699 lnet_peer_is_pref_nid_locked(struct lnet_peer_ni *lpni, lnet_nid_t nid)
 700 {
 701         int i;
 702
 703         if (lpni->lpni_pref_nnids == 0)
 704                 return false;
 705         if (lpni->lpni_pref_nnids == 1)
 706                 return lpni->lpni_pref.nid == nid;
 707         for (i = 0; i < lpni->lpni_pref_nnids; i++) {
 708                 if (lpni->lpni_pref.nids[i] == nid)
 709                         return true;
 710         }
 711         return false;
 712 }
 713
 714 /*
 715  * Set a single ni as preferred, provided no preferred ni is already
 716  * defined. Only to be used for non-multi-rail peer_ni.
 717  */
 718 int
 719 lnet_peer_ni_set_non_mr_pref_nid(struct lnet_peer_ni *lpni, lnet_nid_t nid)
 720 {
 721         int rc = 0;
 722
 723         spin_lock(&lpni->lpni_lock);
 724         if (nid == LNET_NID_ANY) {
 725                 rc = -EINVAL;
 726         } else if (lpni->lpni_pref_nnids > 0) {
 727                 rc = -EPERM;
 728         } else if (lpni->lpni_pref_nnids == 0) {
 729                 lpni->lpni_pref.nid = nid;
 730                 lpni->lpni_pref_nnids = 1;
 731                 lpni->lpni_state |= LNET_PEER_NI_NON_MR_PREF;
 732         }
 733         spin_unlock(&lpni->lpni_lock);
 734
 735         CDEBUG(D_NET, "peer %s nid %s: %d\n",
 736                libcfs_nid2str(lpni->lpni_nid), libcfs_nid2str(nid), rc);
 737         return rc;
 738 }
 739
 740 /*
 741  * Clear the preferred NID from a non-multi-rail peer_ni, provided
 742  * this preference was set by lnet_peer_ni_set_non_mr_pref_nid().
 743  */
 744 int
 745 lnet_peer_ni_clr_non_mr_pref_nid(struct lnet_peer_ni *lpni)
 746 {
 747         int rc = 0;
 748
 749         spin_lock(&lpni->lpni_lock);
 750         if (lpni->lpni_state & LNET_PEER_NI_NON_MR_PREF) {
 751                 lpni->lpni_pref_nnids = 0;
 752                 lpni->lpni_state &= ~LNET_PEER_NI_NON_MR_PREF;
 753         } else if (lpni->lpni_pref_nnids == 0) {
 754                 rc = -ENOENT;
 755         } else {
 756                 rc = -EPERM;
 757         }
 758         spin_unlock(&lpni->lpni_lock);
 759
 760         CDEBUG(D_NET, "peer %s: %d\n",
 761                libcfs_nid2str(lpni->lpni_nid), rc);
 762         return rc;
 763 }
 764
 765 /*
 766  * Clear the preferred NIDs from a non-multi-rail peer.
 767  */
 768 void
 769 lnet_peer_clr_non_mr_pref_nids(struct lnet_peer *lp)
 770 {
 771         struct lnet_peer_ni *lpni = NULL;
 772
 773         while ((lpni = lnet_get_next_peer_ni_locked(lp, NULL, lpni)) != NULL)
 774                 lnet_peer_ni_clr_non_mr_pref_nid(lpni);
 775 }
 776
 777 int
 778 lnet_peer_add_pref_nid(struct lnet_peer_ni *lpni, lnet_nid_t nid)
 779 {
 780         lnet_nid_t *nids = NULL;
 781         lnet_nid_t *oldnids = NULL;
 782         struct lnet_peer *lp = lpni->lpni_peer_net->lpn_peer;
 783         int size;
 784         int i;
 785         int rc = 0;
 786
 787         if (nid == LNET_NID_ANY) {
 788                 rc = -EINVAL;
 789                 goto out;
 790         }
 791
 792         if (lpni->lpni_pref_nnids == 1 && lpni->lpni_pref.nid == nid) {
 793                 rc = -EEXIST;
 794                 goto out;
 795         }
 796
 797         /* A non-MR node may have only one preferred NI per peer_ni */
 798         if (lpni->lpni_pref_nnids > 0) {
 799                 if (!(lp->lp_state & LNET_PEER_MULTI_RAIL)) {
 800                         rc = -EPERM;
 801                         goto out;
 802                 }
 803         }
 804
 805         if (lpni->lpni_pref_nnids != 0) {
 806                 size = sizeof(*nids) * (lpni->lpni_pref_nnids + 1);
 807                 LIBCFS_CPT_ALLOC(nids, lnet_cpt_table(), lpni->lpni_cpt, size);
 808                 if (!nids) {
 809                         rc = -ENOMEM;
 810                         goto out;
 811                 }
 812                 for (i = 0; i < lpni->lpni_pref_nnids; i++) {
 813                         if (lpni->lpni_pref.nids[i] == nid) {
 814                                 LIBCFS_FREE(nids, size);
 815                                 rc = -EEXIST;
 816                                 goto out;
 817                         }
 818                         nids[i] = lpni->lpni_pref.nids[i];
 819                 }
 820                 nids[i] = nid;
 821         }
 822
 823         lnet_net_lock(LNET_LOCK_EX);
 824         spin_lock(&lpni->lpni_lock);
 825         if (lpni->lpni_pref_nnids == 0) {
 826                 lpni->lpni_pref.nid = nid;
 827         } else {
 828                 oldnids = lpni->lpni_pref.nids;
 829                 lpni->lpni_pref.nids = nids;
 830         }
 831         lpni->lpni_pref_nnids++;
 832         lpni->lpni_state &= ~LNET_PEER_NI_NON_MR_PREF;
 833         spin_unlock(&lpni->lpni_lock);
 834         lnet_net_unlock(LNET_LOCK_EX);
 835
 836         if (oldnids) {
 837                 size = sizeof(*nids) * (lpni->lpni_pref_nnids - 1);
 838                 LIBCFS_FREE(oldnids, sizeof(*oldnids) * size);
 839         }
 840 out:
 841         if (rc == -EEXIST && (lpni->lpni_state & LNET_PEER_NI_NON_MR_PREF)) {
 842                 spin_lock(&lpni->lpni_lock);
 843                 lpni->lpni_state &= ~LNET_PEER_NI_NON_MR_PREF;
 844                 spin_unlock(&lpni->lpni_lock);
 845         }
 846         CDEBUG(D_NET, "peer %s nid %s: %d\n",
 847                libcfs_nid2str(lp->lp_primary_nid), libcfs_nid2str(nid), rc);
 848         return rc;
 849 }
 850
 851 int
 852 lnet_peer_del_pref_nid(struct lnet_peer_ni *lpni, lnet_nid_t nid)
 853 {
 854         lnet_nid_t *nids = NULL;
 855         lnet_nid_t *oldnids = NULL;
 856         struct lnet_peer *lp = lpni->lpni_peer_net->lpn_peer;
 857         int size;
 858         int i, j;
 859         int rc = 0;
 860
 861         if (lpni->lpni_pref_nnids == 0) {
 862                 rc = -ENOENT;
 863                 goto out;
 864         }
 865
 866         if (lpni->lpni_pref_nnids == 1) {
 867                 if (lpni->lpni_pref.nid != nid) {
 868                         rc = -ENOENT;
 869                         goto out;
 870                 }
 871         } else if (lpni->lpni_pref_nnids == 2) {
 872                 if (lpni->lpni_pref.nids[0] != nid &&
 873                     lpni->lpni_pref.nids[1] != nid) {
 874                         rc = -ENOENT;
 875                         goto out;
 876                 }
 877         } else {
 878                 size = sizeof(*nids) * (lpni->lpni_pref_nnids - 1);
 879                 LIBCFS_CPT_ALLOC(nids, lnet_cpt_table(), lpni->lpni_cpt, size);
 880                 if (!nids) {
 881                         rc = -ENOMEM;
 882                         goto out;
 883                 }
 884                 for (i = 0, j = 0; i < lpni->lpni_pref_nnids; i++) {
 885                         if (lpni->lpni_pref.nids[i] != nid)
 886                                 continue;
 887                         nids[j++] = lpni->lpni_pref.nids[i];
 888                 }
 889                 /* Check if we actually removed a nid. */
 890                 if (j == lpni->lpni_pref_nnids) {
 891                         LIBCFS_FREE(nids, size);
 892                         rc = -ENOENT;
 893                         goto out;
 894                 }
 895         }
 896
 897         lnet_net_lock(LNET_LOCK_EX);
 898         spin_lock(&lpni->lpni_lock);
 899         if (lpni->lpni_pref_nnids == 1) {
 900                 lpni->lpni_pref.nid = LNET_NID_ANY;
 901         } else if (lpni->lpni_pref_nnids == 2) {
 902                 oldnids = lpni->lpni_pref.nids;
 903                 if (oldnids[0] == nid)
 904                         lpni->lpni_pref.nid = oldnids[1];
 905                 else
 906                         lpni->lpni_pref.nid = oldnids[2];
 907         } else {
 908                 oldnids = lpni->lpni_pref.nids;
 909                 lpni->lpni_pref.nids = nids;
 910         }
 911         lpni->lpni_pref_nnids--;
 912         lpni->lpni_state &= ~LNET_PEER_NI_NON_MR_PREF;
 913         spin_unlock(&lpni->lpni_lock);
 914         lnet_net_unlock(LNET_LOCK_EX);
 915
 916         if (oldnids) {
 917                 size = sizeof(*nids) * (lpni->lpni_pref_nnids + 1);
 918                 LIBCFS_FREE(oldnids, sizeof(*oldnids) * size);
 919         }
 920 out:
 921         CDEBUG(D_NET, "peer %s nid %s: %d\n",
 922                libcfs_nid2str(lp->lp_primary_nid), libcfs_nid2str(nid), rc);
 923         return rc;
 924 }
 925
 926 lnet_nid_t
 927 lnet_peer_primary_nid(lnet_nid_t nid)
 928 {
 929         struct lnet_peer_ni *lpni;
 930         lnet_nid_t primary_nid = nid;
 931         int cpt;
 932
 933         cpt = lnet_net_lock_current();
 934         lpni = lnet_find_peer_ni_locked(nid);
 935         if (lpni) {
 936                 primary_nid = lpni->lpni_peer_net->lpn_peer->lp_primary_nid;
 937                 lnet_peer_ni_decref_locked(lpni);
 938         }
 939         lnet_net_unlock(cpt);
 940
 941         return primary_nid;
 942 }
 943
 944 lnet_nid_t
 945 LNetPrimaryNID(lnet_nid_t nid)
 946 {
 947         struct lnet_peer_ni *lpni;
 948         lnet_nid_t primary_nid = nid;
 949         int rc = 0;
 950         int cpt;
 951
 952         cpt = lnet_net_lock_current();
 953         lpni = lnet_nid2peerni_locked(nid, LNET_NID_ANY, cpt);
 954         if (IS_ERR(lpni)) {
 955                 rc = PTR_ERR(lpni);
 956                 goto out_unlock;
 957         }
 958         primary_nid = lpni->lpni_peer_net->lpn_peer->lp_primary_nid;
 959         lnet_peer_ni_decref_locked(lpni);
 960 out_unlock:
 961         lnet_net_unlock(cpt);
 962
 963         CDEBUG(D_NET, "NID %s primary NID %s rc %d\n", libcfs_nid2str(nid),
 964                libcfs_nid2str(primary_nid), rc);
 965         return primary_nid;
 966 }
 967 EXPORT_SYMBOL(LNetPrimaryNID);
 968
 969 struct lnet_peer_net *
 970 lnet_peer_get_net_locked(struct lnet_peer *peer, __u32 net_id)
 971 {
 972         struct lnet_peer_net *peer_net;
 973         list_for_each_entry(peer_net, &peer->lp_peer_nets, lpn_peer_nets) {
 974                 if (peer_net->lpn_net_id == net_id)
 975                         return peer_net;
 976         }
 977         return NULL;
 978 }
 979
 980 /*
 981  * Attach a peer_ni to a peer_net and peer. This function assumes
 982  * peer_ni is not already attached to the peer_net/peer. The peer_ni
 983  * may be attached to a different peer, in which case it will be
 984  * properly detached first. The whole operation is done atomically.
 985  *
 986  * Always returns 0.  This is the last function called from functions
 987  * that do return an int, so returning 0 here allows the compiler to
 988  * do a tail call.
 989  */
 990 static int
 991 lnet_peer_attach_peer_ni(struct lnet_peer *lp,
 992                                 struct lnet_peer_net *lpn,
 993                                 struct lnet_peer_ni *lpni,
 994                                 unsigned flags)
 995 {
 996         struct lnet_peer_table *ptable;
 997
 998         /* Install the new peer_ni */
 999         lnet_net_lock(LNET_LOCK_EX);
1000         /* Add peer_ni to global peer table hash, if necessary. */
1001         if (list_empty(&lpni->lpni_hashlist)) {
1002                 int hash = lnet_nid2peerhash(lpni->lpni_nid);
1003
1004                 ptable = the_lnet.ln_peer_tables[lpni->lpni_cpt];
1005                 list_add_tail(&lpni->lpni_hashlist, &ptable->pt_hash[hash]);
1006                 ptable->pt_version++;
1007                 ptable->pt_number++;
1008                 /* This is the 1st refcount on lpni. */
1009                 atomic_inc(&lpni->lpni_refcount);
1010         }
1011
1012         /* Detach the peer_ni from an existing peer, if necessary. */
1013         if (lpni->lpni_peer_net) {
1014                 LASSERT(lpni->lpni_peer_net != lpn);
1015                 LASSERT(lpni->lpni_peer_net->lpn_peer != lp);
1016                 lnet_peer_detach_peer_ni_locked(lpni);
1017                 lnet_peer_net_decref_locked(lpni->lpni_peer_net);
1018                 lpni->lpni_peer_net = NULL;
1019         }
1020
1021         /* Add peer_ni to peer_net */
1022         lpni->lpni_peer_net = lpn;
1023         list_add_tail(&lpni->lpni_peer_nis, &lpn->lpn_peer_nis);
1024         lnet_peer_net_addref_locked(lpn);
1025
1026         /* Add peer_net to peer */
1027         if (!lpn->lpn_peer) {
1028                 lpn->lpn_peer = lp;
1029                 list_add_tail(&lpn->lpn_peer_nets, &lp->lp_peer_nets);
1030                 lnet_peer_addref_locked(lp);
1031         }
1032
1033         /* Add peer to global peer list, if necessary */
1034         ptable = the_lnet.ln_peer_tables[lp->lp_cpt];
1035         if (list_empty(&lp->lp_peer_list)) {
1036                 list_add_tail(&lp->lp_peer_list, &ptable->pt_peer_list);
1037                 ptable->pt_peers++;
1038         }
1039
1040
1041         /* Update peer state */
1042         spin_lock(&lp->lp_lock);
1043         if (flags & LNET_PEER_CONFIGURED) {
1044                 if (!(lp->lp_state & LNET_PEER_CONFIGURED))
1045                         lp->lp_state |= LNET_PEER_CONFIGURED;
1046         }
1047         if (flags & LNET_PEER_MULTI_RAIL) {
1048                 if (!(lp->lp_state & LNET_PEER_MULTI_RAIL)) {
1049                         lp->lp_state |= LNET_PEER_MULTI_RAIL;
1050                         lnet_peer_clr_non_mr_pref_nids(lp);
1051                 }
1052         }
1053         spin_unlock(&lp->lp_lock);
1054
1055         lp->lp_nnis++;
1056         the_lnet.ln_peer_tables[lp->lp_cpt]->pt_peer_nnids++;
1057         lnet_net_unlock(LNET_LOCK_EX);
1058
1059         CDEBUG(D_NET, "peer %s NID %s flags %#x\n",
1060                libcfs_nid2str(lp->lp_primary_nid),
1061                libcfs_nid2str(lpni->lpni_nid), flags);
1062
1063         return 0;
1064 }
1065
1066 /*
1067  * Create a new peer, with nid as its primary nid.
1068  *
1069  * Call with the lnet_api_mutex held.
1070  */
1071 static int
1072 lnet_peer_add(lnet_nid_t nid, unsigned flags)
1073 {
1074         struct lnet_peer *lp;
1075         struct lnet_peer_net *lpn;
1076         struct lnet_peer_ni *lpni;
1077         int rc = 0;
1078
1079         LASSERT(nid != LNET_NID_ANY);
1080
1081         /*
1082          * No need for the lnet_net_lock here, because the
1083          * lnet_api_mutex is held.
1084          */
1085         lpni = lnet_find_peer_ni_locked(nid);
1086         if (lpni) {
1087                 /* A peer with this NID already exists. */
1088                 lp = lpni->lpni_peer_net->lpn_peer;
1089                 lnet_peer_ni_decref_locked(lpni);
1090                 /*
1091                  * This is an error if the peer was configured and the
1092                  * primary NID differs or an attempt is made to change
1093                  * the Multi-Rail flag. Otherwise the assumption is
1094                  * that an existing peer is being modified.
1095                  */
1096                 if (lp->lp_state & LNET_PEER_CONFIGURED) {
1097                         if (lp->lp_primary_nid != nid)
1098                                 rc = -EEXIST;
1099                         else if ((lp->lp_state ^ flags) & LNET_PEER_MULTI_RAIL)
1100                                 rc = -EPERM;
1101                         goto out;
1102                 }
1103                 /* Delete and recreate as a configured peer. */
1104                 lnet_peer_del(lp);
1105         }
1106
1107         /* Create peer, peer_net, and peer_ni. */
1108         rc = -ENOMEM;
1109         lp = lnet_peer_alloc(nid);
1110         if (!lp)
1111                 goto out;
1112         lpn = lnet_peer_net_alloc(LNET_NIDNET(nid));
1113         if (!lpn)
1114                 goto out_free_lp;
1115         lpni = lnet_peer_ni_alloc(nid);
1116         if (!lpni)
1117                 goto out_free_lpn;
1118
1119         return lnet_peer_attach_peer_ni(lp, lpn, lpni, flags);
1120
1121 out_free_lpn:
1122         LIBCFS_FREE(lpn, sizeof(*lpn));
1123 out_free_lp:
1124         LIBCFS_FREE(lp, sizeof(*lp));
1125 out:
1126         CDEBUG(D_NET, "peer %s NID flags %#x: %d\n",
1127                libcfs_nid2str(nid), flags, rc);
1128         return rc;
1129 }
1130
1131 /*
1132  * Add a NID to a peer. Call with ln_api_mutex held.
1133  *
1134  * Error codes:
1135  *  -EPERM:    Non-DLC addition to a DLC-configured peer.
1136  *  -EEXIST:   The NID was configured by DLC for a different peer.
1137  *  -ENOMEM:   Out of memory.
1138  *  -ENOTUNIQ: Adding a second peer NID on a single network on a
1139  *             non-multi-rail peer.
1140  */
1141 static int
1142 lnet_peer_add_nid(struct lnet_peer *lp, lnet_nid_t nid, unsigned flags)
1143 {
1144         struct lnet_peer_net *lpn;
1145         struct lnet_peer_ni *lpni;
1146         int rc = 0;
1147
1148         LASSERT(lp);
1149         LASSERT(nid != LNET_NID_ANY);
1150
1151         /* A configured peer can only be updated through configuration. */
1152         if (!(flags & LNET_PEER_CONFIGURED)) {
1153                 if (lp->lp_state & LNET_PEER_CONFIGURED) {
1154                         rc = -EPERM;
1155                         goto out;
1156                 }
1157         }
1158
1159         /*
1160          * The MULTI_RAIL flag can be set but not cleared, because
1161          * that would leave the peer struct in an invalid state.
1162          */
1163         if (flags & LNET_PEER_MULTI_RAIL) {
1164                 spin_lock(&lp->lp_lock);
1165                 if (!(lp->lp_state & LNET_PEER_MULTI_RAIL)) {
1166                         lp->lp_state |= LNET_PEER_MULTI_RAIL;
1167                         lnet_peer_clr_non_mr_pref_nids(lp);
1168                 }
1169                 spin_unlock(&lp->lp_lock);
1170         } else if (lp->lp_state & LNET_PEER_MULTI_RAIL) {
1171                 rc = -EPERM;
1172                 goto out;
1173         }
1174
1175         lpni = lnet_find_peer_ni_locked(nid);
1176         if (lpni) {
1177                 /*
1178                  * A peer_ni already exists. This is only a problem if
1179                  * it is not connected to this peer and was configured
1180                  * by DLC.
1181                  */
1182                 lnet_peer_ni_decref_locked(lpni);
1183                 if (lpni->lpni_peer_net->lpn_peer == lp)
1184                         goto out;
1185                 if (lnet_peer_ni_is_configured(lpni)) {
1186                         rc = -EEXIST;
1187                         goto out;
1188                 }
1189                 /* If this is the primary NID, destroy the peer. */
1190                 if (lnet_peer_ni_is_primary(lpni)) {
1191                         lnet_peer_del(lpni->lpni_peer_net->lpn_peer);
1192                         lpni = lnet_peer_ni_alloc(nid);
1193                         if (!lpni) {
1194                                 rc = -ENOMEM;
1195                                 goto out;
1196                         }
1197                 }
1198         } else {
1199                 lpni = lnet_peer_ni_alloc(nid);
1200                 if (!lpni) {
1201                         rc = -ENOMEM;
1202                         goto out;
1203                 }
1204         }
1205
1206         /*
1207          * Get the peer_net. Check that we're not adding a second
1208          * peer_ni on a peer_net of a non-multi-rail peer.
1209          */
1210         lpn = lnet_peer_get_net_locked(lp, LNET_NIDNET(nid));
1211         if (!lpn) {
1212                 lpn = lnet_peer_net_alloc(LNET_NIDNET(nid));
1213                 if (!lpn) {
1214                         rc = -ENOMEM;
1215                         goto out_free_lpni;
1216                 }
1217         } else if (!(lp->lp_state & LNET_PEER_MULTI_RAIL)) {
1218                 rc = -ENOTUNIQ;
1219                 goto out_free_lpni;
1220         }
1221
1222         return lnet_peer_attach_peer_ni(lp, lpn, lpni, flags);
1223
1224 out_free_lpni:
1225         /* If the peer_ni was allocated above its peer_net pointer is NULL */
1226         if (!lpni->lpni_peer_net)
1227                 LIBCFS_FREE(lpni, sizeof(*lpni));
1228 out:
1229         CDEBUG(D_NET, "peer %s NID %s flags %#x: %d\n",
1230                libcfs_nid2str(lp->lp_primary_nid), libcfs_nid2str(nid),
1231                flags, rc);
1232         return rc;
1233 }
1234
1235 /*
1236  * lpni creation initiated due to traffic either sending or receiving.
1237  */
1238 static int
1239 lnet_peer_ni_traffic_add(lnet_nid_t nid, lnet_nid_t pref)
1240 {
1241         struct lnet_peer *lp;
1242         struct lnet_peer_net *lpn;
1243         struct lnet_peer_ni *lpni;
1244         unsigned flags = 0;
1245         int rc = 0;
1246
1247         if (nid == LNET_NID_ANY) {
1248                 rc = -EINVAL;
1249                 goto out;
1250         }
1251
1252         /* lnet_net_lock is not needed here because ln_api_lock is held */
1253         lpni = lnet_find_peer_ni_locked(nid);
1254         if (lpni) {
1255                 /*
1256                  * We must have raced with another thread. Since we
1257                  * know next to nothing about a peer_ni created by
1258                  * traffic, we just assume everything is ok and
1259                  * return.
1260                  */
1261                 lnet_peer_ni_decref_locked(lpni);
1262                 goto out;
1263         }
1264
1265         /* Create peer, peer_net, and peer_ni. */
1266         rc = -ENOMEM;
1267         lp = lnet_peer_alloc(nid);
1268         if (!lp)
1269                 goto out;
1270         lpn = lnet_peer_net_alloc(LNET_NIDNET(nid));
1271         if (!lpn)
1272                 goto out_free_lp;
1273         lpni = lnet_peer_ni_alloc(nid);
1274         if (!lpni)
1275                 goto out_free_lpn;
1276         if (pref != LNET_NID_ANY)
1277                 lnet_peer_ni_set_non_mr_pref_nid(lpni, pref);
1278
1279         return lnet_peer_attach_peer_ni(lp, lpn, lpni, flags);
1280
1281 out_free_lpn:
1282         LIBCFS_FREE(lpn, sizeof(*lpn));
1283 out_free_lp:
1284         LIBCFS_FREE(lp, sizeof(*lp));
1285 out:
1286         CDEBUG(D_NET, "peer %s: %d\n", libcfs_nid2str(nid), rc);
1287         return rc;
1288 }
1289
1290 /*
1291  * Implementation of IOC_LIBCFS_ADD_PEER_NI.
1292  *
1293  * This API handles the following combinations:
1294  *   Create a peer with its primary NI if only the prim_nid is provided
1295  *   Add a NID to a peer identified by the prim_nid. The peer identified
1296  *   by the prim_nid must already exist.
1297  *   The peer being created may be non-MR.
1298  *
1299  * The caller must hold ln_api_mutex. This prevents the peer from
1300  * being created/modified/deleted by a different thread.
1301  */
1302 int
1303 lnet_add_peer_ni(lnet_nid_t prim_nid, lnet_nid_t nid, bool mr)
1304 {
1305         struct lnet_peer *lp = NULL;
1306         struct lnet_peer_ni *lpni;
1307         unsigned flags;
1308
1309         /* The prim_nid must always be specified */
1310         if (prim_nid == LNET_NID_ANY)
1311                 return -EINVAL;
1312
1313         flags = LNET_PEER_CONFIGURED;
1314         if (mr)
1315                 flags |= LNET_PEER_MULTI_RAIL;
1316
1317         /*
1318          * If nid isn't specified, we must create a new peer with
1319          * prim_nid as its primary nid.
1320          */
1321         if (nid == LNET_NID_ANY)
1322                 return lnet_peer_add(prim_nid, flags);
1323
1324         /* Look up the prim_nid, which must exist. */
1325         lpni = lnet_find_peer_ni_locked(prim_nid);
1326         if (!lpni)
1327                 return -ENOENT;
1328         lnet_peer_ni_decref_locked(lpni);
1329         lp = lpni->lpni_peer_net->lpn_peer;
1330
1331         /* Peer must have been configured. */
1332         if (!(lp->lp_state & LNET_PEER_CONFIGURED)) {
1333                 CDEBUG(D_NET, "peer %s was not configured\n",
1334                        libcfs_nid2str(prim_nid));
1335                 return -ENOENT;
1336         }
1337
1338         /* Primary NID must match */
1339         if (lp->lp_primary_nid != prim_nid) {
1340                 CDEBUG(D_NET, "prim_nid %s is not primary for peer %s\n",
1341                        libcfs_nid2str(prim_nid),
1342                        libcfs_nid2str(lp->lp_primary_nid));
1343                 return -ENODEV;
1344         }
1345
1346         /* Multi-Rail flag must match. */
1347         if ((lp->lp_state ^ flags) & LNET_PEER_MULTI_RAIL) {
1348                 CDEBUG(D_NET, "multi-rail state mismatch for peer %s\n",
1349                        libcfs_nid2str(prim_nid));
1350                 return -EPERM;
1351         }
1352
1353         return lnet_peer_add_nid(lp, nid, flags);
1354 }
1355
1356 /*
1357  * Implementation of IOC_LIBCFS_DEL_PEER_NI.
1358  *
1359  * This API handles the following combinations:
1360  *   Delete a NI from a peer if both prim_nid and nid are provided.
1361  *   Delete a peer if only prim_nid is provided.
1362  *   Delete a peer if its primary nid is provided.
1363  *
1364  * The caller must hold ln_api_mutex. This prevents the peer from
1365  * being modified/deleted by a different thread.
1366  */
1367 int
1368 lnet_del_peer_ni(lnet_nid_t prim_nid, lnet_nid_t nid)
1369 {
1370         struct lnet_peer *lp;
1371         struct lnet_peer_ni *lpni;
1372         unsigned flags;
1373
1374         if (prim_nid == LNET_NID_ANY)
1375                 return -EINVAL;
1376
1377         lpni = lnet_find_peer_ni_locked(prim_nid);
1378         if (!lpni)
1379                 return -ENOENT;
1380         lnet_peer_ni_decref_locked(lpni);
1381         lp = lpni->lpni_peer_net->lpn_peer;
1382
1383         if (prim_nid != lp->lp_primary_nid) {
1384                 CDEBUG(D_NET, "prim_nid %s is not primary for peer %s\n",
1385                        libcfs_nid2str(prim_nid),
1386                        libcfs_nid2str(lp->lp_primary_nid));
1387                 return -ENODEV;
1388         }
1389
1390         if (nid == LNET_NID_ANY || nid == lp->lp_primary_nid)
1391                 return lnet_peer_del(lp);
1392
1393         flags = LNET_PEER_CONFIGURED;
1394         if (lp->lp_state & LNET_PEER_MULTI_RAIL)
1395                 flags |= LNET_PEER_MULTI_RAIL;
1396
1397         return lnet_peer_del_nid(lp, nid, flags);
1398 }
1399
1400 void
1401 lnet_destroy_peer_ni_locked(struct lnet_peer_ni *lpni)
1402 {
1403         struct lnet_peer_table *ptable;
1404         struct lnet_peer_net *lpn;
1405
1406         CDEBUG(D_NET, "%p nid %s\n", lpni, libcfs_nid2str(lpni->lpni_nid));
1407
1408         LASSERT(atomic_read(&lpni->lpni_refcount) == 0);
1409         LASSERT(lpni->lpni_rtr_refcount == 0);
1410         LASSERT(list_empty(&lpni->lpni_txq));
1411         LASSERT(lpni->lpni_txqnob == 0);
1412
1413         lpn = lpni->lpni_peer_net;
1414         lpni->lpni_peer_net = NULL;
1415         lpni->lpni_net = NULL;
1416
1417         /* remove the peer ni from the zombie list */
1418         ptable = the_lnet.ln_peer_tables[lpni->lpni_cpt];
1419         spin_lock(&ptable->pt_zombie_lock);
1420         list_del_init(&lpni->lpni_hashlist);
1421         ptable->pt_zombies--;
1422         spin_unlock(&ptable->pt_zombie_lock);
1423
1424         if (lpni->lpni_pref_nnids > 1) {
1425                 LIBCFS_FREE(lpni->lpni_pref.nids,
1426                         sizeof(*lpni->lpni_pref.nids) * lpni->lpni_pref_nnids);
1427         }
1428         LIBCFS_FREE(lpni, sizeof(*lpni));
1429
1430         lnet_peer_net_decref_locked(lpn);
1431 }
1432
1433 struct lnet_peer_ni *
1434 lnet_nid2peerni_ex(lnet_nid_t nid, int cpt)
1435 {
1436         struct lnet_peer_ni *lpni = NULL;
1437         int rc;
1438
1439         if (the_lnet.ln_state != LNET_STATE_RUNNING)
1440                 return ERR_PTR(-ESHUTDOWN);
1441
1442         /*
1443          * find if a peer_ni already exists.
1444          * If so then just return that.
1445          */
1446         lpni = lnet_find_peer_ni_locked(nid);
1447         if (lpni)
1448                 return lpni;
1449
1450         lnet_net_unlock(cpt);
1451
1452         rc = lnet_peer_ni_traffic_add(nid, LNET_NID_ANY);
1453         if (rc) {
1454                 lpni = ERR_PTR(rc);
1455                 goto out_net_relock;
1456         }
1457
1458         lpni = lnet_find_peer_ni_locked(nid);
1459         LASSERT(lpni);
1460
1461 out_net_relock:
1462         lnet_net_lock(cpt);
1463
1464         return lpni;
1465 }
1466
1467 struct lnet_peer_ni *
1468 lnet_nid2peerni_locked(lnet_nid_t nid, lnet_nid_t pref, int cpt)
1469 {
1470         struct lnet_peer_ni *lpni = NULL;
1471         int rc;
1472
1473         if (the_lnet.ln_state != LNET_STATE_RUNNING)
1474                 return ERR_PTR(-ESHUTDOWN);
1475
1476         /*
1477          * find if a peer_ni already exists.
1478          * If so then just return that.
1479          */
1480         lpni = lnet_find_peer_ni_locked(nid);
1481         if (lpni)
1482                 return lpni;
1483
1484         /*
1485          * Slow path:
1486          * use the lnet_api_mutex to serialize the creation of the peer_ni
1487          * and the creation/deletion of the local ni/net. When a local ni is
1488          * created, if there exists a set of peer_nis on that network,
1489          * they need to be traversed and updated. When a local NI is
1490          * deleted, which could result in a network being deleted, then
1491          * all peer nis on that network need to be removed as well.
1492          *
1493          * Creation through traffic should also be serialized with
1494          * creation through DLC.
1495          */
1496         lnet_net_unlock(cpt);
1497         mutex_lock(&the_lnet.ln_api_mutex);
1498         /*
1499          * Shutdown is only set under the ln_api_lock, so a single
1500          * check here is sufficent.
1501          */
1502         if (the_lnet.ln_state != LNET_STATE_RUNNING) {
1503                 lpni = ERR_PTR(-ESHUTDOWN);
1504                 goto out_mutex_unlock;
1505         }
1506
1507         rc = lnet_peer_ni_traffic_add(nid, pref);
1508         if (rc) {
1509                 lpni = ERR_PTR(rc);
1510                 goto out_mutex_unlock;
1511         }
1512
1513         lpni = lnet_find_peer_ni_locked(nid);
1514         LASSERT(lpni);
1515
1516 out_mutex_unlock:
1517         mutex_unlock(&the_lnet.ln_api_mutex);
1518         lnet_net_lock(cpt);
1519
1520         return lpni;
1521 }
1522
1523 void
1524 lnet_debug_peer(lnet_nid_t nid)
1525 {
1526         char                    *aliveness = "NA";
1527         struct lnet_peer_ni     *lp;
1528         int                     cpt;
1529
1530         cpt = lnet_cpt_of_nid(nid, NULL);
1531         lnet_net_lock(cpt);
1532
1533         lp = lnet_nid2peerni_locked(nid, LNET_NID_ANY, cpt);
1534         if (IS_ERR(lp)) {
1535                 lnet_net_unlock(cpt);
1536                 CDEBUG(D_WARNING, "No peer %s\n", libcfs_nid2str(nid));
1537                 return;
1538         }
1539
1540         if (lnet_isrouter(lp) || lnet_peer_aliveness_enabled(lp))
1541                 aliveness = lp->lpni_alive ? "up" : "down";
1542
1543         CDEBUG(D_WARNING, "%-24s %4d %5s %5d %5d %5d %5d %5d %ld\n",
1544                libcfs_nid2str(lp->lpni_nid), atomic_read(&lp->lpni_refcount),
1545                aliveness, lp->lpni_net->net_tunables.lct_peer_tx_credits,
1546                lp->lpni_rtrcredits, lp->lpni_minrtrcredits,
1547                lp->lpni_txcredits, lp->lpni_mintxcredits, lp->lpni_txqnob);
1548
1549         lnet_peer_ni_decref_locked(lp);
1550
1551         lnet_net_unlock(cpt);
1552 }
1553
1554 int lnet_get_peer_ni_info(__u32 peer_index, __u64 *nid,
1555                           char aliveness[LNET_MAX_STR_LEN],
1556                           __u32 *cpt_iter, __u32 *refcount,
1557                           __u32 *ni_peer_tx_credits, __u32 *peer_tx_credits,
1558                           __u32 *peer_rtr_credits, __u32 *peer_min_rtr_credits,
1559                           __u32 *peer_tx_qnob)
1560 {
1561         struct lnet_peer_table          *peer_table;
1562         struct lnet_peer_ni             *lp;
1563         int                             j;
1564         int                             lncpt;
1565         bool                            found = false;
1566
1567         /* get the number of CPTs */
1568         lncpt = cfs_percpt_number(the_lnet.ln_peer_tables);
1569
1570         /* if the cpt number to be examined is >= the number of cpts in
1571          * the system then indicate that there are no more cpts to examin
1572          */
1573         if (*cpt_iter >= lncpt)
1574                 return -ENOENT;
1575
1576         /* get the current table */
1577         peer_table = the_lnet.ln_peer_tables[*cpt_iter];
1578         /* if the ptable is NULL then there are no more cpts to examine */
1579         if (peer_table == NULL)
1580                 return -ENOENT;
1581
1582         lnet_net_lock(*cpt_iter);
1583
1584         for (j = 0; j < LNET_PEER_HASH_SIZE && !found; j++) {
1585                 struct list_head *peers = &peer_table->pt_hash[j];
1586
1587                 list_for_each_entry(lp, peers, lpni_hashlist) {
1588                         if (peer_index-- > 0)
1589                                 continue;
1590
1591                         snprintf(aliveness, LNET_MAX_STR_LEN, "NA");
1592                         if (lnet_isrouter(lp) ||
1593                                 lnet_peer_aliveness_enabled(lp))
1594                                 snprintf(aliveness, LNET_MAX_STR_LEN,
1595                                          lp->lpni_alive ? "up" : "down");
1596
1597                         *nid = lp->lpni_nid;
1598                         *refcount = atomic_read(&lp->lpni_refcount);
1599                         *ni_peer_tx_credits =
1600                                 lp->lpni_net->net_tunables.lct_peer_tx_credits;
1601                         *peer_tx_credits = lp->lpni_txcredits;
1602                         *peer_rtr_credits = lp->lpni_rtrcredits;
1603                         *peer_min_rtr_credits = lp->lpni_mintxcredits;
1604                         *peer_tx_qnob = lp->lpni_txqnob;
1605
1606                         found = true;
1607                 }
1608
1609         }
1610         lnet_net_unlock(*cpt_iter);
1611
1612         *cpt_iter = lncpt;
1613
1614         return found ? 0 : -ENOENT;
1615 }
1616
1617 /* ln_api_mutex is held, which keeps the peer list stable */
1618 int lnet_get_peer_info(__u32 idx, lnet_nid_t *primary_nid, lnet_nid_t *nid,
1619                        bool *mr, struct lnet_peer_ni_credit_info *peer_ni_info,
1620                        struct lnet_ioctl_element_stats *peer_ni_stats)
1621 {
1622         struct lnet_peer_ni *lpni = NULL;
1623         struct lnet_peer_net *lpn = NULL;
1624         struct lnet_peer *lp = NULL;
1625
1626         lpni = lnet_get_peer_ni_idx_locked(idx, &lpn, &lp);
1627
1628         if (!lpni)
1629                 return -ENOENT;
1630
1631         *primary_nid = lp->lp_primary_nid;
1632         *mr = lnet_peer_is_multi_rail(lp);
1633         *nid = lpni->lpni_nid;
1634         snprintf(peer_ni_info->cr_aliveness, LNET_MAX_STR_LEN, "NA");
1635         if (lnet_isrouter(lpni) ||
1636                 lnet_peer_aliveness_enabled(lpni))
1637                 snprintf(peer_ni_info->cr_aliveness, LNET_MAX_STR_LEN,
1638                          lpni->lpni_alive ? "up" : "down");
1639
1640         peer_ni_info->cr_refcount = atomic_read(&lpni->lpni_refcount);
1641         peer_ni_info->cr_ni_peer_tx_credits = (lpni->lpni_net != NULL) ?
1642                 lpni->lpni_net->net_tunables.lct_peer_tx_credits : 0;
1643         peer_ni_info->cr_peer_tx_credits = lpni->lpni_txcredits;
1644         peer_ni_info->cr_peer_rtr_credits = lpni->lpni_rtrcredits;
1645         peer_ni_info->cr_peer_min_rtr_credits = lpni->lpni_minrtrcredits;
1646         peer_ni_info->cr_peer_min_tx_credits = lpni->lpni_mintxcredits;
1647         peer_ni_info->cr_peer_tx_qnob = lpni->lpni_txqnob;
1648
1649         peer_ni_stats->send_count = atomic_read(&lpni->lpni_stats.send_count);
1650         peer_ni_stats->recv_count = atomic_read(&lpni->lpni_stats.recv_count);
1651         peer_ni_stats->drop_count = atomic_read(&lpni->lpni_stats.drop_count);
1652
1653         return 0;
1654 }