lnet/klnds/ralnd/ralnd.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.sun.com/software/products/lustre/docs/GPLv2.pdf
  19  *
  20  * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
  21  * CA 95054 USA or visit www.sun.com if you need additional information or
  22  * have any questions.
  23  *
  24  * GPL HEADER END
  25  */
  26 /*
  27  * Copyright (c) 2004, 2010, Oracle and/or its affiliates. All rights reserved.
  28  * Use is subject to license terms.
  29  *
  30  * Copyright (c) 2012, Intel Corporation.
  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         cfs_list_t         *ctmp;
 185         cfs_list_t         *cnxt;
 186         int                 loopback;
 187         int                 count = 0;
 188
 189         loopback = peer->rap_nid == kranal_data.kra_ni->ni_nid;
 190
 191         cfs_list_for_each_safe (ctmp, cnxt, &peer->rap_conns) {
 192                 conn = cfs_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         cfs_list_t       *tmp;
 235         int               loopback;
 236
 237         loopback = peer->rap_nid == kranal_data.kra_ni->ni_nid;
 238
 239         cfs_list_for_each(tmp, &peer->rap_conns) {
 240                 conn = cfs_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 (!cfs_in_interrupt());
 301         LIBCFS_ALLOC(conn, sizeof(*conn));
 302
 303         if (conn == NULL)
 304                 return -ENOMEM;
 305
 306         memset(conn, 0, sizeof(*conn));
 307         cfs_atomic_set(&conn->rac_refcount, 1);
 308         CFS_INIT_LIST_HEAD(&conn->rac_list);
 309         CFS_INIT_LIST_HEAD(&conn->rac_hashlist);
 310         CFS_INIT_LIST_HEAD(&conn->rac_schedlist);
 311         CFS_INIT_LIST_HEAD(&conn->rac_fmaq);
 312         CFS_INIT_LIST_HEAD(&conn->rac_rdmaq);
 313         CFS_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         cfs_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 (!cfs_in_interrupt());
 341         LASSERT (!conn->rac_scheduled);
 342         LASSERT (cfs_list_empty(&conn->rac_list));
 343         LASSERT (cfs_list_empty(&conn->rac_hashlist));
 344         LASSERT (cfs_list_empty(&conn->rac_schedlist));
 345         LASSERT (cfs_atomic_read(&conn->rac_refcount) == 0);
 346         LASSERT (cfs_list_empty(&conn->rac_fmaq));
 347         LASSERT (cfs_list_empty(&conn->rac_rdmaq));
 348         LASSERT (cfs_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         cfs_atomic_dec(&kranal_data.kra_nconns);
 359 }
 360
 361 void
 362 kranal_terminate_conn_locked (kra_conn_t *conn)
 363 {
 364         LASSERT (!cfs_in_interrupt());
 365         LASSERT (conn->rac_state == RANAL_CONN_CLOSING);
 366         LASSERT (!cfs_list_empty(&conn->rac_hashlist));
 367         LASSERT (cfs_list_empty(&conn->rac_list));
 368
 369         /* Remove from conn hash table: no new callbacks */
 370         cfs_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_LIMIT(error == 0 ? D_NET : D_NETERROR,
 386                      "closing conn to %s: error %d\n",
 387                      libcfs_nid2str(peer->rap_nid), error);
 388
 389         LASSERT (!cfs_in_interrupt());
 390         LASSERT (conn->rac_state == RANAL_CONN_ESTABLISHED);
 391         LASSERT (!cfs_list_empty(&conn->rac_hashlist));
 392         LASSERT (!cfs_list_empty(&conn->rac_list));
 393
 394         cfs_list_del_init(&conn->rac_list);
 395
 396         if (cfs_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         cfs_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         cfs_list_add_tail(&conn->rac_schedlist, &dev->rad_new_conns);
 453         cfs_waitq_signal(&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. \
 678                                                     kra_global_lock,
 679                                                     flags);
 680                         kranal_conn_decref(conn);
 681                         return -ESTALE;
 682                 }
 683
 684                 peer_nid = peer->rap_nid;
 685         } else {
 686                 /* passive: listener accepted 'sock' */
 687                 LASSERT (peer == NULL);
 688
 689                 rc = kranal_passive_conn_handshake(sock, &peer_nid,
 690                                                    &dst_nid, &conn);
 691                 if (rc != 0)
 692                         return rc;
 693
 694                 /* assume this is a new peer */
 695                 rc = kranal_create_peer(&peer, peer_nid);
 696                 if (rc != 0) {
 697                         CERROR("Can't create conn for %s\n",
 698                                libcfs_nid2str(peer_nid));
 699                         kranal_conn_decref(conn);
 700                         return -ENOMEM;
 701                 }
 702
 703                 write_lock_irqsave(&kranal_data.kra_global_lock, flags);
 704
 705                 peer2 = kranal_find_peer_locked(peer_nid);
 706                 if (peer2 == NULL) {
 707                         new_peer = 1;
 708                 } else {
 709                         /* peer_nid already in the peer table */
 710                         kranal_peer_decref(peer);
 711                         peer = peer2;
 712                 }
 713         }
 714
 715         LASSERT ((!new_peer) != (!kranal_peer_active(peer)));
 716
 717         /* Refuse connection if peer thinks we are a different NID.  We check
 718          * this while holding the global lock, to synch with connection
 719          * destruction on NID change. */
 720         if (kranal_data.kra_ni->ni_nid != dst_nid) {
 721                 write_unlock_irqrestore(&kranal_data.kra_global_lock,
 722                                             flags);
 723
 724                 CERROR("Stale/bad connection with %s: dst_nid %s, expected %s\n",
 725                        libcfs_nid2str(peer_nid), libcfs_nid2str(dst_nid),
 726                        libcfs_nid2str(kranal_data.kra_ni->ni_nid));
 727                 rc = -ESTALE;
 728                 goto failed;
 729         }
 730
 731         /* Refuse to duplicate an existing connection (both sides might try to
 732          * connect at once).  NB we return success!  We _are_ connected so we
 733          * _don't_ have any blocked txs to complete with failure. */
 734         rc = kranal_conn_isdup_locked(peer, conn);
 735         if (rc != 0) {
 736                 LASSERT (!cfs_list_empty(&peer->rap_conns));
 737                 LASSERT (cfs_list_empty(&peer->rap_tx_queue));
 738                 write_unlock_irqrestore(&kranal_data.kra_global_lock,
 739                                             flags);
 740                 CWARN("Not creating duplicate connection to %s: %d\n",
 741                       libcfs_nid2str(peer_nid), rc);
 742                 rc = 0;
 743                 goto failed;
 744         }
 745
 746         if (new_peer) {
 747                 /* peer table takes my ref on the new peer */
 748                 cfs_list_add_tail(&peer->rap_list,
 749                                   kranal_nid2peerlist(peer_nid));
 750         }
 751
 752         /* initialise timestamps before reaper looks at them */
 753         conn->rac_last_tx = conn->rac_last_rx = jiffies;
 754
 755         kranal_peer_addref(peer);               /* +1 ref for conn */
 756         conn->rac_peer = peer;
 757         cfs_list_add_tail(&conn->rac_list, &peer->rap_conns);
 758
 759         kranal_conn_addref(conn);               /* +1 ref for conn table */
 760         cfs_list_add_tail(&conn->rac_hashlist,
 761                           kranal_cqid2connlist(conn->rac_cqid));
 762
 763         /* Schedule all packets blocking for a connection */
 764         while (!cfs_list_empty(&peer->rap_tx_queue)) {
 765                 tx = cfs_list_entry(peer->rap_tx_queue.next,
 766                                     kra_tx_t, tx_list);
 767
 768                 cfs_list_del(&tx->tx_list);
 769                 kranal_post_fma(conn, tx);
 770         }
 771
 772         nstale = kranal_close_stale_conns_locked(peer, conn);
 773
 774         write_unlock_irqrestore(&kranal_data.kra_global_lock, flags);
 775
 776         /* CAVEAT EMPTOR: passive peer can disappear NOW */
 777
 778         if (nstale != 0)
 779                 CWARN("Closed %d stale conns to %s\n", nstale,
 780                       libcfs_nid2str(peer_nid));
 781
 782         CWARN("New connection to %s on devid[%d] = %d\n",
 783                libcfs_nid2str(peer_nid),
 784                conn->rac_device->rad_idx, conn->rac_device->rad_id);
 785
 786         /* Ensure conn gets checked.  Transmits may have been queued and an
 787          * FMA event may have happened before it got in the cq hash table */
 788         kranal_schedule_conn(conn);
 789         return 0;
 790
 791  failed:
 792         if (new_peer)
 793                 kranal_peer_decref(peer);
 794         kranal_conn_decref(conn);
 795         return rc;
 796 }
 797
 798 void
 799 kranal_connect (kra_peer_t *peer)
 800 {
 801         kra_tx_t          *tx;
 802         unsigned long      flags;
 803         cfs_list_t         zombies;
 804         int                rc;
 805
 806         LASSERT (peer->rap_connecting);
 807
 808         CDEBUG(D_NET, "About to handshake %s\n",
 809                libcfs_nid2str(peer->rap_nid));
 810
 811         rc = kranal_conn_handshake(NULL, peer);
 812
 813         CDEBUG(D_NET, "Done handshake %s:%d \n",
 814                libcfs_nid2str(peer->rap_nid), rc);
 815
 816         write_lock_irqsave(&kranal_data.kra_global_lock, flags);
 817
 818         LASSERT (peer->rap_connecting);
 819         peer->rap_connecting = 0;
 820
 821         if (rc == 0) {
 822                 /* kranal_conn_handshake() queues blocked txs immediately on
 823                  * success to avoid messages jumping the queue */
 824                 LASSERT (cfs_list_empty(&peer->rap_tx_queue));
 825
 826                 peer->rap_reconnect_interval = 0; /* OK to reconnect at any time */
 827
 828                 write_unlock_irqrestore(&kranal_data.kra_global_lock,
 829                                             flags);
 830                 return;
 831         }
 832
 833         peer->rap_reconnect_interval *= 2;
 834         peer->rap_reconnect_interval =
 835                 MAX(peer->rap_reconnect_interval,
 836                     *kranal_tunables.kra_min_reconnect_interval);
 837         peer->rap_reconnect_interval =
 838                 MIN(peer->rap_reconnect_interval,
 839                     *kranal_tunables.kra_max_reconnect_interval);
 840
 841         peer->rap_reconnect_time = jiffies + peer->rap_reconnect_interval *
 842                 CFS_HZ;
 843
 844         /* Grab all blocked packets while we have the global lock */
 845         cfs_list_add(&zombies, &peer->rap_tx_queue);
 846         cfs_list_del_init(&peer->rap_tx_queue);
 847
 848         write_unlock_irqrestore(&kranal_data.kra_global_lock, flags);
 849
 850         if (cfs_list_empty(&zombies))
 851                 return;
 852
 853         CNETERR("Dropping packets for %s: connection failed\n",
 854                 libcfs_nid2str(peer->rap_nid));
 855
 856         do {
 857                 tx = cfs_list_entry(zombies.next, kra_tx_t, tx_list);
 858
 859                 cfs_list_del(&tx->tx_list);
 860                 kranal_tx_done(tx, -EHOSTUNREACH);
 861
 862         } while (!cfs_list_empty(&zombies));
 863 }
 864
 865 void
 866 kranal_free_acceptsock (kra_acceptsock_t *ras)
 867 {
 868         libcfs_sock_release(ras->ras_sock);
 869         LIBCFS_FREE(ras, sizeof(*ras));
 870 }
 871
 872 int
 873 kranal_accept (lnet_ni_t *ni, struct socket *sock)
 874 {
 875         kra_acceptsock_t  *ras;
 876         int                rc;
 877         __u32              peer_ip;
 878         int                peer_port;
 879         unsigned long      flags;
 880
 881         rc = libcfs_sock_getaddr(sock, 1, &peer_ip, &peer_port);
 882         LASSERT (rc == 0);                      /* we succeeded before */
 883
 884         LIBCFS_ALLOC(ras, sizeof(*ras));
 885         if (ras == NULL) {
 886                 CERROR("ENOMEM allocating connection request from "
 887                        "%u.%u.%u.%u\n", HIPQUAD(peer_ip));
 888                 return -ENOMEM;
 889         }
 890
 891         ras->ras_sock = sock;
 892
 893         spin_lock_irqsave(&kranal_data.kra_connd_lock, flags);
 894
 895         cfs_list_add_tail(&ras->ras_list, &kranal_data.kra_connd_acceptq);
 896         cfs_waitq_signal(&kranal_data.kra_connd_waitq);
 897
 898         spin_unlock_irqrestore(&kranal_data.kra_connd_lock, flags);
 899         return 0;
 900 }
 901
 902 int
 903 kranal_create_peer (kra_peer_t **peerp, lnet_nid_t nid)
 904 {
 905         kra_peer_t    *peer;
 906         unsigned long  flags;
 907
 908         LASSERT (nid != LNET_NID_ANY);
 909
 910         LIBCFS_ALLOC(peer, sizeof(*peer));
 911         if (peer == NULL)
 912                 return -ENOMEM;
 913
 914         memset(peer, 0, sizeof(*peer));         /* zero flags etc */
 915
 916         peer->rap_nid = nid;
 917         cfs_atomic_set(&peer->rap_refcount, 1);     /* 1 ref for caller */
 918
 919         CFS_INIT_LIST_HEAD(&peer->rap_list);
 920         CFS_INIT_LIST_HEAD(&peer->rap_connd_list);
 921         CFS_INIT_LIST_HEAD(&peer->rap_conns);
 922         CFS_INIT_LIST_HEAD(&peer->rap_tx_queue);
 923
 924         peer->rap_reconnect_interval = 0;       /* OK to connect at any time */
 925
 926         write_lock_irqsave(&kranal_data.kra_global_lock, flags);
 927
 928         if (kranal_data.kra_nonewpeers) {
 929                 /* shutdown has started already */
 930                 write_unlock_irqrestore(&kranal_data.kra_global_lock,
 931                                             flags);
 932
 933                 LIBCFS_FREE(peer, sizeof(*peer));
 934                 CERROR("Can't create peer: network shutdown\n");
 935                 return -ESHUTDOWN;
 936         }
 937
 938         cfs_atomic_inc(&kranal_data.kra_npeers);
 939
 940         write_unlock_irqrestore(&kranal_data.kra_global_lock, flags);
 941
 942         *peerp = peer;
 943         return 0;
 944 }
 945
 946 void
 947 kranal_destroy_peer (kra_peer_t *peer)
 948 {
 949         CDEBUG(D_NET, "peer %s %p deleted\n",
 950                libcfs_nid2str(peer->rap_nid), peer);
 951
 952         LASSERT (cfs_atomic_read(&peer->rap_refcount) == 0);
 953         LASSERT (peer->rap_persistence == 0);
 954         LASSERT (!kranal_peer_active(peer));
 955         LASSERT (!peer->rap_connecting);
 956         LASSERT (cfs_list_empty(&peer->rap_conns));
 957         LASSERT (cfs_list_empty(&peer->rap_tx_queue));
 958         LASSERT (cfs_list_empty(&peer->rap_connd_list));
 959
 960         LIBCFS_FREE(peer, sizeof(*peer));
 961
 962         /* NB a peer's connections keep a reference on their peer until
 963          * they are destroyed, so we can be assured that _all_ state to do
 964          * with this peer has been cleaned up when its refcount drops to
 965          * zero. */
 966         cfs_atomic_dec(&kranal_data.kra_npeers);
 967 }
 968
 969 kra_peer_t *
 970 kranal_find_peer_locked (lnet_nid_t nid)
 971 {
 972         cfs_list_t       *peer_list = kranal_nid2peerlist(nid);
 973         cfs_list_t       *tmp;
 974         kra_peer_t       *peer;
 975
 976         cfs_list_for_each (tmp, peer_list) {
 977
 978                 peer = cfs_list_entry(tmp, kra_peer_t, rap_list);
 979
 980                 LASSERT (peer->rap_persistence > 0 ||     /* persistent peer */
 981                          !cfs_list_empty(&peer->rap_conns));  /* active conn */
 982
 983                 if (peer->rap_nid != nid)
 984                         continue;
 985
 986                 CDEBUG(D_NET, "got peer [%p] -> %s (%d)\n",
 987                        peer, libcfs_nid2str(nid),
 988                        cfs_atomic_read(&peer->rap_refcount));
 989                 return peer;
 990         }
 991         return NULL;
 992 }
 993
 994 kra_peer_t *
 995 kranal_find_peer (lnet_nid_t nid)
 996 {
 997         kra_peer_t     *peer;
 998
 999         read_lock(&kranal_data.kra_global_lock);
1000         peer = kranal_find_peer_locked(nid);
1001         if (peer != NULL)                       /* +1 ref for caller? */
1002                 kranal_peer_addref(peer);
1003         read_unlock(&kranal_data.kra_global_lock);
1004
1005         return peer;
1006 }
1007
1008 void
1009 kranal_unlink_peer_locked (kra_peer_t *peer)
1010 {
1011         LASSERT (peer->rap_persistence == 0);
1012         LASSERT (cfs_list_empty(&peer->rap_conns));
1013
1014         LASSERT (kranal_peer_active(peer));
1015         cfs_list_del_init(&peer->rap_list);
1016
1017         /* lose peerlist's ref */
1018         kranal_peer_decref(peer);
1019 }
1020
1021 int
1022 kranal_get_peer_info (int index, lnet_nid_t *nidp, __u32 *ipp, int *portp,
1023                       int *persistencep)
1024 {
1025         kra_peer_t        *peer;
1026         cfs_list_t        *ptmp;
1027         int                i;
1028
1029         read_lock(&kranal_data.kra_global_lock);
1030
1031         for (i = 0; i < kranal_data.kra_peer_hash_size; i++) {
1032
1033                 cfs_list_for_each(ptmp, &kranal_data.kra_peers[i]) {
1034
1035                         peer = cfs_list_entry(ptmp, kra_peer_t, rap_list);
1036                         LASSERT (peer->rap_persistence > 0 ||
1037                                  !cfs_list_empty(&peer->rap_conns));
1038
1039                         if (index-- > 0)
1040                                 continue;
1041
1042                         *nidp = peer->rap_nid;
1043                         *ipp = peer->rap_ip;
1044                         *portp = peer->rap_port;
1045                         *persistencep = peer->rap_persistence;
1046
1047                         read_unlock(&kranal_data.kra_global_lock);
1048                         return 0;
1049                 }
1050         }
1051
1052         read_unlock(&kranal_data.kra_global_lock);
1053         return -ENOENT;
1054 }
1055
1056 int
1057 kranal_add_persistent_peer (lnet_nid_t nid, __u32 ip, int port)
1058 {
1059         unsigned long      flags;
1060         kra_peer_t        *peer;
1061         kra_peer_t        *peer2;
1062         int                rc;
1063
1064         if (nid == LNET_NID_ANY)
1065                 return -EINVAL;
1066
1067         rc = kranal_create_peer(&peer, nid);
1068         if (rc != 0)
1069                 return rc;
1070
1071         write_lock_irqsave(&kranal_data.kra_global_lock, flags);
1072
1073         peer2 = kranal_find_peer_locked(nid);
1074         if (peer2 != NULL) {
1075                 kranal_peer_decref(peer);
1076                 peer = peer2;
1077         } else {
1078                 /* peer table takes existing ref on peer */
1079                 cfs_list_add_tail(&peer->rap_list,
1080                               kranal_nid2peerlist(nid));
1081         }
1082
1083         peer->rap_ip = ip;
1084         peer->rap_port = port;
1085         peer->rap_persistence++;
1086
1087         write_unlock_irqrestore(&kranal_data.kra_global_lock, flags);
1088         return 0;
1089 }
1090
1091 void
1092 kranal_del_peer_locked (kra_peer_t *peer)
1093 {
1094         cfs_list_t       *ctmp;
1095         cfs_list_t       *cnxt;
1096         kra_conn_t       *conn;
1097
1098         peer->rap_persistence = 0;
1099
1100         if (cfs_list_empty(&peer->rap_conns)) {
1101                 kranal_unlink_peer_locked(peer);
1102         } else {
1103                 cfs_list_for_each_safe(ctmp, cnxt, &peer->rap_conns) {
1104                         conn = cfs_list_entry(ctmp, kra_conn_t, rac_list);
1105
1106                         kranal_close_conn_locked(conn, 0);
1107                 }
1108                 /* peer unlinks itself when last conn is closed */
1109         }
1110 }
1111
1112 int
1113 kranal_del_peer (lnet_nid_t nid)
1114 {
1115         unsigned long      flags;
1116         cfs_list_t        *ptmp;
1117         cfs_list_t        *pnxt;
1118         kra_peer_t        *peer;
1119         int                lo;
1120         int                hi;
1121         int                i;
1122         int                rc = -ENOENT;
1123
1124         write_lock_irqsave(&kranal_data.kra_global_lock, flags);
1125
1126         if (nid != LNET_NID_ANY)
1127                 lo = hi = kranal_nid2peerlist(nid) - kranal_data.kra_peers;
1128         else {
1129                 lo = 0;
1130                 hi = kranal_data.kra_peer_hash_size - 1;
1131         }
1132
1133         for (i = lo; i <= hi; i++) {
1134                 cfs_list_for_each_safe (ptmp, pnxt, &kranal_data.kra_peers[i]) {
1135                         peer = cfs_list_entry(ptmp, kra_peer_t, rap_list);
1136                         LASSERT (peer->rap_persistence > 0 ||
1137                                  !cfs_list_empty(&peer->rap_conns));
1138
1139                         if (!(nid == LNET_NID_ANY || peer->rap_nid == nid))
1140                                 continue;
1141
1142                         kranal_del_peer_locked(peer);
1143                         rc = 0;         /* matched something */
1144                 }
1145         }
1146
1147         write_unlock_irqrestore(&kranal_data.kra_global_lock, flags);
1148
1149         return rc;
1150 }
1151
1152 kra_conn_t *
1153 kranal_get_conn_by_idx (int index)
1154 {
1155         kra_peer_t        *peer;
1156         cfs_list_t        *ptmp;
1157         kra_conn_t        *conn;
1158         cfs_list_t        *ctmp;
1159         int                i;
1160
1161         read_lock(&kranal_data.kra_global_lock);
1162
1163         for (i = 0; i < kranal_data.kra_peer_hash_size; i++) {
1164                 cfs_list_for_each (ptmp, &kranal_data.kra_peers[i]) {
1165
1166                         peer = cfs_list_entry(ptmp, kra_peer_t, rap_list);
1167                         LASSERT (peer->rap_persistence > 0 ||
1168                                  !cfs_list_empty(&peer->rap_conns));
1169
1170                         cfs_list_for_each (ctmp, &peer->rap_conns) {
1171                                 if (index-- > 0)
1172                                         continue;
1173
1174                                 conn = cfs_list_entry(ctmp, kra_conn_t,
1175                                                       rac_list);
1176                                 CDEBUG(D_NET, "++conn[%p] -> %s (%d)\n", conn,
1177                                        libcfs_nid2str(conn->rac_peer->rap_nid),
1178                                        cfs_atomic_read(&conn->rac_refcount));
1179                                 cfs_atomic_inc(&conn->rac_refcount);
1180                                 read_unlock(&kranal_data.kra_global_lock);
1181                                 return conn;
1182                         }
1183                 }
1184         }
1185
1186         read_unlock(&kranal_data.kra_global_lock);
1187         return NULL;
1188 }
1189
1190 int
1191 kranal_close_peer_conns_locked (kra_peer_t *peer, int why)
1192 {
1193         kra_conn_t         *conn;
1194         cfs_list_t         *ctmp;
1195         cfs_list_t         *cnxt;
1196         int                 count = 0;
1197
1198         cfs_list_for_each_safe (ctmp, cnxt, &peer->rap_conns) {
1199                 conn = cfs_list_entry(ctmp, kra_conn_t, rac_list);
1200
1201                 count++;
1202                 kranal_close_conn_locked(conn, why);
1203         }
1204
1205         return count;
1206 }
1207
1208 int
1209 kranal_close_matching_conns (lnet_nid_t nid)
1210 {
1211         unsigned long       flags;
1212         kra_peer_t         *peer;
1213         cfs_list_t         *ptmp;
1214         cfs_list_t         *pnxt;
1215         int                 lo;
1216         int                 hi;
1217         int                 i;
1218         int                 count = 0;
1219
1220         write_lock_irqsave(&kranal_data.kra_global_lock, flags);
1221
1222         if (nid != LNET_NID_ANY)
1223                 lo = hi = kranal_nid2peerlist(nid) - kranal_data.kra_peers;
1224         else {
1225                 lo = 0;
1226                 hi = kranal_data.kra_peer_hash_size - 1;
1227         }
1228
1229         for (i = lo; i <= hi; i++) {
1230                 cfs_list_for_each_safe (ptmp, pnxt, &kranal_data.kra_peers[i]) {
1231
1232                         peer = cfs_list_entry(ptmp, kra_peer_t, rap_list);
1233                         LASSERT (peer->rap_persistence > 0 ||
1234                                  !cfs_list_empty(&peer->rap_conns));
1235
1236                         if (!(nid == LNET_NID_ANY || nid == peer->rap_nid))
1237                                 continue;
1238
1239                         count += kranal_close_peer_conns_locked(peer, 0);
1240                 }
1241         }
1242
1243         write_unlock_irqrestore(&kranal_data.kra_global_lock, flags);
1244
1245         /* wildcards always succeed */
1246         if (nid == LNET_NID_ANY)
1247                 return 0;
1248
1249         return (count == 0) ? -ENOENT : 0;
1250 }
1251
1252 int
1253 kranal_ctl(lnet_ni_t *ni, unsigned int cmd, void *arg)
1254 {
1255         struct libcfs_ioctl_data *data = arg;
1256         int                       rc = -EINVAL;
1257
1258         LASSERT (ni == kranal_data.kra_ni);
1259
1260         switch(cmd) {
1261         case IOC_LIBCFS_GET_PEER: {
1262                 lnet_nid_t   nid = 0;
1263                 __u32       ip = 0;
1264                 int         port = 0;
1265                 int         share_count = 0;
1266
1267                 rc = kranal_get_peer_info(data->ioc_count,
1268                                           &nid, &ip, &port, &share_count);
1269                 data->ioc_nid    = nid;
1270                 data->ioc_count  = share_count;
1271                 data->ioc_u32[0] = ip;
1272                 data->ioc_u32[1] = port;
1273                 break;
1274         }
1275         case IOC_LIBCFS_ADD_PEER: {
1276                 rc = kranal_add_persistent_peer(data->ioc_nid,
1277                                                 data->ioc_u32[0], /* IP */
1278                                                 data->ioc_u32[1]); /* port */
1279                 break;
1280         }
1281         case IOC_LIBCFS_DEL_PEER: {
1282                 rc = kranal_del_peer(data->ioc_nid);
1283                 break;
1284         }
1285         case IOC_LIBCFS_GET_CONN: {
1286                 kra_conn_t *conn = kranal_get_conn_by_idx(data->ioc_count);
1287
1288                 if (conn == NULL)
1289                         rc = -ENOENT;
1290                 else {
1291                         rc = 0;
1292                         data->ioc_nid    = conn->rac_peer->rap_nid;
1293                         data->ioc_u32[0] = conn->rac_device->rad_id;
1294                         kranal_conn_decref(conn);
1295                 }
1296                 break;
1297         }
1298         case IOC_LIBCFS_CLOSE_CONNECTION: {
1299                 rc = kranal_close_matching_conns(data->ioc_nid);
1300                 break;
1301         }
1302         case IOC_LIBCFS_REGISTER_MYNID: {
1303                 /* Ignore if this is a noop */
1304                 if (data->ioc_nid == ni->ni_nid) {
1305                         rc = 0;
1306                 } else {
1307                         CERROR("obsolete IOC_LIBCFS_REGISTER_MYNID: %s(%s)\n",
1308                                libcfs_nid2str(data->ioc_nid),
1309                                libcfs_nid2str(ni->ni_nid));
1310                         rc = -EINVAL;
1311                 }
1312                 break;
1313         }
1314         }
1315
1316         return rc;
1317 }
1318
1319 void
1320 kranal_free_txdescs(cfs_list_t *freelist)
1321 {
1322         kra_tx_t    *tx;
1323
1324         while (!cfs_list_empty(freelist)) {
1325                 tx = cfs_list_entry(freelist->next, kra_tx_t, tx_list);
1326
1327                 cfs_list_del(&tx->tx_list);
1328                 LIBCFS_FREE(tx->tx_phys, LNET_MAX_IOV * sizeof(*tx->tx_phys));
1329                 LIBCFS_FREE(tx, sizeof(*tx));
1330         }
1331 }
1332
1333 int
1334 kranal_alloc_txdescs(cfs_list_t *freelist, int n)
1335 {
1336         int            i;
1337         kra_tx_t      *tx;
1338
1339         LASSERT (freelist == &kranal_data.kra_idle_txs);
1340         LASSERT (cfs_list_empty(freelist));
1341
1342         for (i = 0; i < n; i++) {
1343
1344                 LIBCFS_ALLOC(tx, sizeof(*tx));
1345                 if (tx == NULL) {
1346                         CERROR("Can't allocate tx[%d]\n", i);
1347                         kranal_free_txdescs(freelist);
1348                         return -ENOMEM;
1349                 }
1350
1351                 LIBCFS_ALLOC(tx->tx_phys,
1352                              LNET_MAX_IOV * sizeof(*tx->tx_phys));
1353                 if (tx->tx_phys == NULL) {
1354                         CERROR("Can't allocate tx[%d]->tx_phys\n", i);
1355
1356                         LIBCFS_FREE(tx, sizeof(*tx));
1357                         kranal_free_txdescs(freelist);
1358                         return -ENOMEM;
1359                 }
1360
1361                 tx->tx_buftype = RANAL_BUF_NONE;
1362                 tx->tx_msg.ram_type = RANAL_MSG_NONE;
1363
1364                 cfs_list_add(&tx->tx_list, freelist);
1365         }
1366
1367         return 0;
1368 }
1369
1370 int
1371 kranal_device_init(int id, kra_device_t *dev)
1372 {
1373         int               total_ntx = *kranal_tunables.kra_ntx;
1374         RAP_RETURN        rrc;
1375
1376         dev->rad_id = id;
1377         rrc = RapkGetDeviceByIndex(id, kranal_device_callback,
1378                                    &dev->rad_handle);
1379         if (rrc != RAP_SUCCESS) {
1380                 CERROR("Can't get Rapidarray Device %d: %d\n", id, rrc);
1381                 goto failed_0;
1382         }
1383
1384         rrc = RapkReserveRdma(dev->rad_handle, total_ntx);
1385         if (rrc != RAP_SUCCESS) {
1386                 CERROR("Can't reserve %d RDMA descriptors"
1387                        " for device %d: %d\n", total_ntx, id, rrc);
1388                 goto failed_1;
1389         }
1390
1391         rrc = RapkCreateCQ(dev->rad_handle, total_ntx, RAP_CQTYPE_SEND,
1392                            &dev->rad_rdma_cqh);
1393         if (rrc != RAP_SUCCESS) {
1394                 CERROR("Can't create rdma cq size %d for device %d: %d\n",
1395                        total_ntx, id, rrc);
1396                 goto failed_1;
1397         }
1398
1399         rrc = RapkCreateCQ(dev->rad_handle,
1400                            *kranal_tunables.kra_fma_cq_size,
1401                            RAP_CQTYPE_RECV, &dev->rad_fma_cqh);
1402         if (rrc != RAP_SUCCESS) {
1403                 CERROR("Can't create fma cq size %d for device %d: %d\n",
1404                        *kranal_tunables.kra_fma_cq_size, id, rrc);
1405                 goto failed_2;
1406         }
1407
1408         return 0;
1409
1410  failed_2:
1411         RapkDestroyCQ(dev->rad_handle, dev->rad_rdma_cqh);
1412  failed_1:
1413         RapkReleaseDevice(dev->rad_handle);
1414  failed_0:
1415         return -ENODEV;
1416 }
1417
1418 void
1419 kranal_device_fini(kra_device_t *dev)
1420 {
1421         LASSERT (cfs_list_empty(&dev->rad_ready_conns));
1422         LASSERT (cfs_list_empty(&dev->rad_new_conns));
1423         LASSERT (dev->rad_nphysmap == 0);
1424         LASSERT (dev->rad_nppphysmap == 0);
1425         LASSERT (dev->rad_nvirtmap == 0);
1426         LASSERT (dev->rad_nobvirtmap == 0);
1427
1428         LASSERT(dev->rad_scheduler == NULL);
1429         RapkDestroyCQ(dev->rad_handle, dev->rad_fma_cqh);
1430         RapkDestroyCQ(dev->rad_handle, dev->rad_rdma_cqh);
1431         RapkReleaseDevice(dev->rad_handle);
1432 }
1433
1434 void
1435 kranal_shutdown (lnet_ni_t *ni)
1436 {
1437         int           i;
1438         unsigned long flags;
1439
1440         CDEBUG(D_MALLOC, "before NAL cleanup: kmem %d\n",
1441                cfs_atomic_read(&libcfs_kmemory));
1442
1443         LASSERT (ni == kranal_data.kra_ni);
1444         LASSERT (ni->ni_data == &kranal_data);
1445
1446         switch (kranal_data.kra_init) {
1447         default:
1448                 CERROR("Unexpected state %d\n", kranal_data.kra_init);
1449                 LBUG();
1450
1451         case RANAL_INIT_ALL:
1452                 /* Prevent new peers from being created */
1453                 write_lock_irqsave(&kranal_data.kra_global_lock, flags);
1454                 kranal_data.kra_nonewpeers = 1;
1455                 write_unlock_irqrestore(&kranal_data.kra_global_lock,
1456                                             flags);
1457
1458                 /* Remove all existing peers from the peer table */
1459                 kranal_del_peer(LNET_NID_ANY);
1460
1461                 /* Wait for pending conn reqs to be handled */
1462                 i = 2;
1463                 spin_lock_irqsave(&kranal_data.kra_connd_lock, flags);
1464                 while (!cfs_list_empty(&kranal_data.kra_connd_acceptq)) {
1465                         spin_unlock_irqrestore(&kranal_data.kra_connd_lock,
1466                                                    flags);
1467                         i++;
1468                         CDEBUG(((i & (-i)) == i) ? D_WARNING : D_NET, /* 2**n */
1469                                "waiting for conn reqs to clean up\n");
1470                         cfs_pause(cfs_time_seconds(1));
1471
1472                         spin_lock_irqsave(&kranal_data.kra_connd_lock,
1473                                               flags);
1474                 }
1475                 spin_unlock_irqrestore(&kranal_data.kra_connd_lock, flags);
1476
1477                 /* Wait for all peers to be freed */
1478                 i = 2;
1479                 while (cfs_atomic_read(&kranal_data.kra_npeers) != 0) {
1480                         i++;
1481                         CDEBUG(((i & (-i)) == i) ? D_WARNING : D_NET, /* 2**n */
1482                                "waiting for %d peers to close down\n",
1483                                cfs_atomic_read(&kranal_data.kra_npeers));
1484                         cfs_pause(cfs_time_seconds(1));
1485                 }
1486                 /* fall through */
1487
1488         case RANAL_INIT_DATA:
1489                 break;
1490         }
1491
1492         /* Peer state all cleaned up BEFORE setting shutdown, so threads don't
1493          * have to worry about shutdown races.  NB connections may be created
1494          * while there are still active connds, but these will be temporary
1495          * since peer creation always fails after the listener has started to
1496          * shut down. */
1497         LASSERT (cfs_atomic_read(&kranal_data.kra_npeers) == 0);
1498
1499         /* Flag threads to terminate */
1500         kranal_data.kra_shutdown = 1;
1501
1502         for (i = 0; i < kranal_data.kra_ndevs; i++) {
1503                 kra_device_t *dev = &kranal_data.kra_devices[i];
1504
1505                 spin_lock_irqsave(&dev->rad_lock, flags);
1506                 cfs_waitq_signal(&dev->rad_waitq);
1507                 spin_unlock_irqrestore(&dev->rad_lock, flags);
1508         }
1509
1510         spin_lock_irqsave(&kranal_data.kra_reaper_lock, flags);
1511         cfs_waitq_broadcast(&kranal_data.kra_reaper_waitq);
1512         spin_unlock_irqrestore(&kranal_data.kra_reaper_lock, flags);
1513
1514         LASSERT (cfs_list_empty(&kranal_data.kra_connd_peers));
1515         spin_lock_irqsave(&kranal_data.kra_connd_lock, flags);
1516         cfs_waitq_broadcast(&kranal_data.kra_connd_waitq);
1517         spin_unlock_irqrestore(&kranal_data.kra_connd_lock, flags);
1518
1519         /* Wait for threads to exit */
1520         i = 2;
1521         while (cfs_atomic_read(&kranal_data.kra_nthreads) != 0) {
1522                 i++;
1523                 CDEBUG(((i & (-i)) == i) ? D_WARNING : D_NET, /* power of 2? */
1524                        "Waiting for %d threads to terminate\n",
1525                        cfs_atomic_read(&kranal_data.kra_nthreads));
1526                 cfs_pause(cfs_time_seconds(1));
1527         }
1528
1529         LASSERT (cfs_atomic_read(&kranal_data.kra_npeers) == 0);
1530         if (kranal_data.kra_peers != NULL) {
1531                 for (i = 0; i < kranal_data.kra_peer_hash_size; i++)
1532                         LASSERT (cfs_list_empty(&kranal_data.kra_peers[i]));
1533
1534                 LIBCFS_FREE(kranal_data.kra_peers,
1535                             sizeof (cfs_list_t) *
1536                             kranal_data.kra_peer_hash_size);
1537         }
1538
1539         LASSERT (cfs_atomic_read(&kranal_data.kra_nconns) == 0);
1540         if (kranal_data.kra_conns != NULL) {
1541                 for (i = 0; i < kranal_data.kra_conn_hash_size; i++)
1542                         LASSERT (cfs_list_empty(&kranal_data.kra_conns[i]));
1543
1544                 LIBCFS_FREE(kranal_data.kra_conns,
1545                             sizeof (cfs_list_t) *
1546                             kranal_data.kra_conn_hash_size);
1547         }
1548
1549         for (i = 0; i < kranal_data.kra_ndevs; i++)
1550                 kranal_device_fini(&kranal_data.kra_devices[i]);
1551
1552         kranal_free_txdescs(&kranal_data.kra_idle_txs);
1553
1554         CDEBUG(D_MALLOC, "after NAL cleanup: kmem %d\n",
1555                cfs_atomic_read(&libcfs_kmemory));
1556
1557         kranal_data.kra_init = RANAL_INIT_NOTHING;
1558         PORTAL_MODULE_UNUSE;
1559 }
1560
1561 int
1562 kranal_startup (lnet_ni_t *ni)
1563 {
1564         struct timeval    tv;
1565         int               pkmem = cfs_atomic_read(&libcfs_kmemory);
1566         int               rc;
1567         int               i;
1568         kra_device_t     *dev;
1569
1570         LASSERT (ni->ni_lnd == &the_kralnd);
1571
1572         /* Only 1 instance supported */
1573         if (kranal_data.kra_init != RANAL_INIT_NOTHING) {
1574                 CERROR ("Only 1 instance supported\n");
1575                 return -EPERM;
1576         }
1577
1578         if (lnet_set_ip_niaddr(ni) != 0) {
1579                 CERROR ("Can't determine my NID\n");
1580                 return -EPERM;
1581         }
1582
1583         if (*kranal_tunables.kra_credits > *kranal_tunables.kra_ntx) {
1584                 CERROR ("Can't set credits(%d) > ntx(%d)\n",
1585                         *kranal_tunables.kra_credits,
1586                         *kranal_tunables.kra_ntx);
1587                 return -EINVAL;
1588         }
1589
1590         memset(&kranal_data, 0, sizeof(kranal_data)); /* zero pointers, flags etc */
1591
1592         ni->ni_maxtxcredits = *kranal_tunables.kra_credits;
1593         ni->ni_peertxcredits = *kranal_tunables.kra_peercredits;
1594
1595         ni->ni_data = &kranal_data;
1596         kranal_data.kra_ni = ni;
1597
1598         /* CAVEAT EMPTOR: Every 'Fma' message includes the sender's NID and
1599          * a unique (for all time) connstamp so we can uniquely identify
1600          * the sender.  The connstamp is an incrementing counter
1601          * initialised with seconds + microseconds at startup time.  So we
1602          * rely on NOT creating connections more frequently on average than
1603          * 1MHz to ensure we don't use old connstamps when we reboot. */
1604         cfs_gettimeofday(&tv);
1605         kranal_data.kra_connstamp =
1606         kranal_data.kra_peerstamp = (((__u64)tv.tv_sec) * 1000000) + tv.tv_usec;
1607
1608         rwlock_init(&kranal_data.kra_global_lock);
1609
1610         for (i = 0; i < RANAL_MAXDEVS; i++ ) {
1611                 kra_device_t  *dev = &kranal_data.kra_devices[i];
1612
1613                 dev->rad_idx = i;
1614                 CFS_INIT_LIST_HEAD(&dev->rad_ready_conns);
1615                 CFS_INIT_LIST_HEAD(&dev->rad_new_conns);
1616                 cfs_waitq_init(&dev->rad_waitq);
1617                 spin_lock_init(&dev->rad_lock);
1618         }
1619
1620         kranal_data.kra_new_min_timeout = CFS_MAX_SCHEDULE_TIMEOUT;
1621         cfs_waitq_init(&kranal_data.kra_reaper_waitq);
1622         spin_lock_init(&kranal_data.kra_reaper_lock);
1623
1624         CFS_INIT_LIST_HEAD(&kranal_data.kra_connd_acceptq);
1625         CFS_INIT_LIST_HEAD(&kranal_data.kra_connd_peers);
1626         cfs_waitq_init(&kranal_data.kra_connd_waitq);
1627         spin_lock_init(&kranal_data.kra_connd_lock);
1628
1629         CFS_INIT_LIST_HEAD(&kranal_data.kra_idle_txs);
1630         spin_lock_init(&kranal_data.kra_tx_lock);
1631
1632         /* OK to call kranal_api_shutdown() to cleanup now */
1633         kranal_data.kra_init = RANAL_INIT_DATA;
1634         PORTAL_MODULE_USE;
1635
1636         kranal_data.kra_peer_hash_size = RANAL_PEER_HASH_SIZE;
1637         LIBCFS_ALLOC(kranal_data.kra_peers,
1638                      sizeof(cfs_list_t) *
1639                             kranal_data.kra_peer_hash_size);
1640         if (kranal_data.kra_peers == NULL)
1641                 goto failed;
1642
1643         for (i = 0; i < kranal_data.kra_peer_hash_size; i++)
1644                 CFS_INIT_LIST_HEAD(&kranal_data.kra_peers[i]);
1645
1646         kranal_data.kra_conn_hash_size = RANAL_PEER_HASH_SIZE;
1647         LIBCFS_ALLOC(kranal_data.kra_conns,
1648                      sizeof(cfs_list_t) *
1649                             kranal_data.kra_conn_hash_size);
1650         if (kranal_data.kra_conns == NULL)
1651                 goto failed;
1652
1653         for (i = 0; i < kranal_data.kra_conn_hash_size; i++)
1654                 CFS_INIT_LIST_HEAD(&kranal_data.kra_conns[i]);
1655
1656         rc = kranal_alloc_txdescs(&kranal_data.kra_idle_txs,
1657                                   *kranal_tunables.kra_ntx);
1658         if (rc != 0)
1659                 goto failed;
1660
1661         rc = kranal_thread_start(kranal_reaper, NULL);
1662         if (rc != 0) {
1663                 CERROR("Can't spawn ranal reaper: %d\n", rc);
1664                 goto failed;
1665         }
1666
1667         for (i = 0; i < *kranal_tunables.kra_n_connd; i++) {
1668                 rc = kranal_thread_start(kranal_connd, (void *)(unsigned long)i);
1669                 if (rc != 0) {
1670                         CERROR("Can't spawn ranal connd[%d]: %d\n",
1671                                i, rc);
1672                         goto failed;
1673                 }
1674         }
1675
1676         LASSERT (kranal_data.kra_ndevs == 0);
1677
1678         /* Use all available RapidArray devices */
1679         for (i = 0; i < RANAL_MAXDEVS; i++) {
1680                 dev = &kranal_data.kra_devices[kranal_data.kra_ndevs];
1681
1682                 rc = kranal_device_init(kranal_devids[i], dev);
1683                 if (rc == 0)
1684                         kranal_data.kra_ndevs++;
1685         }
1686
1687         if (kranal_data.kra_ndevs == 0) {
1688                 CERROR("Can't initialise any RapidArray devices\n");
1689                 goto failed;
1690         }
1691
1692         for (i = 0; i < kranal_data.kra_ndevs; i++) {
1693                 dev = &kranal_data.kra_devices[i];
1694                 rc = kranal_thread_start(kranal_scheduler, dev);
1695                 if (rc != 0) {
1696                         CERROR("Can't spawn ranal scheduler[%d]: %d\n",
1697                                i, rc);
1698                         goto failed;
1699                 }
1700         }
1701
1702         /* flag everything initialised */
1703         kranal_data.kra_init = RANAL_INIT_ALL;
1704         /*****************************************************/
1705
1706         CDEBUG(D_MALLOC, "initial kmem %d\n", pkmem);
1707         return 0;
1708
1709  failed:
1710         kranal_shutdown(ni);
1711         return -ENETDOWN;
1712 }
1713
1714 void __exit
1715 kranal_module_fini (void)
1716 {
1717         lnet_unregister_lnd(&the_kralnd);
1718         kranal_tunables_fini();
1719 }
1720
1721 int __init
1722 kranal_module_init (void)
1723 {
1724         int    rc;
1725
1726         rc = kranal_tunables_init();
1727         if (rc != 0)
1728                 return rc;
1729
1730         lnet_register_lnd(&the_kralnd);
1731
1732         return 0;
1733 }
1734
1735 MODULE_AUTHOR("Sun Microsystems, Inc. <http://www.lustre.org/>");
1736 MODULE_DESCRIPTION("Kernel RapidArray LND v0.01");
1737 MODULE_LICENSE("GPL");
1738
1739 module_init(kranal_module_init);
1740 module_exit(kranal_module_fini);