lnet/klnds/ralnd/ralnd.c

   1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
   2  * vim:expandtab:shiftwidth=8:tabstop=8:
   3  *
   4  * GPL HEADER START
   5  *
   6  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
   7  *
   8  * This program is free software; you can redistribute it and/or modify
   9  * it under the terms of the GNU General Public License version 2 only,
  10  * as published by the Free Software Foundation.
  11  *
  12  * This program is distributed in the hope that it will be useful, but
  13  * WITHOUT ANY WARRANTY; without even the implied warranty of
  14  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  15  * General Public License version 2 for more details (a copy is included
  16  * in the LICENSE file that accompanied this code).
  17  *
  18  * You should have received a copy of the GNU General Public License
  19  * version 2 along with this program; If not, see
  20  * http://www.sun.com/software/products/lustre/docs/GPLv2.pdf
  21  *
  22  * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
  23  * CA 95054 USA or visit www.sun.com if you need additional information or
  24  * have any questions.
  25  *
  26  * GPL HEADER END
  27  */
  28 /*
  29  * Copyright  2008 Sun Microsystems, Inc. All rights reserved
  30  * Use is subject to license terms.
  31  */
  32 /*
  33  * This file is part of Lustre, http://www.lustre.org/
  34  * Lustre is a trademark of Sun Microsystems, Inc.
  35  *
  36  * lnet/klnds/ralnd/ralnd.c
  37  *
  38  * Author: Eric Barton <eric@bartonsoftware.com>
  39  */
  40 #include "ralnd.h"
  41
  42 static int        kranal_devids[RANAL_MAXDEVS] = {RAPK_MAIN_DEVICE_ID,
  43                                                   RAPK_EXPANSION_DEVICE_ID};
  44
  45 lnd_t the_kralnd = {
  46         .lnd_type       = RALND,
  47         .lnd_startup    = kranal_startup,
  48         .lnd_shutdown   = kranal_shutdown,
  49         .lnd_ctl        = kranal_ctl,
  50         .lnd_send       = kranal_send,
  51         .lnd_recv       = kranal_recv,
  52         .lnd_eager_recv = kranal_eager_recv,
  53         .lnd_accept     = kranal_accept,
  54 };
  55
  56 kra_data_t              kranal_data;
  57
  58 void
  59 kranal_pack_connreq(kra_connreq_t *connreq, kra_conn_t *conn, lnet_nid_t dstnid)
  60 {
  61         RAP_RETURN   rrc;
  62
  63         memset(connreq, 0, sizeof(*connreq));
  64
  65         connreq->racr_magic     = RANAL_MSG_MAGIC;
  66         connreq->racr_version   = RANAL_MSG_VERSION;
  67
  68         if (conn == NULL)                       /* prepping a "stub" reply */
  69                 return;
  70
  71         connreq->racr_devid     = conn->rac_device->rad_id;
  72         connreq->racr_srcnid    = kranal_data.kra_ni->ni_nid;
  73         connreq->racr_dstnid    = dstnid;
  74         connreq->racr_peerstamp = kranal_data.kra_peerstamp;
  75         connreq->racr_connstamp = conn->rac_my_connstamp;
  76         connreq->racr_timeout   = conn->rac_timeout;
  77
  78         rrc = RapkGetRiParams(conn->rac_rihandle, &connreq->racr_riparams);
  79         LASSERT(rrc == RAP_SUCCESS);
  80 }
  81
  82 int
  83 kranal_recv_connreq(struct socket *sock, kra_connreq_t *connreq, int active)
  84 {
  85         int         timeout = active ? *kranal_tunables.kra_timeout :
  86                                         lnet_acceptor_timeout();
  87         int         swab;
  88         int         rc;
  89
  90         /* return 0 on success, -ve on error, +ve to tell the peer I'm "old" */
  91
  92         rc = libcfs_sock_read(sock, &connreq->racr_magic,
  93                               sizeof(connreq->racr_magic), timeout);
  94         if (rc != 0) {
  95                 CERROR("Read(magic) failed(1): %d\n", rc);
  96                 return -EIO;
  97         }
  98
  99         if (connreq->racr_magic != RANAL_MSG_MAGIC &&
 100             connreq->racr_magic != __swab32(RANAL_MSG_MAGIC)) {
 101                 /* Unexpected magic! */
 102                 if (!active &&
 103                     (connreq->racr_magic == LNET_PROTO_MAGIC ||
 104                      connreq->racr_magic == __swab32(LNET_PROTO_MAGIC))) {
 105                         /* future protocol version compatibility!
 106                          * When LNET unifies protocols over all LNDs, the first
 107                          * thing sent will be a version query.  +ve rc means I
 108                          * reply with my current magic/version */
 109                         return EPROTO;
 110                 }
 111
 112                 CERROR("Unexpected magic %08x (%s)\n",
 113                        connreq->racr_magic, active ? "active" : "passive");
 114                 return -EPROTO;
 115         }
 116
 117         swab = (connreq->racr_magic == __swab32(RANAL_MSG_MAGIC));
 118
 119         rc = libcfs_sock_read(sock, &connreq->racr_version,
 120                               sizeof(connreq->racr_version), timeout);
 121         if (rc != 0) {
 122                 CERROR("Read(version) failed: %d\n", rc);
 123                 return -EIO;
 124         }
 125
 126         if (swab)
 127                 __swab16s(&connreq->racr_version);
 128
 129         if (connreq->racr_version != RANAL_MSG_VERSION) {
 130                 if (active) {
 131                         CERROR("Unexpected version %d\n", connreq->racr_version);
 132                         return -EPROTO;
 133                 }
 134                 /* If this is a future version of the ralnd protocol, and I'm
 135                  * passive (accepted the connection), tell my peer I'm "old"
 136                  * (+ve rc) */
 137                 return EPROTO;
 138         }
 139
 140         rc = libcfs_sock_read(sock, &connreq->racr_devid,
 141                               sizeof(connreq->racr_version) -
 142                               offsetof(kra_connreq_t, racr_devid),
 143                               timeout);
 144         if (rc != 0) {
 145                 CERROR("Read(body) failed: %d\n", rc);
 146                 return -EIO;
 147         }
 148
 149         if (swab) {
 150                 __swab32s(&connreq->racr_magic);
 151                 __swab16s(&connreq->racr_version);
 152                 __swab16s(&connreq->racr_devid);
 153                 __swab64s(&connreq->racr_srcnid);
 154                 __swab64s(&connreq->racr_dstnid);
 155                 __swab64s(&connreq->racr_peerstamp);
 156                 __swab64s(&connreq->racr_connstamp);
 157                 __swab32s(&connreq->racr_timeout);
 158
 159                 __swab32s(&connreq->racr_riparams.HostId);
 160                 __swab32s(&connreq->racr_riparams.FmaDomainHndl);
 161                 __swab32s(&connreq->racr_riparams.PTag);
 162                 __swab32s(&connreq->racr_riparams.CompletionCookie);
 163         }
 164
 165         if (connreq->racr_srcnid == LNET_NID_ANY ||
 166             connreq->racr_dstnid == LNET_NID_ANY) {
 167                 CERROR("Received LNET_NID_ANY\n");
 168                 return -EPROTO;
 169         }
 170
 171         if (connreq->racr_timeout < RANAL_MIN_TIMEOUT) {
 172                 CERROR("Received timeout %d < MIN %d\n",
 173                        connreq->racr_timeout, RANAL_MIN_TIMEOUT);
 174                 return -EPROTO;
 175         }
 176
 177         return 0;
 178 }
 179
 180 int
 181 kranal_close_stale_conns_locked (kra_peer_t *peer, kra_conn_t *newconn)
 182 {
 183         kra_conn_t         *conn;
 184         struct list_head   *ctmp;
 185         struct list_head   *cnxt;
 186         int                 loopback;
 187         int                 count = 0;
 188
 189         loopback = peer->rap_nid == kranal_data.kra_ni->ni_nid;
 190
 191         list_for_each_safe (ctmp, cnxt, &peer->rap_conns) {
 192                 conn = list_entry(ctmp, kra_conn_t, rac_list);
 193
 194                 if (conn == newconn)
 195                         continue;
 196
 197                 if (conn->rac_peerstamp != newconn->rac_peerstamp) {
 198                         CDEBUG(D_NET, "Closing stale conn nid: %s "
 199                                " peerstamp:"LPX64"("LPX64")\n",
 200                                libcfs_nid2str(peer->rap_nid),
 201                                conn->rac_peerstamp, newconn->rac_peerstamp);
 202                         LASSERT (conn->rac_peerstamp < newconn->rac_peerstamp);
 203                         count++;
 204                         kranal_close_conn_locked(conn, -ESTALE);
 205                         continue;
 206                 }
 207
 208                 if (conn->rac_device != newconn->rac_device)
 209                         continue;
 210
 211                 if (loopback &&
 212                     newconn->rac_my_connstamp == conn->rac_peer_connstamp &&
 213                     newconn->rac_peer_connstamp == conn->rac_my_connstamp)
 214                         continue;
 215
 216                 LASSERT (conn->rac_peer_connstamp < newconn->rac_peer_connstamp);
 217
 218                 CDEBUG(D_NET, "Closing stale conn nid: %s"
 219                        " connstamp:"LPX64"("LPX64")\n",
 220                        libcfs_nid2str(peer->rap_nid),
 221                        conn->rac_peer_connstamp, newconn->rac_peer_connstamp);
 222
 223                 count++;
 224                 kranal_close_conn_locked(conn, -ESTALE);
 225         }
 226
 227         return count;
 228 }
 229
 230 int
 231 kranal_conn_isdup_locked(kra_peer_t *peer, kra_conn_t *newconn)
 232 {
 233         kra_conn_t       *conn;
 234         struct list_head *tmp;
 235         int               loopback;
 236
 237         loopback = peer->rap_nid == kranal_data.kra_ni->ni_nid;
 238
 239         list_for_each(tmp, &peer->rap_conns) {
 240                 conn = list_entry(tmp, kra_conn_t, rac_list);
 241
 242                 /* 'newconn' is from an earlier version of 'peer'!!! */
 243                 if (newconn->rac_peerstamp < conn->rac_peerstamp)
 244                         return 1;
 245
 246                 /* 'conn' is from an earlier version of 'peer': it will be
 247                  * removed when we cull stale conns later on... */
 248                 if (newconn->rac_peerstamp > conn->rac_peerstamp)
 249                         continue;
 250
 251                 /* Different devices are OK */
 252                 if (conn->rac_device != newconn->rac_device)
 253                         continue;
 254
 255                 /* It's me connecting to myself */
 256                 if (loopback &&
 257                     newconn->rac_my_connstamp == conn->rac_peer_connstamp &&
 258                     newconn->rac_peer_connstamp == conn->rac_my_connstamp)
 259                         continue;
 260
 261                 /* 'newconn' is an earlier connection from 'peer'!!! */
 262                 if (newconn->rac_peer_connstamp < conn->rac_peer_connstamp)
 263                         return 2;
 264
 265                 /* 'conn' is an earlier connection from 'peer': it will be
 266                  * removed when we cull stale conns later on... */
 267                 if (newconn->rac_peer_connstamp > conn->rac_peer_connstamp)
 268                         continue;
 269
 270                 /* 'newconn' has the SAME connection stamp; 'peer' isn't
 271                  * playing the game... */
 272                 return 3;
 273         }
 274
 275         return 0;
 276 }
 277
 278 void
 279 kranal_set_conn_uniqueness (kra_conn_t *conn)
 280 {
 281         unsigned long  flags;
 282
 283         write_lock_irqsave(&kranal_data.kra_global_lock, flags);
 284
 285         conn->rac_my_connstamp = kranal_data.kra_connstamp++;
 286
 287         do {    /* allocate a unique cqid */
 288                 conn->rac_cqid = kranal_data.kra_next_cqid++;
 289         } while (kranal_cqid2conn_locked(conn->rac_cqid) != NULL);
 290
 291         write_unlock_irqrestore(&kranal_data.kra_global_lock, flags);
 292 }
 293
 294 int
 295 kranal_create_conn(kra_conn_t **connp, kra_device_t *dev)
 296 {
 297         kra_conn_t    *conn;
 298         RAP_RETURN     rrc;
 299
 300         LASSERT (!in_interrupt());
 301         LIBCFS_ALLOC(conn, sizeof(*conn));
 302
 303         if (conn == NULL)
 304                 return -ENOMEM;
 305
 306         memset(conn, 0, sizeof(*conn));
 307         atomic_set(&conn->rac_refcount, 1);
 308         INIT_LIST_HEAD(&conn->rac_list);
 309         INIT_LIST_HEAD(&conn->rac_hashlist);
 310         INIT_LIST_HEAD(&conn->rac_schedlist);
 311         INIT_LIST_HEAD(&conn->rac_fmaq);
 312         INIT_LIST_HEAD(&conn->rac_rdmaq);
 313         INIT_LIST_HEAD(&conn->rac_replyq);
 314         spin_lock_init(&conn->rac_lock);
 315
 316         kranal_set_conn_uniqueness(conn);
 317
 318         conn->rac_device = dev;
 319         conn->rac_timeout = MAX(*kranal_tunables.kra_timeout, RANAL_MIN_TIMEOUT);
 320         kranal_update_reaper_timeout(conn->rac_timeout);
 321
 322         rrc = RapkCreateRi(dev->rad_handle, conn->rac_cqid,
 323                            &conn->rac_rihandle);
 324         if (rrc != RAP_SUCCESS) {
 325                 CERROR("RapkCreateRi failed: %d\n", rrc);
 326                 LIBCFS_FREE(conn, sizeof(*conn));
 327                 return -ENETDOWN;
 328         }
 329
 330         atomic_inc(&kranal_data.kra_nconns);
 331         *connp = conn;
 332         return 0;
 333 }
 334
 335 void
 336 kranal_destroy_conn(kra_conn_t *conn)
 337 {
 338         RAP_RETURN         rrc;
 339
 340         LASSERT (!in_interrupt());
 341         LASSERT (!conn->rac_scheduled);
 342         LASSERT (list_empty(&conn->rac_list));
 343         LASSERT (list_empty(&conn->rac_hashlist));
 344         LASSERT (list_empty(&conn->rac_schedlist));
 345         LASSERT (atomic_read(&conn->rac_refcount) == 0);
 346         LASSERT (list_empty(&conn->rac_fmaq));
 347         LASSERT (list_empty(&conn->rac_rdmaq));
 348         LASSERT (list_empty(&conn->rac_replyq));
 349
 350         rrc = RapkDestroyRi(conn->rac_device->rad_handle,
 351                             conn->rac_rihandle);
 352         LASSERT (rrc == RAP_SUCCESS);
 353
 354         if (conn->rac_peer != NULL)
 355                 kranal_peer_decref(conn->rac_peer);
 356
 357         LIBCFS_FREE(conn, sizeof(*conn));
 358         atomic_dec(&kranal_data.kra_nconns);
 359 }
 360
 361 void
 362 kranal_terminate_conn_locked (kra_conn_t *conn)
 363 {
 364         LASSERT (!in_interrupt());
 365         LASSERT (conn->rac_state == RANAL_CONN_CLOSING);
 366         LASSERT (!list_empty(&conn->rac_hashlist));
 367         LASSERT (list_empty(&conn->rac_list));
 368
 369         /* Remove from conn hash table: no new callbacks */
 370         list_del_init(&conn->rac_hashlist);
 371         kranal_conn_decref(conn);
 372
 373         conn->rac_state = RANAL_CONN_CLOSED;
 374
 375         /* schedule to clear out all uncompleted comms in context of dev's
 376          * scheduler */
 377         kranal_schedule_conn(conn);
 378 }
 379
 380 void
 381 kranal_close_conn_locked (kra_conn_t *conn, int error)
 382 {
 383         kra_peer_t        *peer = conn->rac_peer;
 384
 385         CDEBUG(error == 0 ? D_NET : D_NETERROR,
 386                "closing conn to %s: error %d\n",
 387                libcfs_nid2str(peer->rap_nid), error);
 388
 389         LASSERT (!in_interrupt());
 390         LASSERT (conn->rac_state == RANAL_CONN_ESTABLISHED);
 391         LASSERT (!list_empty(&conn->rac_hashlist));
 392         LASSERT (!list_empty(&conn->rac_list));
 393
 394         list_del_init(&conn->rac_list);
 395
 396         if (list_empty(&peer->rap_conns) &&
 397             peer->rap_persistence == 0) {
 398                 /* Non-persistent peer with no more conns... */
 399                 kranal_unlink_peer_locked(peer);
 400         }
 401
 402         /* Reset RX timeout to ensure we wait for an incoming CLOSE for the
 403          * full timeout.  If we get a CLOSE we know the peer has stopped all
 404          * RDMA.  Otherwise if we wait for the full timeout we can also be sure
 405          * all RDMA has stopped. */
 406         conn->rac_last_rx = jiffies;
 407         mb();
 408
 409         conn->rac_state = RANAL_CONN_CLOSING;
 410         kranal_schedule_conn(conn);             /* schedule sending CLOSE */
 411
 412         kranal_conn_decref(conn);               /* lose peer's ref */
 413 }
 414
 415 void
 416 kranal_close_conn (kra_conn_t *conn, int error)
 417 {
 418         unsigned long    flags;
 419
 420
 421         write_lock_irqsave(&kranal_data.kra_global_lock, flags);
 422
 423         if (conn->rac_state == RANAL_CONN_ESTABLISHED)
 424                 kranal_close_conn_locked(conn, error);
 425
 426         write_unlock_irqrestore(&kranal_data.kra_global_lock, flags);
 427 }
 428
 429 int
 430 kranal_set_conn_params(kra_conn_t *conn, kra_connreq_t *connreq,
 431                        __u32 peer_ip, int peer_port)
 432 {
 433         kra_device_t  *dev = conn->rac_device;
 434         unsigned long  flags;
 435         RAP_RETURN     rrc;
 436
 437         /* CAVEAT EMPTOR: we're really overloading rac_last_tx + rac_keepalive
 438          * to do RapkCompleteSync() timekeeping (see kibnal_scheduler). */
 439         conn->rac_last_tx = jiffies;
 440         conn->rac_keepalive = 0;
 441
 442         rrc = RapkSetRiParams(conn->rac_rihandle, &connreq->racr_riparams);
 443         if (rrc != RAP_SUCCESS) {
 444                 CERROR("Error setting riparams from %u.%u.%u.%u/%d: %d\n",
 445                        HIPQUAD(peer_ip), peer_port, rrc);
 446                 return -ECONNABORTED;
 447         }
 448
 449         /* Schedule conn on rad_new_conns */
 450         kranal_conn_addref(conn);
 451         spin_lock_irqsave(&dev->rad_lock, flags);
 452         list_add_tail(&conn->rac_schedlist, &dev->rad_new_conns);
 453         wake_up(&dev->rad_waitq);
 454         spin_unlock_irqrestore(&dev->rad_lock, flags);
 455
 456         rrc = RapkWaitToConnect(conn->rac_rihandle);
 457         if (rrc != RAP_SUCCESS) {
 458                 CERROR("Error waiting to connect to %u.%u.%u.%u/%d: %d\n",
 459                        HIPQUAD(peer_ip), peer_port, rrc);
 460                 return -ECONNABORTED;
 461         }
 462
 463         /* Scheduler doesn't touch conn apart from to deschedule and decref it
 464          * after RapkCompleteSync() return success, so conn is all mine */
 465
 466         conn->rac_peerstamp = connreq->racr_peerstamp;
 467         conn->rac_peer_connstamp = connreq->racr_connstamp;
 468         conn->rac_keepalive = RANAL_TIMEOUT2KEEPALIVE(connreq->racr_timeout);
 469         kranal_update_reaper_timeout(conn->rac_keepalive);
 470         return 0;
 471 }
 472
 473 int
 474 kranal_passive_conn_handshake (struct socket *sock, lnet_nid_t *src_nidp,
 475                                lnet_nid_t *dst_nidp, kra_conn_t **connp)
 476 {
 477         __u32                peer_ip;
 478         unsigned int         peer_port;
 479         kra_connreq_t        rx_connreq;
 480         kra_connreq_t        tx_connreq;
 481         kra_conn_t          *conn;
 482         kra_device_t        *dev;
 483         int                  rc;
 484         int                  i;
 485
 486         rc = libcfs_sock_getaddr(sock, 1, &peer_ip, &peer_port);
 487         if (rc != 0) {
 488                 CERROR("Can't get peer's IP: %d\n", rc);
 489                 return rc;
 490         }
 491
 492         rc = kranal_recv_connreq(sock, &rx_connreq, 0);
 493
 494         if (rc < 0) {
 495                 CERROR("Can't rx connreq from %u.%u.%u.%u/%d: %d\n",
 496                        HIPQUAD(peer_ip), peer_port, rc);
 497                 return rc;
 498         }
 499
 500         if (rc > 0) {
 501                 /* Request from "new" peer: send reply with my MAGIC/VERSION to
 502                  * tell her I'm old... */
 503                 kranal_pack_connreq(&tx_connreq, NULL, LNET_NID_ANY);
 504
 505                 rc = libcfs_sock_write(sock, &tx_connreq, sizeof(tx_connreq),
 506                                        lnet_acceptor_timeout());
 507                 if (rc != 0)
 508                         CERROR("Can't tx stub connreq to %u.%u.%u.%u/%d: %d\n",
 509                                HIPQUAD(peer_ip), peer_port, rc);
 510
 511                 return -EPROTO;
 512         }
 513
 514         for (i = 0;;i++) {
 515                 if (i == kranal_data.kra_ndevs) {
 516                         CERROR("Can't match dev %d from %u.%u.%u.%u/%d\n",
 517                                rx_connreq.racr_devid, HIPQUAD(peer_ip), peer_port);
 518                         return -ENODEV;
 519                 }
 520                 dev = &kranal_data.kra_devices[i];
 521                 if (dev->rad_id == rx_connreq.racr_devid)
 522                         break;
 523         }
 524
 525         rc = kranal_create_conn(&conn, dev);
 526         if (rc != 0)
 527                 return rc;
 528
 529         kranal_pack_connreq(&tx_connreq, conn, rx_connreq.racr_srcnid);
 530
 531         rc = libcfs_sock_write(sock, &tx_connreq, sizeof(tx_connreq),
 532                                lnet_acceptor_timeout());
 533         if (rc != 0) {
 534                 CERROR("Can't tx connreq to %u.%u.%u.%u/%d: %d\n",
 535                        HIPQUAD(peer_ip), peer_port, rc);
 536                 kranal_conn_decref(conn);
 537                 return rc;
 538         }
 539
 540         rc = kranal_set_conn_params(conn, &rx_connreq, peer_ip, peer_port);
 541         if (rc != 0) {
 542                 kranal_conn_decref(conn);
 543                 return rc;
 544         }
 545
 546         *connp = conn;
 547         *src_nidp = rx_connreq.racr_srcnid;
 548         *dst_nidp = rx_connreq.racr_dstnid;
 549         return 0;
 550 }
 551
 552 int
 553 kranal_active_conn_handshake(kra_peer_t *peer,
 554                              lnet_nid_t *dst_nidp, kra_conn_t **connp)
 555 {
 556         kra_connreq_t       connreq;
 557         kra_conn_t         *conn;
 558         kra_device_t       *dev;
 559         struct socket      *sock;
 560         int                 rc;
 561         unsigned int        idx;
 562
 563         /* spread connections over all devices using both peer NIDs to ensure
 564          * all nids use all devices */
 565         idx = peer->rap_nid + kranal_data.kra_ni->ni_nid;
 566         dev = &kranal_data.kra_devices[idx % kranal_data.kra_ndevs];
 567
 568         rc = kranal_create_conn(&conn, dev);
 569         if (rc != 0)
 570                 return rc;
 571
 572         kranal_pack_connreq(&connreq, conn, peer->rap_nid);
 573
 574         if (the_lnet.ln_testprotocompat != 0) {
 575                 /* single-shot proto test */
 576                 LNET_LOCK();
 577                 if ((the_lnet.ln_testprotocompat & 1) != 0) {
 578                         connreq.racr_version++;
 579                         the_lnet.ln_testprotocompat &= ~1;
 580                 }
 581                 if ((the_lnet.ln_testprotocompat & 2) != 0) {
 582                         connreq.racr_magic = LNET_PROTO_MAGIC;
 583                         the_lnet.ln_testprotocompat &= ~2;
 584                 }
 585                 LNET_UNLOCK();
 586         }
 587
 588         rc = lnet_connect(&sock, peer->rap_nid,
 589                          0, peer->rap_ip, peer->rap_port);
 590         if (rc != 0)
 591                 goto failed_0;
 592
 593         /* CAVEAT EMPTOR: the passive side receives with a SHORT rx timeout
 594          * immediately after accepting a connection, so we connect and then
 595          * send immediately. */
 596
 597         rc = libcfs_sock_write(sock, &connreq, sizeof(connreq),
 598                                lnet_acceptor_timeout());
 599         if (rc != 0) {
 600                 CERROR("Can't tx connreq to %u.%u.%u.%u/%d: %d\n",
 601                        HIPQUAD(peer->rap_ip), peer->rap_port, rc);
 602                 goto failed_2;
 603         }
 604
 605         rc = kranal_recv_connreq(sock, &connreq, 1);
 606         if (rc != 0) {
 607                 CERROR("Can't rx connreq from %u.%u.%u.%u/%d: %d\n",
 608                        HIPQUAD(peer->rap_ip), peer->rap_port, rc);
 609                 goto failed_2;
 610         }
 611
 612         libcfs_sock_release(sock);
 613         rc = -EPROTO;
 614
 615         if (connreq.racr_srcnid != peer->rap_nid) {
 616                 CERROR("Unexpected srcnid from %u.%u.%u.%u/%d: "
 617                        "received %s expected %s\n",
 618                        HIPQUAD(peer->rap_ip), peer->rap_port,
 619                        libcfs_nid2str(connreq.racr_srcnid),
 620                        libcfs_nid2str(peer->rap_nid));
 621                 goto failed_1;
 622         }
 623
 624         if (connreq.racr_devid != dev->rad_id) {
 625                 CERROR("Unexpected device id from %u.%u.%u.%u/%d: "
 626                        "received %d expected %d\n",
 627                        HIPQUAD(peer->rap_ip), peer->rap_port,
 628                        connreq.racr_devid, dev->rad_id);
 629                 goto failed_1;
 630         }
 631
 632         rc = kranal_set_conn_params(conn, &connreq,
 633                                     peer->rap_ip, peer->rap_port);
 634         if (rc != 0)
 635                 goto failed_1;
 636
 637         *connp = conn;
 638         *dst_nidp = connreq.racr_dstnid;
 639         return 0;
 640
 641  failed_2:
 642         libcfs_sock_release(sock);
 643  failed_1:
 644         lnet_connect_console_error(rc, peer->rap_nid,
 645                                   peer->rap_ip, peer->rap_port);
 646  failed_0:
 647         kranal_conn_decref(conn);
 648         return rc;
 649 }
 650
 651 int
 652 kranal_conn_handshake (struct socket *sock, kra_peer_t *peer)
 653 {
 654         kra_peer_t        *peer2;
 655         kra_tx_t          *tx;
 656         lnet_nid_t          peer_nid;
 657         lnet_nid_t          dst_nid;
 658         unsigned long      flags;
 659         kra_conn_t        *conn;
 660         int                rc;
 661         int                nstale;
 662         int                new_peer = 0;
 663
 664         if (sock == NULL) {
 665                 /* active: connd wants to connect to 'peer' */
 666                 LASSERT (peer != NULL);
 667                 LASSERT (peer->rap_connecting);
 668
 669                 rc = kranal_active_conn_handshake(peer, &dst_nid, &conn);
 670                 if (rc != 0)
 671                         return rc;
 672
 673                 write_lock_irqsave(&kranal_data.kra_global_lock, flags);
 674
 675                 if (!kranal_peer_active(peer)) {
 676                         /* raced with peer getting unlinked */
 677                         write_unlock_irqrestore(&kranal_data.kra_global_lock,
 678                                                 flags);
 679                         kranal_conn_decref(conn);
 680                         return -ESTALE;
 681                 }
 682
 683                 peer_nid = peer->rap_nid;
 684         } else {
 685                 /* passive: listener accepted 'sock' */
 686                 LASSERT (peer == NULL);
 687
 688                 rc = kranal_passive_conn_handshake(sock, &peer_nid,
 689                                                    &dst_nid, &conn);
 690                 if (rc != 0)
 691                         return rc;
 692
 693                 /* assume this is a new peer */
 694                 rc = kranal_create_peer(&peer, peer_nid);
 695                 if (rc != 0) {
 696                         CERROR("Can't create conn for %s\n",
 697                                libcfs_nid2str(peer_nid));
 698                         kranal_conn_decref(conn);
 699                         return -ENOMEM;
 700                 }
 701
 702                 write_lock_irqsave(&kranal_data.kra_global_lock, flags);
 703
 704                 peer2 = kranal_find_peer_locked(peer_nid);
 705                 if (peer2 == NULL) {
 706                         new_peer = 1;
 707                 } else {
 708                         /* peer_nid already in the peer table */
 709                         kranal_peer_decref(peer);
 710                         peer = peer2;
 711                 }
 712         }
 713
 714         LASSERT ((!new_peer) != (!kranal_peer_active(peer)));
 715
 716         /* Refuse connection if peer thinks we are a different NID.  We check
 717          * this while holding the global lock, to synch with connection
 718          * destruction on NID change. */
 719         if (kranal_data.kra_ni->ni_nid != dst_nid) {
 720                 write_unlock_irqrestore(&kranal_data.kra_global_lock, flags);
 721
 722                 CERROR("Stale/bad connection with %s: dst_nid %s, expected %s\n",
 723                        libcfs_nid2str(peer_nid), libcfs_nid2str(dst_nid),
 724                        libcfs_nid2str(kranal_data.kra_ni->ni_nid));
 725                 rc = -ESTALE;
 726                 goto failed;
 727         }
 728
 729         /* Refuse to duplicate an existing connection (both sides might try to
 730          * connect at once).  NB we return success!  We _are_ connected so we
 731          * _don't_ have any blocked txs to complete with failure. */
 732         rc = kranal_conn_isdup_locked(peer, conn);
 733         if (rc != 0) {
 734                 LASSERT (!list_empty(&peer->rap_conns));
 735                 LASSERT (list_empty(&peer->rap_tx_queue));
 736                 write_unlock_irqrestore(&kranal_data.kra_global_lock, flags);
 737                 CWARN("Not creating duplicate connection to %s: %d\n",
 738                       libcfs_nid2str(peer_nid), rc);
 739                 rc = 0;
 740                 goto failed;
 741         }
 742
 743         if (new_peer) {
 744                 /* peer table takes my ref on the new peer */
 745                 list_add_tail(&peer->rap_list,
 746                               kranal_nid2peerlist(peer_nid));
 747         }
 748
 749         /* initialise timestamps before reaper looks at them */
 750         conn->rac_last_tx = conn->rac_last_rx = jiffies;
 751
 752         kranal_peer_addref(peer);               /* +1 ref for conn */
 753         conn->rac_peer = peer;
 754         list_add_tail(&conn->rac_list, &peer->rap_conns);
 755
 756         kranal_conn_addref(conn);               /* +1 ref for conn table */
 757         list_add_tail(&conn->rac_hashlist,
 758                       kranal_cqid2connlist(conn->rac_cqid));
 759
 760         /* Schedule all packets blocking for a connection */
 761         while (!list_empty(&peer->rap_tx_queue)) {
 762                 tx = list_entry(peer->rap_tx_queue.next,
 763                                 kra_tx_t, tx_list);
 764
 765                 list_del(&tx->tx_list);
 766                 kranal_post_fma(conn, tx);
 767         }
 768
 769         nstale = kranal_close_stale_conns_locked(peer, conn);
 770
 771         write_unlock_irqrestore(&kranal_data.kra_global_lock, flags);
 772
 773         /* CAVEAT EMPTOR: passive peer can disappear NOW */
 774
 775         if (nstale != 0)
 776                 CWARN("Closed %d stale conns to %s\n", nstale,
 777                       libcfs_nid2str(peer_nid));
 778
 779         CWARN("New connection to %s on devid[%d] = %d\n",
 780                libcfs_nid2str(peer_nid),
 781                conn->rac_device->rad_idx, conn->rac_device->rad_id);
 782
 783         /* Ensure conn gets checked.  Transmits may have been queued and an
 784          * FMA event may have happened before it got in the cq hash table */
 785         kranal_schedule_conn(conn);
 786         return 0;
 787
 788  failed:
 789         if (new_peer)
 790                 kranal_peer_decref(peer);
 791         kranal_conn_decref(conn);
 792         return rc;
 793 }
 794
 795 void
 796 kranal_connect (kra_peer_t *peer)
 797 {
 798         kra_tx_t          *tx;
 799         unsigned long      flags;
 800         struct list_head   zombies;
 801         int                rc;
 802
 803         LASSERT (peer->rap_connecting);
 804
 805         CDEBUG(D_NET, "About to handshake %s\n",
 806                libcfs_nid2str(peer->rap_nid));
 807
 808         rc = kranal_conn_handshake(NULL, peer);
 809
 810         CDEBUG(D_NET, "Done handshake %s:%d \n",
 811                libcfs_nid2str(peer->rap_nid), rc);
 812
 813         write_lock_irqsave(&kranal_data.kra_global_lock, flags);
 814
 815         LASSERT (peer->rap_connecting);
 816         peer->rap_connecting = 0;
 817
 818         if (rc == 0) {
 819                 /* kranal_conn_handshake() queues blocked txs immediately on
 820                  * success to avoid messages jumping the queue */
 821                 LASSERT (list_empty(&peer->rap_tx_queue));
 822
 823                 peer->rap_reconnect_interval = 0; /* OK to reconnect at any time */
 824
 825                 write_unlock_irqrestore(&kranal_data.kra_global_lock, flags);
 826                 return;
 827         }
 828
 829         peer->rap_reconnect_interval *= 2;
 830         peer->rap_reconnect_interval =
 831                 MAX(peer->rap_reconnect_interval,
 832                     *kranal_tunables.kra_min_reconnect_interval);
 833         peer->rap_reconnect_interval =
 834                 MIN(peer->rap_reconnect_interval,
 835                     *kranal_tunables.kra_max_reconnect_interval);
 836
 837         peer->rap_reconnect_time = jiffies + peer->rap_reconnect_interval * HZ;
 838
 839         /* Grab all blocked packets while we have the global lock */
 840         list_add(&zombies, &peer->rap_tx_queue);
 841         list_del_init(&peer->rap_tx_queue);
 842
 843         write_unlock_irqrestore(&kranal_data.kra_global_lock, flags);
 844
 845         if (list_empty(&zombies))
 846                 return;
 847
 848         CDEBUG(D_NETERROR, "Dropping packets for %s: connection failed\n",
 849                libcfs_nid2str(peer->rap_nid));
 850
 851         do {
 852                 tx = list_entry(zombies.next, kra_tx_t, tx_list);
 853
 854                 list_del(&tx->tx_list);
 855                 kranal_tx_done(tx, -EHOSTUNREACH);
 856
 857         } while (!list_empty(&zombies));
 858 }
 859
 860 void
 861 kranal_free_acceptsock (kra_acceptsock_t *ras)
 862 {
 863         libcfs_sock_release(ras->ras_sock);
 864         LIBCFS_FREE(ras, sizeof(*ras));
 865 }
 866
 867 int
 868 kranal_accept (lnet_ni_t *ni, struct socket *sock)
 869 {
 870         kra_acceptsock_t  *ras;
 871         int                rc;
 872         __u32              peer_ip;
 873         int                peer_port;
 874         unsigned long      flags;
 875
 876         rc = libcfs_sock_getaddr(sock, 1, &peer_ip, &peer_port);
 877         LASSERT (rc == 0);                      /* we succeeded before */
 878
 879         LIBCFS_ALLOC(ras, sizeof(*ras));
 880         if (ras == NULL) {
 881                 CERROR("ENOMEM allocating connection request from "
 882                        "%u.%u.%u.%u\n", HIPQUAD(peer_ip));
 883                 return -ENOMEM;
 884         }
 885
 886         ras->ras_sock = sock;
 887
 888         spin_lock_irqsave(&kranal_data.kra_connd_lock, flags);
 889
 890         list_add_tail(&ras->ras_list, &kranal_data.kra_connd_acceptq);
 891         wake_up(&kranal_data.kra_connd_waitq);
 892
 893         spin_unlock_irqrestore(&kranal_data.kra_connd_lock, flags);
 894         return 0;
 895 }
 896
 897 int
 898 kranal_create_peer (kra_peer_t **peerp, lnet_nid_t nid)
 899 {
 900         kra_peer_t    *peer;
 901         unsigned long  flags;
 902
 903         LASSERT (nid != LNET_NID_ANY);
 904
 905         LIBCFS_ALLOC(peer, sizeof(*peer));
 906         if (peer == NULL)
 907                 return -ENOMEM;
 908
 909         memset(peer, 0, sizeof(*peer));         /* zero flags etc */
 910
 911         peer->rap_nid = nid;
 912         atomic_set(&peer->rap_refcount, 1);     /* 1 ref for caller */
 913
 914         INIT_LIST_HEAD(&peer->rap_list);
 915         INIT_LIST_HEAD(&peer->rap_connd_list);
 916         INIT_LIST_HEAD(&peer->rap_conns);
 917         INIT_LIST_HEAD(&peer->rap_tx_queue);
 918
 919         peer->rap_reconnect_interval = 0;       /* OK to connect at any time */
 920
 921         write_lock_irqsave(&kranal_data.kra_global_lock, flags);
 922
 923         if (kranal_data.kra_nonewpeers) {
 924                 /* shutdown has started already */
 925                 write_unlock_irqrestore(&kranal_data.kra_global_lock, flags);
 926
 927                 LIBCFS_FREE(peer, sizeof(*peer));
 928                 CERROR("Can't create peer: network shutdown\n");
 929                 return -ESHUTDOWN;
 930         }
 931
 932         atomic_inc(&kranal_data.kra_npeers);
 933
 934         write_unlock_irqrestore(&kranal_data.kra_global_lock, flags);
 935
 936         *peerp = peer;
 937         return 0;
 938 }
 939
 940 void
 941 kranal_destroy_peer (kra_peer_t *peer)
 942 {
 943         CDEBUG(D_NET, "peer %s %p deleted\n",
 944                libcfs_nid2str(peer->rap_nid), peer);
 945
 946         LASSERT (atomic_read(&peer->rap_refcount) == 0);
 947         LASSERT (peer->rap_persistence == 0);
 948         LASSERT (!kranal_peer_active(peer));
 949         LASSERT (!peer->rap_connecting);
 950         LASSERT (list_empty(&peer->rap_conns));
 951         LASSERT (list_empty(&peer->rap_tx_queue));
 952         LASSERT (list_empty(&peer->rap_connd_list));
 953
 954         LIBCFS_FREE(peer, sizeof(*peer));
 955
 956         /* NB a peer's connections keep a reference on their peer until
 957          * they are destroyed, so we can be assured that _all_ state to do
 958          * with this peer has been cleaned up when its refcount drops to
 959          * zero. */
 960         atomic_dec(&kranal_data.kra_npeers);
 961 }
 962
 963 kra_peer_t *
 964 kranal_find_peer_locked (lnet_nid_t nid)
 965 {
 966         struct list_head *peer_list = kranal_nid2peerlist(nid);
 967         struct list_head *tmp;
 968         kra_peer_t       *peer;
 969
 970         list_for_each (tmp, peer_list) {
 971
 972                 peer = list_entry(tmp, kra_peer_t, rap_list);
 973
 974                 LASSERT (peer->rap_persistence > 0 ||     /* persistent peer */
 975                          !list_empty(&peer->rap_conns));  /* active conn */
 976
 977                 if (peer->rap_nid != nid)
 978                         continue;
 979
 980                 CDEBUG(D_NET, "got peer [%p] -> %s (%d)\n",
 981                        peer, libcfs_nid2str(nid),
 982                        atomic_read(&peer->rap_refcount));
 983                 return peer;
 984         }
 985         return NULL;
 986 }
 987
 988 kra_peer_t *
 989 kranal_find_peer (lnet_nid_t nid)
 990 {
 991         kra_peer_t     *peer;
 992
 993         read_lock(&kranal_data.kra_global_lock);
 994         peer = kranal_find_peer_locked(nid);
 995         if (peer != NULL)                       /* +1 ref for caller? */
 996                 kranal_peer_addref(peer);
 997         read_unlock(&kranal_data.kra_global_lock);
 998
 999         return peer;
1000 }
1001
1002 void
1003 kranal_unlink_peer_locked (kra_peer_t *peer)
1004 {
1005         LASSERT (peer->rap_persistence == 0);
1006         LASSERT (list_empty(&peer->rap_conns));
1007
1008         LASSERT (kranal_peer_active(peer));
1009         list_del_init(&peer->rap_list);
1010
1011         /* lose peerlist's ref */
1012         kranal_peer_decref(peer);
1013 }
1014
1015 int
1016 kranal_get_peer_info (int index, lnet_nid_t *nidp, __u32 *ipp, int *portp,
1017                       int *persistencep)
1018 {
1019         kra_peer_t        *peer;
1020         struct list_head  *ptmp;
1021         int                i;
1022
1023         read_lock(&kranal_data.kra_global_lock);
1024
1025         for (i = 0; i < kranal_data.kra_peer_hash_size; i++) {
1026
1027                 list_for_each(ptmp, &kranal_data.kra_peers[i]) {
1028
1029                         peer = list_entry(ptmp, kra_peer_t, rap_list);
1030                         LASSERT (peer->rap_persistence > 0 ||
1031                                  !list_empty(&peer->rap_conns));
1032
1033                         if (index-- > 0)
1034                                 continue;
1035
1036                         *nidp = peer->rap_nid;
1037                         *ipp = peer->rap_ip;
1038                         *portp = peer->rap_port;
1039                         *persistencep = peer->rap_persistence;
1040
1041                         read_unlock(&kranal_data.kra_global_lock);
1042                         return 0;
1043                 }
1044         }
1045
1046         read_unlock(&kranal_data.kra_global_lock);
1047         return -ENOENT;
1048 }
1049
1050 int
1051 kranal_add_persistent_peer (lnet_nid_t nid, __u32 ip, int port)
1052 {
1053         unsigned long      flags;
1054         kra_peer_t        *peer;
1055         kra_peer_t        *peer2;
1056         int                rc;
1057
1058         if (nid == LNET_NID_ANY)
1059                 return -EINVAL;
1060
1061         rc = kranal_create_peer(&peer, nid);
1062         if (rc != 0)
1063                 return rc;
1064
1065         write_lock_irqsave(&kranal_data.kra_global_lock, flags);
1066
1067         peer2 = kranal_find_peer_locked(nid);
1068         if (peer2 != NULL) {
1069                 kranal_peer_decref(peer);
1070                 peer = peer2;
1071         } else {
1072                 /* peer table takes existing ref on peer */
1073                 list_add_tail(&peer->rap_list,
1074                               kranal_nid2peerlist(nid));
1075         }
1076
1077         peer->rap_ip = ip;
1078         peer->rap_port = port;
1079         peer->rap_persistence++;
1080
1081         write_unlock_irqrestore(&kranal_data.kra_global_lock, flags);
1082         return 0;
1083 }
1084
1085 void
1086 kranal_del_peer_locked (kra_peer_t *peer)
1087 {
1088         struct list_head *ctmp;
1089         struct list_head *cnxt;
1090         kra_conn_t       *conn;
1091
1092         peer->rap_persistence = 0;
1093
1094         if (list_empty(&peer->rap_conns)) {
1095                 kranal_unlink_peer_locked(peer);
1096         } else {
1097                 list_for_each_safe(ctmp, cnxt, &peer->rap_conns) {
1098                         conn = list_entry(ctmp, kra_conn_t, rac_list);
1099
1100                         kranal_close_conn_locked(conn, 0);
1101                 }
1102                 /* peer unlinks itself when last conn is closed */
1103         }
1104 }
1105
1106 int
1107 kranal_del_peer (lnet_nid_t nid)
1108 {
1109         unsigned long      flags;
1110         struct list_head  *ptmp;
1111         struct list_head  *pnxt;
1112         kra_peer_t        *peer;
1113         int                lo;
1114         int                hi;
1115         int                i;
1116         int                rc = -ENOENT;
1117
1118         write_lock_irqsave(&kranal_data.kra_global_lock, flags);
1119
1120         if (nid != LNET_NID_ANY)
1121                 lo = hi = kranal_nid2peerlist(nid) - kranal_data.kra_peers;
1122         else {
1123                 lo = 0;
1124                 hi = kranal_data.kra_peer_hash_size - 1;
1125         }
1126
1127         for (i = lo; i <= hi; i++) {
1128                 list_for_each_safe (ptmp, pnxt, &kranal_data.kra_peers[i]) {
1129                         peer = list_entry(ptmp, kra_peer_t, rap_list);
1130                         LASSERT (peer->rap_persistence > 0 ||
1131                                  !list_empty(&peer->rap_conns));
1132
1133                         if (!(nid == LNET_NID_ANY || peer->rap_nid == nid))
1134                                 continue;
1135
1136                         kranal_del_peer_locked(peer);
1137                         rc = 0;         /* matched something */
1138                 }
1139         }
1140
1141         write_unlock_irqrestore(&kranal_data.kra_global_lock, flags);
1142
1143         return rc;
1144 }
1145
1146 kra_conn_t *
1147 kranal_get_conn_by_idx (int index)
1148 {
1149         kra_peer_t        *peer;
1150         struct list_head  *ptmp;
1151         kra_conn_t        *conn;
1152         struct list_head  *ctmp;
1153         int                i;
1154
1155         read_lock (&kranal_data.kra_global_lock);
1156
1157         for (i = 0; i < kranal_data.kra_peer_hash_size; i++) {
1158                 list_for_each (ptmp, &kranal_data.kra_peers[i]) {
1159
1160                         peer = list_entry(ptmp, kra_peer_t, rap_list);
1161                         LASSERT (peer->rap_persistence > 0 ||
1162                                  !list_empty(&peer->rap_conns));
1163
1164                         list_for_each (ctmp, &peer->rap_conns) {
1165                                 if (index-- > 0)
1166                                         continue;
1167
1168                                 conn = list_entry(ctmp, kra_conn_t, rac_list);
1169                                 CDEBUG(D_NET, "++conn[%p] -> %s (%d)\n", conn,
1170                                        libcfs_nid2str(conn->rac_peer->rap_nid),
1171                                        atomic_read(&conn->rac_refcount));
1172                                 atomic_inc(&conn->rac_refcount);
1173                                 read_unlock(&kranal_data.kra_global_lock);
1174                                 return conn;
1175                         }
1176                 }
1177         }
1178
1179         read_unlock(&kranal_data.kra_global_lock);
1180         return NULL;
1181 }
1182
1183 int
1184 kranal_close_peer_conns_locked (kra_peer_t *peer, int why)
1185 {
1186         kra_conn_t         *conn;
1187         struct list_head   *ctmp;
1188         struct list_head   *cnxt;
1189         int                 count = 0;
1190
1191         list_for_each_safe (ctmp, cnxt, &peer->rap_conns) {
1192                 conn = list_entry(ctmp, kra_conn_t, rac_list);
1193
1194                 count++;
1195                 kranal_close_conn_locked(conn, why);
1196         }
1197
1198         return count;
1199 }
1200
1201 int
1202 kranal_close_matching_conns (lnet_nid_t nid)
1203 {
1204         unsigned long       flags;
1205         kra_peer_t         *peer;
1206         struct list_head   *ptmp;
1207         struct list_head   *pnxt;
1208         int                 lo;
1209         int                 hi;
1210         int                 i;
1211         int                 count = 0;
1212
1213         write_lock_irqsave(&kranal_data.kra_global_lock, flags);
1214
1215         if (nid != LNET_NID_ANY)
1216                 lo = hi = kranal_nid2peerlist(nid) - kranal_data.kra_peers;
1217         else {
1218                 lo = 0;
1219                 hi = kranal_data.kra_peer_hash_size - 1;
1220         }
1221
1222         for (i = lo; i <= hi; i++) {
1223                 list_for_each_safe (ptmp, pnxt, &kranal_data.kra_peers[i]) {
1224
1225                         peer = list_entry(ptmp, kra_peer_t, rap_list);
1226                         LASSERT (peer->rap_persistence > 0 ||
1227                                  !list_empty(&peer->rap_conns));
1228
1229                         if (!(nid == LNET_NID_ANY || nid == peer->rap_nid))
1230                                 continue;
1231
1232                         count += kranal_close_peer_conns_locked(peer, 0);
1233                 }
1234         }
1235
1236         write_unlock_irqrestore(&kranal_data.kra_global_lock, flags);
1237
1238         /* wildcards always succeed */
1239         if (nid == LNET_NID_ANY)
1240                 return 0;
1241
1242         return (count == 0) ? -ENOENT : 0;
1243 }
1244
1245 int
1246 kranal_ctl(lnet_ni_t *ni, unsigned int cmd, void *arg)
1247 {
1248         struct libcfs_ioctl_data *data = arg;
1249         int                       rc = -EINVAL;
1250
1251         LASSERT (ni == kranal_data.kra_ni);
1252
1253         switch(cmd) {
1254         case IOC_LIBCFS_GET_PEER: {
1255                 lnet_nid_t   nid = 0;
1256                 __u32       ip = 0;
1257                 int         port = 0;
1258                 int         share_count = 0;
1259
1260                 rc = kranal_get_peer_info(data->ioc_count,
1261                                           &nid, &ip, &port, &share_count);
1262                 data->ioc_nid    = nid;
1263                 data->ioc_count  = share_count;
1264                 data->ioc_u32[0] = ip;
1265                 data->ioc_u32[1] = port;
1266                 break;
1267         }
1268         case IOC_LIBCFS_ADD_PEER: {
1269                 rc = kranal_add_persistent_peer(data->ioc_nid,
1270                                                 data->ioc_u32[0], /* IP */
1271                                                 data->ioc_u32[1]); /* port */
1272                 break;
1273         }
1274         case IOC_LIBCFS_DEL_PEER: {
1275                 rc = kranal_del_peer(data->ioc_nid);
1276                 break;
1277         }
1278         case IOC_LIBCFS_GET_CONN: {
1279                 kra_conn_t *conn = kranal_get_conn_by_idx(data->ioc_count);
1280
1281                 if (conn == NULL)
1282                         rc = -ENOENT;
1283                 else {
1284                         rc = 0;
1285                         data->ioc_nid    = conn->rac_peer->rap_nid;
1286                         data->ioc_u32[0] = conn->rac_device->rad_id;
1287                         kranal_conn_decref(conn);
1288                 }
1289                 break;
1290         }
1291         case IOC_LIBCFS_CLOSE_CONNECTION: {
1292                 rc = kranal_close_matching_conns(data->ioc_nid);
1293                 break;
1294         }
1295         case IOC_LIBCFS_REGISTER_MYNID: {
1296                 /* Ignore if this is a noop */
1297                 if (data->ioc_nid == ni->ni_nid) {
1298                         rc = 0;
1299                 } else {
1300                         CERROR("obsolete IOC_LIBCFS_REGISTER_MYNID: %s(%s)\n",
1301                                libcfs_nid2str(data->ioc_nid),
1302                                libcfs_nid2str(ni->ni_nid));
1303                         rc = -EINVAL;
1304                 }
1305                 break;
1306         }
1307         }
1308
1309         return rc;
1310 }
1311
1312 void
1313 kranal_free_txdescs(struct list_head *freelist)
1314 {
1315         kra_tx_t    *tx;
1316
1317         while (!list_empty(freelist)) {
1318                 tx = list_entry(freelist->next, kra_tx_t, tx_list);
1319
1320                 list_del(&tx->tx_list);
1321                 LIBCFS_FREE(tx->tx_phys, LNET_MAX_IOV * sizeof(*tx->tx_phys));
1322                 LIBCFS_FREE(tx, sizeof(*tx));
1323         }
1324 }
1325
1326 int
1327 kranal_alloc_txdescs(struct list_head *freelist, int n)
1328 {
1329         int            i;
1330         kra_tx_t      *tx;
1331
1332         LASSERT (freelist == &kranal_data.kra_idle_txs);
1333         LASSERT (list_empty(freelist));
1334
1335         for (i = 0; i < n; i++) {
1336
1337                 LIBCFS_ALLOC(tx, sizeof(*tx));
1338                 if (tx == NULL) {
1339                         CERROR("Can't allocate tx[%d]\n", i);
1340                         kranal_free_txdescs(freelist);
1341                         return -ENOMEM;
1342                 }
1343
1344                 LIBCFS_ALLOC(tx->tx_phys,
1345                              LNET_MAX_IOV * sizeof(*tx->tx_phys));
1346                 if (tx->tx_phys == NULL) {
1347                         CERROR("Can't allocate tx[%d]->tx_phys\n", i);
1348
1349                         LIBCFS_FREE(tx, sizeof(*tx));
1350                         kranal_free_txdescs(freelist);
1351                         return -ENOMEM;
1352                 }
1353
1354                 tx->tx_buftype = RANAL_BUF_NONE;
1355                 tx->tx_msg.ram_type = RANAL_MSG_NONE;
1356
1357                 list_add(&tx->tx_list, freelist);
1358         }
1359
1360         return 0;
1361 }
1362
1363 int
1364 kranal_device_init(int id, kra_device_t *dev)
1365 {
1366         int               total_ntx = *kranal_tunables.kra_ntx;
1367         RAP_RETURN        rrc;
1368
1369         dev->rad_id = id;
1370         rrc = RapkGetDeviceByIndex(id, kranal_device_callback,
1371                                    &dev->rad_handle);
1372         if (rrc != RAP_SUCCESS) {
1373                 CERROR("Can't get Rapidarray Device %d: %d\n", id, rrc);
1374                 goto failed_0;
1375         }
1376
1377         rrc = RapkReserveRdma(dev->rad_handle, total_ntx);
1378         if (rrc != RAP_SUCCESS) {
1379                 CERROR("Can't reserve %d RDMA descriptors"
1380                        " for device %d: %d\n", total_ntx, id, rrc);
1381                 goto failed_1;
1382         }
1383
1384         rrc = RapkCreateCQ(dev->rad_handle, total_ntx, RAP_CQTYPE_SEND,
1385                            &dev->rad_rdma_cqh);
1386         if (rrc != RAP_SUCCESS) {
1387                 CERROR("Can't create rdma cq size %d for device %d: %d\n",
1388                        total_ntx, id, rrc);
1389                 goto failed_1;
1390         }
1391
1392         rrc = RapkCreateCQ(dev->rad_handle,
1393                            *kranal_tunables.kra_fma_cq_size,
1394                            RAP_CQTYPE_RECV, &dev->rad_fma_cqh);
1395         if (rrc != RAP_SUCCESS) {
1396                 CERROR("Can't create fma cq size %d for device %d: %d\n",
1397                        *kranal_tunables.kra_fma_cq_size, id, rrc);
1398                 goto failed_2;
1399         }
1400
1401         return 0;
1402
1403  failed_2:
1404         RapkDestroyCQ(dev->rad_handle, dev->rad_rdma_cqh);
1405  failed_1:
1406         RapkReleaseDevice(dev->rad_handle);
1407  failed_0:
1408         return -ENODEV;
1409 }
1410
1411 void
1412 kranal_device_fini(kra_device_t *dev)
1413 {
1414         LASSERT (list_empty(&dev->rad_ready_conns));
1415         LASSERT (list_empty(&dev->rad_new_conns));
1416         LASSERT (dev->rad_nphysmap == 0);
1417         LASSERT (dev->rad_nppphysmap == 0);
1418         LASSERT (dev->rad_nvirtmap == 0);
1419         LASSERT (dev->rad_nobvirtmap == 0);
1420
1421         LASSERT(dev->rad_scheduler == NULL);
1422         RapkDestroyCQ(dev->rad_handle, dev->rad_fma_cqh);
1423         RapkDestroyCQ(dev->rad_handle, dev->rad_rdma_cqh);
1424         RapkReleaseDevice(dev->rad_handle);
1425 }
1426
1427 void
1428 kranal_shutdown (lnet_ni_t *ni)
1429 {
1430         int           i;
1431         unsigned long flags;
1432
1433         CDEBUG(D_MALLOC, "before NAL cleanup: kmem %d\n",
1434                atomic_read(&libcfs_kmemory));
1435
1436         LASSERT (ni == kranal_data.kra_ni);
1437         LASSERT (ni->ni_data == &kranal_data);
1438
1439         switch (kranal_data.kra_init) {
1440         default:
1441                 CERROR("Unexpected state %d\n", kranal_data.kra_init);
1442                 LBUG();
1443
1444         case RANAL_INIT_ALL:
1445                 /* Prevent new peers from being created */
1446                 write_lock_irqsave(&kranal_data.kra_global_lock, flags);
1447                 kranal_data.kra_nonewpeers = 1;
1448                 write_unlock_irqrestore(&kranal_data.kra_global_lock, flags);
1449
1450                 /* Remove all existing peers from the peer table */
1451                 kranal_del_peer(LNET_NID_ANY);
1452
1453                 /* Wait for pending conn reqs to be handled */
1454                 i = 2;
1455                 spin_lock_irqsave(&kranal_data.kra_connd_lock, flags);
1456                 while (!list_empty(&kranal_data.kra_connd_acceptq)) {
1457                         spin_unlock_irqrestore(&kranal_data.kra_connd_lock,
1458                                                flags);
1459                         i++;
1460                         CDEBUG(((i & (-i)) == i) ? D_WARNING : D_NET, /* 2**n */
1461                                "waiting for conn reqs to clean up\n");
1462                         cfs_pause(cfs_time_seconds(1));
1463
1464                         spin_lock_irqsave(&kranal_data.kra_connd_lock, flags);
1465                 }
1466                 spin_unlock_irqrestore(&kranal_data.kra_connd_lock, flags);
1467
1468                 /* Wait for all peers to be freed */
1469                 i = 2;
1470                 while (atomic_read(&kranal_data.kra_npeers) != 0) {
1471                         i++;
1472                         CDEBUG(((i & (-i)) == i) ? D_WARNING : D_NET, /* 2**n */
1473                                "waiting for %d peers to close down\n",
1474                                atomic_read(&kranal_data.kra_npeers));
1475                         cfs_pause(cfs_time_seconds(1));
1476                 }
1477                 /* fall through */
1478
1479         case RANAL_INIT_DATA:
1480                 break;
1481         }
1482
1483         /* Peer state all cleaned up BEFORE setting shutdown, so threads don't
1484          * have to worry about shutdown races.  NB connections may be created
1485          * while there are still active connds, but these will be temporary
1486          * since peer creation always fails after the listener has started to
1487          * shut down. */
1488         LASSERT (atomic_read(&kranal_data.kra_npeers) == 0);
1489
1490         /* Flag threads to terminate */
1491         kranal_data.kra_shutdown = 1;
1492
1493         for (i = 0; i < kranal_data.kra_ndevs; i++) {
1494                 kra_device_t *dev = &kranal_data.kra_devices[i];
1495
1496                 spin_lock_irqsave(&dev->rad_lock, flags);
1497                 wake_up(&dev->rad_waitq);
1498                 spin_unlock_irqrestore(&dev->rad_lock, flags);
1499         }
1500
1501         spin_lock_irqsave(&kranal_data.kra_reaper_lock, flags);
1502         wake_up_all(&kranal_data.kra_reaper_waitq);
1503         spin_unlock_irqrestore(&kranal_data.kra_reaper_lock, flags);
1504
1505         LASSERT (list_empty(&kranal_data.kra_connd_peers));
1506         spin_lock_irqsave(&kranal_data.kra_connd_lock, flags);
1507         wake_up_all(&kranal_data.kra_connd_waitq);
1508         spin_unlock_irqrestore(&kranal_data.kra_connd_lock, flags);
1509
1510         /* Wait for threads to exit */
1511         i = 2;
1512         while (atomic_read(&kranal_data.kra_nthreads) != 0) {
1513                 i++;
1514                 CDEBUG(((i & (-i)) == i) ? D_WARNING : D_NET, /* power of 2? */
1515                        "Waiting for %d threads to terminate\n",
1516                        atomic_read(&kranal_data.kra_nthreads));
1517                 cfs_pause(cfs_time_seconds(1));
1518         }
1519
1520         LASSERT (atomic_read(&kranal_data.kra_npeers) == 0);
1521         if (kranal_data.kra_peers != NULL) {
1522                 for (i = 0; i < kranal_data.kra_peer_hash_size; i++)
1523                         LASSERT (list_empty(&kranal_data.kra_peers[i]));
1524
1525                 LIBCFS_FREE(kranal_data.kra_peers,
1526                             sizeof (struct list_head) *
1527                             kranal_data.kra_peer_hash_size);
1528         }
1529
1530         LASSERT (atomic_read(&kranal_data.kra_nconns) == 0);
1531         if (kranal_data.kra_conns != NULL) {
1532                 for (i = 0; i < kranal_data.kra_conn_hash_size; i++)
1533                         LASSERT (list_empty(&kranal_data.kra_conns[i]));
1534
1535                 LIBCFS_FREE(kranal_data.kra_conns,
1536                             sizeof (struct list_head) *
1537                             kranal_data.kra_conn_hash_size);
1538         }
1539
1540         for (i = 0; i < kranal_data.kra_ndevs; i++)
1541                 kranal_device_fini(&kranal_data.kra_devices[i]);
1542
1543         kranal_free_txdescs(&kranal_data.kra_idle_txs);
1544
1545         CDEBUG(D_MALLOC, "after NAL cleanup: kmem %d\n",
1546                atomic_read(&libcfs_kmemory));
1547
1548         kranal_data.kra_init = RANAL_INIT_NOTHING;
1549         PORTAL_MODULE_UNUSE;
1550 }
1551
1552 int
1553 kranal_startup (lnet_ni_t *ni)
1554 {
1555         struct timeval    tv;
1556         int               pkmem = atomic_read(&libcfs_kmemory);
1557         int               rc;
1558         int               i;
1559         kra_device_t     *dev;
1560
1561         LASSERT (ni->ni_lnd == &the_kralnd);
1562
1563         /* Only 1 instance supported */
1564         if (kranal_data.kra_init != RANAL_INIT_NOTHING) {
1565                 CERROR ("Only 1 instance supported\n");
1566                 return -EPERM;
1567         }
1568
1569         if (lnet_set_ip_niaddr(ni) != 0) {
1570                 CERROR ("Can't determine my NID\n");
1571                 return -EPERM;
1572         }
1573
1574         if (*kranal_tunables.kra_credits > *kranal_tunables.kra_ntx) {
1575                 CERROR ("Can't set credits(%d) > ntx(%d)\n",
1576                         *kranal_tunables.kra_credits,
1577                         *kranal_tunables.kra_ntx);
1578                 return -EINVAL;
1579         }
1580
1581         memset(&kranal_data, 0, sizeof(kranal_data)); /* zero pointers, flags etc */
1582
1583         ni->ni_maxtxcredits = *kranal_tunables.kra_credits;
1584         ni->ni_peertxcredits = *kranal_tunables.kra_peercredits;
1585
1586         ni->ni_data = &kranal_data;
1587         kranal_data.kra_ni = ni;
1588
1589         /* CAVEAT EMPTOR: Every 'Fma' message includes the sender's NID and
1590          * a unique (for all time) connstamp so we can uniquely identify
1591          * the sender.  The connstamp is an incrementing counter
1592          * initialised with seconds + microseconds at startup time.  So we
1593          * rely on NOT creating connections more frequently on average than
1594          * 1MHz to ensure we don't use old connstamps when we reboot. */
1595         do_gettimeofday(&tv);
1596         kranal_data.kra_connstamp =
1597         kranal_data.kra_peerstamp = (((__u64)tv.tv_sec) * 1000000) + tv.tv_usec;
1598
1599         rwlock_init(&kranal_data.kra_global_lock);
1600
1601         for (i = 0; i < RANAL_MAXDEVS; i++ ) {
1602                 kra_device_t  *dev = &kranal_data.kra_devices[i];
1603
1604                 dev->rad_idx = i;
1605                 INIT_LIST_HEAD(&dev->rad_ready_conns);
1606                 INIT_LIST_HEAD(&dev->rad_new_conns);
1607                 init_waitqueue_head(&dev->rad_waitq);
1608                 spin_lock_init(&dev->rad_lock);
1609         }
1610
1611         kranal_data.kra_new_min_timeout = MAX_SCHEDULE_TIMEOUT;
1612         init_waitqueue_head(&kranal_data.kra_reaper_waitq);
1613         spin_lock_init(&kranal_data.kra_reaper_lock);
1614
1615         INIT_LIST_HEAD(&kranal_data.kra_connd_acceptq);
1616         INIT_LIST_HEAD(&kranal_data.kra_connd_peers);
1617         init_waitqueue_head(&kranal_data.kra_connd_waitq);
1618         spin_lock_init(&kranal_data.kra_connd_lock);
1619
1620         INIT_LIST_HEAD(&kranal_data.kra_idle_txs);
1621         spin_lock_init(&kranal_data.kra_tx_lock);
1622
1623         /* OK to call kranal_api_shutdown() to cleanup now */
1624         kranal_data.kra_init = RANAL_INIT_DATA;
1625         PORTAL_MODULE_USE;
1626
1627         kranal_data.kra_peer_hash_size = RANAL_PEER_HASH_SIZE;
1628         LIBCFS_ALLOC(kranal_data.kra_peers,
1629                      sizeof(struct list_head) * kranal_data.kra_peer_hash_size);
1630         if (kranal_data.kra_peers == NULL)
1631                 goto failed;
1632
1633         for (i = 0; i < kranal_data.kra_peer_hash_size; i++)
1634                 INIT_LIST_HEAD(&kranal_data.kra_peers[i]);
1635
1636         kranal_data.kra_conn_hash_size = RANAL_PEER_HASH_SIZE;
1637         LIBCFS_ALLOC(kranal_data.kra_conns,
1638                      sizeof(struct list_head) * kranal_data.kra_conn_hash_size);
1639         if (kranal_data.kra_conns == NULL)
1640                 goto failed;
1641
1642         for (i = 0; i < kranal_data.kra_conn_hash_size; i++)
1643                 INIT_LIST_HEAD(&kranal_data.kra_conns[i]);
1644
1645         rc = kranal_alloc_txdescs(&kranal_data.kra_idle_txs,
1646                                   *kranal_tunables.kra_ntx);
1647         if (rc != 0)
1648                 goto failed;
1649
1650         rc = kranal_thread_start(kranal_reaper, NULL);
1651         if (rc != 0) {
1652                 CERROR("Can't spawn ranal reaper: %d\n", rc);
1653                 goto failed;
1654         }
1655
1656         for (i = 0; i < *kranal_tunables.kra_n_connd; i++) {
1657                 rc = kranal_thread_start(kranal_connd, (void *)(unsigned long)i);
1658                 if (rc != 0) {
1659                         CERROR("Can't spawn ranal connd[%d]: %d\n",
1660                                i, rc);
1661                         goto failed;
1662                 }
1663         }
1664
1665         LASSERT (kranal_data.kra_ndevs == 0);
1666
1667         /* Use all available RapidArray devices */
1668         for (i = 0; i < RANAL_MAXDEVS; i++) {
1669                 dev = &kranal_data.kra_devices[kranal_data.kra_ndevs];
1670
1671                 rc = kranal_device_init(kranal_devids[i], dev);
1672                 if (rc == 0)
1673                         kranal_data.kra_ndevs++;
1674         }
1675
1676         if (kranal_data.kra_ndevs == 0) {
1677                 CERROR("Can't initialise any RapidArray devices\n");
1678                 goto failed;
1679         }
1680
1681         for (i = 0; i < kranal_data.kra_ndevs; i++) {
1682                 dev = &kranal_data.kra_devices[i];
1683                 rc = kranal_thread_start(kranal_scheduler, dev);
1684                 if (rc != 0) {
1685                         CERROR("Can't spawn ranal scheduler[%d]: %d\n",
1686                                i, rc);
1687                         goto failed;
1688                 }
1689         }
1690
1691         /* flag everything initialised */
1692         kranal_data.kra_init = RANAL_INIT_ALL;
1693         /*****************************************************/
1694
1695         CDEBUG(D_MALLOC, "initial kmem %d\n", pkmem);
1696         return 0;
1697
1698  failed:
1699         kranal_shutdown(ni);
1700         return -ENETDOWN;
1701 }
1702
1703 void __exit
1704 kranal_module_fini (void)
1705 {
1706         lnet_unregister_lnd(&the_kralnd);
1707         kranal_tunables_fini();
1708 }
1709
1710 int __init
1711 kranal_module_init (void)
1712 {
1713         int    rc;
1714
1715         rc = kranal_tunables_init();
1716         if (rc != 0)
1717                 return rc;
1718
1719         lnet_register_lnd(&the_kralnd);
1720
1721         return 0;
1722 }
1723
1724 MODULE_AUTHOR("Sun Microsystems, Inc. <http://www.lustre.org/>");
1725 MODULE_DESCRIPTION("Kernel RapidArray LND v0.01");
1726 MODULE_LICENSE("GPL");
1727
1728 module_init(kranal_module_init);
1729 module_exit(kranal_module_fini);