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 <uapi/linux/lnet/lnet-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_power_of_2(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_locked(lnet_nid_t nid)
 928 {
 929         struct lnet_peer_ni *lpni;
 930         lnet_nid_t primary_nid = nid;
 931
 932         lpni = lnet_find_peer_ni_locked(nid);
 933         if (lpni) {
 934                 primary_nid = lpni->lpni_peer_net->lpn_peer->lp_primary_nid;
 935                 lnet_peer_ni_decref_locked(lpni);
 936         }
 937
 938         return primary_nid;
 939 }
 940
 941 lnet_nid_t
 942 LNetPrimaryNID(lnet_nid_t nid)
 943 {
 944         struct lnet_peer_ni *lpni;
 945         lnet_nid_t primary_nid = nid;
 946         int rc = 0;
 947         int cpt;
 948
 949         cpt = lnet_net_lock_current();
 950         lpni = lnet_nid2peerni_locked(nid, LNET_NID_ANY, cpt);
 951         if (IS_ERR(lpni)) {
 952                 rc = PTR_ERR(lpni);
 953                 goto out_unlock;
 954         }
 955         primary_nid = lpni->lpni_peer_net->lpn_peer->lp_primary_nid;
 956         lnet_peer_ni_decref_locked(lpni);
 957 out_unlock:
 958         lnet_net_unlock(cpt);
 959
 960         CDEBUG(D_NET, "NID %s primary NID %s rc %d\n", libcfs_nid2str(nid),
 961                libcfs_nid2str(primary_nid), rc);
 962         return primary_nid;
 963 }
 964 EXPORT_SYMBOL(LNetPrimaryNID);
 965
 966 struct lnet_peer_net *
 967 lnet_peer_get_net_locked(struct lnet_peer *peer, __u32 net_id)
 968 {
 969         struct lnet_peer_net *peer_net;
 970         list_for_each_entry(peer_net, &peer->lp_peer_nets, lpn_peer_nets) {
 971                 if (peer_net->lpn_net_id == net_id)
 972                         return peer_net;
 973         }
 974         return NULL;
 975 }
 976
 977 /*
 978  * Attach a peer_ni to a peer_net and peer. This function assumes
 979  * peer_ni is not already attached to the peer_net/peer. The peer_ni
 980  * may be attached to a different peer, in which case it will be
 981  * properly detached first. The whole operation is done atomically.
 982  *
 983  * Always returns 0.  This is the last function called from functions
 984  * that do return an int, so returning 0 here allows the compiler to
 985  * do a tail call.
 986  */
 987 static int
 988 lnet_peer_attach_peer_ni(struct lnet_peer *lp,
 989                                 struct lnet_peer_net *lpn,
 990                                 struct lnet_peer_ni *lpni,
 991                                 unsigned flags)
 992 {
 993         struct lnet_peer_table *ptable;
 994
 995         /* Install the new peer_ni */
 996         lnet_net_lock(LNET_LOCK_EX);
 997         /* Add peer_ni to global peer table hash, if necessary. */
 998         if (list_empty(&lpni->lpni_hashlist)) {
 999                 int hash = lnet_nid2peerhash(lpni->lpni_nid);
1000
1001                 ptable = the_lnet.ln_peer_tables[lpni->lpni_cpt];
1002                 list_add_tail(&lpni->lpni_hashlist, &ptable->pt_hash[hash]);
1003                 ptable->pt_version++;
1004                 ptable->pt_number++;
1005                 /* This is the 1st refcount on lpni. */
1006                 atomic_inc(&lpni->lpni_refcount);
1007         }
1008
1009         /* Detach the peer_ni from an existing peer, if necessary. */
1010         if (lpni->lpni_peer_net) {
1011                 LASSERT(lpni->lpni_peer_net != lpn);
1012                 LASSERT(lpni->lpni_peer_net->lpn_peer != lp);
1013                 lnet_peer_detach_peer_ni_locked(lpni);
1014                 lnet_peer_net_decref_locked(lpni->lpni_peer_net);
1015                 lpni->lpni_peer_net = NULL;
1016         }
1017
1018         /* Add peer_ni to peer_net */
1019         lpni->lpni_peer_net = lpn;
1020         list_add_tail(&lpni->lpni_peer_nis, &lpn->lpn_peer_nis);
1021         lnet_peer_net_addref_locked(lpn);
1022
1023         /* Add peer_net to peer */
1024         if (!lpn->lpn_peer) {
1025                 lpn->lpn_peer = lp;
1026                 list_add_tail(&lpn->lpn_peer_nets, &lp->lp_peer_nets);
1027                 lnet_peer_addref_locked(lp);
1028         }
1029
1030         /* Add peer to global peer list, if necessary */
1031         ptable = the_lnet.ln_peer_tables[lp->lp_cpt];
1032         if (list_empty(&lp->lp_peer_list)) {
1033                 list_add_tail(&lp->lp_peer_list, &ptable->pt_peer_list);
1034                 ptable->pt_peers++;
1035         }
1036
1037
1038         /* Update peer state */
1039         spin_lock(&lp->lp_lock);
1040         if (flags & LNET_PEER_CONFIGURED) {
1041                 if (!(lp->lp_state & LNET_PEER_CONFIGURED))
1042                         lp->lp_state |= LNET_PEER_CONFIGURED;
1043         }
1044         if (flags & LNET_PEER_MULTI_RAIL) {
1045                 if (!(lp->lp_state & LNET_PEER_MULTI_RAIL)) {
1046                         lp->lp_state |= LNET_PEER_MULTI_RAIL;
1047                         lnet_peer_clr_non_mr_pref_nids(lp);
1048                 }
1049         }
1050         spin_unlock(&lp->lp_lock);
1051
1052         lp->lp_nnis++;
1053         the_lnet.ln_peer_tables[lp->lp_cpt]->pt_peer_nnids++;
1054         lnet_net_unlock(LNET_LOCK_EX);
1055
1056         CDEBUG(D_NET, "peer %s NID %s flags %#x\n",
1057                libcfs_nid2str(lp->lp_primary_nid),
1058                libcfs_nid2str(lpni->lpni_nid), flags);
1059
1060         return 0;
1061 }
1062
1063 /*
1064  * Create a new peer, with nid as its primary nid.
1065  *
1066  * Call with the lnet_api_mutex held.
1067  */
1068 static int
1069 lnet_peer_add(lnet_nid_t nid, unsigned flags)
1070 {
1071         struct lnet_peer *lp;
1072         struct lnet_peer_net *lpn;
1073         struct lnet_peer_ni *lpni;
1074         int rc = 0;
1075
1076         LASSERT(nid != LNET_NID_ANY);
1077
1078         /*
1079          * No need for the lnet_net_lock here, because the
1080          * lnet_api_mutex is held.
1081          */
1082         lpni = lnet_find_peer_ni_locked(nid);
1083         if (lpni) {
1084                 /* A peer with this NID already exists. */
1085                 lp = lpni->lpni_peer_net->lpn_peer;
1086                 lnet_peer_ni_decref_locked(lpni);
1087                 /*
1088                  * This is an error if the peer was configured and the
1089                  * primary NID differs or an attempt is made to change
1090                  * the Multi-Rail flag. Otherwise the assumption is
1091                  * that an existing peer is being modified.
1092                  */
1093                 if (lp->lp_state & LNET_PEER_CONFIGURED) {
1094                         if (lp->lp_primary_nid != nid)
1095                                 rc = -EEXIST;
1096                         else if ((lp->lp_state ^ flags) & LNET_PEER_MULTI_RAIL)
1097                                 rc = -EPERM;
1098                         goto out;
1099                 }
1100                 /* Delete and recreate as a configured peer. */
1101                 lnet_peer_del(lp);
1102         }
1103
1104         /* Create peer, peer_net, and peer_ni. */
1105         rc = -ENOMEM;
1106         lp = lnet_peer_alloc(nid);
1107         if (!lp)
1108                 goto out;
1109         lpn = lnet_peer_net_alloc(LNET_NIDNET(nid));
1110         if (!lpn)
1111                 goto out_free_lp;
1112         lpni = lnet_peer_ni_alloc(nid);
1113         if (!lpni)
1114                 goto out_free_lpn;
1115
1116         return lnet_peer_attach_peer_ni(lp, lpn, lpni, flags);
1117
1118 out_free_lpn:
1119         LIBCFS_FREE(lpn, sizeof(*lpn));
1120 out_free_lp:
1121         LIBCFS_FREE(lp, sizeof(*lp));
1122 out:
1123         CDEBUG(D_NET, "peer %s NID flags %#x: %d\n",
1124                libcfs_nid2str(nid), flags, rc);
1125         return rc;
1126 }
1127
1128 /*
1129  * Add a NID to a peer. Call with ln_api_mutex held.
1130  *
1131  * Error codes:
1132  *  -EPERM:    Non-DLC addition to a DLC-configured peer.
1133  *  -EEXIST:   The NID was configured by DLC for a different peer.
1134  *  -ENOMEM:   Out of memory.
1135  *  -ENOTUNIQ: Adding a second peer NID on a single network on a
1136  *             non-multi-rail peer.
1137  */
1138 static int
1139 lnet_peer_add_nid(struct lnet_peer *lp, lnet_nid_t nid, unsigned flags)
1140 {
1141         struct lnet_peer_net *lpn;
1142         struct lnet_peer_ni *lpni;
1143         int rc = 0;
1144
1145         LASSERT(lp);
1146         LASSERT(nid != LNET_NID_ANY);
1147
1148         /* A configured peer can only be updated through configuration. */
1149         if (!(flags & LNET_PEER_CONFIGURED)) {
1150                 if (lp->lp_state & LNET_PEER_CONFIGURED) {
1151                         rc = -EPERM;
1152                         goto out;
1153                 }
1154         }
1155
1156         /*
1157          * The MULTI_RAIL flag can be set but not cleared, because
1158          * that would leave the peer struct in an invalid state.
1159          */
1160         if (flags & LNET_PEER_MULTI_RAIL) {
1161                 spin_lock(&lp->lp_lock);
1162                 if (!(lp->lp_state & LNET_PEER_MULTI_RAIL)) {
1163                         lp->lp_state |= LNET_PEER_MULTI_RAIL;
1164                         lnet_peer_clr_non_mr_pref_nids(lp);
1165                 }
1166                 spin_unlock(&lp->lp_lock);
1167         } else if (lp->lp_state & LNET_PEER_MULTI_RAIL) {
1168                 rc = -EPERM;
1169                 goto out;
1170         }
1171
1172         lpni = lnet_find_peer_ni_locked(nid);
1173         if (lpni) {
1174                 /*
1175                  * A peer_ni already exists. This is only a problem if
1176                  * it is not connected to this peer and was configured
1177                  * by DLC.
1178                  */
1179                 lnet_peer_ni_decref_locked(lpni);
1180                 if (lpni->lpni_peer_net->lpn_peer == lp)
1181                         goto out;
1182                 if (lnet_peer_ni_is_configured(lpni)) {
1183                         rc = -EEXIST;
1184                         goto out;
1185                 }
1186                 /* If this is the primary NID, destroy the peer. */
1187                 if (lnet_peer_ni_is_primary(lpni)) {
1188                         lnet_peer_del(lpni->lpni_peer_net->lpn_peer);
1189                         lpni = lnet_peer_ni_alloc(nid);
1190                         if (!lpni) {
1191                                 rc = -ENOMEM;
1192                                 goto out;
1193                         }
1194                 }
1195         } else {
1196                 lpni = lnet_peer_ni_alloc(nid);
1197                 if (!lpni) {
1198                         rc = -ENOMEM;
1199                         goto out;
1200                 }
1201         }
1202
1203         /*
1204          * Get the peer_net. Check that we're not adding a second
1205          * peer_ni on a peer_net of a non-multi-rail peer.
1206          */
1207         lpn = lnet_peer_get_net_locked(lp, LNET_NIDNET(nid));
1208         if (!lpn) {
1209                 lpn = lnet_peer_net_alloc(LNET_NIDNET(nid));
1210                 if (!lpn) {
1211                         rc = -ENOMEM;
1212                         goto out_free_lpni;
1213                 }
1214         } else if (!(lp->lp_state & LNET_PEER_MULTI_RAIL)) {
1215                 rc = -ENOTUNIQ;
1216                 goto out_free_lpni;
1217         }
1218
1219         return lnet_peer_attach_peer_ni(lp, lpn, lpni, flags);
1220
1221 out_free_lpni:
1222         /* If the peer_ni was allocated above its peer_net pointer is NULL */
1223         if (!lpni->lpni_peer_net)
1224                 LIBCFS_FREE(lpni, sizeof(*lpni));
1225 out:
1226         CDEBUG(D_NET, "peer %s NID %s flags %#x: %d\n",
1227                libcfs_nid2str(lp->lp_primary_nid), libcfs_nid2str(nid),
1228                flags, rc);
1229         return rc;
1230 }
1231
1232 /*
1233  * lpni creation initiated due to traffic either sending or receiving.
1234  */
1235 static int
1236 lnet_peer_ni_traffic_add(lnet_nid_t nid, lnet_nid_t pref)
1237 {
1238         struct lnet_peer *lp;
1239         struct lnet_peer_net *lpn;
1240         struct lnet_peer_ni *lpni;
1241         unsigned flags = 0;
1242         int rc = 0;
1243
1244         if (nid == LNET_NID_ANY) {
1245                 rc = -EINVAL;
1246                 goto out;
1247         }
1248
1249         /* lnet_net_lock is not needed here because ln_api_lock is held */
1250         lpni = lnet_find_peer_ni_locked(nid);
1251         if (lpni) {
1252                 /*
1253                  * We must have raced with another thread. Since we
1254                  * know next to nothing about a peer_ni created by
1255                  * traffic, we just assume everything is ok and
1256                  * return.
1257                  */
1258                 lnet_peer_ni_decref_locked(lpni);
1259                 goto out;
1260         }
1261
1262         /* Create peer, peer_net, and peer_ni. */
1263         rc = -ENOMEM;
1264         lp = lnet_peer_alloc(nid);
1265         if (!lp)
1266                 goto out;
1267         lpn = lnet_peer_net_alloc(LNET_NIDNET(nid));
1268         if (!lpn)
1269                 goto out_free_lp;
1270         lpni = lnet_peer_ni_alloc(nid);
1271         if (!lpni)
1272                 goto out_free_lpn;
1273         if (pref != LNET_NID_ANY)
1274                 lnet_peer_ni_set_non_mr_pref_nid(lpni, pref);
1275
1276         return lnet_peer_attach_peer_ni(lp, lpn, lpni, flags);
1277
1278 out_free_lpn:
1279         LIBCFS_FREE(lpn, sizeof(*lpn));
1280 out_free_lp:
1281         LIBCFS_FREE(lp, sizeof(*lp));
1282 out:
1283         CDEBUG(D_NET, "peer %s: %d\n", libcfs_nid2str(nid), rc);
1284         return rc;
1285 }
1286
1287 /*
1288  * Implementation of IOC_LIBCFS_ADD_PEER_NI.
1289  *
1290  * This API handles the following combinations:
1291  *   Create a peer with its primary NI if only the prim_nid is provided
1292  *   Add a NID to a peer identified by the prim_nid. The peer identified
1293  *   by the prim_nid must already exist.
1294  *   The peer being created may be non-MR.
1295  *
1296  * The caller must hold ln_api_mutex. This prevents the peer from
1297  * being created/modified/deleted by a different thread.
1298  */
1299 int
1300 lnet_add_peer_ni(lnet_nid_t prim_nid, lnet_nid_t nid, bool mr)
1301 {
1302         struct lnet_peer *lp = NULL;
1303         struct lnet_peer_ni *lpni;
1304         unsigned flags;
1305
1306         /* The prim_nid must always be specified */
1307         if (prim_nid == LNET_NID_ANY)
1308                 return -EINVAL;
1309
1310         flags = LNET_PEER_CONFIGURED;
1311         if (mr)
1312                 flags |= LNET_PEER_MULTI_RAIL;
1313
1314         /*
1315          * If nid isn't specified, we must create a new peer with
1316          * prim_nid as its primary nid.
1317          */
1318         if (nid == LNET_NID_ANY)
1319                 return lnet_peer_add(prim_nid, flags);
1320
1321         /* Look up the prim_nid, which must exist. */
1322         lpni = lnet_find_peer_ni_locked(prim_nid);
1323         if (!lpni)
1324                 return -ENOENT;
1325         lnet_peer_ni_decref_locked(lpni);
1326         lp = lpni->lpni_peer_net->lpn_peer;
1327
1328         /* Peer must have been configured. */
1329         if (!(lp->lp_state & LNET_PEER_CONFIGURED)) {
1330                 CDEBUG(D_NET, "peer %s was not configured\n",
1331                        libcfs_nid2str(prim_nid));
1332                 return -ENOENT;
1333         }
1334
1335         /* Primary NID must match */
1336         if (lp->lp_primary_nid != prim_nid) {
1337                 CDEBUG(D_NET, "prim_nid %s is not primary for peer %s\n",
1338                        libcfs_nid2str(prim_nid),
1339                        libcfs_nid2str(lp->lp_primary_nid));
1340                 return -ENODEV;
1341         }
1342
1343         /* Multi-Rail flag must match. */
1344         if ((lp->lp_state ^ flags) & LNET_PEER_MULTI_RAIL) {
1345                 CDEBUG(D_NET, "multi-rail state mismatch for peer %s\n",
1346                        libcfs_nid2str(prim_nid));
1347                 return -EPERM;
1348         }
1349
1350         return lnet_peer_add_nid(lp, nid, flags);
1351 }
1352
1353 /*
1354  * Implementation of IOC_LIBCFS_DEL_PEER_NI.
1355  *
1356  * This API handles the following combinations:
1357  *   Delete a NI from a peer if both prim_nid and nid are provided.
1358  *   Delete a peer if only prim_nid is provided.
1359  *   Delete a peer if its primary nid is provided.
1360  *
1361  * The caller must hold ln_api_mutex. This prevents the peer from
1362  * being modified/deleted by a different thread.
1363  */
1364 int
1365 lnet_del_peer_ni(lnet_nid_t prim_nid, lnet_nid_t nid)
1366 {
1367         struct lnet_peer *lp;
1368         struct lnet_peer_ni *lpni;
1369         unsigned flags;
1370
1371         if (prim_nid == LNET_NID_ANY)
1372                 return -EINVAL;
1373
1374         lpni = lnet_find_peer_ni_locked(prim_nid);
1375         if (!lpni)
1376                 return -ENOENT;
1377         lnet_peer_ni_decref_locked(lpni);
1378         lp = lpni->lpni_peer_net->lpn_peer;
1379
1380         if (prim_nid != lp->lp_primary_nid) {
1381                 CDEBUG(D_NET, "prim_nid %s is not primary for peer %s\n",
1382                        libcfs_nid2str(prim_nid),
1383                        libcfs_nid2str(lp->lp_primary_nid));
1384                 return -ENODEV;
1385         }
1386
1387         if (nid == LNET_NID_ANY || nid == lp->lp_primary_nid)
1388                 return lnet_peer_del(lp);
1389
1390         flags = LNET_PEER_CONFIGURED;
1391         if (lp->lp_state & LNET_PEER_MULTI_RAIL)
1392                 flags |= LNET_PEER_MULTI_RAIL;
1393
1394         return lnet_peer_del_nid(lp, nid, flags);
1395 }
1396
1397 void
1398 lnet_destroy_peer_ni_locked(struct lnet_peer_ni *lpni)
1399 {
1400         struct lnet_peer_table *ptable;
1401         struct lnet_peer_net *lpn;
1402
1403         CDEBUG(D_NET, "%p nid %s\n", lpni, libcfs_nid2str(lpni->lpni_nid));
1404
1405         LASSERT(atomic_read(&lpni->lpni_refcount) == 0);
1406         LASSERT(lpni->lpni_rtr_refcount == 0);
1407         LASSERT(list_empty(&lpni->lpni_txq));
1408         LASSERT(lpni->lpni_txqnob == 0);
1409
1410         lpn = lpni->lpni_peer_net;
1411         lpni->lpni_peer_net = NULL;
1412         lpni->lpni_net = NULL;
1413
1414         /* remove the peer ni from the zombie list */
1415         ptable = the_lnet.ln_peer_tables[lpni->lpni_cpt];
1416         spin_lock(&ptable->pt_zombie_lock);
1417         list_del_init(&lpni->lpni_hashlist);
1418         ptable->pt_zombies--;
1419         spin_unlock(&ptable->pt_zombie_lock);
1420
1421         if (lpni->lpni_pref_nnids > 1) {
1422                 LIBCFS_FREE(lpni->lpni_pref.nids,
1423                         sizeof(*lpni->lpni_pref.nids) * lpni->lpni_pref_nnids);
1424         }
1425         LIBCFS_FREE(lpni, sizeof(*lpni));
1426
1427         lnet_peer_net_decref_locked(lpn);
1428 }
1429
1430 struct lnet_peer_ni *
1431 lnet_nid2peerni_ex(lnet_nid_t nid, int cpt)
1432 {
1433         struct lnet_peer_ni *lpni = NULL;
1434         int rc;
1435
1436         if (the_lnet.ln_state != LNET_STATE_RUNNING)
1437                 return ERR_PTR(-ESHUTDOWN);
1438
1439         /*
1440          * find if a peer_ni already exists.
1441          * If so then just return that.
1442          */
1443         lpni = lnet_find_peer_ni_locked(nid);
1444         if (lpni)
1445                 return lpni;
1446
1447         lnet_net_unlock(cpt);
1448
1449         rc = lnet_peer_ni_traffic_add(nid, LNET_NID_ANY);
1450         if (rc) {
1451                 lpni = ERR_PTR(rc);
1452                 goto out_net_relock;
1453         }
1454
1455         lpni = lnet_find_peer_ni_locked(nid);
1456         LASSERT(lpni);
1457
1458 out_net_relock:
1459         lnet_net_lock(cpt);
1460
1461         return lpni;
1462 }
1463
1464 struct lnet_peer_ni *
1465 lnet_nid2peerni_locked(lnet_nid_t nid, lnet_nid_t pref, int cpt)
1466 {
1467         struct lnet_peer_ni *lpni = NULL;
1468         int rc;
1469
1470         if (the_lnet.ln_state != LNET_STATE_RUNNING)
1471                 return ERR_PTR(-ESHUTDOWN);
1472
1473         /*
1474          * find if a peer_ni already exists.
1475          * If so then just return that.
1476          */
1477         lpni = lnet_find_peer_ni_locked(nid);
1478         if (lpni)
1479                 return lpni;
1480
1481         /*
1482          * Slow path:
1483          * use the lnet_api_mutex to serialize the creation of the peer_ni
1484          * and the creation/deletion of the local ni/net. When a local ni is
1485          * created, if there exists a set of peer_nis on that network,
1486          * they need to be traversed and updated. When a local NI is
1487          * deleted, which could result in a network being deleted, then
1488          * all peer nis on that network need to be removed as well.
1489          *
1490          * Creation through traffic should also be serialized with
1491          * creation through DLC.
1492          */
1493         lnet_net_unlock(cpt);
1494         mutex_lock(&the_lnet.ln_api_mutex);
1495         /*
1496          * Shutdown is only set under the ln_api_lock, so a single
1497          * check here is sufficent.
1498          */
1499         if (the_lnet.ln_state != LNET_STATE_RUNNING) {
1500                 lpni = ERR_PTR(-ESHUTDOWN);
1501                 goto out_mutex_unlock;
1502         }
1503
1504         rc = lnet_peer_ni_traffic_add(nid, pref);
1505         if (rc) {
1506                 lpni = ERR_PTR(rc);
1507                 goto out_mutex_unlock;
1508         }
1509
1510         lpni = lnet_find_peer_ni_locked(nid);
1511         LASSERT(lpni);
1512
1513 out_mutex_unlock:
1514         mutex_unlock(&the_lnet.ln_api_mutex);
1515         lnet_net_lock(cpt);
1516
1517         return lpni;
1518 }
1519
1520 void
1521 lnet_debug_peer(lnet_nid_t nid)
1522 {
1523         char                    *aliveness = "NA";
1524         struct lnet_peer_ni     *lp;
1525         int                     cpt;
1526
1527         cpt = lnet_cpt_of_nid(nid, NULL);
1528         lnet_net_lock(cpt);
1529
1530         lp = lnet_nid2peerni_locked(nid, LNET_NID_ANY, cpt);
1531         if (IS_ERR(lp)) {
1532                 lnet_net_unlock(cpt);
1533                 CDEBUG(D_WARNING, "No peer %s\n", libcfs_nid2str(nid));
1534                 return;
1535         }
1536
1537         if (lnet_isrouter(lp) || lnet_peer_aliveness_enabled(lp))
1538                 aliveness = lp->lpni_alive ? "up" : "down";
1539
1540         CDEBUG(D_WARNING, "%-24s %4d %5s %5d %5d %5d %5d %5d %ld\n",
1541                libcfs_nid2str(lp->lpni_nid), atomic_read(&lp->lpni_refcount),
1542                aliveness, lp->lpni_net->net_tunables.lct_peer_tx_credits,
1543                lp->lpni_rtrcredits, lp->lpni_minrtrcredits,
1544                lp->lpni_txcredits, lp->lpni_mintxcredits, lp->lpni_txqnob);
1545
1546         lnet_peer_ni_decref_locked(lp);
1547
1548         lnet_net_unlock(cpt);
1549 }
1550
1551 int lnet_get_peer_ni_info(__u32 peer_index, __u64 *nid,
1552                           char aliveness[LNET_MAX_STR_LEN],
1553                           __u32 *cpt_iter, __u32 *refcount,
1554                           __u32 *ni_peer_tx_credits, __u32 *peer_tx_credits,
1555                           __u32 *peer_rtr_credits, __u32 *peer_min_rtr_credits,
1556                           __u32 *peer_tx_qnob)
1557 {
1558         struct lnet_peer_table          *peer_table;
1559         struct lnet_peer_ni             *lp;
1560         int                             j;
1561         int                             lncpt;
1562         bool                            found = false;
1563
1564         /* get the number of CPTs */
1565         lncpt = cfs_percpt_number(the_lnet.ln_peer_tables);
1566
1567         /* if the cpt number to be examined is >= the number of cpts in
1568          * the system then indicate that there are no more cpts to examin
1569          */
1570         if (*cpt_iter >= lncpt)
1571                 return -ENOENT;
1572
1573         /* get the current table */
1574         peer_table = the_lnet.ln_peer_tables[*cpt_iter];
1575         /* if the ptable is NULL then there are no more cpts to examine */
1576         if (peer_table == NULL)
1577                 return -ENOENT;
1578
1579         lnet_net_lock(*cpt_iter);
1580
1581         for (j = 0; j < LNET_PEER_HASH_SIZE && !found; j++) {
1582                 struct list_head *peers = &peer_table->pt_hash[j];
1583
1584                 list_for_each_entry(lp, peers, lpni_hashlist) {
1585                         if (peer_index-- > 0)
1586                                 continue;
1587
1588                         snprintf(aliveness, LNET_MAX_STR_LEN, "NA");
1589                         if (lnet_isrouter(lp) ||
1590                                 lnet_peer_aliveness_enabled(lp))
1591                                 snprintf(aliveness, LNET_MAX_STR_LEN,
1592                                          lp->lpni_alive ? "up" : "down");
1593
1594                         *nid = lp->lpni_nid;
1595                         *refcount = atomic_read(&lp->lpni_refcount);
1596                         *ni_peer_tx_credits =
1597                                 lp->lpni_net->net_tunables.lct_peer_tx_credits;
1598                         *peer_tx_credits = lp->lpni_txcredits;
1599                         *peer_rtr_credits = lp->lpni_rtrcredits;
1600                         *peer_min_rtr_credits = lp->lpni_mintxcredits;
1601                         *peer_tx_qnob = lp->lpni_txqnob;
1602
1603                         found = true;
1604                 }
1605
1606         }
1607         lnet_net_unlock(*cpt_iter);
1608
1609         *cpt_iter = lncpt;
1610
1611         return found ? 0 : -ENOENT;
1612 }
1613
1614 /* ln_api_mutex is held, which keeps the peer list stable */
1615 int lnet_get_peer_info(__u32 idx, lnet_nid_t *primary_nid, lnet_nid_t *nid,
1616                        bool *mr,
1617                        struct lnet_peer_ni_credit_info __user *peer_ni_info,
1618                        struct lnet_ioctl_element_stats __user *peer_ni_stats)
1619 {
1620         struct lnet_peer_ni *lpni = NULL;
1621         struct lnet_peer_net *lpn = NULL;
1622         struct lnet_peer *lp = NULL;
1623         struct lnet_peer_ni_credit_info ni_info;
1624         struct lnet_ioctl_element_stats ni_stats;
1625         int rc;
1626
1627         lpni = lnet_get_peer_ni_idx_locked(idx, &lpn, &lp);
1628
1629         if (!lpni)
1630                 return -ENOENT;
1631
1632         *primary_nid = lp->lp_primary_nid;
1633         *mr = lnet_peer_is_multi_rail(lp);
1634         *nid = lpni->lpni_nid;
1635         snprintf(ni_info.cr_aliveness, LNET_MAX_STR_LEN, "NA");
1636         if (lnet_isrouter(lpni) ||
1637                 lnet_peer_aliveness_enabled(lpni))
1638                 snprintf(ni_info.cr_aliveness, LNET_MAX_STR_LEN,
1639                          lpni->lpni_alive ? "up" : "down");
1640
1641         ni_info.cr_refcount = atomic_read(&lpni->lpni_refcount);
1642         ni_info.cr_ni_peer_tx_credits = (lpni->lpni_net != NULL) ?
1643                 lpni->lpni_net->net_tunables.lct_peer_tx_credits : 0;
1644         ni_info.cr_peer_tx_credits = lpni->lpni_txcredits;
1645         ni_info.cr_peer_rtr_credits = lpni->lpni_rtrcredits;
1646         ni_info.cr_peer_min_rtr_credits = lpni->lpni_minrtrcredits;
1647         ni_info.cr_peer_min_tx_credits = lpni->lpni_mintxcredits;
1648         ni_info.cr_peer_tx_qnob = lpni->lpni_txqnob;
1649         ni_info.cr_ncpt = lpni->lpni_cpt;
1650
1651         ni_stats.iel_send_count = atomic_read(&lpni->lpni_stats.send_count);
1652         ni_stats.iel_recv_count = atomic_read(&lpni->lpni_stats.recv_count);
1653         ni_stats.iel_drop_count = atomic_read(&lpni->lpni_stats.drop_count);
1654
1655         /* If copy_to_user fails */
1656         rc = -EFAULT;
1657         if (copy_to_user(peer_ni_info, &ni_info, sizeof(ni_info)))
1658                 goto copy_failed;
1659
1660         if (copy_to_user(peer_ni_stats, &ni_stats, sizeof(ni_stats)))
1661                 goto copy_failed;
1662
1663         rc = 0;
1664
1665 copy_failed:
1666         return rc;
1667 }