lnet/klnds/openiblnd/openiblnd.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  *
   7  *   This file is part of Lustre, http://www.lustre.org.
   8  *
   9  *   Lustre is free software; you can redistribute it and/or
  10  *   modify it under the terms of version 2 of the GNU General Public
  11  *   License as published by the Free Software Foundation.
  12  *
  13  *   Lustre is distributed in the hope that it will be useful,
  14  *   but WITHOUT ANY WARRANTY; without even the implied warranty of
  15  *   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  16  *   GNU General Public License for more details.
  17  *
  18  *   You should have received a copy of the GNU General Public License
  19  *   along with Lustre; if not, write to the Free Software
  20  *   Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
  21  *
  22  */
  23
  24 #include "openibnal.h"
  25
  26 nal_t                   kibnal_api;
  27 ptl_handle_ni_t         kibnal_ni;
  28 kib_data_t              kibnal_data;
  29 kib_tunables_t          kibnal_tunables;
  30
  31 #define IBNAL_SYSCTL             202
  32
  33 enum {
  34         IBNAL_SYSCTL_TIMEOUT=1,
  35         IBNAL_SYSCTL_LISTENER_TIMEOUT,
  36         IBNAL_SYSCTL_BACKLOG,
  37         IBNAL_SYSCTL_PORT
  38 };
  39
  40 static ctl_table kibnal_ctl_table[] = {
  41         {IBNAL_SYSCTL_TIMEOUT, "timeout",
  42          &kibnal_tunables.kib_io_timeout, sizeof (int),
  43          0644, NULL, &proc_dointvec},
  44         {IBNAL_SYSCTL_LISTENER_TIMEOUT, "listener_timeout",
  45          &kibnal_tunables.kib_listener_timeout, sizeof(int),
  46          0644, NULL, &proc_dointvec},
  47         {IBNAL_SYSCTL_BACKLOG, "backlog",
  48          &kibnal_tunables.kib_backlog, sizeof(int),
  49          0644, NULL, kibnal_listener_procint},
  50         {IBNAL_SYSCTL_PORT, "port",
  51          &kibnal_tunables.kib_port, sizeof(int),
  52          0644, NULL, kibnal_listener_procint},
  53         { 0 }
  54 };
  55
  56 static ctl_table kibnal_top_ctl_table[] = {
  57         {IBNAL_SYSCTL, "openibnal", NULL, 0, 0555, kibnal_ctl_table},
  58         { 0 }
  59 };
  60
  61 __u32
  62 kibnal_cksum (void *ptr, int nob)
  63 {
  64         char  *c  = ptr;
  65         __u32  sum = 0;
  66
  67         while (nob-- > 0)
  68                 sum = ((sum << 1) | (sum >> 31)) + *c++;
  69
  70         /* ensure I don't return 0 (== no checksum) */
  71         return (sum == 0) ? 1 : sum;
  72 }
  73
  74 void
  75 kibnal_init_msg(kib_msg_t *msg, int type, int body_nob)
  76 {
  77         msg->ibm_type = type;
  78         msg->ibm_nob  = offsetof(kib_msg_t, ibm_u) + body_nob;
  79 }
  80
  81 void
  82 kibnal_pack_msg(kib_msg_t *msg, int credits, ptl_nid_t dstnid, __u64 dststamp)
  83 {
  84         /* CAVEAT EMPTOR! all message fields not set here should have been
  85          * initialised previously. */
  86         msg->ibm_magic    = IBNAL_MSG_MAGIC;
  87         msg->ibm_version  = IBNAL_MSG_VERSION;
  88         /*   ibm_type */
  89         msg->ibm_credits  = credits;
  90         /*   ibm_nob */
  91         msg->ibm_cksum    = 0;
  92         msg->ibm_srcnid   = kibnal_lib.libnal_ni.ni_pid.nid;
  93         msg->ibm_srcstamp = kibnal_data.kib_incarnation;
  94         msg->ibm_dstnid   = dstnid;
  95         msg->ibm_dststamp = dststamp;
  96 #if IBNAL_CKSUM
  97         /* NB ibm_cksum zero while computing cksum */
  98         msg->ibm_cksum    = kibnal_cksum(msg, msg->ibm_nob);
  99 #endif
 100 }
 101
 102 int
 103 kibnal_unpack_msg(kib_msg_t *msg, int nob)
 104 {
 105         const int hdr_size = offsetof(kib_msg_t, ibm_u);
 106         __u32     msg_cksum;
 107         int       flip;
 108         int       msg_nob;
 109
 110         if (nob < 6) {
 111                 CERROR("Short message: %d\n", nob);
 112                 return -EPROTO;
 113         }
 114
 115         if (msg->ibm_magic == IBNAL_MSG_MAGIC) {
 116                 flip = 0;
 117         } else if (msg->ibm_magic == __swab32(IBNAL_MSG_MAGIC)) {
 118                 flip = 1;
 119         } else {
 120                 CERROR("Bad magic: %08x\n", msg->ibm_magic);
 121                 return -EPROTO;
 122         }
 123
 124         if (msg->ibm_version !=
 125             (flip ? __swab16(IBNAL_MSG_VERSION) : IBNAL_MSG_VERSION)) {
 126                 CERROR("Bad version: %d\n", msg->ibm_version);
 127                 return -EPROTO;
 128         }
 129
 130         if (nob < hdr_size) {
 131                 CERROR("Short message: %d\n", nob);
 132                 return -EPROTO;
 133         }
 134
 135         msg_nob = flip ? __swab32(msg->ibm_nob) : msg->ibm_nob;
 136         if (msg_nob > nob) {
 137                 CERROR("Short message: got %d, wanted %d\n", nob, msg_nob);
 138                 return -EPROTO;
 139         }
 140
 141         /* checksum must be computed with ibm_cksum zero and BEFORE anything
 142          * gets flipped */
 143         msg_cksum = flip ? __swab32(msg->ibm_cksum) : msg->ibm_cksum;
 144         msg->ibm_cksum = 0;
 145         if (msg_cksum != 0 &&
 146             msg_cksum != kibnal_cksum(msg, msg_nob)) {
 147                 CERROR("Bad checksum\n");
 148                 return -EPROTO;
 149         }
 150         msg->ibm_cksum = msg_cksum;
 151
 152         if (flip) {
 153                 /* leave magic unflipped as a clue to peer endianness */
 154                 __swab16s(&msg->ibm_version);
 155                 LASSERT (sizeof(msg->ibm_type) == 1);
 156                 LASSERT (sizeof(msg->ibm_credits) == 1);
 157                 msg->ibm_nob = msg_nob;
 158                 __swab64s(&msg->ibm_srcnid);
 159                 __swab64s(&msg->ibm_srcstamp);
 160                 __swab64s(&msg->ibm_dstnid);
 161                 __swab64s(&msg->ibm_dststamp);
 162         }
 163
 164         if (msg->ibm_srcnid == PTL_NID_ANY) {
 165                 CERROR("Bad src nid: "LPX64"\n", msg->ibm_srcnid);
 166                 return -EPROTO;
 167         }
 168
 169         switch (msg->ibm_type) {
 170         default:
 171                 CERROR("Unknown message type %x\n", msg->ibm_type);
 172                 return -EPROTO;
 173
 174         case IBNAL_MSG_SVCQRY:
 175         case IBNAL_MSG_NOOP:
 176                 break;
 177
 178         case IBNAL_MSG_SVCRSP:
 179                 if (msg_nob < hdr_size + sizeof(msg->ibm_u.svcrsp)) {
 180                         CERROR("Short SVCRSP: %d(%d)\n", msg_nob,
 181                                (int)(hdr_size + sizeof(msg->ibm_u.svcrsp)));
 182                         return -EPROTO;
 183                 }
 184                 if (flip) {
 185                         __swab64s(&msg->ibm_u.svcrsp.ibsr_svc_id);
 186                         __swab16s(&msg->ibm_u.svcrsp.ibsr_svc_pkey);
 187                 }
 188                 break;
 189
 190         case IBNAL_MSG_CONNREQ:
 191         case IBNAL_MSG_CONNACK:
 192                 if (msg_nob < hdr_size + sizeof(msg->ibm_u.connparams)) {
 193                         CERROR("Short CONNREQ: %d(%d)\n", msg_nob,
 194                                (int)(hdr_size + sizeof(msg->ibm_u.connparams)));
 195                         return -EPROTO;
 196                 }
 197                 if (flip)
 198                         __swab32s(&msg->ibm_u.connparams.ibcp_queue_depth);
 199                 break;
 200
 201         case IBNAL_MSG_IMMEDIATE:
 202                 if (msg_nob < offsetof(kib_msg_t, ibm_u.immediate.ibim_payload[0])) {
 203                         CERROR("Short IMMEDIATE: %d(%d)\n", msg_nob,
 204                                (int)offsetof(kib_msg_t, ibm_u.immediate.ibim_payload[0]));
 205                         return -EPROTO;
 206                 }
 207                 break;
 208
 209         case IBNAL_MSG_PUT_RDMA:
 210         case IBNAL_MSG_GET_RDMA:
 211                 if (msg_nob < hdr_size + sizeof(msg->ibm_u.rdma)) {
 212                         CERROR("Short RDMA req: %d(%d)\n", msg_nob,
 213                                (int)(hdr_size + sizeof(msg->ibm_u.rdma)));
 214                         return -EPROTO;
 215                 }
 216                 if (flip) {
 217                         __swab32s(&msg->ibm_u.rdma.ibrm_desc.rd_key);
 218                         __swab32s(&msg->ibm_u.rdma.ibrm_desc.rd_nob);
 219                         __swab64s(&msg->ibm_u.rdma.ibrm_desc.rd_addr);
 220                 }
 221                 break;
 222
 223         case IBNAL_MSG_PUT_DONE:
 224         case IBNAL_MSG_GET_DONE:
 225                 if (msg_nob < hdr_size + sizeof(msg->ibm_u.completion)) {
 226                         CERROR("Short RDMA completion: %d(%d)\n", msg_nob,
 227                                (int)(hdr_size + sizeof(msg->ibm_u.completion)));
 228                         return -EPROTO;
 229                 }
 230                 if (flip)
 231                         __swab32s(&msg->ibm_u.completion.ibcm_status);
 232                 break;
 233         }
 234         return 0;
 235 }
 236
 237 int
 238 kibnal_sock_write (struct socket *sock, void *buffer, int nob)
 239 {
 240         int           rc;
 241         mm_segment_t  oldmm = get_fs();
 242         struct iovec  iov = {
 243                 .iov_base = buffer,
 244                 .iov_len  = nob
 245         };
 246         struct msghdr msg = {
 247                 .msg_name       = NULL,
 248                 .msg_namelen    = 0,
 249                 .msg_iov        = &iov,
 250                 .msg_iovlen     = 1,
 251                 .msg_control    = NULL,
 252                 .msg_controllen = 0,
 253                 .msg_flags      = MSG_DONTWAIT
 254         };
 255
 256         /* We've set up the socket's send buffer to be large enough for
 257          * everything we send, so a single non-blocking send should
 258          * complete without error. */
 259
 260         set_fs(KERNEL_DS);
 261         rc = sock_sendmsg(sock, &msg, iov.iov_len);
 262         set_fs(oldmm);
 263
 264         if (rc == nob)
 265                 return 0;
 266
 267         if (rc >= 0)
 268                 return -EAGAIN;
 269
 270         return rc;
 271 }
 272
 273 int
 274 kibnal_sock_read (struct socket *sock, void *buffer, int nob, int timeout)
 275 {
 276         int            rc;
 277         mm_segment_t   oldmm = get_fs();
 278         long           ticks = timeout * HZ;
 279         unsigned long  then;
 280         struct timeval tv;
 281
 282         LASSERT (nob > 0);
 283         LASSERT (ticks > 0);
 284
 285         for (;;) {
 286                 struct iovec  iov = {
 287                         .iov_base = buffer,
 288                         .iov_len  = nob
 289                 };
 290                 struct msghdr msg = {
 291                         .msg_name       = NULL,
 292                         .msg_namelen    = 0,
 293                         .msg_iov        = &iov,
 294                         .msg_iovlen     = 1,
 295                         .msg_control    = NULL,
 296                         .msg_controllen = 0,
 297                         .msg_flags      = 0
 298                 };
 299
 300                 /* Set receive timeout to remaining time */
 301                 tv = (struct timeval) {
 302                         .tv_sec = ticks / HZ,
 303                         .tv_usec = ((ticks % HZ) * 1000000) / HZ
 304                 };
 305                 set_fs(KERNEL_DS);
 306                 rc = sock_setsockopt(sock, SOL_SOCKET, SO_RCVTIMEO,
 307                                      (char *)&tv, sizeof(tv));
 308                 set_fs(oldmm);
 309                 if (rc != 0) {
 310                         CERROR("Can't set socket recv timeout %d: %d\n",
 311                                timeout, rc);
 312                         return rc;
 313                 }
 314
 315                 set_fs(KERNEL_DS);
 316                 then = jiffies;
 317                 rc = sock_recvmsg(sock, &msg, iov.iov_len, 0);
 318                 ticks -= jiffies - then;
 319                 set_fs(oldmm);
 320
 321                 if (rc < 0)
 322                         return rc;
 323
 324                 if (rc == 0)
 325                         return -ECONNABORTED;
 326
 327                 buffer = ((char *)buffer) + rc;
 328                 nob -= rc;
 329
 330                 if (nob == 0)
 331                         return 0;
 332
 333                 if (ticks <= 0)
 334                         return -ETIMEDOUT;
 335         }
 336 }
 337
 338 int
 339 kibnal_create_sock(struct socket **sockp)
 340 {
 341         struct socket       *sock;
 342         int                  rc;
 343         int                  option;
 344         mm_segment_t         oldmm = get_fs();
 345
 346         rc = sock_create(PF_INET, SOCK_STREAM, 0, &sock);
 347         if (rc != 0) {
 348                 CERROR("Can't create socket: %d\n", rc);
 349                 return rc;
 350         }
 351
 352         /* Ensure sends will not block */
 353         option = 2 * sizeof(kib_msg_t);
 354         set_fs(KERNEL_DS);
 355         rc = sock_setsockopt(sock, SOL_SOCKET, SO_SNDBUF,
 356                              (char *)&option, sizeof(option));
 357         set_fs(oldmm);
 358         if (rc != 0) {
 359                 CERROR("Can't set send buffer %d: %d\n", option, rc);
 360                 goto failed;
 361         }
 362
 363         option = 1;
 364         set_fs(KERNEL_DS);
 365         rc = sock_setsockopt(sock, SOL_SOCKET, SO_REUSEADDR,
 366                              (char *)&option, sizeof(option));
 367         set_fs(oldmm);
 368         if (rc != 0) {
 369                 CERROR("Can't set SO_REUSEADDR: %d\n", rc);
 370                 goto failed;
 371         }
 372
 373         *sockp = sock;
 374         return 0;
 375
 376  failed:
 377         sock_release(sock);
 378         return rc;
 379 }
 380
 381 void
 382 kibnal_pause(int ticks)
 383 {
 384         set_current_state(TASK_UNINTERRUPTIBLE);
 385         schedule_timeout(ticks);
 386 }
 387
 388 int
 389 kibnal_connect_sock(kib_peer_t *peer, struct socket **sockp)
 390 {
 391         struct sockaddr_in  locaddr;
 392         struct sockaddr_in  srvaddr;
 393         struct socket      *sock;
 394         unsigned int        port;
 395         int                 rc;
 396
 397         for (port = 1023; port >= 512; port--) {
 398
 399                 memset(&locaddr, 0, sizeof(locaddr));
 400                 locaddr.sin_family      = AF_INET;
 401                 locaddr.sin_port        = htons(port);
 402                 locaddr.sin_addr.s_addr = htonl(INADDR_ANY);
 403
 404                 memset (&srvaddr, 0, sizeof (srvaddr));
 405                 srvaddr.sin_family      = AF_INET;
 406                 srvaddr.sin_port        = htons (peer->ibp_port);
 407                 srvaddr.sin_addr.s_addr = htonl (peer->ibp_ip);
 408
 409                 rc = kibnal_create_sock(&sock);
 410                 if (rc != 0)
 411                         return rc;
 412
 413                 rc = sock->ops->bind(sock,
 414                                      (struct sockaddr *)&locaddr, sizeof(locaddr));
 415                 if (rc != 0) {
 416                         sock_release(sock);
 417
 418                         if (rc == -EADDRINUSE) {
 419                                 CDEBUG(D_NET, "Port %d already in use\n", port);
 420                                 continue;
 421                         }
 422
 423                         CERROR("Can't bind to reserved port %d: %d\n", port, rc);
 424                         return rc;
 425                 }
 426
 427                 rc = sock->ops->connect(sock,
 428                                         (struct sockaddr *)&srvaddr, sizeof(srvaddr),
 429                                         0);
 430                 if (rc == 0) {
 431                         *sockp = sock;
 432                         return 0;
 433                 }
 434
 435                 sock_release(sock);
 436
 437                 if (rc != -EADDRNOTAVAIL) {
 438                         CERROR("Can't connect port %d to %u.%u.%u.%u/%d: %d\n",
 439                                port, HIPQUAD(peer->ibp_ip), peer->ibp_port, rc);
 440                         return rc;
 441                 }
 442
 443                 CDEBUG(D_NET, "Port %d not available for %u.%u.%u.%u/%d\n",
 444                        port, HIPQUAD(peer->ibp_ip), peer->ibp_port);
 445         }
 446
 447         /* all ports busy */
 448         return -EHOSTUNREACH;
 449 }
 450
 451 int
 452 kibnal_make_svcqry (kib_conn_t *conn)
 453 {
 454         kib_peer_t    *peer = conn->ibc_peer;
 455         kib_msg_t     *msg;
 456         struct socket *sock;
 457         int            rc;
 458         int            nob;
 459
 460         LASSERT (conn->ibc_connreq != NULL);
 461         msg = &conn->ibc_connreq->cr_msg;
 462
 463         kibnal_init_msg(msg, IBNAL_MSG_SVCQRY, 0);
 464         kibnal_pack_msg(msg, 0, peer->ibp_nid, 0);
 465
 466         rc = kibnal_connect_sock(peer, &sock);
 467         if (rc != 0)
 468                 return rc;
 469
 470         rc = kibnal_sock_write(sock, msg, msg->ibm_nob);
 471         if (rc != 0) {
 472                 CERROR("Error %d sending svcqry to "
 473                        LPX64"@%u.%u.%u.%u/%d\n", rc,
 474                        peer->ibp_nid, HIPQUAD(peer->ibp_ip), peer->ibp_port);
 475                 goto out;
 476         }
 477
 478         nob = offsetof(kib_msg_t, ibm_u) + sizeof(msg->ibm_u.svcrsp);
 479         rc = kibnal_sock_read(sock, msg, nob, kibnal_tunables.kib_io_timeout);
 480         if (rc != 0) {
 481                 CERROR("Error %d receiving svcrsp from "
 482                        LPX64"@%u.%u.%u.%u/%d\n", rc,
 483                        peer->ibp_nid, HIPQUAD(peer->ibp_ip), peer->ibp_port);
 484                 goto out;
 485         }
 486
 487         rc = kibnal_unpack_msg(msg, nob);
 488         if (rc != 0) {
 489                 CERROR("Error %d unpacking svcrsp from "
 490                        LPX64"@%u.%u.%u.%u/%d\n", rc,
 491                        peer->ibp_nid, HIPQUAD(peer->ibp_ip), peer->ibp_port);
 492                 goto out;
 493         }
 494
 495         if (msg->ibm_type != IBNAL_MSG_SVCRSP) {
 496                 CERROR("Unexpected response type %d from "
 497                        LPX64"@%u.%u.%u.%u/%d\n", msg->ibm_type,
 498                        peer->ibp_nid, HIPQUAD(peer->ibp_ip), peer->ibp_port);
 499                 rc = -EPROTO;
 500                 goto out;
 501         }
 502
 503         if (msg->ibm_dstnid != kibnal_lib.libnal_ni.ni_pid.nid ||
 504             msg->ibm_dststamp != kibnal_data.kib_incarnation) {
 505                 CERROR("Unexpected dst NID/stamp "LPX64"/"LPX64" from "
 506                        LPX64"@%u.%u.%u.%u/%d\n",
 507                        msg->ibm_dstnid, msg->ibm_dststamp,
 508                        peer->ibp_nid, HIPQUAD(peer->ibp_ip), peer->ibp_port);
 509                 rc = -EPROTO;
 510                 goto out;
 511         }
 512
 513         if (msg->ibm_srcnid != peer->ibp_nid) {
 514                 CERROR("Unexpected src NID "LPX64" from "
 515                        LPX64"@%u.%u.%u.%u/%d\n", msg->ibm_srcnid,
 516                        peer->ibp_nid, HIPQUAD(peer->ibp_ip), peer->ibp_port);
 517                 rc = -EPROTO;
 518                 goto out;
 519         }
 520
 521         conn->ibc_incarnation = msg->ibm_srcstamp;
 522         conn->ibc_connreq->cr_svcrsp = msg->ibm_u.svcrsp;
 523  out:
 524         sock_release(sock);
 525         return rc;
 526 }
 527
 528 void
 529 kibnal_handle_svcqry (struct socket *sock)
 530 {
 531         struct sockaddr_in   addr;
 532         __u32                peer_ip;
 533         unsigned int         peer_port;
 534         kib_msg_t           *msg;
 535         __u64                srcnid;
 536         __u64                srcstamp;
 537         int                  len;
 538         int                  rc;
 539
 540         len = sizeof(addr);
 541         rc = sock->ops->getname(sock, (struct sockaddr *)&addr, &len, 2);
 542         if (rc != 0) {
 543                 CERROR("Can't get peer's IP: %d\n", rc);
 544                 return;
 545         }
 546
 547         peer_ip = ntohl(addr.sin_addr.s_addr);
 548         peer_port = ntohs(addr.sin_port);
 549
 550         if (peer_port >= 1024) {
 551                 CERROR("Refusing unprivileged connection from %u.%u.%u.%u/%d\n",
 552                        HIPQUAD(peer_ip), peer_port);
 553                 return;
 554         }
 555
 556         PORTAL_ALLOC(msg, sizeof(*msg));
 557         if (msg == NULL) {
 558                 CERROR("Can't allocate msgs for %u.%u.%u.%u/%d\n",
 559                        HIPQUAD(peer_ip), peer_port);
 560                 goto out;
 561         }
 562
 563         rc = kibnal_sock_read(sock, msg, offsetof(kib_msg_t, ibm_u),
 564                               kibnal_tunables.kib_listener_timeout);
 565         if (rc != 0) {
 566                 CERROR("Error %d receiving svcqry from %u.%u.%u.%u/%d\n",
 567                        rc, HIPQUAD(peer_ip), peer_port);
 568                 goto out;
 569         }
 570
 571         rc = kibnal_unpack_msg(msg, offsetof(kib_msg_t, ibm_u));
 572         if (rc != 0) {
 573                 CERROR("Error %d unpacking svcqry from %u.%u.%u.%u/%d\n",
 574                        rc, HIPQUAD(peer_ip), peer_port);
 575                 goto out;
 576         }
 577
 578         if (msg->ibm_type != IBNAL_MSG_SVCQRY) {
 579                 CERROR("Unexpected message %d from %u.%u.%u.%u/%d\n",
 580                        msg->ibm_type, HIPQUAD(peer_ip), peer_port);
 581                 goto out;
 582         }
 583
 584         if (msg->ibm_dstnid != kibnal_lib.libnal_ni.ni_pid.nid) {
 585                 CERROR("Unexpected dstnid "LPX64"(expected "LPX64" "
 586                        "from %u.%u.%u.%u/%d\n", msg->ibm_dstnid,
 587                        kibnal_lib.libnal_ni.ni_pid.nid,
 588                        HIPQUAD(peer_ip), peer_port);
 589                 goto out;
 590         }
 591
 592         srcnid = msg->ibm_srcnid;
 593         srcstamp = msg->ibm_srcstamp;
 594
 595         kibnal_init_msg(msg, IBNAL_MSG_SVCRSP, sizeof(msg->ibm_u.svcrsp));
 596
 597         msg->ibm_u.svcrsp.ibsr_svc_id = kibnal_data.kib_svc_id;
 598         memcpy(msg->ibm_u.svcrsp.ibsr_svc_gid, kibnal_data.kib_svc_gid,
 599                sizeof(kibnal_data.kib_svc_gid));
 600         msg->ibm_u.svcrsp.ibsr_svc_pkey = kibnal_data.kib_svc_pkey;
 601
 602         kibnal_pack_msg(msg, 0, srcnid, srcstamp);
 603
 604         rc = kibnal_sock_write (sock, msg, msg->ibm_nob);
 605         if (rc != 0) {
 606                 CERROR("Error %d replying to svcqry from %u.%u.%u.%u/%d\n",
 607                        rc, HIPQUAD(peer_ip), peer_port);
 608                 goto out;
 609         }
 610
 611  out:
 612         PORTAL_FREE(msg, sizeof(*msg));
 613 }
 614
 615 void
 616 kibnal_free_acceptsock (kib_acceptsock_t *as)
 617 {
 618         sock_release(as->ibas_sock);
 619         PORTAL_FREE(as, sizeof(*as));
 620 }
 621
 622 int
 623 kibnal_ip_listener(void *arg)
 624 {
 625         struct sockaddr_in addr;
 626         wait_queue_t       wait;
 627         struct socket     *sock;
 628         kib_acceptsock_t  *as;
 629         int                port;
 630         char               name[16];
 631         int                rc;
 632         unsigned long      flags;
 633
 634         /* Parent thread holds kib_nid_mutex, and is, or is about to
 635          * block on kib_listener_signal */
 636
 637         port = kibnal_tunables.kib_port;
 638         snprintf(name, sizeof(name), "kibnal_lstn%03d", port);
 639         kportal_daemonize(name);
 640         kportal_blockallsigs();
 641
 642         init_waitqueue_entry(&wait, current);
 643
 644         rc = kibnal_create_sock(&sock);
 645         if (rc != 0)
 646                 goto out_0;
 647
 648         memset(&addr, 0, sizeof(addr));
 649         addr.sin_family      = AF_INET;
 650         addr.sin_port        = htons(port);
 651         addr.sin_addr.s_addr = INADDR_ANY;
 652
 653         rc = sock->ops->bind(sock, (struct sockaddr *)&addr, sizeof(addr));
 654         if (rc != 0) {
 655                 CERROR("Can't bind to port %d\n", port);
 656                 goto out_1;
 657         }
 658
 659         rc = sock->ops->listen(sock, kibnal_tunables.kib_backlog);
 660         if (rc != 0) {
 661                 CERROR("Can't set listen backlog %d: %d\n",
 662                        kibnal_tunables.kib_backlog, rc);
 663                 goto out_1;
 664         }
 665
 666         LASSERT (kibnal_data.kib_listener_sock == NULL);
 667         kibnal_data.kib_listener_sock = sock;
 668
 669         /* unblock waiting parent */
 670         LASSERT (kibnal_data.kib_listener_shutdown == 0);
 671         up(&kibnal_data.kib_listener_signal);
 672
 673         /* Wake me any time something happens on my socket */
 674         add_wait_queue(sock->sk->sk_sleep, &wait);
 675         as = NULL;
 676
 677         while (kibnal_data.kib_listener_shutdown == 0) {
 678
 679                 if (as == NULL) {
 680                         PORTAL_ALLOC(as, sizeof(*as));
 681                         if (as == NULL) {
 682                                 CERROR("Out of Memory: pausing...\n");
 683                                 kibnal_pause(HZ);
 684                                 continue;
 685                         }
 686                         as->ibas_sock = NULL;
 687                 }
 688
 689                 if (as->ibas_sock == NULL) {
 690                         as->ibas_sock = sock_alloc();
 691                         if (as->ibas_sock == NULL) {
 692                                 CERROR("Can't allocate socket: pausing...\n");
 693                                 kibnal_pause(HZ);
 694                                 continue;
 695                         }
 696                         /* XXX this should add a ref to sock->ops->owner, if
 697                          * TCP could be a module */
 698                         as->ibas_sock->type = sock->type;
 699                         as->ibas_sock->ops = sock->ops;
 700                 }
 701
 702                 set_current_state(TASK_INTERRUPTIBLE);
 703
 704                 rc = sock->ops->accept(sock, as->ibas_sock, O_NONBLOCK);
 705
 706                 /* Sleep for socket activity? */
 707                 if (rc == -EAGAIN &&
 708                     kibnal_data.kib_listener_shutdown == 0)
 709                         schedule();
 710
 711                 set_current_state(TASK_RUNNING);
 712
 713                 if (rc == 0) {
 714                         spin_lock_irqsave(&kibnal_data.kib_connd_lock, flags);
 715
 716                         list_add_tail(&as->ibas_list,
 717                                       &kibnal_data.kib_connd_acceptq);
 718
 719                         spin_unlock_irqrestore(&kibnal_data.kib_connd_lock, flags);
 720                         wake_up(&kibnal_data.kib_connd_waitq);
 721
 722                         as = NULL;
 723                         continue;
 724                 }
 725
 726                 if (rc != -EAGAIN) {
 727                         CERROR("Accept failed: %d, pausing...\n", rc);
 728                         kibnal_pause(HZ);
 729                 }
 730         }
 731
 732         if (as != NULL) {
 733                 if (as->ibas_sock != NULL)
 734                         sock_release(as->ibas_sock);
 735                 PORTAL_FREE(as, sizeof(*as));
 736         }
 737
 738         rc = 0;
 739         remove_wait_queue(sock->sk->sk_sleep, &wait);
 740  out_1:
 741         sock_release(sock);
 742         kibnal_data.kib_listener_sock = NULL;
 743  out_0:
 744         /* set completion status and unblock thread waiting for me
 745          * (parent on startup failure, executioner on normal shutdown) */
 746         kibnal_data.kib_listener_shutdown = rc;
 747         up(&kibnal_data.kib_listener_signal);
 748
 749         return 0;
 750 }
 751
 752 int
 753 kibnal_start_ip_listener (void)
 754 {
 755         long           pid;
 756         int            rc;
 757
 758         CDEBUG(D_NET, "Starting listener\n");
 759
 760         /* Called holding kib_nid_mutex: listener stopped */
 761         LASSERT (kibnal_data.kib_listener_sock == NULL);
 762
 763         kibnal_data.kib_listener_shutdown = 0;
 764         pid = kernel_thread(kibnal_ip_listener, NULL, 0);
 765         if (pid < 0) {
 766                 CERROR("Can't spawn listener: %ld\n", pid);
 767                 return (int)pid;
 768         }
 769
 770         /* Block until listener has started up. */
 771         down(&kibnal_data.kib_listener_signal);
 772
 773         rc = kibnal_data.kib_listener_shutdown;
 774         LASSERT ((rc != 0) == (kibnal_data.kib_listener_sock == NULL));
 775
 776         CDEBUG((rc == 0) ? D_WARNING : D_ERROR,
 777                "Listener %s: pid:%ld port:%d backlog:%d\n",
 778                (rc == 0) ? "started OK" : "startup failed",
 779                pid, kibnal_tunables.kib_port, kibnal_tunables.kib_backlog);
 780
 781         return rc;
 782 }
 783
 784 void
 785 kibnal_stop_ip_listener(int clear_acceptq)
 786 {
 787         struct list_head  zombie_accepts;
 788         kib_acceptsock_t *as;
 789         unsigned long     flags;
 790
 791         CDEBUG(D_NET, "Stopping listener\n");
 792
 793         /* Called holding kib_nid_mutex: listener running */
 794         LASSERT (kibnal_data.kib_listener_sock != NULL);
 795
 796         kibnal_data.kib_listener_shutdown = 1;
 797         wake_up_all(kibnal_data.kib_listener_sock->sk->sk_sleep);
 798
 799         /* Block until listener has torn down. */
 800         down(&kibnal_data.kib_listener_signal);
 801
 802         LASSERT (kibnal_data.kib_listener_sock == NULL);
 803         CDEBUG(D_WARNING, "Listener stopped\n");
 804
 805         if (!clear_acceptq)
 806                 return;
 807
 808         /* Close any unhandled accepts */
 809         spin_lock_irqsave(&kibnal_data.kib_connd_lock, flags);
 810
 811         list_add(&zombie_accepts, &kibnal_data.kib_connd_acceptq);
 812         list_del_init(&kibnal_data.kib_connd_acceptq);
 813
 814         spin_unlock_irqrestore(&kibnal_data.kib_connd_lock, flags);
 815
 816         while (!list_empty(&zombie_accepts)) {
 817                 as = list_entry(zombie_accepts.next,
 818                                 kib_acceptsock_t, ibas_list);
 819                 list_del(&as->ibas_list);
 820                 kibnal_free_acceptsock(as);
 821         }
 822 }
 823
 824 int
 825 kibnal_listener_procint(ctl_table *table, int write, struct file *filp,
 826                         void *buffer, size_t *lenp)
 827 {
 828         int   *tunable = (int *)table->data;
 829         int    old_val;
 830         int    rc;
 831
 832         /* No race with nal initialisation since the nal is setup all the time
 833          * it's loaded.  When that changes, change this! */
 834         LASSERT (kibnal_data.kib_init == IBNAL_INIT_ALL);
 835
 836         down(&kibnal_data.kib_nid_mutex);
 837
 838         LASSERT (tunable == &kibnal_tunables.kib_port ||
 839                  tunable == &kibnal_tunables.kib_backlog);
 840         old_val = *tunable;
 841
 842         rc = proc_dointvec(table, write, filp, buffer, lenp);
 843
 844         if (write &&
 845             (*tunable != old_val ||
 846              kibnal_data.kib_listener_sock == NULL)) {
 847
 848                 if (kibnal_data.kib_listener_sock != NULL)
 849                         kibnal_stop_ip_listener(0);
 850
 851                 rc = kibnal_start_ip_listener();
 852                 if (rc != 0) {
 853                         CERROR("Unable to restart listener with new tunable:"
 854                                " reverting to old value\n");
 855                         *tunable = old_val;
 856                         kibnal_start_ip_listener();
 857                 }
 858         }
 859
 860         up(&kibnal_data.kib_nid_mutex);
 861
 862         LASSERT (kibnal_data.kib_init == IBNAL_INIT_ALL);
 863         return rc;
 864 }
 865
 866 int
 867 kibnal_start_ib_listener (void)
 868 {
 869         int    rc;
 870
 871         LASSERT (kibnal_data.kib_listen_handle == NULL);
 872
 873         kibnal_data.kib_svc_id = ib_cm_service_assign();
 874         CDEBUG(D_NET, "svc id "LPX64"\n", kibnal_data.kib_svc_id);
 875
 876         rc = ib_cached_gid_get(kibnal_data.kib_device,
 877                                kibnal_data.kib_port, 0,
 878                                kibnal_data.kib_svc_gid);
 879         if (rc != 0) {
 880                 CERROR("Can't get port %d GID: %d\n",
 881                        kibnal_data.kib_port, rc);
 882                 return rc;
 883         }
 884
 885         rc = ib_cached_pkey_get(kibnal_data.kib_device,
 886                                 kibnal_data.kib_port, 0,
 887                                 &kibnal_data.kib_svc_pkey);
 888         if (rc != 0) {
 889                 CERROR ("Can't get port %d PKEY: %d\n",
 890                         kibnal_data.kib_port, rc);
 891                 return rc;
 892         }
 893
 894         rc = ib_cm_listen(kibnal_data.kib_svc_id,
 895                           TS_IB_CM_SERVICE_EXACT_MASK,
 896                           kibnal_passive_conn_callback, NULL,
 897                           &kibnal_data.kib_listen_handle);
 898         if (rc != 0) {
 899                 kibnal_data.kib_listen_handle = NULL;
 900                 CERROR ("Can't create IB listener: %d\n", rc);
 901                 return rc;
 902         }
 903
 904         LASSERT (kibnal_data.kib_listen_handle != NULL);
 905         return 0;
 906 }
 907
 908 void
 909 kibnal_stop_ib_listener (void)
 910 {
 911         int    rc;
 912
 913         LASSERT (kibnal_data.kib_listen_handle != NULL);
 914
 915         rc = ib_cm_listen_stop (kibnal_data.kib_listen_handle);
 916         if (rc != 0)
 917                 CERROR("Error stopping IB listener: %d\n", rc);
 918
 919         kibnal_data.kib_listen_handle = NULL;
 920 }
 921
 922 int
 923 kibnal_set_mynid (ptl_nid_t nid)
 924 {
 925         lib_ni_t         *ni = &kibnal_lib.libnal_ni;
 926         int               rc;
 927
 928         CDEBUG(D_IOCTL, "setting mynid to "LPX64" (old nid="LPX64")\n",
 929                nid, ni->ni_pid.nid);
 930
 931         down (&kibnal_data.kib_nid_mutex);
 932
 933         if (nid == kibnal_data.kib_nid) {
 934                 /* no change of NID */
 935                 up (&kibnal_data.kib_nid_mutex);
 936                 return (0);
 937         }
 938
 939         CDEBUG(D_NET, "NID "LPX64"("LPX64")\n",
 940                kibnal_data.kib_nid, nid);
 941
 942         if (kibnal_data.kib_listener_sock != NULL)
 943                 kibnal_stop_ip_listener(1);
 944
 945         if (kibnal_data.kib_listen_handle != NULL)
 946                 kibnal_stop_ib_listener();
 947
 948         ni->ni_pid.nid = nid;
 949         kibnal_data.kib_incarnation++;
 950         mb();
 951         /* Delete all existing peers and their connections after new
 952          * NID/incarnation set to ensure no old connections in our brave new
 953          * world. */
 954         kibnal_del_peer (PTL_NID_ANY, 0);
 955
 956         if (ni->ni_pid.nid != PTL_NID_ANY) {
 957                 /* got a new NID to install */
 958                 rc = kibnal_start_ib_listener();
 959                 if (rc != 0) {
 960                         CERROR("Can't start IB listener: %d\n", rc);
 961                         goto failed_0;
 962                 }
 963
 964                 rc = kibnal_start_ip_listener();
 965                 if (rc != 0) {
 966                         CERROR("Can't start IP listener: %d\n", rc);
 967                         goto failed_1;
 968                 }
 969         }
 970
 971         up(&kibnal_data.kib_nid_mutex);
 972         return 0;
 973
 974  failed_1:
 975         kibnal_stop_ib_listener();
 976  failed_0:
 977         ni->ni_pid.nid = PTL_NID_ANY;
 978         kibnal_data.kib_incarnation++;
 979         mb();
 980         kibnal_del_peer (PTL_NID_ANY, 0);
 981         up(&kibnal_data.kib_nid_mutex);
 982         return rc;
 983 }
 984
 985 kib_peer_t *
 986 kibnal_create_peer (ptl_nid_t nid)
 987 {
 988         kib_peer_t *peer;
 989
 990         LASSERT (nid != PTL_NID_ANY);
 991
 992         PORTAL_ALLOC (peer, sizeof (*peer));
 993         if (peer == NULL)
 994                 return (NULL);
 995
 996         memset(peer, 0, sizeof(*peer));         /* zero flags etc */
 997
 998         peer->ibp_nid = nid;
 999         atomic_set (&peer->ibp_refcount, 1);    /* 1 ref for caller */
1000
1001         INIT_LIST_HEAD (&peer->ibp_list);       /* not in the peer table yet */
1002         INIT_LIST_HEAD (&peer->ibp_conns);
1003         INIT_LIST_HEAD (&peer->ibp_tx_queue);
1004
1005         peer->ibp_reconnect_time = jiffies;
1006         peer->ibp_reconnect_interval = IBNAL_MIN_RECONNECT_INTERVAL;
1007
1008         atomic_inc (&kibnal_data.kib_npeers);
1009         CDEBUG(D_NET, "peer %p "LPX64"\n", peer, nid);
1010
1011         return (peer);
1012 }
1013
1014 void
1015 kibnal_destroy_peer (kib_peer_t *peer)
1016 {
1017         CDEBUG (D_NET, "peer "LPX64" %p deleted\n", peer->ibp_nid, peer);
1018
1019         LASSERT (atomic_read (&peer->ibp_refcount) == 0);
1020         LASSERT (peer->ibp_persistence == 0);
1021         LASSERT (!kibnal_peer_active(peer));
1022         LASSERT (peer->ibp_connecting == 0);
1023         LASSERT (list_empty (&peer->ibp_conns));
1024         LASSERT (list_empty (&peer->ibp_tx_queue));
1025
1026         PORTAL_FREE (peer, sizeof (*peer));
1027
1028         /* NB a peer's connections keep a reference on their peer until
1029          * they are destroyed, so we can be assured that _all_ state to do
1030          * with this peer has been cleaned up when its refcount drops to
1031          * zero. */
1032         atomic_dec (&kibnal_data.kib_npeers);
1033 }
1034
1035 void
1036 kibnal_put_peer (kib_peer_t *peer)
1037 {
1038         CDEBUG (D_OTHER, "putting peer[%p] -> "LPX64" (%d)\n",
1039                 peer, peer->ibp_nid,
1040                 atomic_read (&peer->ibp_refcount));
1041
1042         LASSERT (atomic_read (&peer->ibp_refcount) > 0);
1043         if (!atomic_dec_and_test (&peer->ibp_refcount))
1044                 return;
1045
1046         kibnal_destroy_peer (peer);
1047 }
1048
1049 kib_peer_t *
1050 kibnal_find_peer_locked (ptl_nid_t nid)
1051 {
1052         struct list_head *peer_list = kibnal_nid2peerlist (nid);
1053         struct list_head *tmp;
1054         kib_peer_t       *peer;
1055
1056         list_for_each (tmp, peer_list) {
1057
1058                 peer = list_entry (tmp, kib_peer_t, ibp_list);
1059
1060                 LASSERT (peer->ibp_persistence != 0 || /* persistent peer */
1061                          peer->ibp_connecting != 0 || /* creating conns */
1062                          !list_empty (&peer->ibp_conns));  /* active conn */
1063
1064                 if (peer->ibp_nid != nid)
1065                         continue;
1066
1067                 CDEBUG(D_NET, "got peer [%p] -> "LPX64" (%d)\n",
1068                        peer, nid, atomic_read (&peer->ibp_refcount));
1069                 return (peer);
1070         }
1071         return (NULL);
1072 }
1073
1074 kib_peer_t *
1075 kibnal_get_peer (ptl_nid_t nid)
1076 {
1077         kib_peer_t     *peer;
1078         unsigned long   flags;
1079
1080         read_lock_irqsave(&kibnal_data.kib_global_lock, flags);
1081         peer = kibnal_find_peer_locked (nid);
1082         if (peer != NULL)                       /* +1 ref for caller? */
1083                 atomic_inc (&peer->ibp_refcount);
1084         read_unlock_irqrestore(&kibnal_data.kib_global_lock, flags);
1085
1086         return (peer);
1087 }
1088
1089 void
1090 kibnal_unlink_peer_locked (kib_peer_t *peer)
1091 {
1092         LASSERT (peer->ibp_persistence == 0);
1093         LASSERT (list_empty(&peer->ibp_conns));
1094
1095         LASSERT (kibnal_peer_active(peer));
1096         list_del_init (&peer->ibp_list);
1097         /* lose peerlist's ref */
1098         kibnal_put_peer (peer);
1099 }
1100
1101 int
1102 kibnal_get_peer_info (int index, ptl_nid_t *nidp, __u32 *ipp, int *portp,
1103                       int *persistencep)
1104 {
1105         kib_peer_t        *peer;
1106         struct list_head  *ptmp;
1107         unsigned long      flags;
1108         int                i;
1109
1110         read_lock_irqsave(&kibnal_data.kib_global_lock, flags);
1111
1112         for (i = 0; i < kibnal_data.kib_peer_hash_size; i++) {
1113
1114                 list_for_each (ptmp, &kibnal_data.kib_peers[i]) {
1115
1116                         peer = list_entry (ptmp, kib_peer_t, ibp_list);
1117                         LASSERT (peer->ibp_persistence != 0 ||
1118                                  peer->ibp_connecting != 0 ||
1119                                  !list_empty (&peer->ibp_conns));
1120
1121                         if (index-- > 0)
1122                                 continue;
1123
1124                         *nidp = peer->ibp_nid;
1125                         *ipp = peer->ibp_ip;
1126                         *portp = peer->ibp_port;
1127                         *persistencep = peer->ibp_persistence;
1128
1129                         read_unlock_irqrestore(&kibnal_data.kib_global_lock,
1130                                                flags);
1131                         return (0);
1132                 }
1133         }
1134
1135         read_unlock_irqrestore(&kibnal_data.kib_global_lock, flags);
1136         return (-ENOENT);
1137 }
1138
1139 int
1140 kibnal_add_persistent_peer (ptl_nid_t nid, __u32 ip, int port)
1141 {
1142         unsigned long      flags;
1143         kib_peer_t        *peer;
1144         kib_peer_t        *peer2;
1145
1146         if (nid == PTL_NID_ANY)
1147                 return (-EINVAL);
1148
1149         peer = kibnal_create_peer (nid);
1150         if (peer == NULL)
1151                 return (-ENOMEM);
1152
1153         write_lock_irqsave (&kibnal_data.kib_global_lock, flags);
1154
1155         peer2 = kibnal_find_peer_locked (nid);
1156         if (peer2 != NULL) {
1157                 kibnal_put_peer (peer);
1158                 peer = peer2;
1159         } else {
1160                 /* peer table takes existing ref on peer */
1161                 list_add_tail (&peer->ibp_list,
1162                                kibnal_nid2peerlist (nid));
1163         }
1164
1165         peer->ibp_ip = ip;
1166         peer->ibp_port = port;
1167         peer->ibp_persistence++;
1168
1169         write_unlock_irqrestore (&kibnal_data.kib_global_lock, flags);
1170         return (0);
1171 }
1172
1173 void
1174 kibnal_del_peer_locked (kib_peer_t *peer, int single_share)
1175 {
1176         struct list_head *ctmp;
1177         struct list_head *cnxt;
1178         kib_conn_t       *conn;
1179
1180         if (!single_share)
1181                 peer->ibp_persistence = 0;
1182         else if (peer->ibp_persistence > 0)
1183                 peer->ibp_persistence--;
1184
1185         if (peer->ibp_persistence != 0)
1186                 return;
1187
1188         if (list_empty(&peer->ibp_conns)) {
1189                 kibnal_unlink_peer_locked(peer);
1190         } else {
1191                 list_for_each_safe (ctmp, cnxt, &peer->ibp_conns) {
1192                         conn = list_entry(ctmp, kib_conn_t, ibc_list);
1193
1194                         kibnal_close_conn_locked (conn, 0);
1195                 }
1196                 /* NB peer is no longer persistent; closing its last conn
1197                  * unlinked it. */
1198         }
1199         /* NB peer now unlinked; might even be freed if the peer table had the
1200          * last ref on it. */
1201 }
1202
1203 int
1204 kibnal_del_peer (ptl_nid_t nid, int single_share)
1205 {
1206         unsigned long      flags;
1207         struct list_head  *ptmp;
1208         struct list_head  *pnxt;
1209         kib_peer_t        *peer;
1210         int                lo;
1211         int                hi;
1212         int                i;
1213         int                rc = -ENOENT;
1214
1215         write_lock_irqsave (&kibnal_data.kib_global_lock, flags);
1216
1217         if (nid != PTL_NID_ANY)
1218                 lo = hi = kibnal_nid2peerlist(nid) - kibnal_data.kib_peers;
1219         else {
1220                 lo = 0;
1221                 hi = kibnal_data.kib_peer_hash_size - 1;
1222         }
1223
1224         for (i = lo; i <= hi; i++) {
1225                 list_for_each_safe (ptmp, pnxt, &kibnal_data.kib_peers[i]) {
1226                         peer = list_entry (ptmp, kib_peer_t, ibp_list);
1227                         LASSERT (peer->ibp_persistence != 0 ||
1228                                  peer->ibp_connecting != 0 ||
1229                                  !list_empty (&peer->ibp_conns));
1230
1231                         if (!(nid == PTL_NID_ANY || peer->ibp_nid == nid))
1232                                 continue;
1233
1234                         kibnal_del_peer_locked (peer, single_share);
1235                         rc = 0;         /* matched something */
1236
1237                         if (single_share)
1238                                 goto out;
1239                 }
1240         }
1241  out:
1242         write_unlock_irqrestore (&kibnal_data.kib_global_lock, flags);
1243
1244         return (rc);
1245 }
1246
1247 kib_conn_t *
1248 kibnal_get_conn_by_idx (int index)
1249 {
1250         kib_peer_t        *peer;
1251         struct list_head  *ptmp;
1252         kib_conn_t        *conn;
1253         struct list_head  *ctmp;
1254         unsigned long      flags;
1255         int                i;
1256
1257         read_lock_irqsave(&kibnal_data.kib_global_lock, flags);
1258
1259         for (i = 0; i < kibnal_data.kib_peer_hash_size; i++) {
1260                 list_for_each (ptmp, &kibnal_data.kib_peers[i]) {
1261
1262                         peer = list_entry (ptmp, kib_peer_t, ibp_list);
1263                         LASSERT (peer->ibp_persistence > 0 ||
1264                                  peer->ibp_connecting != 0 ||
1265                                  !list_empty (&peer->ibp_conns));
1266
1267                         list_for_each (ctmp, &peer->ibp_conns) {
1268                                 if (index-- > 0)
1269                                         continue;
1270
1271                                 conn = list_entry (ctmp, kib_conn_t, ibc_list);
1272                                 CDEBUG(D_NET, "++conn[%p] state %d -> "LPX64" (%d)\n",
1273                                        conn, conn->ibc_state, conn->ibc_peer->ibp_nid,
1274                                        atomic_read (&conn->ibc_refcount));
1275                                 atomic_inc (&conn->ibc_refcount);
1276                                 read_unlock_irqrestore(&kibnal_data.kib_global_lock,
1277                                                        flags);
1278                                 return (conn);
1279                         }
1280                 }
1281         }
1282
1283         read_unlock_irqrestore(&kibnal_data.kib_global_lock, flags);
1284         return (NULL);
1285 }
1286
1287 kib_conn_t *
1288 kibnal_create_conn (void)
1289 {
1290         kib_conn_t  *conn;
1291         int          i;
1292         __u64        vaddr = 0;
1293         __u64        vaddr_base;
1294         int          page_offset;
1295         int          ipage;
1296         int          rc;
1297         union {
1298                 struct ib_qp_create_param  qp_create;
1299                 struct ib_qp_attribute     qp_attr;
1300         } params;
1301
1302         PORTAL_ALLOC (conn, sizeof (*conn));
1303         if (conn == NULL) {
1304                 CERROR ("Can't allocate connection\n");
1305                 return (NULL);
1306         }
1307
1308         /* zero flags, NULL pointers etc... */
1309         memset (conn, 0, sizeof (*conn));
1310
1311         INIT_LIST_HEAD (&conn->ibc_tx_queue);
1312         INIT_LIST_HEAD (&conn->ibc_active_txs);
1313         spin_lock_init (&conn->ibc_lock);
1314
1315         atomic_inc (&kibnal_data.kib_nconns);
1316         /* well not really, but I call destroy() on failure, which decrements */
1317
1318         PORTAL_ALLOC (conn->ibc_rxs, IBNAL_RX_MSGS * sizeof (kib_rx_t));
1319         if (conn->ibc_rxs == NULL)
1320                 goto failed;
1321         memset (conn->ibc_rxs, 0, IBNAL_RX_MSGS * sizeof(kib_rx_t));
1322
1323         rc = kibnal_alloc_pages(&conn->ibc_rx_pages,
1324                                 IBNAL_RX_MSG_PAGES,
1325                                 IB_ACCESS_LOCAL_WRITE);
1326         if (rc != 0)
1327                 goto failed;
1328
1329         vaddr_base = vaddr = conn->ibc_rx_pages->ibp_vaddr;
1330
1331         for (i = ipage = page_offset = 0; i < IBNAL_RX_MSGS; i++) {
1332                 struct page *page = conn->ibc_rx_pages->ibp_pages[ipage];
1333                 kib_rx_t   *rx = &conn->ibc_rxs[i];
1334
1335                 rx->rx_conn = conn;
1336                 rx->rx_vaddr = vaddr;
1337                 rx->rx_msg = (kib_msg_t *)(((char *)page_address(page)) + page_offset);
1338
1339                 vaddr += IBNAL_MSG_SIZE;
1340                 LASSERT (vaddr <= vaddr_base + IBNAL_RX_MSG_BYTES);
1341
1342                 page_offset += IBNAL_MSG_SIZE;
1343                 LASSERT (page_offset <= PAGE_SIZE);
1344
1345                 if (page_offset == PAGE_SIZE) {
1346                         page_offset = 0;
1347                         ipage++;
1348                         LASSERT (ipage <= IBNAL_RX_MSG_PAGES);
1349                 }
1350         }
1351
1352         params.qp_create = (struct ib_qp_create_param) {
1353                 .limit = {
1354                         /* Sends have an optional RDMA */
1355                         .max_outstanding_send_request    = 2 * IBNAL_MSG_QUEUE_SIZE,
1356                         .max_outstanding_receive_request = IBNAL_MSG_QUEUE_SIZE,
1357                         .max_send_gather_element         = 1,
1358                         .max_receive_scatter_element     = 1,
1359                 },
1360                 .pd              = kibnal_data.kib_pd,
1361                 .send_queue      = kibnal_data.kib_cq,
1362                 .receive_queue   = kibnal_data.kib_cq,
1363                 .send_policy     = IB_WQ_SIGNAL_SELECTABLE,
1364                 .receive_policy  = IB_WQ_SIGNAL_SELECTABLE,
1365                 .rd_domain       = 0,
1366                 .transport       = IB_TRANSPORT_RC,
1367                 .device_specific = NULL,
1368         };
1369
1370         rc = ib_qp_create (&params.qp_create, &conn->ibc_qp, &conn->ibc_qpn);
1371         if (rc != 0) {
1372                 CERROR ("Failed to create queue pair: %d\n", rc);
1373                 goto failed;
1374         }
1375
1376         /* Mark QP created */
1377         conn->ibc_state = IBNAL_CONN_INIT_QP;
1378
1379         params.qp_attr = (struct ib_qp_attribute) {
1380                 .state             = IB_QP_STATE_INIT,
1381                 .port              = kibnal_data.kib_port,
1382                 .enable_rdma_read  = 1,
1383                 .enable_rdma_write = 1,
1384                 .valid_fields      = (IB_QP_ATTRIBUTE_STATE |
1385                                       IB_QP_ATTRIBUTE_PORT |
1386                                       IB_QP_ATTRIBUTE_PKEY_INDEX |
1387                                       IB_QP_ATTRIBUTE_RDMA_ATOMIC_ENABLE),
1388         };
1389         rc = ib_qp_modify(conn->ibc_qp, &params.qp_attr);
1390         if (rc != 0) {
1391                 CERROR ("Failed to modify queue pair: %d\n", rc);
1392                 goto failed;
1393         }
1394
1395         /* 1 ref for caller */
1396         atomic_set (&conn->ibc_refcount, 1);
1397         return (conn);
1398
1399  failed:
1400         kibnal_destroy_conn (conn);
1401         return (NULL);
1402 }
1403
1404 void
1405 kibnal_destroy_conn (kib_conn_t *conn)
1406 {
1407         int    rc;
1408
1409         CDEBUG (D_NET, "connection %p\n", conn);
1410
1411         LASSERT (atomic_read (&conn->ibc_refcount) == 0);
1412         LASSERT (list_empty(&conn->ibc_tx_queue));
1413         LASSERT (list_empty(&conn->ibc_active_txs));
1414         LASSERT (conn->ibc_nsends_posted == 0);
1415         LASSERT (conn->ibc_connreq == NULL);
1416
1417         switch (conn->ibc_state) {
1418         case IBNAL_CONN_ZOMBIE:
1419                 /* called after connection sequence initiated */
1420
1421         case IBNAL_CONN_INIT_QP:
1422                 rc = ib_qp_destroy(conn->ibc_qp);
1423                 if (rc != 0)
1424                         CERROR("Can't destroy QP: %d\n", rc);
1425                 /* fall through */
1426
1427         case IBNAL_CONN_INIT_NOTHING:
1428                 break;
1429
1430         default:
1431                 LASSERT (0);
1432         }
1433
1434         if (conn->ibc_rx_pages != NULL)
1435                 kibnal_free_pages(conn->ibc_rx_pages);
1436
1437         if (conn->ibc_rxs != NULL)
1438                 PORTAL_FREE(conn->ibc_rxs,
1439                             IBNAL_RX_MSGS * sizeof(kib_rx_t));
1440
1441         if (conn->ibc_peer != NULL)
1442                 kibnal_put_peer(conn->ibc_peer);
1443
1444         PORTAL_FREE(conn, sizeof (*conn));
1445
1446         atomic_dec(&kibnal_data.kib_nconns);
1447
1448         if (atomic_read (&kibnal_data.kib_nconns) == 0 &&
1449             kibnal_data.kib_shutdown) {
1450                 /* I just nuked the last connection on shutdown; wake up
1451                  * everyone so they can exit. */
1452                 wake_up_all(&kibnal_data.kib_sched_waitq);
1453                 wake_up_all(&kibnal_data.kib_reaper_waitq);
1454         }
1455 }
1456
1457 void
1458 kibnal_put_conn (kib_conn_t *conn)
1459 {
1460         unsigned long flags;
1461
1462         CDEBUG (D_NET, "putting conn[%p] state %d -> "LPX64" (%d)\n",
1463                 conn, conn->ibc_state, conn->ibc_peer->ibp_nid,
1464                 atomic_read (&conn->ibc_refcount));
1465
1466         LASSERT (atomic_read (&conn->ibc_refcount) > 0);
1467         if (!atomic_dec_and_test (&conn->ibc_refcount))
1468                 return;
1469
1470         /* last ref only goes on zombies */
1471         LASSERT (conn->ibc_state == IBNAL_CONN_ZOMBIE);
1472
1473         spin_lock_irqsave (&kibnal_data.kib_reaper_lock, flags);
1474
1475         list_add (&conn->ibc_list, &kibnal_data.kib_reaper_conns);
1476         wake_up (&kibnal_data.kib_reaper_waitq);
1477
1478         spin_unlock_irqrestore (&kibnal_data.kib_reaper_lock, flags);
1479 }
1480
1481 int
1482 kibnal_close_peer_conns_locked (kib_peer_t *peer, int why)
1483 {
1484         kib_conn_t         *conn;
1485         struct list_head   *ctmp;
1486         struct list_head   *cnxt;
1487         int                 count = 0;
1488
1489         list_for_each_safe (ctmp, cnxt, &peer->ibp_conns) {
1490                 conn = list_entry (ctmp, kib_conn_t, ibc_list);
1491
1492                 count++;
1493                 kibnal_close_conn_locked (conn, why);
1494         }
1495
1496         return (count);
1497 }
1498
1499 int
1500 kibnal_close_stale_conns_locked (kib_peer_t *peer, __u64 incarnation)
1501 {
1502         kib_conn_t         *conn;
1503         struct list_head   *ctmp;
1504         struct list_head   *cnxt;
1505         int                 count = 0;
1506
1507         list_for_each_safe (ctmp, cnxt, &peer->ibp_conns) {
1508                 conn = list_entry (ctmp, kib_conn_t, ibc_list);
1509
1510                 if (conn->ibc_incarnation == incarnation)
1511                         continue;
1512
1513                 CDEBUG(D_NET, "Closing stale conn nid:"LPX64" incarnation:"LPX64"("LPX64")\n",
1514                        peer->ibp_nid, conn->ibc_incarnation, incarnation);
1515
1516                 count++;
1517                 kibnal_close_conn_locked (conn, -ESTALE);
1518         }
1519
1520         return (count);
1521 }
1522
1523 int
1524 kibnal_close_matching_conns (ptl_nid_t nid)
1525 {
1526         unsigned long       flags;
1527         kib_peer_t         *peer;
1528         struct list_head   *ptmp;
1529         struct list_head   *pnxt;
1530         int                 lo;
1531         int                 hi;
1532         int                 i;
1533         int                 count = 0;
1534
1535         write_lock_irqsave (&kibnal_data.kib_global_lock, flags);
1536
1537         if (nid != PTL_NID_ANY)
1538                 lo = hi = kibnal_nid2peerlist(nid) - kibnal_data.kib_peers;
1539         else {
1540                 lo = 0;
1541                 hi = kibnal_data.kib_peer_hash_size - 1;
1542         }
1543
1544         for (i = lo; i <= hi; i++) {
1545                 list_for_each_safe (ptmp, pnxt, &kibnal_data.kib_peers[i]) {
1546
1547                         peer = list_entry (ptmp, kib_peer_t, ibp_list);
1548                         LASSERT (peer->ibp_persistence != 0 ||
1549                                  peer->ibp_connecting != 0 ||
1550                                  !list_empty (&peer->ibp_conns));
1551
1552                         if (!(nid == PTL_NID_ANY || nid == peer->ibp_nid))
1553                                 continue;
1554
1555                         count += kibnal_close_peer_conns_locked (peer, 0);
1556                 }
1557         }
1558
1559         write_unlock_irqrestore (&kibnal_data.kib_global_lock, flags);
1560
1561         /* wildcards always succeed */
1562         if (nid == PTL_NID_ANY)
1563                 return (0);
1564
1565         return (count == 0 ? -ENOENT : 0);
1566 }
1567
1568 int
1569 kibnal_cmd(struct portals_cfg *pcfg, void * private)
1570 {
1571         int rc = -EINVAL;
1572
1573         LASSERT (pcfg != NULL);
1574
1575         switch(pcfg->pcfg_command) {
1576         case NAL_CMD_GET_PEER: {
1577                 ptl_nid_t   nid = 0;
1578                 __u32       ip = 0;
1579                 int         port = 0;
1580                 int         share_count = 0;
1581
1582                 rc = kibnal_get_peer_info(pcfg->pcfg_count,
1583                                           &nid, &ip, &port, &share_count);
1584                 pcfg->pcfg_nid   = nid;
1585                 pcfg->pcfg_size  = 0;
1586                 pcfg->pcfg_id    = ip;
1587                 pcfg->pcfg_misc  = port;
1588                 pcfg->pcfg_count = 0;
1589                 pcfg->pcfg_wait  = share_count;
1590                 break;
1591         }
1592         case NAL_CMD_ADD_PEER: {
1593                 rc = kibnal_add_persistent_peer (pcfg->pcfg_nid,
1594                                                  pcfg->pcfg_id, /* IP */
1595                                                  pcfg->pcfg_misc); /* port */
1596                 break;
1597         }
1598         case NAL_CMD_DEL_PEER: {
1599                 rc = kibnal_del_peer (pcfg->pcfg_nid,
1600                                        /* flags == single_share */
1601                                        pcfg->pcfg_flags != 0);
1602                 break;
1603         }
1604         case NAL_CMD_GET_CONN: {
1605                 kib_conn_t *conn = kibnal_get_conn_by_idx (pcfg->pcfg_count);
1606
1607                 if (conn == NULL)
1608                         rc = -ENOENT;
1609                 else {
1610                         rc = 0;
1611                         pcfg->pcfg_nid   = conn->ibc_peer->ibp_nid;
1612                         pcfg->pcfg_id    = 0;
1613                         pcfg->pcfg_misc  = 0;
1614                         pcfg->pcfg_flags = 0;
1615                         kibnal_put_conn (conn);
1616                 }
1617                 break;
1618         }
1619         case NAL_CMD_CLOSE_CONNECTION: {
1620                 rc = kibnal_close_matching_conns (pcfg->pcfg_nid);
1621                 break;
1622         }
1623         case NAL_CMD_REGISTER_MYNID: {
1624                 if (pcfg->pcfg_nid == PTL_NID_ANY)
1625                         rc = -EINVAL;
1626                 else
1627                         rc = kibnal_set_mynid (pcfg->pcfg_nid);
1628                 break;
1629         }
1630         }
1631
1632         return rc;
1633 }
1634
1635 void
1636 kibnal_free_pages (kib_pages_t *p)
1637 {
1638         int     npages = p->ibp_npages;
1639         int     rc;
1640         int     i;
1641
1642         if (p->ibp_mapped) {
1643                 rc = ib_memory_deregister(p->ibp_handle);
1644                 if (rc != 0)
1645                         CERROR ("Deregister error: %d\n", rc);
1646         }
1647
1648         for (i = 0; i < npages; i++)
1649                 if (p->ibp_pages[i] != NULL)
1650                         __free_page(p->ibp_pages[i]);
1651
1652         PORTAL_FREE (p, offsetof(kib_pages_t, ibp_pages[npages]));
1653 }
1654
1655 int
1656 kibnal_alloc_pages (kib_pages_t **pp, int npages, int access)
1657 {
1658         kib_pages_t                *p;
1659         struct ib_physical_buffer  *phys_pages;
1660         int                         i;
1661         int                         rc;
1662
1663         PORTAL_ALLOC(p, offsetof(kib_pages_t, ibp_pages[npages]));
1664         if (p == NULL) {
1665                 CERROR ("Can't allocate buffer %d\n", npages);
1666                 return (-ENOMEM);
1667         }
1668
1669         memset (p, 0, offsetof(kib_pages_t, ibp_pages[npages]));
1670         p->ibp_npages = npages;
1671
1672         for (i = 0; i < npages; i++) {
1673                 p->ibp_pages[i] = alloc_page (GFP_KERNEL);
1674                 if (p->ibp_pages[i] == NULL) {
1675                         CERROR ("Can't allocate page %d of %d\n", i, npages);
1676                         kibnal_free_pages(p);
1677                         return (-ENOMEM);
1678                 }
1679         }
1680
1681         PORTAL_ALLOC(phys_pages, npages * sizeof(*phys_pages));
1682         if (phys_pages == NULL) {
1683                 CERROR ("Can't allocate physarray for %d pages\n", npages);
1684                 kibnal_free_pages(p);
1685                 return (-ENOMEM);
1686         }
1687
1688         for (i = 0; i < npages; i++) {
1689                 phys_pages[i].size = PAGE_SIZE;
1690                 phys_pages[i].address =
1691                         kibnal_page2phys(p->ibp_pages[i]);
1692         }
1693
1694         p->ibp_vaddr = 0;
1695         rc = ib_memory_register_physical(kibnal_data.kib_pd,
1696                                          phys_pages, npages,
1697                                          &p->ibp_vaddr,
1698                                          npages * PAGE_SIZE, 0,
1699                                          access,
1700                                          &p->ibp_handle,
1701                                          &p->ibp_lkey,
1702                                          &p->ibp_rkey);
1703
1704         PORTAL_FREE(phys_pages, npages * sizeof(*phys_pages));
1705
1706         if (rc != 0) {
1707                 CERROR ("Error %d mapping %d pages\n", rc, npages);
1708                 kibnal_free_pages(p);
1709                 return (rc);
1710         }
1711
1712         p->ibp_mapped = 1;
1713         *pp = p;
1714         return (0);
1715 }
1716
1717 int
1718 kibnal_setup_tx_descs (void)
1719 {
1720         int           ipage = 0;
1721         int           page_offset = 0;
1722         __u64         vaddr;
1723         __u64         vaddr_base;
1724         struct page  *page;
1725         kib_tx_t     *tx;
1726         int           i;
1727         int           rc;
1728
1729         /* pre-mapped messages are not bigger than 1 page */
1730         LASSERT (IBNAL_MSG_SIZE <= PAGE_SIZE);
1731
1732         /* No fancy arithmetic when we do the buffer calculations */
1733         LASSERT (PAGE_SIZE % IBNAL_MSG_SIZE == 0);
1734
1735         rc = kibnal_alloc_pages(&kibnal_data.kib_tx_pages,
1736                                 IBNAL_TX_MSG_PAGES,
1737                                 0);            /* local read access only */
1738         if (rc != 0)
1739                 return (rc);
1740
1741         vaddr = vaddr_base = kibnal_data.kib_tx_pages->ibp_vaddr;
1742
1743         for (i = 0; i < IBNAL_TX_MSGS; i++) {
1744                 page = kibnal_data.kib_tx_pages->ibp_pages[ipage];
1745                 tx = &kibnal_data.kib_tx_descs[i];
1746
1747                 memset (tx, 0, sizeof(*tx));    /* zero flags etc */
1748
1749                 tx->tx_msg = (kib_msg_t *)(((char *)page_address(page)) + page_offset);
1750                 tx->tx_vaddr = vaddr;
1751                 tx->tx_isnblk = (i >= IBNAL_NTX);
1752                 tx->tx_mapped = KIB_TX_UNMAPPED;
1753
1754                 CDEBUG(D_NET, "Tx[%d] %p->%p - "LPX64"\n",
1755                        i, tx, tx->tx_msg, tx->tx_vaddr);
1756
1757                 if (tx->tx_isnblk)
1758                         list_add (&tx->tx_list,
1759                                   &kibnal_data.kib_idle_nblk_txs);
1760                 else
1761                         list_add (&tx->tx_list,
1762                                   &kibnal_data.kib_idle_txs);
1763
1764                 vaddr += IBNAL_MSG_SIZE;
1765                 LASSERT (vaddr <= vaddr_base + IBNAL_TX_MSG_BYTES);
1766
1767                 page_offset += IBNAL_MSG_SIZE;
1768                 LASSERT (page_offset <= PAGE_SIZE);
1769
1770                 if (page_offset == PAGE_SIZE) {
1771                         page_offset = 0;
1772                         ipage++;
1773                         LASSERT (ipage <= IBNAL_TX_MSG_PAGES);
1774                 }
1775         }
1776
1777         return (0);
1778 }
1779
1780 void
1781 kibnal_api_shutdown (nal_t *nal)
1782 {
1783         int   i;
1784         int   rc;
1785
1786         if (nal->nal_refct != 0) {
1787                 /* This module got the first ref */
1788                 PORTAL_MODULE_UNUSE;
1789                 return;
1790         }
1791
1792         CDEBUG(D_MALLOC, "before NAL cleanup: kmem %d\n",
1793                atomic_read (&portal_kmemory));
1794
1795         LASSERT(nal == &kibnal_api);
1796
1797         switch (kibnal_data.kib_init) {
1798         default:
1799                 CERROR ("Unexpected state %d\n", kibnal_data.kib_init);
1800                 LBUG();
1801
1802         case IBNAL_INIT_ALL:
1803                 /* stop calls to nal_cmd */
1804                 libcfs_nal_cmd_unregister(OPENIBNAL);
1805                 /* No new peers */
1806
1807                 /* resetting my NID unadvertises me, removes my
1808                  * listener and nukes all current peers */
1809                 kibnal_set_mynid (PTL_NID_ANY);
1810
1811                 /* Wait for all peer state to clean up */
1812                 i = 2;
1813                 while (atomic_read (&kibnal_data.kib_npeers) != 0) {
1814                         i++;
1815                         CDEBUG(((i & (-i)) == i) ? D_WARNING : D_NET, /* power of 2? */
1816                                "waiting for %d peers to close down\n",
1817                                atomic_read (&kibnal_data.kib_npeers));
1818                         set_current_state (TASK_INTERRUPTIBLE);
1819                         schedule_timeout (HZ);
1820                 }
1821                 /* fall through */
1822
1823         case IBNAL_INIT_CQ:
1824                 rc = ib_cq_destroy (kibnal_data.kib_cq);
1825                 if (rc != 0)
1826                         CERROR ("Destroy CQ error: %d\n", rc);
1827                 /* fall through */
1828
1829         case IBNAL_INIT_TXD:
1830                 kibnal_free_pages (kibnal_data.kib_tx_pages);
1831                 /* fall through */
1832 #if IBNAL_FMR
1833         case IBNAL_INIT_FMR:
1834                 rc = ib_fmr_pool_destroy (kibnal_data.kib_fmr_pool);
1835                 if (rc != 0)
1836                         CERROR ("Destroy FMR pool error: %d\n", rc);
1837                 /* fall through */
1838 #endif
1839         case IBNAL_INIT_PD:
1840                 rc = ib_pd_destroy(kibnal_data.kib_pd);
1841                 if (rc != 0)
1842                         CERROR ("Destroy PD error: %d\n", rc);
1843                 /* fall through */
1844
1845         case IBNAL_INIT_LIB:
1846                 lib_fini(&kibnal_lib);
1847                 /* fall through */
1848
1849         case IBNAL_INIT_DATA:
1850                 /* Module refcount only gets to zero when all peers
1851                  * have been closed so all lists must be empty */
1852                 LASSERT (atomic_read (&kibnal_data.kib_npeers) == 0);
1853                 LASSERT (kibnal_data.kib_peers != NULL);
1854                 for (i = 0; i < kibnal_data.kib_peer_hash_size; i++) {
1855                         LASSERT (list_empty (&kibnal_data.kib_peers[i]));
1856                 }
1857                 LASSERT (atomic_read (&kibnal_data.kib_nconns) == 0);
1858                 LASSERT (list_empty (&kibnal_data.kib_sched_rxq));
1859                 LASSERT (list_empty (&kibnal_data.kib_sched_txq));
1860                 LASSERT (list_empty (&kibnal_data.kib_reaper_conns));
1861                 LASSERT (list_empty (&kibnal_data.kib_connd_peers));
1862                 LASSERT (list_empty (&kibnal_data.kib_connd_acceptq));
1863
1864                 /* flag threads to terminate; wake and wait for them to die */
1865                 kibnal_data.kib_shutdown = 1;
1866                 wake_up_all (&kibnal_data.kib_sched_waitq);
1867                 wake_up_all (&kibnal_data.kib_reaper_waitq);
1868                 wake_up_all (&kibnal_data.kib_connd_waitq);
1869
1870                 i = 2;
1871                 while (atomic_read (&kibnal_data.kib_nthreads) != 0) {
1872                         i++;
1873                         CDEBUG(((i & (-i)) == i) ? D_WARNING : D_NET, /* power of 2? */
1874                                "Waiting for %d threads to terminate\n",
1875                                atomic_read (&kibnal_data.kib_nthreads));
1876                         set_current_state (TASK_INTERRUPTIBLE);
1877                         schedule_timeout (HZ);
1878                 }
1879                 /* fall through */
1880
1881         case IBNAL_INIT_NOTHING:
1882                 break;
1883         }
1884
1885         if (kibnal_data.kib_tx_descs != NULL)
1886                 PORTAL_FREE (kibnal_data.kib_tx_descs,
1887                              IBNAL_TX_MSGS * sizeof(kib_tx_t));
1888
1889         if (kibnal_data.kib_peers != NULL)
1890                 PORTAL_FREE (kibnal_data.kib_peers,
1891                              sizeof (struct list_head) *
1892                              kibnal_data.kib_peer_hash_size);
1893
1894         CDEBUG(D_MALLOC, "after NAL cleanup: kmem %d\n",
1895                atomic_read (&portal_kmemory));
1896         printk(KERN_INFO "Lustre: OpenIB NAL unloaded (final mem %d)\n",
1897                atomic_read(&portal_kmemory));
1898
1899         kibnal_data.kib_init = IBNAL_INIT_NOTHING;
1900 }
1901
1902 int
1903 kibnal_api_startup (nal_t *nal, ptl_pid_t requested_pid,
1904                      ptl_ni_limits_t *requested_limits,
1905                      ptl_ni_limits_t *actual_limits)
1906 {
1907         struct timeval    tv;
1908         ptl_process_id_t  process_id;
1909         int               pkmem = atomic_read(&portal_kmemory);
1910         int               rc;
1911         int               i;
1912
1913         LASSERT (nal == &kibnal_api);
1914
1915         if (nal->nal_refct != 0) {
1916                 if (actual_limits != NULL)
1917                         *actual_limits = kibnal_lib.libnal_ni.ni_actual_limits;
1918                 /* This module got the first ref */
1919                 PORTAL_MODULE_USE;
1920                 return (PTL_OK);
1921         }
1922
1923         LASSERT (kibnal_data.kib_init == IBNAL_INIT_NOTHING);
1924
1925         memset (&kibnal_data, 0, sizeof (kibnal_data)); /* zero pointers, flags etc */
1926
1927         do_gettimeofday(&tv);
1928         kibnal_data.kib_incarnation = (((__u64)tv.tv_sec) * 1000000) + tv.tv_usec;
1929
1930         init_MUTEX (&kibnal_data.kib_nid_mutex);
1931         init_MUTEX_LOCKED (&kibnal_data.kib_listener_signal);
1932
1933         rwlock_init(&kibnal_data.kib_global_lock);
1934
1935         kibnal_data.kib_peer_hash_size = IBNAL_PEER_HASH_SIZE;
1936         PORTAL_ALLOC (kibnal_data.kib_peers,
1937                       sizeof (struct list_head) * kibnal_data.kib_peer_hash_size);
1938         if (kibnal_data.kib_peers == NULL) {
1939                 goto failed;
1940         }
1941         for (i = 0; i < kibnal_data.kib_peer_hash_size; i++)
1942                 INIT_LIST_HEAD(&kibnal_data.kib_peers[i]);
1943
1944         spin_lock_init (&kibnal_data.kib_reaper_lock);
1945         INIT_LIST_HEAD (&kibnal_data.kib_reaper_conns);
1946         init_waitqueue_head (&kibnal_data.kib_reaper_waitq);
1947
1948         spin_lock_init (&kibnal_data.kib_connd_lock);
1949         INIT_LIST_HEAD (&kibnal_data.kib_connd_acceptq);
1950         INIT_LIST_HEAD (&kibnal_data.kib_connd_peers);
1951         init_waitqueue_head (&kibnal_data.kib_connd_waitq);
1952
1953         spin_lock_init (&kibnal_data.kib_sched_lock);
1954         INIT_LIST_HEAD (&kibnal_data.kib_sched_txq);
1955         INIT_LIST_HEAD (&kibnal_data.kib_sched_rxq);
1956         init_waitqueue_head (&kibnal_data.kib_sched_waitq);
1957
1958         spin_lock_init (&kibnal_data.kib_tx_lock);
1959         INIT_LIST_HEAD (&kibnal_data.kib_idle_txs);
1960         INIT_LIST_HEAD (&kibnal_data.kib_idle_nblk_txs);
1961         init_waitqueue_head(&kibnal_data.kib_idle_tx_waitq);
1962
1963         PORTAL_ALLOC (kibnal_data.kib_tx_descs,
1964                       IBNAL_TX_MSGS * sizeof(kib_tx_t));
1965         if (kibnal_data.kib_tx_descs == NULL) {
1966                 CERROR ("Can't allocate tx descs\n");
1967                 goto failed;
1968         }
1969
1970         /* lists/ptrs/locks initialised */
1971         kibnal_data.kib_init = IBNAL_INIT_DATA;
1972         /*****************************************************/
1973
1974
1975         process_id.pid = requested_pid;
1976         process_id.nid = PTL_NID_ANY;           /* don't know my NID yet */
1977
1978         rc = lib_init(&kibnal_lib, nal, process_id,
1979                       requested_limits, actual_limits);
1980         if (rc != PTL_OK) {
1981                 CERROR("lib_init failed: error %d\n", rc);
1982                 goto failed;
1983         }
1984
1985         /* lib interface initialised */
1986         kibnal_data.kib_init = IBNAL_INIT_LIB;
1987         /*****************************************************/
1988
1989         for (i = 0; i < IBNAL_N_SCHED; i++) {
1990                 rc = kibnal_thread_start (kibnal_scheduler,
1991                                           (void *)((unsigned long)i));
1992                 if (rc != 0) {
1993                         CERROR("Can't spawn openibnal scheduler[%d]: %d\n",
1994                                i, rc);
1995                         goto failed;
1996                 }
1997         }
1998
1999         for (i = 0; i < IBNAL_N_CONND; i++) {
2000                 rc = kibnal_thread_start (kibnal_connd,
2001                                           (void *)((unsigned long)i));
2002                 if (rc != 0) {
2003                         CERROR("Can't spawn openibnal connd[%d]: %d\n",
2004                                i, rc);
2005                         goto failed;
2006                 }
2007         }
2008
2009         rc = kibnal_thread_start (kibnal_reaper, NULL);
2010         if (rc != 0) {
2011                 CERROR ("Can't spawn openibnal reaper: %d\n", rc);
2012                 goto failed;
2013         }
2014
2015         kibnal_data.kib_device = ib_device_get_by_index(0);
2016         if (kibnal_data.kib_device == NULL) {
2017                 CERROR ("Can't open ib device 0\n");
2018                 goto failed;
2019         }
2020
2021         rc = ib_device_properties_get(kibnal_data.kib_device,
2022                                       &kibnal_data.kib_device_props);
2023         if (rc != 0) {
2024                 CERROR ("Can't get device props: %d\n", rc);
2025                 goto failed;
2026         }
2027
2028         CDEBUG(D_NET, "Max Initiator: %d Max Responder %d\n",
2029                kibnal_data.kib_device_props.max_initiator_per_qp,
2030                kibnal_data.kib_device_props.max_responder_per_qp);
2031
2032         kibnal_data.kib_port = 0;
2033         for (i = 1; i <= 2; i++) {
2034                 rc = ib_port_properties_get(kibnal_data.kib_device, i,
2035                                             &kibnal_data.kib_port_props);
2036                 if (rc == 0) {
2037                         kibnal_data.kib_port = i;
2038                         break;
2039                 }
2040         }
2041         if (kibnal_data.kib_port == 0) {
2042                 CERROR ("Can't find a port\n");
2043                 goto failed;
2044         }
2045
2046         rc = ib_pd_create(kibnal_data.kib_device,
2047                           NULL, &kibnal_data.kib_pd);
2048         if (rc != 0) {
2049                 CERROR ("Can't create PD: %d\n", rc);
2050                 goto failed;
2051         }
2052
2053         /* flag PD initialised */
2054         kibnal_data.kib_init = IBNAL_INIT_PD;
2055         /*****************************************************/
2056 #if IBNAL_FMR
2057         {
2058                 const int pool_size = IBNAL_NTX + IBNAL_NTX_NBLK;
2059                 struct ib_fmr_pool_param params = {
2060                         .max_pages_per_fmr = PTL_MTU/PAGE_SIZE,
2061                         .access            = (IB_ACCESS_LOCAL_WRITE |
2062                                               IB_ACCESS_REMOTE_WRITE |
2063                                               IB_ACCESS_REMOTE_READ),
2064                         .pool_size         = pool_size,
2065                         .dirty_watermark   = (pool_size * 3)/4,
2066                         .flush_function    = NULL,
2067                         .flush_arg         = NULL,
2068                         .cache             = 1,
2069                 };
2070                 rc = ib_fmr_pool_create(kibnal_data.kib_pd, &params,
2071                                         &kibnal_data.kib_fmr_pool);
2072                 if (rc != 0) {
2073                         CERROR ("Can't create FMR pool size %d: %d\n",
2074                                 pool_size, rc);
2075                         goto failed;
2076                 }
2077         }
2078
2079         /* flag FMR pool initialised */
2080         kibnal_data.kib_init = IBNAL_INIT_FMR;
2081 #endif
2082         /*****************************************************/
2083
2084         rc = kibnal_setup_tx_descs();
2085         if (rc != 0) {
2086                 CERROR ("Can't register tx descs: %d\n", rc);
2087                 goto failed;
2088         }
2089
2090         /* flag TX descs initialised */
2091         kibnal_data.kib_init = IBNAL_INIT_TXD;
2092         /*****************************************************/
2093
2094         {
2095                 struct ib_cq_callback callback = {
2096                         .context        = IBNAL_CALLBACK_CTXT,
2097                         .policy         = IB_CQ_PROVIDER_REARM,
2098                         .function       = {
2099                                 .entry  = kibnal_callback,
2100                         },
2101                         .arg            = NULL,
2102                 };
2103                 int  nentries = IBNAL_CQ_ENTRIES;
2104
2105                 rc = ib_cq_create (kibnal_data.kib_device,
2106                                    &nentries, &callback, NULL,
2107                                    &kibnal_data.kib_cq);
2108                 if (rc != 0) {
2109                         CERROR ("Can't create CQ: %d\n", rc);
2110                         goto failed;
2111                 }
2112
2113                 /* I only want solicited events */
2114                 rc = ib_cq_request_notification(kibnal_data.kib_cq, 1);
2115                 LASSERT (rc == 0);
2116         }
2117
2118         /* flag CQ initialised */
2119         kibnal_data.kib_init = IBNAL_INIT_CQ;
2120         /*****************************************************/
2121
2122         rc = libcfs_nal_cmd_register(OPENIBNAL, &kibnal_cmd, NULL);
2123         if (rc != 0) {
2124                 CERROR ("Can't initialise command interface (rc = %d)\n", rc);
2125                 goto failed;
2126         }
2127
2128         /* flag everything initialised */
2129         kibnal_data.kib_init = IBNAL_INIT_ALL;
2130         /*****************************************************/
2131
2132         printk(KERN_INFO "Lustre: OpenIB NAL loaded "
2133                "(initial mem %d)\n", pkmem);
2134
2135         return (PTL_OK);
2136
2137  failed:
2138         kibnal_api_shutdown (&kibnal_api);
2139         return (PTL_FAIL);
2140 }
2141
2142 void __exit
2143 kibnal_module_fini (void)
2144 {
2145         if (kibnal_tunables.kib_sysctl != NULL)
2146                 unregister_sysctl_table (kibnal_tunables.kib_sysctl);
2147         PtlNIFini(kibnal_ni);
2148
2149         ptl_unregister_nal(OPENIBNAL);
2150 }
2151
2152 int __init
2153 kibnal_module_init (void)
2154 {
2155         int    rc;
2156
2157         /* the following must be sizeof(int) for proc_dointvec() */
2158         LASSERT (sizeof(kibnal_tunables.kib_io_timeout) == sizeof(int));
2159         LASSERT (sizeof(kibnal_tunables.kib_listener_timeout) == sizeof(int));
2160         LASSERT (sizeof(kibnal_tunables.kib_backlog) == sizeof(int));
2161         LASSERT (sizeof(kibnal_tunables.kib_port) == sizeof(int));
2162
2163         kibnal_api.nal_ni_init = kibnal_api_startup;
2164         kibnal_api.nal_ni_fini = kibnal_api_shutdown;
2165
2166         /* Initialise dynamic tunables to defaults once only */
2167         kibnal_tunables.kib_io_timeout = IBNAL_IO_TIMEOUT;
2168         kibnal_tunables.kib_listener_timeout = IBNAL_LISTENER_TIMEOUT;
2169         kibnal_tunables.kib_backlog = IBNAL_BACKLOG;
2170         kibnal_tunables.kib_port = IBNAL_PORT;
2171
2172         rc = ptl_register_nal(OPENIBNAL, &kibnal_api);
2173         if (rc != PTL_OK) {
2174                 CERROR("Can't register IBNAL: %d\n", rc);
2175                 return (-ENOMEM);               /* or something... */
2176         }
2177
2178         /* Pure gateways want the NAL started up at module load time... */
2179         rc = PtlNIInit(OPENIBNAL, LUSTRE_SRV_PTL_PID, NULL, NULL, &kibnal_ni);
2180         if (rc != PTL_OK && rc != PTL_IFACE_DUP) {
2181                 ptl_unregister_nal(OPENIBNAL);
2182                 return (-ENODEV);
2183         }
2184
2185         kibnal_tunables.kib_sysctl =
2186                 register_sysctl_table (kibnal_top_ctl_table, 0);
2187         if (kibnal_tunables.kib_sysctl == NULL) {
2188                 CERROR("Can't register sysctl table\n");
2189                 PtlNIFini(kibnal_ni);
2190                 ptl_unregister_nal(OPENIBNAL);
2191                 return (-ENOMEM);
2192         }
2193
2194         return (0);
2195 }
2196
2197 MODULE_AUTHOR("Cluster File Systems, Inc. <info@clusterfs.com>");
2198 MODULE_DESCRIPTION("Kernel OpenIB NAL v0.01");
2199 MODULE_LICENSE("GPL");
2200
2201 module_init(kibnal_module_init);
2202 module_exit(kibnal_module_fini);
2203