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