lnet/klnds/viblnd/viblnd.c

   1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
   2  * vim:expandtab:shiftwidth=8:tabstop=8:
   3  *
   4  * Copyright (C) 2004 Cluster File Systems, Inc.
   5  *   Author: Eric Barton <eric@bartonsoftware.com>
   6  *   Author: Frank Zago <fzago@systemfabricworks.com>
   7  *
   8  *   This file is part of Lustre, http://www.lustre.org.
   9  *
  10  *   Lustre is free software; you can redistribute it and/or
  11  *   modify it under the terms of version 2 of the GNU General Public
  12  *   License as published by the Free Software Foundation.
  13  *
  14  *   Lustre is distributed in the hope that it will be useful,
  15  *   but WITHOUT ANY WARRANTY; without even the implied warranty of
  16  *   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  17  *   GNU General Public License for more details.
  18  *
  19  *   You should have received a copy of the GNU General Public License
  20  *   along with Lustre; if not, write to the Free Software
  21  *   Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
  22  *
  23  */
  24
  25 #include "viblnd.h"
  26
  27 lnd_t the_kiblnd = {
  28         .lnd_type       = VIBLND,
  29         .lnd_startup    = kibnal_startup,
  30         .lnd_shutdown   = kibnal_shutdown,
  31         .lnd_ctl        = kibnal_ctl,
  32         .lnd_send       = kibnal_send,
  33         .lnd_recv       = kibnal_recv,
  34         .lnd_eager_recv = kibnal_eager_recv,
  35 };
  36
  37 kib_data_t              kibnal_data;
  38
  39 void vibnal_assert_wire_constants (void)
  40 {
  41         /* Wire protocol assertions generated by 'wirecheck'
  42          * running on Linux robert 2.6.11-1.27_FC3 #1 Tue May 17 20:27:37 EDT 2005 i686 athlon i386 G
  43          * with gcc version 3.4.3 20050227 (Red Hat 3.4.3-22.fc3) */
  44
  45
  46         /* Constants... */
  47         CLASSERT (IBNAL_MSG_MAGIC == 0x0be91b91);
  48         CLASSERT (IBNAL_MSG_VERSION == 0x11);
  49         CLASSERT (IBNAL_MSG_CONNREQ == 0xc0);
  50         CLASSERT (IBNAL_MSG_CONNACK == 0xc1);
  51         CLASSERT (IBNAL_MSG_NOOP == 0xd0);
  52         CLASSERT (IBNAL_MSG_IMMEDIATE == 0xd1);
  53         CLASSERT (IBNAL_MSG_PUT_REQ == 0xd2);
  54         CLASSERT (IBNAL_MSG_PUT_NAK == 0xd3);
  55         CLASSERT (IBNAL_MSG_PUT_ACK == 0xd4);
  56         CLASSERT (IBNAL_MSG_PUT_DONE == 0xd5);
  57         CLASSERT (IBNAL_MSG_GET_REQ == 0xd6);
  58         CLASSERT (IBNAL_MSG_GET_DONE == 0xd7);
  59
  60         /* Checks for struct kib_connparams_t */
  61         CLASSERT ((int)sizeof(kib_connparams_t) == 12);
  62         CLASSERT ((int)offsetof(kib_connparams_t, ibcp_queue_depth) == 0);
  63         CLASSERT ((int)sizeof(((kib_connparams_t *)0)->ibcp_queue_depth) == 4);
  64         CLASSERT ((int)offsetof(kib_connparams_t, ibcp_max_msg_size) == 4);
  65         CLASSERT ((int)sizeof(((kib_connparams_t *)0)->ibcp_max_msg_size) == 4);
  66         CLASSERT ((int)offsetof(kib_connparams_t, ibcp_max_frags) == 8);
  67         CLASSERT ((int)sizeof(((kib_connparams_t *)0)->ibcp_max_frags) == 4);
  68
  69         /* Checks for struct kib_immediate_msg_t */
  70         CLASSERT ((int)sizeof(kib_immediate_msg_t) == 72);
  71         CLASSERT ((int)offsetof(kib_immediate_msg_t, ibim_hdr) == 0);
  72         CLASSERT ((int)sizeof(((kib_immediate_msg_t *)0)->ibim_hdr) == 72);
  73         CLASSERT ((int)offsetof(kib_immediate_msg_t, ibim_payload[13]) == 85);
  74         CLASSERT ((int)sizeof(((kib_immediate_msg_t *)0)->ibim_payload[13]) == 1);
  75         CLASSERT (IBNAL_USE_FMR == 1);
  76
  77         /* Checks for struct kib_rdma_desc_t */
  78         CLASSERT ((int)sizeof(kib_rdma_desc_t) == 16);
  79         CLASSERT ((int)offsetof(kib_rdma_desc_t, rd_addr) == 0);
  80         CLASSERT ((int)sizeof(((kib_rdma_desc_t *)0)->rd_addr) == 8);
  81         CLASSERT ((int)offsetof(kib_rdma_desc_t, rd_nob) == 8);
  82         CLASSERT ((int)sizeof(((kib_rdma_desc_t *)0)->rd_nob) == 4);
  83         CLASSERT ((int)offsetof(kib_rdma_desc_t, rd_key) == 12);
  84         CLASSERT ((int)sizeof(((kib_rdma_desc_t *)0)->rd_key) == 4);
  85
  86         /* Checks for struct kib_putreq_msg_t */
  87         CLASSERT ((int)sizeof(kib_putreq_msg_t) == 80);
  88         CLASSERT ((int)offsetof(kib_putreq_msg_t, ibprm_hdr) == 0);
  89         CLASSERT ((int)sizeof(((kib_putreq_msg_t *)0)->ibprm_hdr) == 72);
  90         CLASSERT ((int)offsetof(kib_putreq_msg_t, ibprm_cookie) == 72);
  91         CLASSERT ((int)sizeof(((kib_putreq_msg_t *)0)->ibprm_cookie) == 8);
  92
  93         /* Checks for struct kib_putack_msg_t */
  94         CLASSERT ((int)sizeof(kib_putack_msg_t) == 32);
  95         CLASSERT ((int)offsetof(kib_putack_msg_t, ibpam_src_cookie) == 0);
  96         CLASSERT ((int)sizeof(((kib_putack_msg_t *)0)->ibpam_src_cookie) == 8);
  97         CLASSERT ((int)offsetof(kib_putack_msg_t, ibpam_dst_cookie) == 8);
  98         CLASSERT ((int)sizeof(((kib_putack_msg_t *)0)->ibpam_dst_cookie) == 8);
  99         CLASSERT ((int)offsetof(kib_putack_msg_t, ibpam_rd) == 16);
 100         CLASSERT ((int)sizeof(((kib_putack_msg_t *)0)->ibpam_rd) == 16);
 101
 102         /* Checks for struct kib_get_msg_t */
 103         CLASSERT ((int)sizeof(kib_get_msg_t) == 96);
 104         CLASSERT ((int)offsetof(kib_get_msg_t, ibgm_hdr) == 0);
 105         CLASSERT ((int)sizeof(((kib_get_msg_t *)0)->ibgm_hdr) == 72);
 106         CLASSERT ((int)offsetof(kib_get_msg_t, ibgm_cookie) == 72);
 107         CLASSERT ((int)sizeof(((kib_get_msg_t *)0)->ibgm_cookie) == 8);
 108         CLASSERT ((int)offsetof(kib_get_msg_t, ibgm_rd) == 80);
 109         CLASSERT ((int)sizeof(((kib_get_msg_t *)0)->ibgm_rd) == 16);
 110
 111         /* Checks for struct kib_completion_msg_t */
 112         CLASSERT ((int)sizeof(kib_completion_msg_t) == 12);
 113         CLASSERT ((int)offsetof(kib_completion_msg_t, ibcm_cookie) == 0);
 114         CLASSERT ((int)sizeof(((kib_completion_msg_t *)0)->ibcm_cookie) == 8);
 115         CLASSERT ((int)offsetof(kib_completion_msg_t, ibcm_status) == 8);
 116         CLASSERT ((int)sizeof(((kib_completion_msg_t *)0)->ibcm_status) == 4);
 117
 118         /* Checks for struct kib_msg_t */
 119         CLASSERT ((int)sizeof(kib_msg_t) == 152);
 120         CLASSERT ((int)offsetof(kib_msg_t, ibm_magic) == 0);
 121         CLASSERT ((int)sizeof(((kib_msg_t *)0)->ibm_magic) == 4);
 122         CLASSERT ((int)offsetof(kib_msg_t, ibm_version) == 4);
 123         CLASSERT ((int)sizeof(((kib_msg_t *)0)->ibm_version) == 2);
 124         CLASSERT ((int)offsetof(kib_msg_t, ibm_type) == 6);
 125         CLASSERT ((int)sizeof(((kib_msg_t *)0)->ibm_type) == 1);
 126         CLASSERT ((int)offsetof(kib_msg_t, ibm_credits) == 7);
 127         CLASSERT ((int)sizeof(((kib_msg_t *)0)->ibm_credits) == 1);
 128         CLASSERT ((int)offsetof(kib_msg_t, ibm_nob) == 8);
 129         CLASSERT ((int)sizeof(((kib_msg_t *)0)->ibm_nob) == 4);
 130         CLASSERT ((int)offsetof(kib_msg_t, ibm_cksum) == 12);
 131         CLASSERT ((int)sizeof(((kib_msg_t *)0)->ibm_cksum) == 4);
 132         CLASSERT ((int)offsetof(kib_msg_t, ibm_srcnid) == 16);
 133         CLASSERT ((int)sizeof(((kib_msg_t *)0)->ibm_srcnid) == 8);
 134         CLASSERT ((int)offsetof(kib_msg_t, ibm_srcstamp) == 24);
 135         CLASSERT ((int)sizeof(((kib_msg_t *)0)->ibm_srcstamp) == 8);
 136         CLASSERT ((int)offsetof(kib_msg_t, ibm_dstnid) == 32);
 137         CLASSERT ((int)sizeof(((kib_msg_t *)0)->ibm_dstnid) == 8);
 138         CLASSERT ((int)offsetof(kib_msg_t, ibm_dststamp) == 40);
 139         CLASSERT ((int)sizeof(((kib_msg_t *)0)->ibm_dststamp) == 8);
 140         CLASSERT ((int)offsetof(kib_msg_t, ibm_seq) == 48);
 141         CLASSERT ((int)sizeof(((kib_msg_t *)0)->ibm_seq) == 8);
 142         CLASSERT ((int)offsetof(kib_msg_t, ibm_u.connparams) == 56);
 143         CLASSERT ((int)sizeof(((kib_msg_t *)0)->ibm_u.connparams) == 12);
 144         CLASSERT ((int)offsetof(kib_msg_t, ibm_u.immediate) == 56);
 145         CLASSERT ((int)sizeof(((kib_msg_t *)0)->ibm_u.immediate) == 72);
 146         CLASSERT ((int)offsetof(kib_msg_t, ibm_u.putreq) == 56);
 147         CLASSERT ((int)sizeof(((kib_msg_t *)0)->ibm_u.putreq) == 80);
 148         CLASSERT ((int)offsetof(kib_msg_t, ibm_u.putack) == 56);
 149         CLASSERT ((int)sizeof(((kib_msg_t *)0)->ibm_u.putack) == 32);
 150         CLASSERT ((int)offsetof(kib_msg_t, ibm_u.get) == 56);
 151         CLASSERT ((int)sizeof(((kib_msg_t *)0)->ibm_u.get) == 96);
 152         CLASSERT ((int)offsetof(kib_msg_t, ibm_u.completion) == 56);
 153         CLASSERT ((int)sizeof(((kib_msg_t *)0)->ibm_u.completion) == 12);
 154 }
 155
 156 __u32
 157 kibnal_cksum (void *ptr, int nob)
 158 {
 159         char  *c  = ptr;
 160         __u32  sum = 0;
 161
 162         while (nob-- > 0)
 163                 sum = ((sum << 1) | (sum >> 31)) + *c++;
 164
 165         /* ensure I don't return 0 (== no checksum) */
 166         return (sum == 0) ? 1 : sum;
 167 }
 168
 169 void
 170 kibnal_init_msg(kib_msg_t *msg, int type, int body_nob)
 171 {
 172         msg->ibm_type = type;
 173         msg->ibm_nob  = offsetof(kib_msg_t, ibm_u) + body_nob;
 174 }
 175
 176 void
 177 kibnal_pack_msg(kib_msg_t *msg, __u32 version, int credits,
 178                 lnet_nid_t dstnid, __u64 dststamp, __u64 seq)
 179 {
 180         /* CAVEAT EMPTOR! all message fields not set here should have been
 181          * initialised previously. */
 182         msg->ibm_magic    = IBNAL_MSG_MAGIC;
 183         msg->ibm_version  = version;
 184         /*   ibm_type */
 185         msg->ibm_credits  = credits;
 186         /*   ibm_nob */
 187         msg->ibm_cksum    = 0;
 188         msg->ibm_srcnid   = lnet_ptlcompat_srcnid(kibnal_data.kib_ni->ni_nid,
 189                                                   dstnid);
 190         msg->ibm_srcstamp = kibnal_data.kib_incarnation;
 191         msg->ibm_dstnid   = dstnid;
 192         msg->ibm_dststamp = dststamp;
 193         msg->ibm_seq      = seq;
 194
 195         if (*kibnal_tunables.kib_cksum) {
 196                 /* NB ibm_cksum zero while computing cksum */
 197                 msg->ibm_cksum = kibnal_cksum(msg, msg->ibm_nob);
 198         }
 199 }
 200
 201 int
 202 kibnal_unpack_msg(kib_msg_t *msg, __u32 expected_version, int nob)
 203 {
 204         const int hdr_size = offsetof(kib_msg_t, ibm_u);
 205         __u32     msg_cksum;
 206         __u32     msg_version;
 207         int       flip;
 208         int       msg_nob;
 209 #if !IBNAL_USE_FMR
 210         int       i;
 211         int       n;
 212 #endif
 213         /* 6 bytes are enough to have received magic + version */
 214         if (nob < 6) {
 215                 CERROR("Short message: %d\n", nob);
 216                 return -EPROTO;
 217         }
 218
 219         /* Future protocol version compatibility support!
 220          * If the viblnd-specific protocol changes, or when LNET unifies
 221          * protocols over all LNDs, the initial connection will negotiate a
 222          * protocol version.  If I find this, I avoid any console errors.  If
 223          * my is doing connection establishment, the reject will tell the peer
 224          * which version I'm running. */
 225
 226         if (msg->ibm_magic == IBNAL_MSG_MAGIC) {
 227                 flip = 0;
 228         } else if (msg->ibm_magic == __swab32(IBNAL_MSG_MAGIC)) {
 229                 flip = 1;
 230         } else {
 231                 if (msg->ibm_magic == LNET_PROTO_MAGIC ||
 232                     msg->ibm_magic == __swab32(LNET_PROTO_MAGIC))
 233                         return -EPROTO;
 234
 235                 /* Completely out to lunch */
 236                 CERROR("Bad magic: %08x\n", msg->ibm_magic);
 237                 return -EPROTO;
 238         }
 239
 240         msg_version = flip ? __swab16(msg->ibm_version) : msg->ibm_version;
 241         if (expected_version == 0) {
 242                 if (msg_version != IBNAL_MSG_VERSION_RDMAREPLYNOTRSRVD &&
 243                     msg_version != IBNAL_MSG_VERSION)
 244                         return -EPROTO;
 245         } else if (msg_version != expected_version) {
 246                 CERROR("Bad version: %x(%x expected)\n",
 247                        msg_version, expected_version);
 248                 return -EPROTO;
 249         }
 250
 251         if (nob < hdr_size) {
 252                 CERROR("Short message: %d\n", nob);
 253                 return -EPROTO;
 254         }
 255
 256         msg_nob = flip ? __swab32(msg->ibm_nob) : msg->ibm_nob;
 257         if (msg_nob > nob) {
 258                 CERROR("Short message: got %d, wanted %d\n", nob, msg_nob);
 259                 return -EPROTO;
 260         }
 261
 262         /* checksum must be computed with ibm_cksum zero and BEFORE anything
 263          * gets flipped */
 264         msg_cksum = flip ? __swab32(msg->ibm_cksum) : msg->ibm_cksum;
 265         msg->ibm_cksum = 0;
 266         if (msg_cksum != 0 &&
 267             msg_cksum != kibnal_cksum(msg, msg_nob)) {
 268                 CERROR("Bad checksum\n");
 269                 return -EPROTO;
 270         }
 271         msg->ibm_cksum = msg_cksum;
 272
 273         if (flip) {
 274                 /* leave magic unflipped as a clue to peer endianness */
 275                 msg->ibm_version = msg_version;
 276                 CLASSERT (sizeof(msg->ibm_type) == 1);
 277                 CLASSERT (sizeof(msg->ibm_credits) == 1);
 278                 msg->ibm_nob = msg_nob;
 279                 __swab64s(&msg->ibm_srcnid);
 280                 __swab64s(&msg->ibm_srcstamp);
 281                 __swab64s(&msg->ibm_dstnid);
 282                 __swab64s(&msg->ibm_dststamp);
 283                 __swab64s(&msg->ibm_seq);
 284         }
 285
 286         if (msg->ibm_srcnid == LNET_NID_ANY) {
 287                 CERROR("Bad src nid: %s\n", libcfs_nid2str(msg->ibm_srcnid));
 288                 return -EPROTO;
 289         }
 290
 291         switch (msg->ibm_type) {
 292         default:
 293                 CERROR("Unknown message type %x\n", msg->ibm_type);
 294                 return -EPROTO;
 295
 296         case IBNAL_MSG_NOOP:
 297                 break;
 298
 299         case IBNAL_MSG_IMMEDIATE:
 300                 if (msg_nob < offsetof(kib_msg_t, ibm_u.immediate.ibim_payload[0])) {
 301                         CERROR("Short IMMEDIATE: %d(%d)\n", msg_nob,
 302                                (int)offsetof(kib_msg_t, ibm_u.immediate.ibim_payload[0]));
 303                         return -EPROTO;
 304                 }
 305                 break;
 306
 307         case IBNAL_MSG_PUT_REQ:
 308                 if (msg_nob < hdr_size + sizeof(msg->ibm_u.putreq)) {
 309                         CERROR("Short PUT_REQ: %d(%d)\n", msg_nob,
 310                                (int)(hdr_size + sizeof(msg->ibm_u.putreq)));
 311                         return -EPROTO;
 312                 }
 313                 break;
 314
 315         case IBNAL_MSG_PUT_ACK:
 316                 if (msg_nob < hdr_size + sizeof(msg->ibm_u.putack)) {
 317                         CERROR("Short PUT_ACK: %d(%d)\n", msg_nob,
 318                                (int)(hdr_size + sizeof(msg->ibm_u.putack)));
 319                         return -EPROTO;
 320                 }
 321 #if IBNAL_USE_FMR
 322                 if (flip) {
 323                         __swab64s(&msg->ibm_u.putack.ibpam_rd.rd_addr);
 324                         __swab32s(&msg->ibm_u.putack.ibpam_rd.rd_nob);
 325                         __swab32s(&msg->ibm_u.putack.ibpam_rd.rd_key);
 326                 }
 327 #else
 328                 if (flip) {
 329                         __swab32s(&msg->ibm_u.putack.ibpam_rd.rd_key);
 330                         __swab32s(&msg->ibm_u.putack.ibpam_rd.rd_nfrag);
 331                 }
 332
 333                 n = msg->ibm_u.putack.ibpam_rd.rd_nfrag;
 334                 if (n <= 0 || n > IBNAL_MAX_RDMA_FRAGS) {
 335                         CERROR("Bad PUT_ACK nfrags: %d, should be 0 < n <= %d\n",
 336                                n, IBNAL_MAX_RDMA_FRAGS);
 337                         return -EPROTO;
 338                 }
 339
 340                 if (msg_nob < offsetof(kib_msg_t, ibm_u.putack.ibpam_rd.rd_frags[n])) {
 341                         CERROR("Short PUT_ACK: %d(%d)\n", msg_nob,
 342                                (int)offsetof(kib_msg_t, ibm_u.putack.ibpam_rd.rd_frags[n]));
 343                         return -EPROTO;
 344                 }
 345
 346                 if (flip) {
 347                         for (i = 0; i < n; i++) {
 348                                 __swab32s(&msg->ibm_u.putack.ibpam_rd.rd_frags[i].rf_nob);
 349                                 __swab32s(&msg->ibm_u.putack.ibpam_rd.rd_frags[i].rf_addr_lo);
 350                                 __swab32s(&msg->ibm_u.putack.ibpam_rd.rd_frags[i].rf_addr_hi);
 351                         }
 352                 }
 353 #endif
 354                 break;
 355
 356         case IBNAL_MSG_GET_REQ:
 357                 if (msg_nob < hdr_size + sizeof(msg->ibm_u.get)) {
 358                         CERROR("Short GET_REQ: %d(%d)\n", msg_nob,
 359                                (int)(hdr_size + sizeof(msg->ibm_u.get)));
 360                         return -EPROTO;
 361                 }
 362 #if IBNAL_USE_FMR
 363                 if (flip) {
 364                         __swab64s(&msg->ibm_u.get.ibgm_rd.rd_addr);
 365                         __swab32s(&msg->ibm_u.get.ibgm_rd.rd_nob);
 366                         __swab32s(&msg->ibm_u.get.ibgm_rd.rd_key);
 367                 }
 368 #else
 369                 if (flip) {
 370                         __swab32s(&msg->ibm_u.get.ibgm_rd.rd_key);
 371                         __swab32s(&msg->ibm_u.get.ibgm_rd.rd_nfrag);
 372                 }
 373
 374                 n = msg->ibm_u.get.ibgm_rd.rd_nfrag;
 375                 if (n <= 0 || n > IBNAL_MAX_RDMA_FRAGS) {
 376                         CERROR("Bad GET_REQ nfrags: %d, should be 0 < n <= %d\n",
 377                                n, IBNAL_MAX_RDMA_FRAGS);
 378                         return -EPROTO;
 379                 }
 380
 381                 if (msg_nob < offsetof(kib_msg_t, ibm_u.get.ibgm_rd.rd_frags[n])) {
 382                         CERROR("Short GET_REQ: %d(%d)\n", msg_nob,
 383                                (int)offsetof(kib_msg_t, ibm_u.get.ibgm_rd.rd_frags[n]));
 384                         return -EPROTO;
 385                 }
 386
 387                 if (flip)
 388                         for (i = 0; i < msg->ibm_u.get.ibgm_rd.rd_nfrag; i++) {
 389                                 __swab32s(&msg->ibm_u.get.ibgm_rd.rd_frags[i].rf_nob);
 390                                 __swab32s(&msg->ibm_u.get.ibgm_rd.rd_frags[i].rf_addr_lo);
 391                                 __swab32s(&msg->ibm_u.get.ibgm_rd.rd_frags[i].rf_addr_hi);
 392                         }
 393 #endif
 394                 break;
 395
 396         case IBNAL_MSG_PUT_NAK:
 397         case IBNAL_MSG_PUT_DONE:
 398         case IBNAL_MSG_GET_DONE:
 399                 if (msg_nob < hdr_size + sizeof(msg->ibm_u.completion)) {
 400                         CERROR("Short RDMA completion: %d(%d)\n", msg_nob,
 401                                (int)(hdr_size + sizeof(msg->ibm_u.completion)));
 402                         return -EPROTO;
 403                 }
 404                 if (flip)
 405                         __swab32s(&msg->ibm_u.completion.ibcm_status);
 406                 break;
 407
 408         case IBNAL_MSG_CONNREQ:
 409         case IBNAL_MSG_CONNACK:
 410                 if (msg_nob < hdr_size + sizeof(msg->ibm_u.connparams)) {
 411                         CERROR("Short connreq/ack: %d(%d)\n", msg_nob,
 412                                (int)(hdr_size + sizeof(msg->ibm_u.connparams)));
 413                         return -EPROTO;
 414                 }
 415                 if (flip) {
 416                         __swab32s(&msg->ibm_u.connparams.ibcp_queue_depth);
 417                         __swab32s(&msg->ibm_u.connparams.ibcp_max_msg_size);
 418                         __swab32s(&msg->ibm_u.connparams.ibcp_max_frags);
 419                 }
 420                 break;
 421         }
 422         return 0;
 423 }
 424
 425 int
 426 kibnal_start_listener (lnet_ni_t *ni)
 427 {
 428         static cm_listen_data_t info;
 429
 430         cm_return_t      cmrc;
 431
 432         LASSERT (kibnal_data.kib_listen_handle == NULL);
 433
 434         kibnal_data.kib_listen_handle =
 435                 cm_create_cep(cm_cep_transp_rc);
 436         if (kibnal_data.kib_listen_handle == NULL) {
 437                 CERROR ("Can't create listen CEP\n");
 438                 return -ENOMEM;
 439         }
 440
 441         CDEBUG(D_NET, "Created CEP %p for listening\n",
 442                kibnal_data.kib_listen_handle);
 443
 444         memset(&info, 0, sizeof(info));
 445         info.listen_addr.end_pt.sid =
 446                 (__u64)(*kibnal_tunables.kib_service_number);
 447
 448         cmrc = cm_listen(kibnal_data.kib_listen_handle, &info,
 449                          kibnal_listen_callback, NULL);
 450         if (cmrc == cm_stat_success)
 451                 return 0;
 452
 453         CERROR ("cm_listen error: %d\n", cmrc);
 454
 455         cmrc = cm_destroy_cep(kibnal_data.kib_listen_handle);
 456         LASSERT (cmrc == cm_stat_success);
 457
 458         kibnal_data.kib_listen_handle = NULL;
 459         return -EINVAL;
 460 }
 461
 462 void
 463 kibnal_stop_listener(lnet_ni_t *ni)
 464 {
 465         cm_return_t      cmrc;
 466
 467         LASSERT (kibnal_data.kib_listen_handle != NULL);
 468
 469         cmrc = cm_cancel(kibnal_data.kib_listen_handle);
 470         if (cmrc != cm_stat_success)
 471                 CERROR ("Error %d stopping listener\n", cmrc);
 472
 473         cfs_pause(cfs_time_seconds(1)/10);   /* ensure no more callbacks */
 474
 475         cmrc = cm_destroy_cep(kibnal_data.kib_listen_handle);
 476         if (cmrc != vv_return_ok)
 477                 CERROR ("Error %d destroying CEP\n", cmrc);
 478
 479         kibnal_data.kib_listen_handle = NULL;
 480 }
 481
 482 int
 483 kibnal_create_peer (kib_peer_t **peerp, lnet_nid_t nid)
 484 {
 485         kib_peer_t     *peer;
 486         unsigned long   flags;
 487         int             rc;
 488
 489         LASSERT (nid != LNET_NID_ANY);
 490
 491         LIBCFS_ALLOC(peer, sizeof (*peer));
 492         if (peer == NULL) {
 493                 CERROR("Cannot allocate peer\n");
 494                 return -ENOMEM;
 495         }
 496
 497         memset(peer, 0, sizeof(*peer));         /* zero flags etc */
 498
 499         peer->ibp_nid = nid;
 500         atomic_set (&peer->ibp_refcount, 1);    /* 1 ref for caller */
 501
 502         INIT_LIST_HEAD (&peer->ibp_list);       /* not in the peer table yet */
 503         INIT_LIST_HEAD (&peer->ibp_conns);
 504         INIT_LIST_HEAD (&peer->ibp_tx_queue);
 505
 506         peer->ibp_error = 0;
 507         peer->ibp_last_alive = cfs_time_current();
 508         peer->ibp_reconnect_interval = 0;       /* OK to connect at any time */
 509
 510         write_lock_irqsave(&kibnal_data.kib_global_lock, flags);
 511
 512         if (atomic_read(&kibnal_data.kib_npeers) >=
 513             *kibnal_tunables.kib_concurrent_peers) {
 514                 rc = -EOVERFLOW;        /* !! but at least it distinguishes */
 515         } else if (kibnal_data.kib_listen_handle == NULL) {
 516                 rc = -ESHUTDOWN;        /* shutdown has started */
 517         } else {
 518                 rc = 0;
 519                 /* npeers only grows with the global lock held */
 520                 atomic_inc(&kibnal_data.kib_npeers);
 521         }
 522
 523         write_unlock_irqrestore(&kibnal_data.kib_global_lock, flags);
 524
 525         if (rc != 0) {
 526                 CERROR("Can't create peer: %s\n",
 527                        (rc == -ESHUTDOWN) ? "shutting down" :
 528                        "too many peers");
 529                 LIBCFS_FREE(peer, sizeof(*peer));
 530         } else {
 531                 *peerp = peer;
 532         }
 533
 534         return rc;
 535 }
 536
 537 void
 538 kibnal_destroy_peer (kib_peer_t *peer)
 539 {
 540         LASSERT (atomic_read (&peer->ibp_refcount) == 0);
 541         LASSERT (peer->ibp_persistence == 0);
 542         LASSERT (!kibnal_peer_active(peer));
 543         LASSERT (peer->ibp_connecting == 0);
 544         LASSERT (peer->ibp_accepting == 0);
 545         LASSERT (list_empty (&peer->ibp_conns));
 546         LASSERT (list_empty (&peer->ibp_tx_queue));
 547
 548         LIBCFS_FREE (peer, sizeof (*peer));
 549
 550         /* NB a peer's connections keep a reference on their peer until
 551          * they are destroyed, so we can be assured that _all_ state to do
 552          * with this peer has been cleaned up when its refcount drops to
 553          * zero. */
 554         atomic_dec(&kibnal_data.kib_npeers);
 555 }
 556
 557 kib_peer_t *
 558 kibnal_find_peer_locked (lnet_nid_t nid)
 559 {
 560         /* the caller is responsible for accounting the additional reference
 561          * that this creates */
 562         struct list_head *peer_list = kibnal_nid2peerlist (nid);
 563         struct list_head *tmp;
 564         kib_peer_t       *peer;
 565
 566         list_for_each (tmp, peer_list) {
 567
 568                 peer = list_entry (tmp, kib_peer_t, ibp_list);
 569
 570                 LASSERT (peer->ibp_persistence != 0 || /* persistent peer */
 571                          peer->ibp_connecting != 0 || /* creating conns */
 572                          peer->ibp_accepting != 0 ||
 573                          !list_empty (&peer->ibp_conns));  /* active conn */
 574
 575                 if (peer->ibp_nid != nid)
 576                         continue;
 577
 578                 CDEBUG(D_NET, "got peer [%p] -> %s (%d)\n",
 579                        peer, libcfs_nid2str(nid),
 580                        atomic_read (&peer->ibp_refcount));
 581                 return (peer);
 582         }
 583         return (NULL);
 584 }
 585
 586 void
 587 kibnal_unlink_peer_locked (kib_peer_t *peer)
 588 {
 589         LASSERT (peer->ibp_persistence == 0);
 590         LASSERT (list_empty(&peer->ibp_conns));
 591
 592         LASSERT (kibnal_peer_active(peer));
 593         list_del_init (&peer->ibp_list);
 594         /* lose peerlist's ref */
 595         kibnal_peer_decref(peer);
 596 }
 597
 598 int
 599 kibnal_get_peer_info (int index, lnet_nid_t *nidp, __u32 *ipp,
 600                       int *persistencep)
 601 {
 602         kib_peer_t        *peer;
 603         struct list_head  *ptmp;
 604         int                i;
 605         unsigned long      flags;
 606
 607         read_lock_irqsave(&kibnal_data.kib_global_lock, flags);
 608
 609         for (i = 0; i < kibnal_data.kib_peer_hash_size; i++) {
 610
 611                 list_for_each (ptmp, &kibnal_data.kib_peers[i]) {
 612
 613                         peer = list_entry (ptmp, kib_peer_t, ibp_list);
 614                         LASSERT (peer->ibp_persistence != 0 ||
 615                                  peer->ibp_connecting != 0 ||
 616                                  peer->ibp_accepting != 0 ||
 617                                  !list_empty (&peer->ibp_conns));
 618
 619                         if (index-- > 0)
 620                                 continue;
 621
 622                         *nidp = peer->ibp_nid;
 623                         *ipp = peer->ibp_ip;
 624                         *persistencep = peer->ibp_persistence;
 625
 626                         read_unlock_irqrestore(&kibnal_data.kib_global_lock,
 627                                                flags);
 628                         return (0);
 629                 }
 630         }
 631
 632         read_unlock_irqrestore(&kibnal_data.kib_global_lock, flags);
 633         return (-ENOENT);
 634 }
 635
 636 int
 637 kibnal_add_persistent_peer (lnet_nid_t nid, __u32 ip)
 638 {
 639         kib_peer_t        *peer;
 640         kib_peer_t        *peer2;
 641         unsigned long      flags;
 642         int                rc;
 643
 644         CDEBUG(D_NET, "%s at %u.%u.%u.%u\n",
 645                libcfs_nid2str(nid), HIPQUAD(ip));
 646
 647         if (nid == LNET_NID_ANY)
 648                 return (-EINVAL);
 649
 650         rc = kibnal_create_peer(&peer, nid);
 651         if (rc != 0)
 652                 return rc;
 653
 654         write_lock_irqsave(&kibnal_data.kib_global_lock, flags);
 655
 656         peer2 = kibnal_find_peer_locked (nid);
 657         if (peer2 != NULL) {
 658                 kibnal_peer_decref (peer);
 659                 peer = peer2;
 660         } else {
 661                 /* peer table takes existing ref on peer */
 662                 list_add_tail (&peer->ibp_list,
 663                                kibnal_nid2peerlist (nid));
 664         }
 665
 666         peer->ibp_ip = ip;
 667         peer->ibp_persistence++;
 668
 669         write_unlock_irqrestore(&kibnal_data.kib_global_lock, flags);
 670         return (0);
 671 }
 672
 673 void
 674 kibnal_del_peer_locked (kib_peer_t *peer)
 675 {
 676         struct list_head *ctmp;
 677         struct list_head *cnxt;
 678         kib_conn_t       *conn;
 679
 680         peer->ibp_persistence = 0;
 681
 682         if (list_empty(&peer->ibp_conns)) {
 683                 kibnal_unlink_peer_locked(peer);
 684         } else {
 685                 list_for_each_safe (ctmp, cnxt, &peer->ibp_conns) {
 686                         conn = list_entry(ctmp, kib_conn_t, ibc_list);
 687
 688                         kibnal_close_conn_locked (conn, 0);
 689                 }
 690                 /* NB peer is no longer persistent; closing its last conn
 691                  * unlinked it. */
 692         }
 693         /* NB peer now unlinked; might even be freed if the peer table had the
 694          * last ref on it. */
 695 }
 696
 697 int
 698 kibnal_del_peer (lnet_nid_t nid)
 699 {
 700         CFS_LIST_HEAD     (zombies);
 701         struct list_head  *ptmp;
 702         struct list_head  *pnxt;
 703         kib_peer_t        *peer;
 704         int                lo;
 705         int                hi;
 706         int                i;
 707         unsigned long      flags;
 708         int                rc = -ENOENT;
 709
 710         write_lock_irqsave(&kibnal_data.kib_global_lock, flags);
 711
 712         if (nid != LNET_NID_ANY)
 713                 lo = hi = kibnal_nid2peerlist(nid) - kibnal_data.kib_peers;
 714         else {
 715                 lo = 0;
 716                 hi = kibnal_data.kib_peer_hash_size - 1;
 717         }
 718
 719         for (i = lo; i <= hi; i++) {
 720                 list_for_each_safe (ptmp, pnxt, &kibnal_data.kib_peers[i]) {
 721                         peer = list_entry (ptmp, kib_peer_t, ibp_list);
 722                         LASSERT (peer->ibp_persistence != 0 ||
 723                                  peer->ibp_connecting != 0 ||
 724                                  peer->ibp_accepting != 0 ||
 725                                  !list_empty (&peer->ibp_conns));
 726
 727                         if (!(nid == LNET_NID_ANY || peer->ibp_nid == nid))
 728                                 continue;
 729
 730                         if (!list_empty(&peer->ibp_tx_queue)) {
 731                                 LASSERT (list_empty(&peer->ibp_conns));
 732
 733                                 list_splice_init(&peer->ibp_tx_queue, &zombies);
 734                         }
 735
 736                         kibnal_del_peer_locked (peer);
 737                         rc = 0;         /* matched something */
 738                 }
 739         }
 740
 741         write_unlock_irqrestore(&kibnal_data.kib_global_lock, flags);
 742
 743         kibnal_txlist_done(&zombies, -EIO);
 744
 745         return (rc);
 746 }
 747
 748 kib_conn_t *
 749 kibnal_get_conn_by_idx (int index)
 750 {
 751         kib_peer_t        *peer;
 752         struct list_head  *ptmp;
 753         kib_conn_t        *conn;
 754         struct list_head  *ctmp;
 755         int                i;
 756         unsigned long      flags;
 757
 758         read_lock_irqsave(&kibnal_data.kib_global_lock, flags);
 759
 760         for (i = 0; i < kibnal_data.kib_peer_hash_size; i++) {
 761                 list_for_each (ptmp, &kibnal_data.kib_peers[i]) {
 762
 763                         peer = list_entry (ptmp, kib_peer_t, ibp_list);
 764                         LASSERT (peer->ibp_persistence > 0 ||
 765                                  peer->ibp_connecting != 0 ||
 766                                  peer->ibp_accepting != 0 ||
 767                                  !list_empty (&peer->ibp_conns));
 768
 769                         list_for_each (ctmp, &peer->ibp_conns) {
 770                                 if (index-- > 0)
 771                                         continue;
 772
 773                                 conn = list_entry (ctmp, kib_conn_t, ibc_list);
 774                                 kibnal_conn_addref(conn);
 775                                 read_unlock_irqrestore(&kibnal_data.kib_global_lock,
 776                                                        flags);
 777                                 return (conn);
 778                         }
 779                 }
 780         }
 781
 782         read_unlock_irqrestore(&kibnal_data.kib_global_lock, flags);
 783         return (NULL);
 784 }
 785
 786 void
 787 kibnal_debug_rx (kib_rx_t *rx)
 788 {
 789         CDEBUG(D_CONSOLE, "      %p nob %d msg_type %x "
 790                "cred %d seq "LPD64"\n",
 791                rx, rx->rx_nob, rx->rx_msg->ibm_type,
 792                rx->rx_msg->ibm_credits, rx->rx_msg->ibm_seq);
 793 }
 794
 795 void
 796 kibnal_debug_tx (kib_tx_t *tx)
 797 {
 798         CDEBUG(D_CONSOLE, "      %p snd %d q %d w %d rc %d dl %lx "
 799                "cookie "LPX64" msg %s%s type %x cred %d seq "LPD64"\n",
 800                tx, tx->tx_sending, tx->tx_queued, tx->tx_waiting,
 801                tx->tx_status, tx->tx_deadline, tx->tx_cookie,
 802                tx->tx_lntmsg[0] == NULL ? "-" : "!",
 803                tx->tx_lntmsg[1] == NULL ? "-" : "!",
 804                tx->tx_msg->ibm_type, tx->tx_msg->ibm_credits,
 805                tx->tx_msg->ibm_seq);
 806 }
 807
 808 void
 809 kibnal_debug_conn (kib_conn_t *conn)
 810 {
 811         struct list_head *tmp;
 812         int               i;
 813
 814         spin_lock(&conn->ibc_lock);
 815
 816         CDEBUG(D_CONSOLE, "conn[%d] %p -> %s: \n",
 817                atomic_read(&conn->ibc_refcount), conn,
 818                libcfs_nid2str(conn->ibc_peer->ibp_nid));
 819         CDEBUG(D_CONSOLE, "   txseq "LPD64" rxseq "LPD64" state %d \n",
 820                conn->ibc_txseq, conn->ibc_rxseq, conn->ibc_state);
 821         CDEBUG(D_CONSOLE, "   nposted %d cred %d o_cred %d r_cred %d\n",
 822                conn->ibc_nsends_posted, conn->ibc_credits,
 823                conn->ibc_outstanding_credits, conn->ibc_reserved_credits);
 824         CDEBUG(D_CONSOLE, "   disc %d comms_err %d\n",
 825                conn->ibc_disconnect, conn->ibc_comms_error);
 826
 827         CDEBUG(D_CONSOLE, "   early_rxs:\n");
 828         list_for_each(tmp, &conn->ibc_early_rxs)
 829                 kibnal_debug_rx(list_entry(tmp, kib_rx_t, rx_list));
 830
 831         CDEBUG(D_CONSOLE, "   tx_queue_nocred:\n");
 832         list_for_each(tmp, &conn->ibc_tx_queue_nocred)
 833                 kibnal_debug_tx(list_entry(tmp, kib_tx_t, tx_list));
 834
 835         CDEBUG(D_CONSOLE, "   tx_queue_rsrvd:\n");
 836         list_for_each(tmp, &conn->ibc_tx_queue_rsrvd)
 837                 kibnal_debug_tx(list_entry(tmp, kib_tx_t, tx_list));
 838
 839         CDEBUG(D_CONSOLE, "   tx_queue:\n");
 840         list_for_each(tmp, &conn->ibc_tx_queue)
 841                 kibnal_debug_tx(list_entry(tmp, kib_tx_t, tx_list));
 842
 843         CDEBUG(D_CONSOLE, "   active_txs:\n");
 844         list_for_each(tmp, &conn->ibc_active_txs)
 845                 kibnal_debug_tx(list_entry(tmp, kib_tx_t, tx_list));
 846
 847         CDEBUG(D_CONSOLE, "   rxs:\n");
 848         for (i = 0; i < IBNAL_RX_MSGS; i++)
 849                 kibnal_debug_rx(&conn->ibc_rxs[i]);
 850
 851         spin_unlock(&conn->ibc_lock);
 852 }
 853
 854 int
 855 kibnal_set_qp_state (kib_conn_t *conn, vv_qp_state_t new_state)
 856 {
 857         static vv_qp_attr_t attr;
 858
 859         kib_connvars_t   *cv = conn->ibc_connvars;
 860         vv_return_t       vvrc;
 861
 862         /* Only called by connd => static OK */
 863         LASSERT (!in_interrupt());
 864         LASSERT (current == kibnal_data.kib_connd);
 865
 866         memset(&attr, 0, sizeof(attr));
 867
 868         switch (new_state) {
 869         default:
 870                 LBUG();
 871
 872         case vv_qp_state_init: {
 873                 struct vv_qp_modify_init_st *init = &attr.modify.params.init;
 874
 875                 init->p_key_indx     = cv->cv_pkey_index;
 876                 init->phy_port_num   = cv->cv_port;
 877                 init->q_key          = IBNAL_QKEY; /* XXX but VV_QP_AT_Q_KEY not set! */
 878                 init->access_control = vv_acc_r_mem_read |
 879                                        vv_acc_r_mem_write; /* XXX vv_acc_l_mem_write ? */
 880
 881                 attr.modify.vv_qp_attr_mask = VV_QP_AT_P_KEY_IX |
 882                                               VV_QP_AT_PHY_PORT_NUM |
 883                                               VV_QP_AT_ACCESS_CON_F;
 884                 break;
 885         }
 886         case vv_qp_state_rtr: {
 887                 struct vv_qp_modify_rtr_st *rtr = &attr.modify.params.rtr;
 888                 vv_add_vec_t               *av  = &rtr->remote_add_vec;
 889
 890                 av->dlid                      = cv->cv_path.dlid;
 891                 av->grh_flag                  = (!IBNAL_LOCAL_SUB);
 892                 av->max_static_rate           = IBNAL_R_2_STATIC_RATE(cv->cv_path.rate);
 893                 av->service_level             = cv->cv_path.sl;
 894                 av->source_path_bit           = IBNAL_SOURCE_PATH_BIT;
 895                 av->pmtu                      = cv->cv_path.mtu;
 896                 av->rnr_retry_count           = cv->cv_rnr_count;
 897                 av->global_dest.traffic_class = cv->cv_path.traffic_class;
 898                 av->global_dest.hope_limit    = cv->cv_path.hop_limut;
 899                 av->global_dest.flow_lable    = cv->cv_path.flow_label;
 900                 av->global_dest.s_gid_index   = cv->cv_sgid_index;
 901                 // XXX other av fields zero?
 902
 903                 rtr->destanation_qp            = cv->cv_remote_qpn;
 904                 rtr->receive_psn               = cv->cv_rxpsn;
 905                 rtr->responder_rdma_r_atom_num = IBNAL_OUS_DST_RD;
 906                 rtr->opt_min_rnr_nak_timer     = *kibnal_tunables.kib_rnr_nak_timer;
 907
 908
 909                 // XXX sdp sets VV_QP_AT_OP_F but no actual optional options
 910                 attr.modify.vv_qp_attr_mask = VV_QP_AT_ADD_VEC |
 911                                               VV_QP_AT_DEST_QP |
 912                                               VV_QP_AT_R_PSN |
 913                                               VV_QP_AT_MIN_RNR_NAK_T |
 914                                               VV_QP_AT_RESP_RDMA_ATOM_OUT_NUM |
 915                                               VV_QP_AT_OP_F;
 916                 break;
 917         }
 918         case vv_qp_state_rts: {
 919                 struct vv_qp_modify_rts_st *rts = &attr.modify.params.rts;
 920
 921                 rts->send_psn                 = cv->cv_txpsn;
 922                 rts->local_ack_timeout        = *kibnal_tunables.kib_local_ack_timeout;
 923                 rts->retry_num                = *kibnal_tunables.kib_retry_cnt;
 924                 rts->rnr_num                  = *kibnal_tunables.kib_rnr_cnt;
 925                 rts->dest_out_rdma_r_atom_num = IBNAL_OUS_DST_RD;
 926
 927                 attr.modify.vv_qp_attr_mask = VV_QP_AT_S_PSN |
 928                                               VV_QP_AT_L_ACK_T |
 929                                               VV_QP_AT_RETRY_NUM |
 930                                               VV_QP_AT_RNR_NUM |
 931                                               VV_QP_AT_DEST_RDMA_ATOM_OUT_NUM;
 932                 break;
 933         }
 934         case vv_qp_state_error:
 935         case vv_qp_state_reset:
 936                 attr.modify.vv_qp_attr_mask = 0;
 937                 break;
 938         }
 939
 940         attr.modify.qp_modify_into_state = new_state;
 941         attr.modify.vv_qp_attr_mask |= VV_QP_AT_STATE;
 942
 943         vvrc = vv_qp_modify(kibnal_data.kib_hca, conn->ibc_qp, &attr, NULL);
 944         if (vvrc != vv_return_ok) {
 945                 CERROR("Can't modify qp -> %s state to %d: %d\n",
 946                        libcfs_nid2str(conn->ibc_peer->ibp_nid),
 947                        new_state, vvrc);
 948                 return -EIO;
 949         }
 950
 951         return 0;
 952 }
 953
 954 kib_conn_t *
 955 kibnal_create_conn (cm_cep_handle_t cep)
 956 {
 957         kib_conn_t   *conn;
 958         int           i;
 959         int           page_offset;
 960         int           ipage;
 961         vv_return_t   vvrc;
 962         int           rc;
 963
 964         static vv_qp_attr_t  reqattr;
 965         static vv_qp_attr_t  rspattr;
 966
 967         /* Only the connd creates conns => single threaded */
 968         LASSERT(!in_interrupt());
 969         LASSERT(current == kibnal_data.kib_connd);
 970
 971         LIBCFS_ALLOC(conn, sizeof (*conn));
 972         if (conn == NULL) {
 973                 CERROR ("Can't allocate connection\n");
 974                 return (NULL);
 975         }
 976
 977         /* zero flags, NULL pointers etc... */
 978         memset (conn, 0, sizeof (*conn));
 979
 980         conn->ibc_version = IBNAL_MSG_VERSION;  /* Use latest version at first */
 981
 982         INIT_LIST_HEAD (&conn->ibc_early_rxs);
 983         INIT_LIST_HEAD (&conn->ibc_tx_queue_nocred);
 984         INIT_LIST_HEAD (&conn->ibc_tx_queue);
 985         INIT_LIST_HEAD (&conn->ibc_tx_queue_rsrvd);
 986         INIT_LIST_HEAD (&conn->ibc_active_txs);
 987         spin_lock_init (&conn->ibc_lock);
 988
 989         atomic_inc (&kibnal_data.kib_nconns);
 990         /* well not really, but I call destroy() on failure, which decrements */
 991
 992         conn->ibc_cep = cep;
 993
 994         LIBCFS_ALLOC(conn->ibc_connvars, sizeof(*conn->ibc_connvars));
 995         if (conn->ibc_connvars == NULL) {
 996                 CERROR("Can't allocate in-progress connection state\n");
 997                 goto failed;
 998         }
 999         memset (conn->ibc_connvars, 0, sizeof(*conn->ibc_connvars));
1000         /* Random seed for QP sequence number */
1001         get_random_bytes(&conn->ibc_connvars->cv_rxpsn,
1002                          sizeof(conn->ibc_connvars->cv_rxpsn));
1003
1004         LIBCFS_ALLOC(conn->ibc_rxs, IBNAL_RX_MSGS * sizeof (kib_rx_t));
1005         if (conn->ibc_rxs == NULL) {
1006                 CERROR("Cannot allocate RX buffers\n");
1007                 goto failed;
1008         }
1009         memset (conn->ibc_rxs, 0, IBNAL_RX_MSGS * sizeof(kib_rx_t));
1010
1011         rc = kibnal_alloc_pages(&conn->ibc_rx_pages, IBNAL_RX_MSG_PAGES, 1);
1012         if (rc != 0)
1013                 goto failed;
1014
1015         for (i = ipage = page_offset = 0; i < IBNAL_RX_MSGS; i++) {
1016                 struct page    *page = conn->ibc_rx_pages->ibp_pages[ipage];
1017                 kib_rx_t       *rx = &conn->ibc_rxs[i];
1018                 vv_mem_reg_h_t  mem_h;
1019                 vv_r_key_t      r_key;
1020
1021                 rx->rx_conn = conn;
1022                 rx->rx_msg = (kib_msg_t *)(((char *)page_address(page)) +
1023                              page_offset);
1024
1025                 vvrc = vv_get_gen_mr_attrib(kibnal_data.kib_hca,
1026                                             rx->rx_msg,
1027                                             IBNAL_MSG_SIZE,
1028                                             &mem_h,
1029                                             &rx->rx_lkey,
1030                                             &r_key);
1031                 LASSERT (vvrc == vv_return_ok);
1032
1033                 CDEBUG(D_NET, "Rx[%d] %p->%p[%x]\n", i, rx,
1034                        rx->rx_msg, rx->rx_lkey);
1035
1036                 page_offset += IBNAL_MSG_SIZE;
1037                 LASSERT (page_offset <= PAGE_SIZE);
1038
1039                 if (page_offset == PAGE_SIZE) {
1040                         page_offset = 0;
1041                         ipage++;
1042                         LASSERT (ipage <= IBNAL_RX_MSG_PAGES);
1043                 }
1044         }
1045
1046         memset(&reqattr, 0, sizeof(reqattr));
1047
1048         reqattr.create.qp_type                    = vv_qp_type_r_conn;
1049         reqattr.create.cq_send_h                  = kibnal_data.kib_cq;
1050         reqattr.create.cq_receive_h               = kibnal_data.kib_cq;
1051         reqattr.create.send_max_outstand_wr       = (1 + IBNAL_MAX_RDMA_FRAGS) *
1052                                                     (*kibnal_tunables.kib_concurrent_sends);
1053         reqattr.create.receive_max_outstand_wr    = IBNAL_RX_MSGS;
1054         reqattr.create.max_scatgat_per_send_wr    = 1;
1055         reqattr.create.max_scatgat_per_receive_wr = 1;
1056         reqattr.create.signaling_type             = vv_selectable_signaling;
1057         reqattr.create.pd_h                       = kibnal_data.kib_pd;
1058         reqattr.create.recv_solicited_events      = vv_selectable_signaling; // vv_signal_all;
1059
1060         vvrc = vv_qp_create(kibnal_data.kib_hca, &reqattr, NULL,
1061                             &conn->ibc_qp, &rspattr);
1062         if (vvrc != vv_return_ok) {
1063                 CERROR ("Failed to create queue pair: %d\n", vvrc);
1064                 goto failed;
1065         }
1066
1067         /* Mark QP created */
1068         conn->ibc_state = IBNAL_CONN_INIT_QP;
1069         conn->ibc_connvars->cv_local_qpn = rspattr.create_return.qp_num;
1070
1071         if (rspattr.create_return.receive_max_outstand_wr <
1072             IBNAL_RX_MSGS ||
1073             rspattr.create_return.send_max_outstand_wr <
1074             (1 + IBNAL_MAX_RDMA_FRAGS) * (*kibnal_tunables.kib_concurrent_sends)) {
1075                 CERROR("Insufficient rx/tx work items: wanted %d/%d got %d/%d\n",
1076                        IBNAL_RX_MSGS,
1077                        (1 + IBNAL_MAX_RDMA_FRAGS) *
1078                        (*kibnal_tunables.kib_concurrent_sends),
1079                        rspattr.create_return.receive_max_outstand_wr,
1080                        rspattr.create_return.send_max_outstand_wr);
1081                 goto failed;
1082         }
1083
1084         /* Mark init complete */
1085         conn->ibc_state = IBNAL_CONN_INIT;
1086
1087         /* 1 ref for caller */
1088         atomic_set (&conn->ibc_refcount, 1);
1089         return (conn);
1090
1091  failed:
1092         kibnal_destroy_conn (conn);
1093         return (NULL);
1094 }
1095
1096 void
1097 kibnal_destroy_conn (kib_conn_t *conn)
1098 {
1099         vv_return_t vvrc;
1100
1101         /* Only the connd does this (i.e. single threaded) */
1102         LASSERT (!in_interrupt());
1103         LASSERT (current == kibnal_data.kib_connd);
1104
1105         CDEBUG (D_NET, "connection %p\n", conn);
1106
1107         LASSERT (atomic_read (&conn->ibc_refcount) == 0);
1108         LASSERT (list_empty(&conn->ibc_early_rxs));
1109         LASSERT (list_empty(&conn->ibc_tx_queue));
1110         LASSERT (list_empty(&conn->ibc_tx_queue_rsrvd));
1111         LASSERT (list_empty(&conn->ibc_tx_queue_nocred));
1112         LASSERT (list_empty(&conn->ibc_active_txs));
1113         LASSERT (conn->ibc_nsends_posted == 0);
1114
1115         switch (conn->ibc_state) {
1116         default:
1117                 /* conn must be completely disengaged from the network */
1118                 LBUG();
1119
1120         case IBNAL_CONN_DISCONNECTED:
1121                 /* connvars should have been freed already */
1122                 LASSERT (conn->ibc_connvars == NULL);
1123                 /* fall through */
1124
1125         case IBNAL_CONN_INIT:
1126                 vvrc = cm_destroy_cep(conn->ibc_cep);
1127                 LASSERT (vvrc == vv_return_ok);
1128                 /* fall through */
1129
1130         case IBNAL_CONN_INIT_QP:
1131                 kibnal_set_qp_state(conn, vv_qp_state_reset);
1132                 vvrc = vv_qp_destroy(kibnal_data.kib_hca, conn->ibc_qp);
1133                 if (vvrc != vv_return_ok)
1134                         CERROR("Can't destroy QP: %d\n", vvrc);
1135                 /* fall through */
1136
1137         case IBNAL_CONN_INIT_NOTHING:
1138                 break;
1139         }
1140
1141         if (conn->ibc_rx_pages != NULL)
1142                 kibnal_free_pages(conn->ibc_rx_pages);
1143
1144         if (conn->ibc_rxs != NULL)
1145                 LIBCFS_FREE(conn->ibc_rxs,
1146                             IBNAL_RX_MSGS * sizeof(kib_rx_t));
1147
1148         if (conn->ibc_connvars != NULL)
1149                 LIBCFS_FREE(conn->ibc_connvars, sizeof(*conn->ibc_connvars));
1150
1151         if (conn->ibc_peer != NULL)
1152                 kibnal_peer_decref(conn->ibc_peer);
1153
1154         LIBCFS_FREE(conn, sizeof (*conn));
1155
1156         atomic_dec(&kibnal_data.kib_nconns);
1157 }
1158
1159 int
1160 kibnal_close_peer_conns_locked (kib_peer_t *peer, int why)
1161 {
1162         kib_conn_t         *conn;
1163         struct list_head   *ctmp;
1164         struct list_head   *cnxt;
1165         int                 count = 0;
1166
1167         list_for_each_safe (ctmp, cnxt, &peer->ibp_conns) {
1168                 conn = list_entry (ctmp, kib_conn_t, ibc_list);
1169
1170                 count++;
1171                 kibnal_close_conn_locked (conn, why);
1172         }
1173
1174         return (count);
1175 }
1176
1177 int
1178 kibnal_close_stale_conns_locked (kib_peer_t *peer, __u64 incarnation)
1179 {
1180         kib_conn_t         *conn;
1181         struct list_head   *ctmp;
1182         struct list_head   *cnxt;
1183         int                 count = 0;
1184
1185         list_for_each_safe (ctmp, cnxt, &peer->ibp_conns) {
1186                 conn = list_entry (ctmp, kib_conn_t, ibc_list);
1187
1188                 if (conn->ibc_incarnation == incarnation)
1189                         continue;
1190
1191                 CDEBUG(D_NET, "Closing stale conn -> %s incarnation:"LPX64"("LPX64")\n",
1192                        libcfs_nid2str(peer->ibp_nid),
1193                        conn->ibc_incarnation, incarnation);
1194
1195                 count++;
1196                 kibnal_close_conn_locked (conn, -ESTALE);
1197         }
1198
1199         return (count);
1200 }
1201
1202 int
1203 kibnal_close_matching_conns (lnet_nid_t nid)
1204 {
1205         kib_peer_t         *peer;
1206         struct list_head   *ptmp;
1207         struct list_head   *pnxt;
1208         int                 lo;
1209         int                 hi;
1210         int                 i;
1211         unsigned long       flags;
1212         int                 count = 0;
1213
1214         write_lock_irqsave(&kibnal_data.kib_global_lock, flags);
1215
1216         if (nid != LNET_NID_ANY)
1217                 lo = hi = kibnal_nid2peerlist(nid) - kibnal_data.kib_peers;
1218         else {
1219                 lo = 0;
1220                 hi = kibnal_data.kib_peer_hash_size - 1;
1221         }
1222
1223         for (i = lo; i <= hi; i++) {
1224                 list_for_each_safe (ptmp, pnxt, &kibnal_data.kib_peers[i]) {
1225
1226                         peer = list_entry (ptmp, kib_peer_t, ibp_list);
1227                         LASSERT (peer->ibp_persistence != 0 ||
1228                                  peer->ibp_connecting != 0 ||
1229                                  peer->ibp_accepting != 0 ||
1230                                  !list_empty (&peer->ibp_conns));
1231
1232                         if (!(nid == LNET_NID_ANY || nid == peer->ibp_nid))
1233                                 continue;
1234
1235                         count += kibnal_close_peer_conns_locked (peer, 0);
1236                 }
1237         }
1238
1239         write_unlock_irqrestore(&kibnal_data.kib_global_lock, flags);
1240
1241         /* wildcards always succeed */
1242         if (nid == LNET_NID_ANY)
1243                 return (0);
1244
1245         return (count == 0 ? -ENOENT : 0);
1246 }
1247
1248 int
1249 kibnal_ctl(lnet_ni_t *ni, unsigned int cmd, void *arg)
1250 {
1251         struct libcfs_ioctl_data *data = arg;
1252         int                       rc = -EINVAL;
1253
1254         LASSERT (ni == kibnal_data.kib_ni);
1255
1256         switch(cmd) {
1257         case IOC_LIBCFS_GET_PEER: {
1258                 lnet_nid_t   nid = 0;
1259                 __u32        ip = 0;
1260                 int          share_count = 0;
1261
1262                 rc = kibnal_get_peer_info(data->ioc_count,
1263                                           &nid, &ip, &share_count);
1264                 data->ioc_nid    = nid;
1265                 data->ioc_count  = share_count;
1266                 data->ioc_u32[0] = ip;
1267                 data->ioc_u32[1] = *kibnal_tunables.kib_service_number; /* port */
1268                 break;
1269         }
1270         case IOC_LIBCFS_ADD_PEER: {
1271                 rc = kibnal_add_persistent_peer (data->ioc_nid,
1272                                                  data->ioc_u32[0]); /* IP */
1273                 break;
1274         }
1275         case IOC_LIBCFS_DEL_PEER: {
1276                 rc = kibnal_del_peer (data->ioc_nid);
1277                 break;
1278         }
1279         case IOC_LIBCFS_GET_CONN: {
1280                 kib_conn_t *conn = kibnal_get_conn_by_idx (data->ioc_count);
1281
1282                 if (conn == NULL)
1283                         rc = -ENOENT;
1284                 else {
1285                         // kibnal_debug_conn(conn);
1286                         rc = 0;
1287                         data->ioc_nid = conn->ibc_peer->ibp_nid;
1288                         kibnal_conn_decref(conn);
1289                 }
1290                 break;
1291         }
1292         case IOC_LIBCFS_CLOSE_CONNECTION: {
1293                 rc = kibnal_close_matching_conns (data->ioc_nid);
1294                 break;
1295         }
1296         case IOC_LIBCFS_REGISTER_MYNID: {
1297                 if (ni->ni_nid == data->ioc_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 kibnal_free_pages (kib_pages_t *p)
1314 {
1315         int         npages = p->ibp_npages;
1316         int         i;
1317
1318         for (i = 0; i < npages; i++)
1319                 if (p->ibp_pages[i] != NULL)
1320                         __free_page(p->ibp_pages[i]);
1321
1322         LIBCFS_FREE (p, offsetof(kib_pages_t, ibp_pages[npages]));
1323 }
1324
1325 int
1326 kibnal_alloc_pages (kib_pages_t **pp, int npages, int allow_write)
1327 {
1328         kib_pages_t   *p;
1329         int            i;
1330
1331         LIBCFS_ALLOC(p, offsetof(kib_pages_t, ibp_pages[npages]));
1332         if (p == NULL) {
1333                 CERROR ("Can't allocate buffer %d\n", npages);
1334                 return (-ENOMEM);
1335         }
1336
1337         memset (p, 0, offsetof(kib_pages_t, ibp_pages[npages]));
1338         p->ibp_npages = npages;
1339
1340         for (i = 0; i < npages; i++) {
1341                 p->ibp_pages[i] = alloc_page (GFP_KERNEL);
1342                 if (p->ibp_pages[i] == NULL) {
1343                         CERROR ("Can't allocate page %d of %d\n", i, npages);
1344                         kibnal_free_pages(p);
1345                         return (-ENOMEM);
1346                 }
1347         }
1348
1349         *pp = p;
1350         return (0);
1351 }
1352
1353 int
1354 kibnal_alloc_tx_descs (void)
1355 {
1356         int    i;
1357
1358         LIBCFS_ALLOC (kibnal_data.kib_tx_descs,
1359                       IBNAL_TX_MSGS() * sizeof(kib_tx_t));
1360         if (kibnal_data.kib_tx_descs == NULL)
1361                 return -ENOMEM;
1362
1363         memset(kibnal_data.kib_tx_descs, 0,
1364                IBNAL_TX_MSGS() * sizeof(kib_tx_t));
1365
1366         for (i = 0; i < IBNAL_TX_MSGS(); i++) {
1367                 kib_tx_t *tx = &kibnal_data.kib_tx_descs[i];
1368
1369 #if IBNAL_USE_FMR
1370                 LIBCFS_ALLOC(tx->tx_pages, LNET_MAX_IOV *
1371                              sizeof(*tx->tx_pages));
1372                 if (tx->tx_pages == NULL)
1373                         return -ENOMEM;
1374 #else
1375                 LIBCFS_ALLOC(tx->tx_wrq,
1376                              (1 + IBNAL_MAX_RDMA_FRAGS) *
1377                              sizeof(*tx->tx_wrq));
1378                 if (tx->tx_wrq == NULL)
1379                         return -ENOMEM;
1380
1381                 LIBCFS_ALLOC(tx->tx_gl,
1382                              (1 + IBNAL_MAX_RDMA_FRAGS) *
1383                              sizeof(*tx->tx_gl));
1384                 if (tx->tx_gl == NULL)
1385                         return -ENOMEM;
1386
1387                 LIBCFS_ALLOC(tx->tx_rd,
1388                              offsetof(kib_rdma_desc_t,
1389                                       rd_frags[IBNAL_MAX_RDMA_FRAGS]));
1390                 if (tx->tx_rd == NULL)
1391                         return -ENOMEM;
1392 #endif
1393         }
1394
1395         return 0;
1396 }
1397
1398 void
1399 kibnal_free_tx_descs (void)
1400 {
1401         int    i;
1402
1403         if (kibnal_data.kib_tx_descs == NULL)
1404                 return;
1405
1406         for (i = 0; i < IBNAL_TX_MSGS(); i++) {
1407                 kib_tx_t *tx = &kibnal_data.kib_tx_descs[i];
1408
1409 #if IBNAL_USE_FMR
1410                 if (tx->tx_pages != NULL)
1411                         LIBCFS_FREE(tx->tx_pages, LNET_MAX_IOV *
1412                                     sizeof(*tx->tx_pages));
1413 #else
1414                 if (tx->tx_wrq != NULL)
1415                         LIBCFS_FREE(tx->tx_wrq,
1416                                     (1 + IBNAL_MAX_RDMA_FRAGS) *
1417                                     sizeof(*tx->tx_wrq));
1418
1419                 if (tx->tx_gl != NULL)
1420                         LIBCFS_FREE(tx->tx_gl,
1421                                     (1 + IBNAL_MAX_RDMA_FRAGS) *
1422                                     sizeof(*tx->tx_gl));
1423
1424                 if (tx->tx_rd != NULL)
1425                         LIBCFS_FREE(tx->tx_rd,
1426                                     offsetof(kib_rdma_desc_t,
1427                                              rd_frags[IBNAL_MAX_RDMA_FRAGS]));
1428 #endif
1429         }
1430
1431         LIBCFS_FREE(kibnal_data.kib_tx_descs,
1432                     IBNAL_TX_MSGS() * sizeof(kib_tx_t));
1433 }
1434
1435 #if IBNAL_USE_FMR
1436 void
1437 kibnal_free_fmrs (int n)
1438 {
1439         int             i;
1440         vv_return_t     vvrc;
1441         kib_tx_t       *tx;
1442
1443         for (i = 0; i < n; i++) {
1444                 tx = &kibnal_data.kib_tx_descs[i];
1445
1446                 vvrc = vv_free_fmr(kibnal_data.kib_hca,
1447                                    tx->tx_md.md_fmrhandle);
1448                 if (vvrc != vv_return_ok)
1449                         CWARN("vv_free_fmr[%d]: %d\n", i, vvrc);
1450         }
1451 }
1452 #endif
1453
1454 int
1455 kibnal_setup_tx_descs (void)
1456 {
1457         int             ipage = 0;
1458         int             page_offset = 0;
1459         struct page    *page;
1460         kib_tx_t       *tx;
1461         vv_mem_reg_h_t  mem_h;
1462         vv_r_key_t      rkey;
1463         vv_return_t     vvrc;
1464         int             i;
1465         int             rc;
1466 #if IBNAL_USE_FMR
1467         vv_fmr_t        fmr_props;
1468 #endif
1469
1470         /* pre-mapped messages are not bigger than 1 page */
1471         CLASSERT (IBNAL_MSG_SIZE <= PAGE_SIZE);
1472
1473         /* No fancy arithmetic when we do the buffer calculations */
1474         CLASSERT (PAGE_SIZE % IBNAL_MSG_SIZE == 0);
1475
1476         rc = kibnal_alloc_pages(&kibnal_data.kib_tx_pages,
1477                                 IBNAL_TX_MSG_PAGES(), 0);
1478         if (rc != 0)
1479                 return (rc);
1480
1481         for (i = 0; i < IBNAL_TX_MSGS(); i++) {
1482                 page = kibnal_data.kib_tx_pages->ibp_pages[ipage];
1483                 tx = &kibnal_data.kib_tx_descs[i];
1484
1485 #if IBNAL_USE_FMR
1486                 memset(&fmr_props, 0, sizeof(fmr_props));
1487                 fmr_props.pd_hndl              = kibnal_data.kib_pd;
1488                 fmr_props.acl                  = (vv_acc_r_mem_write |
1489                                                   vv_acc_l_mem_write);
1490                 fmr_props.max_pages            = LNET_MAX_IOV;
1491                 fmr_props.log2_page_sz         = PAGE_SHIFT;
1492                 fmr_props.max_outstanding_maps = *kibnal_tunables.kib_fmr_remaps;
1493
1494                 vvrc = vv_alloc_fmr(kibnal_data.kib_hca,
1495                                     &fmr_props,
1496                                     &tx->tx_md.md_fmrhandle);
1497                 if (vvrc != vv_return_ok) {
1498                         CERROR("Can't allocate fmr %d: %d\n", i, vvrc);
1499
1500                         kibnal_free_fmrs(i);
1501                         kibnal_free_pages (kibnal_data.kib_tx_pages);
1502                         return -ENOMEM;
1503                 }
1504
1505                 tx->tx_md.md_fmrcount = *kibnal_tunables.kib_fmr_remaps;
1506                 tx->tx_md.md_active   = 0;
1507 #endif
1508                 tx->tx_msg = (kib_msg_t *)(((char *)page_address(page)) +
1509                                            page_offset);
1510
1511                 vvrc = vv_get_gen_mr_attrib(kibnal_data.kib_hca,
1512                                             tx->tx_msg,
1513                                             IBNAL_MSG_SIZE,
1514                                             &mem_h,
1515                                             &tx->tx_lkey,
1516                                             &rkey);
1517                 LASSERT (vvrc == vv_return_ok);
1518
1519                 CDEBUG(D_NET, "Tx[%d] %p->%p[%x]\n", i, tx,
1520                        tx->tx_msg, tx->tx_lkey);
1521
1522                 list_add (&tx->tx_list, &kibnal_data.kib_idle_txs);
1523
1524                 page_offset += IBNAL_MSG_SIZE;
1525                 LASSERT (page_offset <= PAGE_SIZE);
1526
1527                 if (page_offset == PAGE_SIZE) {
1528                         page_offset = 0;
1529                         ipage++;
1530                         LASSERT (ipage <= IBNAL_TX_MSG_PAGES());
1531                 }
1532         }
1533
1534         return (0);
1535 }
1536
1537 void
1538 kibnal_shutdown (lnet_ni_t *ni)
1539 {
1540         int           i;
1541         vv_return_t   vvrc;
1542
1543         LASSERT (ni == kibnal_data.kib_ni);
1544         LASSERT (ni->ni_data == &kibnal_data);
1545
1546         CDEBUG(D_MALLOC, "before NAL cleanup: kmem %d\n",
1547                atomic_read (&libcfs_kmemory));
1548
1549         switch (kibnal_data.kib_init) {
1550
1551         case IBNAL_INIT_ALL:
1552                 /* stop accepting connections and prevent new peers */
1553                 kibnal_stop_listener(ni);
1554
1555                 /* nuke all existing peers */
1556                 kibnal_del_peer(LNET_NID_ANY);
1557
1558                 /* Wait for all peer state to clean up */
1559                 i = 2;
1560                 while (atomic_read(&kibnal_data.kib_npeers) != 0) {
1561                         i++;
1562                         CDEBUG(((i & (-i)) == i) ? D_WARNING : D_NET, /* 2**n? */
1563                                "waiting for %d peers to disconnect\n",
1564                                atomic_read(&kibnal_data.kib_npeers));
1565                         cfs_pause(cfs_time_seconds(1));
1566                 }
1567                 /* fall through */
1568
1569         case IBNAL_INIT_CQ:
1570                 vvrc = vv_cq_destroy(kibnal_data.kib_hca, kibnal_data.kib_cq);
1571                 if (vvrc != vv_return_ok)
1572                         CERROR ("Destroy CQ error: %d\n", vvrc);
1573                 /* fall through */
1574
1575         case IBNAL_INIT_TXD:
1576                 kibnal_free_pages (kibnal_data.kib_tx_pages);
1577 #if IBNAL_USE_FMR
1578                 kibnal_free_fmrs(IBNAL_TX_MSGS());
1579 #endif
1580                 /* fall through */
1581
1582         case IBNAL_INIT_PD:
1583 #if 0
1584                 /* Only deallocate a PD if we actually allocated one */
1585                 vvrc = vv_pd_deallocate(kibnal_data.kib_hca,
1586                                         kibnal_data.kib_pd);
1587                 if (vvrc != vv_return_ok)
1588                         CERROR ("Destroy PD error: %d\n", vvrc);
1589 #endif
1590                 /* fall through */
1591
1592         case IBNAL_INIT_ASYNC:
1593                 vvrc = vv_dell_async_event_cb (kibnal_data.kib_hca,
1594                                               kibnal_async_callback);
1595                 if (vvrc != vv_return_ok)
1596                         CERROR("vv_dell_async_event_cb error: %d\n", vvrc);
1597
1598                 /* fall through */
1599
1600         case IBNAL_INIT_HCA:
1601                 vvrc = vv_hca_close(kibnal_data.kib_hca);
1602                 if (vvrc != vv_return_ok)
1603                         CERROR ("Close HCA  error: %d\n", vvrc);
1604                 /* fall through */
1605
1606         case IBNAL_INIT_DATA:
1607                 LASSERT (atomic_read(&kibnal_data.kib_npeers) == 0);
1608                 LASSERT (kibnal_data.kib_peers != NULL);
1609                 for (i = 0; i < kibnal_data.kib_peer_hash_size; i++) {
1610                         LASSERT (list_empty (&kibnal_data.kib_peers[i]));
1611                 }
1612                 LASSERT (atomic_read (&kibnal_data.kib_nconns) == 0);
1613                 LASSERT (list_empty (&kibnal_data.kib_connd_zombies));
1614                 LASSERT (list_empty (&kibnal_data.kib_connd_conns));
1615                 LASSERT (list_empty (&kibnal_data.kib_connd_pcreqs));
1616                 LASSERT (list_empty (&kibnal_data.kib_connd_peers));
1617
1618                 /* flag threads to terminate; wake and wait for them to die */
1619                 kibnal_data.kib_shutdown = 1;
1620                 wake_up_all (&kibnal_data.kib_sched_waitq);
1621                 wake_up_all (&kibnal_data.kib_connd_waitq);
1622
1623                 i = 2;
1624                 while (atomic_read (&kibnal_data.kib_nthreads) != 0) {
1625                         i++;
1626                         CDEBUG(((i & (-i)) == i) ? D_WARNING : D_NET, /* power of 2? */
1627                                "Waiting for %d threads to terminate\n",
1628                                atomic_read (&kibnal_data.kib_nthreads));
1629                         cfs_pause(cfs_time_seconds(1));
1630                 }
1631                 /* fall through */
1632
1633         case IBNAL_INIT_NOTHING:
1634                 break;
1635         }
1636
1637         kibnal_free_tx_descs();
1638
1639         if (kibnal_data.kib_peers != NULL)
1640                 LIBCFS_FREE (kibnal_data.kib_peers,
1641                              sizeof (struct list_head) *
1642                              kibnal_data.kib_peer_hash_size);
1643
1644         CDEBUG(D_MALLOC, "after NAL cleanup: kmem %d\n",
1645                atomic_read (&libcfs_kmemory));
1646
1647         kibnal_data.kib_init = IBNAL_INIT_NOTHING;
1648         PORTAL_MODULE_UNUSE;
1649 }
1650
1651 int
1652 kibnal_startup (lnet_ni_t *ni)
1653 {
1654         char                      scratch[32];
1655         char                      ipif_name[32];
1656         char                     *hca_name;
1657         __u32                     ip;
1658         __u32                     netmask;
1659         int                       up;
1660         int                       nob;
1661         int                       devno;
1662         struct timeval            tv;
1663         int                       rc;
1664         int                       i;
1665         vv_request_event_record_t req_er;
1666         vv_return_t               vvrc;
1667
1668         LASSERT (ni->ni_lnd == &the_kiblnd);
1669
1670         /* Only 1 instance supported */
1671         if (kibnal_data.kib_init != IBNAL_INIT_NOTHING) {
1672                 CERROR ("Only 1 instance supported\n");
1673                 return -EPERM;
1674         }
1675
1676         if (*kibnal_tunables.kib_credits > *kibnal_tunables.kib_ntx) {
1677                 CERROR ("Can't set credits(%d) > ntx(%d)\n",
1678                         *kibnal_tunables.kib_credits,
1679                         *kibnal_tunables.kib_ntx);
1680                 return -EINVAL;
1681         }
1682
1683         ni->ni_maxtxcredits = *kibnal_tunables.kib_credits;
1684         ni->ni_peertxcredits = *kibnal_tunables.kib_peercredits;
1685
1686         CLASSERT (LNET_MAX_INTERFACES > 1);
1687
1688         if (ni->ni_interfaces[0] != NULL) {
1689                 /* Use the HCA specified in 'networks=' */
1690
1691                 if (ni->ni_interfaces[1] != NULL) {
1692                         CERROR("Multiple interfaces not supported\n");
1693                         return -EPERM;
1694                 }
1695
1696                 /* Parse <hca base name><number> */
1697                 hca_name = ni->ni_interfaces[0];
1698                 nob = strlen(*kibnal_tunables.kib_hca_basename);
1699
1700                 if (strncmp(hca_name, *kibnal_tunables.kib_hca_basename, nob) ||
1701                     sscanf(hca_name + nob, "%d%n", &devno, &nob) < 1) {
1702                         CERROR("Unrecognised HCA %s\n", hca_name);
1703                         return -EINVAL;
1704                 }
1705
1706         } else {
1707                 /* Use <hca base name>0 */
1708                 devno = 0;
1709
1710                 hca_name = scratch;
1711                 snprintf(hca_name, sizeof(scratch), "%s%d",
1712                          *kibnal_tunables.kib_hca_basename, devno);
1713                 if (strlen(hca_name) == sizeof(scratch) - 1) {
1714                         CERROR("HCA name %s truncated\n", hca_name);
1715                         return -EINVAL;
1716                 }
1717         }
1718
1719         /* Find IP address from <ipif base name><hca number> */
1720         snprintf(ipif_name, sizeof(ipif_name), "%s%d",
1721                  *kibnal_tunables.kib_ipif_basename, devno);
1722         if (strlen(ipif_name) == sizeof(ipif_name) - 1) {
1723                 CERROR("IPoIB interface name %s truncated\n", ipif_name);
1724                 return -EINVAL;
1725         }
1726
1727         rc = libcfs_ipif_query(ipif_name, &up, &ip, &netmask);
1728         if (rc != 0) {
1729                 CERROR("Can't query IPoIB interface %s: %d\n", ipif_name, rc);
1730                 return -ENETDOWN;
1731         }
1732
1733         if (!up) {
1734                 CERROR("Can't query IPoIB interface %s: it's down\n", ipif_name);
1735                 return -ENETDOWN;
1736         }
1737
1738         ni->ni_nid = LNET_MKNID(LNET_NIDNET(ni->ni_nid), ip);
1739
1740         PORTAL_MODULE_USE;
1741         memset (&kibnal_data, 0, sizeof (kibnal_data)); /* zero pointers, flags etc */
1742
1743         kibnal_data.kib_ni = ni;
1744         ni->ni_data = &kibnal_data;
1745
1746         do_gettimeofday(&tv);
1747         kibnal_data.kib_incarnation = (((__u64)tv.tv_sec) * 1000000) + tv.tv_usec;
1748
1749         rwlock_init(&kibnal_data.kib_global_lock);
1750
1751         kibnal_data.kib_peer_hash_size = IBNAL_PEER_HASH_SIZE;
1752         LIBCFS_ALLOC (kibnal_data.kib_peers,
1753                       sizeof (struct list_head) * kibnal_data.kib_peer_hash_size);
1754         if (kibnal_data.kib_peers == NULL) {
1755                 goto failed;
1756         }
1757         for (i = 0; i < kibnal_data.kib_peer_hash_size; i++)
1758                 INIT_LIST_HEAD(&kibnal_data.kib_peers[i]);
1759
1760         spin_lock_init (&kibnal_data.kib_connd_lock);
1761         INIT_LIST_HEAD (&kibnal_data.kib_connd_peers);
1762         INIT_LIST_HEAD (&kibnal_data.kib_connd_pcreqs);
1763         INIT_LIST_HEAD (&kibnal_data.kib_connd_conns);
1764         INIT_LIST_HEAD (&kibnal_data.kib_connd_zombies);
1765         init_waitqueue_head (&kibnal_data.kib_connd_waitq);
1766
1767         spin_lock_init (&kibnal_data.kib_sched_lock);
1768         init_waitqueue_head (&kibnal_data.kib_sched_waitq);
1769
1770         spin_lock_init (&kibnal_data.kib_tx_lock);
1771         INIT_LIST_HEAD (&kibnal_data.kib_idle_txs);
1772
1773         rc = kibnal_alloc_tx_descs();
1774         if (rc != 0) {
1775                 CERROR("Can't allocate tx descs\n");
1776                 goto failed;
1777         }
1778
1779         /* lists/ptrs/locks initialised */
1780         kibnal_data.kib_init = IBNAL_INIT_DATA;
1781         /*****************************************************/
1782
1783         for (i = 0; i < IBNAL_N_SCHED; i++) {
1784                 rc = kibnal_thread_start (kibnal_scheduler, (void *)((long)i));
1785                 if (rc != 0) {
1786                         CERROR("Can't spawn vibnal scheduler[%d]: %d\n",
1787                                i, rc);
1788                         goto failed;
1789                 }
1790         }
1791
1792         rc = kibnal_thread_start (kibnal_connd, NULL);
1793         if (rc != 0) {
1794                 CERROR ("Can't spawn vibnal connd: %d\n", rc);
1795                 goto failed;
1796         }
1797
1798         vvrc = vv_hca_open(hca_name, NULL, &kibnal_data.kib_hca);
1799         if (vvrc != vv_return_ok) {
1800                 CERROR ("Can't open HCA %s: %d\n", hca_name, vvrc);
1801                 goto failed;
1802         }
1803
1804         /* Channel Adapter opened */
1805         kibnal_data.kib_init = IBNAL_INIT_HCA;
1806
1807         /* register to get HCA's asynchronous events. */
1808         req_er.req_event_type = VV_ASYNC_EVENT_ALL_MASK;
1809         vvrc = vv_set_async_event_cb (kibnal_data.kib_hca, req_er,
1810                                      kibnal_async_callback);
1811         if (vvrc != vv_return_ok) {
1812                 CERROR ("Can't set HCA %s callback: %d\n", hca_name, vvrc);
1813                 goto failed;
1814         }
1815
1816         kibnal_data.kib_init = IBNAL_INIT_ASYNC;
1817
1818         /*****************************************************/
1819
1820         vvrc = vv_hca_query(kibnal_data.kib_hca, &kibnal_data.kib_hca_attrs);
1821         if (vvrc != vv_return_ok) {
1822                 CERROR ("Can't size port attrs for %s: %d\n", hca_name, vvrc);
1823                 goto failed;
1824         }
1825
1826         kibnal_data.kib_port = -1;
1827
1828         for (i = 0; i<kibnal_data.kib_hca_attrs.port_num; i++) {
1829
1830                 int port_num = i+1;
1831                 u_int32_t tbl_count;
1832                 vv_port_attrib_t *pattr = &kibnal_data.kib_port_attr;
1833
1834                 vvrc = vv_port_query(kibnal_data.kib_hca, port_num, pattr);
1835                 if (vvrc != vv_return_ok) {
1836                         CERROR("vv_port_query failed for %s port %d: %d\n",
1837                                hca_name, port_num, vvrc);
1838                         continue;
1839                 }
1840
1841                 switch (pattr->port_state) {
1842                 case vv_state_linkDoun:
1843                         CDEBUG(D_NET, "port[%d] Down\n", port_num);
1844                         continue;
1845                 case vv_state_linkInit:
1846                         CDEBUG(D_NET, "port[%d] Init\n", port_num);
1847                         continue;
1848                 case vv_state_linkArm:
1849                         CDEBUG(D_NET, "port[%d] Armed\n", port_num);
1850                         continue;
1851                 case vv_state_linkActive:
1852                         CDEBUG(D_NET, "port[%d] Active\n", port_num);
1853
1854                         /* Found a suitable port. Get its GUID and PKEY. */
1855                         tbl_count = 1;
1856                         vvrc = vv_get_port_gid_tbl(kibnal_data.kib_hca,
1857                                                    port_num, &tbl_count,
1858                                                    &kibnal_data.kib_port_gid);
1859                         if (vvrc != vv_return_ok) {
1860                                 CERROR("vv_get_port_gid_tbl failed "
1861                                        "for %s port %d: %d\n",
1862                                        hca_name, port_num, vvrc);
1863                                 continue;
1864                         }
1865
1866                         tbl_count = 1;
1867                         vvrc = vv_get_port_partition_tbl(kibnal_data.kib_hca,
1868                                                          port_num, &tbl_count,
1869                                                          &kibnal_data.kib_port_pkey);
1870                         if (vvrc != vv_return_ok) {
1871                                 CERROR("vv_get_port_partition_tbl failed "
1872                                        "for %s port %d: %d\n",
1873                                        hca_name, port_num, vvrc);
1874                                 continue;
1875                         }
1876
1877                         kibnal_data.kib_port = port_num;
1878
1879                         break;
1880                 case vv_state_linkActDefer: /* TODO: correct? */
1881                 case vv_state_linkNoChange:
1882                         CERROR("Unexpected %s port[%d] state %d\n",
1883                                hca_name, i, pattr->port_state);
1884                         continue;
1885                 }
1886                 break;
1887         }
1888
1889         if (kibnal_data.kib_port == -1) {
1890                 CERROR ("Can't find an active port on %s\n", hca_name);
1891                 goto failed;
1892         }
1893
1894         CDEBUG(D_NET, "Using %s port %d - GID="LPX64":"LPX64"\n",
1895                hca_name, kibnal_data.kib_port,
1896                kibnal_data.kib_port_gid.scope.g.subnet,
1897                kibnal_data.kib_port_gid.scope.g.eui64);
1898
1899         /*****************************************************/
1900
1901 #if 1
1902         /* We use a pre-allocated PD */
1903         vvrc = vv_get_gen_pd_h(kibnal_data.kib_hca, &kibnal_data.kib_pd);
1904 #else
1905         vvrc = vv_pd_allocate(kibnal_data.kib_hca, &kibnal_data.kib_pd);
1906 #endif
1907         if (vvrc != vv_return_ok) {
1908                 CERROR ("Can't init PD: %d\n", vvrc);
1909                 goto failed;
1910         }
1911
1912         /* flag PD initialised */
1913         kibnal_data.kib_init = IBNAL_INIT_PD;
1914         /*****************************************************/
1915
1916         rc = kibnal_setup_tx_descs();
1917         if (rc != 0) {
1918                 CERROR ("Can't register tx descs: %d\n", rc);
1919                 goto failed;
1920         }
1921
1922         /* flag TX descs initialised */
1923         kibnal_data.kib_init = IBNAL_INIT_TXD;
1924         /*****************************************************/
1925
1926         {
1927                 uint32_t nentries;
1928
1929                 vvrc = vv_cq_create(kibnal_data.kib_hca, IBNAL_CQ_ENTRIES(),
1930                                     kibnal_cq_callback,
1931                                     NULL, /* context */
1932                                     &kibnal_data.kib_cq, &nentries);
1933                 if (vvrc != 0) {
1934                         CERROR ("Can't create RX CQ: %d\n", vvrc);
1935                         goto failed;
1936                 }
1937
1938                 /* flag CQ initialised */
1939                 kibnal_data.kib_init = IBNAL_INIT_CQ;
1940
1941                 if (nentries < IBNAL_CQ_ENTRIES()) {
1942                         CERROR ("CQ only has %d entries, need %d\n",
1943                                 nentries, IBNAL_CQ_ENTRIES());
1944                         goto failed;
1945                 }
1946
1947                 vvrc = vv_request_completion_notification(kibnal_data.kib_hca,
1948                                                           kibnal_data.kib_cq,
1949                                                           vv_next_solicit_unsolicit_event);
1950                 if (vvrc != 0) {
1951                         CERROR ("Failed to re-arm completion queue: %d\n", rc);
1952                         goto failed;
1953                 }
1954         }
1955
1956         rc = kibnal_start_listener(ni);
1957         if (rc != 0) {
1958                 CERROR("Can't start listener: %d\n", rc);
1959                 goto failed;
1960         }
1961
1962         /* flag everything initialised */
1963         kibnal_data.kib_init = IBNAL_INIT_ALL;
1964         /*****************************************************/
1965
1966         return (0);
1967
1968  failed:
1969         CDEBUG(D_NET, "kibnal_startup failed\n");
1970         kibnal_shutdown (ni);
1971         return (-ENETDOWN);
1972 }
1973
1974 void __exit
1975 kibnal_module_fini (void)
1976 {
1977         lnet_unregister_lnd(&the_kiblnd);
1978         kibnal_tunables_fini();
1979 }
1980
1981 int __init
1982 kibnal_module_init (void)
1983 {
1984         int    rc;
1985
1986         vibnal_assert_wire_constants();
1987
1988         CLASSERT (offsetof(kib_msg_t, ibm_u) + sizeof(kib_connparams_t)
1989                   <= cm_REQ_priv_data_len);
1990         CLASSERT (offsetof(kib_msg_t, ibm_u) + sizeof(kib_connparams_t)
1991                   <= cm_REP_priv_data_len);
1992         CLASSERT (sizeof(kib_msg_t) <= IBNAL_MSG_SIZE);
1993 #if !IBNAL_USE_FMR
1994         CLASSERT (offsetof(kib_msg_t, ibm_u.get.ibgm_rd.rd_frags[IBNAL_MAX_RDMA_FRAGS])
1995                   <= IBNAL_MSG_SIZE);
1996         CLASSERT (offsetof(kib_msg_t, ibm_u.putack.ibpam_rd.rd_frags[IBNAL_MAX_RDMA_FRAGS])
1997                   <= IBNAL_MSG_SIZE);
1998 #endif
1999         rc = kibnal_tunables_init();
2000         if (rc != 0)
2001                 return rc;
2002
2003         lnet_register_lnd(&the_kiblnd);
2004
2005         return 0;
2006 }
2007
2008 MODULE_AUTHOR("Cluster File Systems, Inc. <info@clusterfs.com>");
2009 MODULE_DESCRIPTION("Kernel Voltaire IB LND v1.00");
2010 MODULE_LICENSE("GPL");
2011
2012 module_init(kibnal_module_init);
2013 module_exit(kibnal_module_fini);
2014