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
1079         read_lock (&kibnal_data.kib_global_lock);
1080         peer = kibnal_find_peer_locked (nid);
1081         if (peer != NULL)                       /* +1 ref for caller? */
1082                 atomic_inc (&peer->ibp_refcount);
1083         read_unlock (&kibnal_data.kib_global_lock);
1084
1085         return (peer);
1086 }
1087
1088 void
1089 kibnal_unlink_peer_locked (kib_peer_t *peer)
1090 {
1091         LASSERT (peer->ibp_persistence == 0);
1092         LASSERT (list_empty(&peer->ibp_conns));
1093
1094         LASSERT (kibnal_peer_active(peer));
1095         list_del_init (&peer->ibp_list);
1096         /* lose peerlist's ref */
1097         kibnal_put_peer (peer);
1098 }
1099
1100 int
1101 kibnal_get_peer_info (int index, ptl_nid_t *nidp, __u32 *ipp, int *portp,
1102                       int *persistencep)
1103 {
1104         kib_peer_t        *peer;
1105         struct list_head  *ptmp;
1106         int                i;
1107
1108         read_lock (&kibnal_data.kib_global_lock);
1109
1110         for (i = 0; i < kibnal_data.kib_peer_hash_size; i++) {
1111
1112                 list_for_each (ptmp, &kibnal_data.kib_peers[i]) {
1113
1114                         peer = list_entry (ptmp, kib_peer_t, ibp_list);
1115                         LASSERT (peer->ibp_persistence != 0 ||
1116                                  peer->ibp_connecting != 0 ||
1117                                  !list_empty (&peer->ibp_conns));
1118
1119                         if (index-- > 0)
1120                                 continue;
1121
1122                         *nidp = peer->ibp_nid;
1123                         *ipp = peer->ibp_ip;
1124                         *portp = peer->ibp_port;
1125                         *persistencep = peer->ibp_persistence;
1126
1127                         read_unlock (&kibnal_data.kib_global_lock);
1128                         return (0);
1129                 }
1130         }
1131
1132         read_unlock (&kibnal_data.kib_global_lock);
1133         return (-ENOENT);
1134 }
1135
1136 int
1137 kibnal_add_persistent_peer (ptl_nid_t nid, __u32 ip, int port)
1138 {
1139         unsigned long      flags;
1140         kib_peer_t        *peer;
1141         kib_peer_t        *peer2;
1142
1143         if (nid == PTL_NID_ANY)
1144                 return (-EINVAL);
1145
1146         peer = kibnal_create_peer (nid);
1147         if (peer == NULL)
1148                 return (-ENOMEM);
1149
1150         write_lock_irqsave (&kibnal_data.kib_global_lock, flags);
1151
1152         peer2 = kibnal_find_peer_locked (nid);
1153         if (peer2 != NULL) {
1154                 kibnal_put_peer (peer);
1155                 peer = peer2;
1156         } else {
1157                 /* peer table takes existing ref on peer */
1158                 list_add_tail (&peer->ibp_list,
1159                                kibnal_nid2peerlist (nid));
1160         }
1161
1162         peer->ibp_ip = ip;
1163         peer->ibp_port = port;
1164         peer->ibp_persistence++;
1165
1166         write_unlock_irqrestore (&kibnal_data.kib_global_lock, flags);
1167         return (0);
1168 }
1169
1170 void
1171 kibnal_del_peer_locked (kib_peer_t *peer, int single_share)
1172 {
1173         struct list_head *ctmp;
1174         struct list_head *cnxt;
1175         kib_conn_t       *conn;
1176
1177         if (!single_share)
1178                 peer->ibp_persistence = 0;
1179         else if (peer->ibp_persistence > 0)
1180                 peer->ibp_persistence--;
1181
1182         if (peer->ibp_persistence != 0)
1183                 return;
1184
1185         if (list_empty(&peer->ibp_conns)) {
1186                 kibnal_unlink_peer_locked(peer);
1187         } else {
1188                 list_for_each_safe (ctmp, cnxt, &peer->ibp_conns) {
1189                         conn = list_entry(ctmp, kib_conn_t, ibc_list);
1190
1191                         kibnal_close_conn_locked (conn, 0);
1192                 }
1193                 /* NB peer is no longer persistent; closing its last conn
1194                  * unlinked it. */
1195         }
1196         /* NB peer now unlinked; might even be freed if the peer table had the
1197          * last ref on it. */
1198 }
1199
1200 int
1201 kibnal_del_peer (ptl_nid_t nid, int single_share)
1202 {
1203         unsigned long      flags;
1204         struct list_head  *ptmp;
1205         struct list_head  *pnxt;
1206         kib_peer_t        *peer;
1207         int                lo;
1208         int                hi;
1209         int                i;
1210         int                rc = -ENOENT;
1211
1212         write_lock_irqsave (&kibnal_data.kib_global_lock, flags);
1213
1214         if (nid != PTL_NID_ANY)
1215                 lo = hi = kibnal_nid2peerlist(nid) - kibnal_data.kib_peers;
1216         else {
1217                 lo = 0;
1218                 hi = kibnal_data.kib_peer_hash_size - 1;
1219         }
1220
1221         for (i = lo; i <= hi; i++) {
1222                 list_for_each_safe (ptmp, pnxt, &kibnal_data.kib_peers[i]) {
1223                         peer = list_entry (ptmp, kib_peer_t, ibp_list);
1224                         LASSERT (peer->ibp_persistence != 0 ||
1225                                  peer->ibp_connecting != 0 ||
1226                                  !list_empty (&peer->ibp_conns));
1227
1228                         if (!(nid == PTL_NID_ANY || peer->ibp_nid == nid))
1229                                 continue;
1230
1231                         kibnal_del_peer_locked (peer, single_share);
1232                         rc = 0;         /* matched something */
1233
1234                         if (single_share)
1235                                 goto out;
1236                 }
1237         }
1238  out:
1239         write_unlock_irqrestore (&kibnal_data.kib_global_lock, flags);
1240
1241         return (rc);
1242 }
1243
1244 kib_conn_t *
1245 kibnal_get_conn_by_idx (int index)
1246 {
1247         kib_peer_t        *peer;
1248         struct list_head  *ptmp;
1249         kib_conn_t        *conn;
1250         struct list_head  *ctmp;
1251         int                i;
1252
1253         read_lock (&kibnal_data.kib_global_lock);
1254
1255         for (i = 0; i < kibnal_data.kib_peer_hash_size; i++) {
1256                 list_for_each (ptmp, &kibnal_data.kib_peers[i]) {
1257
1258                         peer = list_entry (ptmp, kib_peer_t, ibp_list);
1259                         LASSERT (peer->ibp_persistence > 0 ||
1260                                  peer->ibp_connecting != 0 ||
1261                                  !list_empty (&peer->ibp_conns));
1262
1263                         list_for_each (ctmp, &peer->ibp_conns) {
1264                                 if (index-- > 0)
1265                                         continue;
1266
1267                                 conn = list_entry (ctmp, kib_conn_t, ibc_list);
1268                                 CDEBUG(D_NET, "++conn[%p] state %d -> "LPX64" (%d)\n",
1269                                        conn, conn->ibc_state, conn->ibc_peer->ibp_nid,
1270                                        atomic_read (&conn->ibc_refcount));
1271                                 atomic_inc (&conn->ibc_refcount);
1272                                 read_unlock (&kibnal_data.kib_global_lock);
1273                                 return (conn);
1274                         }
1275                 }
1276         }
1277
1278         read_unlock (&kibnal_data.kib_global_lock);
1279         return (NULL);
1280 }
1281
1282 kib_conn_t *
1283 kibnal_create_conn (void)
1284 {
1285         kib_conn_t  *conn;
1286         int          i;
1287         __u64        vaddr = 0;
1288         __u64        vaddr_base;
1289         int          page_offset;
1290         int          ipage;
1291         int          rc;
1292         union {
1293                 struct ib_qp_create_param  qp_create;
1294                 struct ib_qp_attribute     qp_attr;
1295         } params;
1296
1297         PORTAL_ALLOC (conn, sizeof (*conn));
1298         if (conn == NULL) {
1299                 CERROR ("Can't allocate connection\n");
1300                 return (NULL);
1301         }
1302
1303         /* zero flags, NULL pointers etc... */
1304         memset (conn, 0, sizeof (*conn));
1305
1306         INIT_LIST_HEAD (&conn->ibc_tx_queue);
1307         INIT_LIST_HEAD (&conn->ibc_active_txs);
1308         spin_lock_init (&conn->ibc_lock);
1309
1310         atomic_inc (&kibnal_data.kib_nconns);
1311         /* well not really, but I call destroy() on failure, which decrements */
1312
1313         PORTAL_ALLOC (conn->ibc_rxs, IBNAL_RX_MSGS * sizeof (kib_rx_t));
1314         if (conn->ibc_rxs == NULL)
1315                 goto failed;
1316         memset (conn->ibc_rxs, 0, IBNAL_RX_MSGS * sizeof(kib_rx_t));
1317
1318         rc = kibnal_alloc_pages(&conn->ibc_rx_pages,
1319                                 IBNAL_RX_MSG_PAGES,
1320                                 IB_ACCESS_LOCAL_WRITE);
1321         if (rc != 0)
1322                 goto failed;
1323
1324         vaddr_base = vaddr = conn->ibc_rx_pages->ibp_vaddr;
1325
1326         for (i = ipage = page_offset = 0; i < IBNAL_RX_MSGS; i++) {
1327                 struct page *page = conn->ibc_rx_pages->ibp_pages[ipage];
1328                 kib_rx_t   *rx = &conn->ibc_rxs[i];
1329
1330                 rx->rx_conn = conn;
1331                 rx->rx_vaddr = vaddr;
1332                 rx->rx_msg = (kib_msg_t *)(((char *)page_address(page)) + page_offset);
1333
1334                 vaddr += IBNAL_MSG_SIZE;
1335                 LASSERT (vaddr <= vaddr_base + IBNAL_RX_MSG_BYTES);
1336
1337                 page_offset += IBNAL_MSG_SIZE;
1338                 LASSERT (page_offset <= PAGE_SIZE);
1339
1340                 if (page_offset == PAGE_SIZE) {
1341                         page_offset = 0;
1342                         ipage++;
1343                         LASSERT (ipage <= IBNAL_RX_MSG_PAGES);
1344                 }
1345         }
1346
1347         params.qp_create = (struct ib_qp_create_param) {
1348                 .limit = {
1349                         /* Sends have an optional RDMA */
1350                         .max_outstanding_send_request    = 2 * IBNAL_MSG_QUEUE_SIZE,
1351                         .max_outstanding_receive_request = IBNAL_MSG_QUEUE_SIZE,
1352                         .max_send_gather_element         = 1,
1353                         .max_receive_scatter_element     = 1,
1354                 },
1355                 .pd              = kibnal_data.kib_pd,
1356                 .send_queue      = kibnal_data.kib_cq,
1357                 .receive_queue   = kibnal_data.kib_cq,
1358                 .send_policy     = IB_WQ_SIGNAL_SELECTABLE,
1359                 .receive_policy  = IB_WQ_SIGNAL_SELECTABLE,
1360                 .rd_domain       = 0,
1361                 .transport       = IB_TRANSPORT_RC,
1362                 .device_specific = NULL,
1363         };
1364
1365         rc = ib_qp_create (&params.qp_create, &conn->ibc_qp, &conn->ibc_qpn);
1366         if (rc != 0) {
1367                 CERROR ("Failed to create queue pair: %d\n", rc);
1368                 goto failed;
1369         }
1370
1371         /* Mark QP created */
1372         conn->ibc_state = IBNAL_CONN_INIT_QP;
1373
1374         params.qp_attr = (struct ib_qp_attribute) {
1375                 .state             = IB_QP_STATE_INIT,
1376                 .port              = kibnal_data.kib_port,
1377                 .enable_rdma_read  = 1,
1378                 .enable_rdma_write = 1,
1379                 .valid_fields      = (IB_QP_ATTRIBUTE_STATE |
1380                                       IB_QP_ATTRIBUTE_PORT |
1381                                       IB_QP_ATTRIBUTE_PKEY_INDEX |
1382                                       IB_QP_ATTRIBUTE_RDMA_ATOMIC_ENABLE),
1383         };
1384         rc = ib_qp_modify(conn->ibc_qp, &params.qp_attr);
1385         if (rc != 0) {
1386                 CERROR ("Failed to modify queue pair: %d\n", rc);
1387                 goto failed;
1388         }
1389
1390         /* 1 ref for caller */
1391         atomic_set (&conn->ibc_refcount, 1);
1392         return (conn);
1393
1394  failed:
1395         kibnal_destroy_conn (conn);
1396         return (NULL);
1397 }
1398
1399 void
1400 kibnal_destroy_conn (kib_conn_t *conn)
1401 {
1402         int    rc;
1403
1404         CDEBUG (D_NET, "connection %p\n", conn);
1405
1406         LASSERT (atomic_read (&conn->ibc_refcount) == 0);
1407         LASSERT (list_empty(&conn->ibc_tx_queue));
1408         LASSERT (list_empty(&conn->ibc_active_txs));
1409         LASSERT (conn->ibc_nsends_posted == 0);
1410         LASSERT (conn->ibc_connreq == NULL);
1411
1412         switch (conn->ibc_state) {
1413         case IBNAL_CONN_ZOMBIE:
1414                 /* called after connection sequence initiated */
1415
1416         case IBNAL_CONN_INIT_QP:
1417                 rc = ib_qp_destroy(conn->ibc_qp);
1418                 if (rc != 0)
1419                         CERROR("Can't destroy QP: %d\n", rc);
1420                 /* fall through */
1421
1422         case IBNAL_CONN_INIT_NOTHING:
1423                 break;
1424
1425         default:
1426                 LASSERT (0);
1427         }
1428
1429         if (conn->ibc_rx_pages != NULL)
1430                 kibnal_free_pages(conn->ibc_rx_pages);
1431
1432         if (conn->ibc_rxs != NULL)
1433                 PORTAL_FREE(conn->ibc_rxs,
1434                             IBNAL_RX_MSGS * sizeof(kib_rx_t));
1435
1436         if (conn->ibc_peer != NULL)
1437                 kibnal_put_peer(conn->ibc_peer);
1438
1439         PORTAL_FREE(conn, sizeof (*conn));
1440
1441         atomic_dec(&kibnal_data.kib_nconns);
1442
1443         if (atomic_read (&kibnal_data.kib_nconns) == 0 &&
1444             kibnal_data.kib_shutdown) {
1445                 /* I just nuked the last connection on shutdown; wake up
1446                  * everyone so they can exit. */
1447                 wake_up_all(&kibnal_data.kib_sched_waitq);
1448                 wake_up_all(&kibnal_data.kib_reaper_waitq);
1449         }
1450 }
1451
1452 void
1453 kibnal_put_conn (kib_conn_t *conn)
1454 {
1455         unsigned long flags;
1456
1457         CDEBUG (D_NET, "putting conn[%p] state %d -> "LPX64" (%d)\n",
1458                 conn, conn->ibc_state, conn->ibc_peer->ibp_nid,
1459                 atomic_read (&conn->ibc_refcount));
1460
1461         LASSERT (atomic_read (&conn->ibc_refcount) > 0);
1462         if (!atomic_dec_and_test (&conn->ibc_refcount))
1463                 return;
1464
1465         /* last ref only goes on zombies */
1466         LASSERT (conn->ibc_state == IBNAL_CONN_ZOMBIE);
1467
1468         spin_lock_irqsave (&kibnal_data.kib_reaper_lock, flags);
1469
1470         list_add (&conn->ibc_list, &kibnal_data.kib_reaper_conns);
1471         wake_up (&kibnal_data.kib_reaper_waitq);
1472
1473         spin_unlock_irqrestore (&kibnal_data.kib_reaper_lock, flags);
1474 }
1475
1476 int
1477 kibnal_close_peer_conns_locked (kib_peer_t *peer, int why)
1478 {
1479         kib_conn_t         *conn;
1480         struct list_head   *ctmp;
1481         struct list_head   *cnxt;
1482         int                 count = 0;
1483
1484         list_for_each_safe (ctmp, cnxt, &peer->ibp_conns) {
1485                 conn = list_entry (ctmp, kib_conn_t, ibc_list);
1486
1487                 count++;
1488                 kibnal_close_conn_locked (conn, why);
1489         }
1490
1491         return (count);
1492 }
1493
1494 int
1495 kibnal_close_stale_conns_locked (kib_peer_t *peer, __u64 incarnation)
1496 {
1497         kib_conn_t         *conn;
1498         struct list_head   *ctmp;
1499         struct list_head   *cnxt;
1500         int                 count = 0;
1501
1502         list_for_each_safe (ctmp, cnxt, &peer->ibp_conns) {
1503                 conn = list_entry (ctmp, kib_conn_t, ibc_list);
1504
1505                 if (conn->ibc_incarnation == incarnation)
1506                         continue;
1507
1508                 CDEBUG(D_NET, "Closing stale conn nid:"LPX64" incarnation:"LPX64"("LPX64")\n",
1509                        peer->ibp_nid, conn->ibc_incarnation, incarnation);
1510
1511                 count++;
1512                 kibnal_close_conn_locked (conn, -ESTALE);
1513         }
1514
1515         return (count);
1516 }
1517
1518 int
1519 kibnal_close_matching_conns (ptl_nid_t nid)
1520 {
1521         unsigned long       flags;
1522         kib_peer_t         *peer;
1523         struct list_head   *ptmp;
1524         struct list_head   *pnxt;
1525         int                 lo;
1526         int                 hi;
1527         int                 i;
1528         int                 count = 0;
1529
1530         write_lock_irqsave (&kibnal_data.kib_global_lock, flags);
1531
1532         if (nid != PTL_NID_ANY)
1533                 lo = hi = kibnal_nid2peerlist(nid) - kibnal_data.kib_peers;
1534         else {
1535                 lo = 0;
1536                 hi = kibnal_data.kib_peer_hash_size - 1;
1537         }
1538
1539         for (i = lo; i <= hi; i++) {
1540                 list_for_each_safe (ptmp, pnxt, &kibnal_data.kib_peers[i]) {
1541
1542                         peer = list_entry (ptmp, kib_peer_t, ibp_list);
1543                         LASSERT (peer->ibp_persistence != 0 ||
1544                                  peer->ibp_connecting != 0 ||
1545                                  !list_empty (&peer->ibp_conns));
1546
1547                         if (!(nid == PTL_NID_ANY || nid == peer->ibp_nid))
1548                                 continue;
1549
1550                         count += kibnal_close_peer_conns_locked (peer, 0);
1551                 }
1552         }
1553
1554         write_unlock_irqrestore (&kibnal_data.kib_global_lock, flags);
1555
1556         /* wildcards always succeed */
1557         if (nid == PTL_NID_ANY)
1558                 return (0);
1559
1560         return (count == 0 ? -ENOENT : 0);
1561 }
1562
1563 int
1564 kibnal_cmd(struct portals_cfg *pcfg, void * private)
1565 {
1566         int rc = -EINVAL;
1567
1568         LASSERT (pcfg != NULL);
1569
1570         switch(pcfg->pcfg_command) {
1571         case NAL_CMD_GET_PEER: {
1572                 ptl_nid_t   nid = 0;
1573                 __u32       ip = 0;
1574                 int         port = 0;
1575                 int         share_count = 0;
1576
1577                 rc = kibnal_get_peer_info(pcfg->pcfg_count,
1578                                           &nid, &ip, &port, &share_count);
1579                 pcfg->pcfg_nid   = nid;
1580                 pcfg->pcfg_size  = 0;
1581                 pcfg->pcfg_id    = ip;
1582                 pcfg->pcfg_misc  = port;
1583                 pcfg->pcfg_count = 0;
1584                 pcfg->pcfg_wait  = share_count;
1585                 break;
1586         }
1587         case NAL_CMD_ADD_PEER: {
1588                 rc = kibnal_add_persistent_peer (pcfg->pcfg_nid,
1589                                                  pcfg->pcfg_id, /* IP */
1590                                                  pcfg->pcfg_misc); /* port */
1591                 break;
1592         }
1593         case NAL_CMD_DEL_PEER: {
1594                 rc = kibnal_del_peer (pcfg->pcfg_nid,
1595                                        /* flags == single_share */
1596                                        pcfg->pcfg_flags != 0);
1597                 break;
1598         }
1599         case NAL_CMD_GET_CONN: {
1600                 kib_conn_t *conn = kibnal_get_conn_by_idx (pcfg->pcfg_count);
1601
1602                 if (conn == NULL)
1603                         rc = -ENOENT;
1604                 else {
1605                         rc = 0;
1606                         pcfg->pcfg_nid   = conn->ibc_peer->ibp_nid;
1607                         pcfg->pcfg_id    = 0;
1608                         pcfg->pcfg_misc  = 0;
1609                         pcfg->pcfg_flags = 0;
1610                         kibnal_put_conn (conn);
1611                 }
1612                 break;
1613         }
1614         case NAL_CMD_CLOSE_CONNECTION: {
1615                 rc = kibnal_close_matching_conns (pcfg->pcfg_nid);
1616                 break;
1617         }
1618         case NAL_CMD_REGISTER_MYNID: {
1619                 if (pcfg->pcfg_nid == PTL_NID_ANY)
1620                         rc = -EINVAL;
1621                 else
1622                         rc = kibnal_set_mynid (pcfg->pcfg_nid);
1623                 break;
1624         }
1625         }
1626
1627         return rc;
1628 }
1629
1630 void
1631 kibnal_free_pages (kib_pages_t *p)
1632 {
1633         int     npages = p->ibp_npages;
1634         int     rc;
1635         int     i;
1636
1637         if (p->ibp_mapped) {
1638                 rc = ib_memory_deregister(p->ibp_handle);
1639                 if (rc != 0)
1640                         CERROR ("Deregister error: %d\n", rc);
1641         }
1642
1643         for (i = 0; i < npages; i++)
1644                 if (p->ibp_pages[i] != NULL)
1645                         __free_page(p->ibp_pages[i]);
1646
1647         PORTAL_FREE (p, offsetof(kib_pages_t, ibp_pages[npages]));
1648 }
1649
1650 int
1651 kibnal_alloc_pages (kib_pages_t **pp, int npages, int access)
1652 {
1653         kib_pages_t                *p;
1654         struct ib_physical_buffer  *phys_pages;
1655         int                         i;
1656         int                         rc;
1657
1658         PORTAL_ALLOC(p, offsetof(kib_pages_t, ibp_pages[npages]));
1659         if (p == NULL) {
1660                 CERROR ("Can't allocate buffer %d\n", npages);
1661                 return (-ENOMEM);
1662         }
1663
1664         memset (p, 0, offsetof(kib_pages_t, ibp_pages[npages]));
1665         p->ibp_npages = npages;
1666
1667         for (i = 0; i < npages; i++) {
1668                 p->ibp_pages[i] = alloc_page (GFP_KERNEL);
1669                 if (p->ibp_pages[i] == NULL) {
1670                         CERROR ("Can't allocate page %d of %d\n", i, npages);
1671                         kibnal_free_pages(p);
1672                         return (-ENOMEM);
1673                 }
1674         }
1675
1676         PORTAL_ALLOC(phys_pages, npages * sizeof(*phys_pages));
1677         if (phys_pages == NULL) {
1678                 CERROR ("Can't allocate physarray for %d pages\n", npages);
1679                 kibnal_free_pages(p);
1680                 return (-ENOMEM);
1681         }
1682
1683         for (i = 0; i < npages; i++) {
1684                 phys_pages[i].size = PAGE_SIZE;
1685                 phys_pages[i].address =
1686                         kibnal_page2phys(p->ibp_pages[i]);
1687         }
1688
1689         p->ibp_vaddr = 0;
1690         rc = ib_memory_register_physical(kibnal_data.kib_pd,
1691                                          phys_pages, npages,
1692                                          &p->ibp_vaddr,
1693                                          npages * PAGE_SIZE, 0,
1694                                          access,
1695                                          &p->ibp_handle,
1696                                          &p->ibp_lkey,
1697                                          &p->ibp_rkey);
1698
1699         PORTAL_FREE(phys_pages, npages * sizeof(*phys_pages));
1700
1701         if (rc != 0) {
1702                 CERROR ("Error %d mapping %d pages\n", rc, npages);
1703                 kibnal_free_pages(p);
1704                 return (rc);
1705         }
1706
1707         p->ibp_mapped = 1;
1708         *pp = p;
1709         return (0);
1710 }
1711
1712 int
1713 kibnal_setup_tx_descs (void)
1714 {
1715         int           ipage = 0;
1716         int           page_offset = 0;
1717         __u64         vaddr;
1718         __u64         vaddr_base;
1719         struct page  *page;
1720         kib_tx_t     *tx;
1721         int           i;
1722         int           rc;
1723
1724         /* pre-mapped messages are not bigger than 1 page */
1725         LASSERT (IBNAL_MSG_SIZE <= PAGE_SIZE);
1726
1727         /* No fancy arithmetic when we do the buffer calculations */
1728         LASSERT (PAGE_SIZE % IBNAL_MSG_SIZE == 0);
1729
1730         rc = kibnal_alloc_pages(&kibnal_data.kib_tx_pages,
1731                                 IBNAL_TX_MSG_PAGES,
1732                                 0);            /* local read access only */
1733         if (rc != 0)
1734                 return (rc);
1735
1736         vaddr = vaddr_base = kibnal_data.kib_tx_pages->ibp_vaddr;
1737
1738         for (i = 0; i < IBNAL_TX_MSGS; i++) {
1739                 page = kibnal_data.kib_tx_pages->ibp_pages[ipage];
1740                 tx = &kibnal_data.kib_tx_descs[i];
1741
1742                 memset (tx, 0, sizeof(*tx));    /* zero flags etc */
1743
1744                 tx->tx_msg = (kib_msg_t *)(((char *)page_address(page)) + page_offset);
1745                 tx->tx_vaddr = vaddr;
1746                 tx->tx_isnblk = (i >= IBNAL_NTX);
1747                 tx->tx_mapped = KIB_TX_UNMAPPED;
1748
1749                 CDEBUG(D_NET, "Tx[%d] %p->%p - "LPX64"\n",
1750                        i, tx, tx->tx_msg, tx->tx_vaddr);
1751
1752                 if (tx->tx_isnblk)
1753                         list_add (&tx->tx_list,
1754                                   &kibnal_data.kib_idle_nblk_txs);
1755                 else
1756                         list_add (&tx->tx_list,
1757                                   &kibnal_data.kib_idle_txs);
1758
1759                 vaddr += IBNAL_MSG_SIZE;
1760                 LASSERT (vaddr <= vaddr_base + IBNAL_TX_MSG_BYTES);
1761
1762                 page_offset += IBNAL_MSG_SIZE;
1763                 LASSERT (page_offset <= PAGE_SIZE);
1764
1765                 if (page_offset == PAGE_SIZE) {
1766                         page_offset = 0;
1767                         ipage++;
1768                         LASSERT (ipage <= IBNAL_TX_MSG_PAGES);
1769                 }
1770         }
1771
1772         return (0);
1773 }
1774
1775 void
1776 kibnal_api_shutdown (nal_t *nal)
1777 {
1778         int   i;
1779         int   rc;
1780
1781         if (nal->nal_refct != 0) {
1782                 /* This module got the first ref */
1783                 PORTAL_MODULE_UNUSE;
1784                 return;
1785         }
1786
1787         CDEBUG(D_MALLOC, "before NAL cleanup: kmem %d\n",
1788                atomic_read (&portal_kmemory));
1789
1790         LASSERT(nal == &kibnal_api);
1791
1792         switch (kibnal_data.kib_init) {
1793         default:
1794                 CERROR ("Unexpected state %d\n", kibnal_data.kib_init);
1795                 LBUG();
1796
1797         case IBNAL_INIT_ALL:
1798                 /* stop calls to nal_cmd */
1799                 libcfs_nal_cmd_unregister(OPENIBNAL);
1800                 /* No new peers */
1801
1802                 /* resetting my NID unadvertises me, removes my
1803                  * listener and nukes all current peers */
1804                 kibnal_set_mynid (PTL_NID_ANY);
1805
1806                 /* Wait for all peer state to clean up */
1807                 i = 2;
1808                 while (atomic_read (&kibnal_data.kib_npeers) != 0) {
1809                         i++;
1810                         CDEBUG(((i & (-i)) == i) ? D_WARNING : D_NET, /* power of 2? */
1811                                "waiting for %d peers to close down\n",
1812                                atomic_read (&kibnal_data.kib_npeers));
1813                         set_current_state (TASK_INTERRUPTIBLE);
1814                         schedule_timeout (HZ);
1815                 }
1816                 /* fall through */
1817
1818         case IBNAL_INIT_CQ:
1819                 rc = ib_cq_destroy (kibnal_data.kib_cq);
1820                 if (rc != 0)
1821                         CERROR ("Destroy CQ error: %d\n", rc);
1822                 /* fall through */
1823
1824         case IBNAL_INIT_TXD:
1825                 kibnal_free_pages (kibnal_data.kib_tx_pages);
1826                 /* fall through */
1827 #if IBNAL_FMR
1828         case IBNAL_INIT_FMR:
1829                 rc = ib_fmr_pool_destroy (kibnal_data.kib_fmr_pool);
1830                 if (rc != 0)
1831                         CERROR ("Destroy FMR pool error: %d\n", rc);
1832                 /* fall through */
1833 #endif
1834         case IBNAL_INIT_PD:
1835                 rc = ib_pd_destroy(kibnal_data.kib_pd);
1836                 if (rc != 0)
1837                         CERROR ("Destroy PD error: %d\n", rc);
1838                 /* fall through */
1839
1840         case IBNAL_INIT_LIB:
1841                 lib_fini(&kibnal_lib);
1842                 /* fall through */
1843
1844         case IBNAL_INIT_DATA:
1845                 /* Module refcount only gets to zero when all peers
1846                  * have been closed so all lists must be empty */
1847                 LASSERT (atomic_read (&kibnal_data.kib_npeers) == 0);
1848                 LASSERT (kibnal_data.kib_peers != NULL);
1849                 for (i = 0; i < kibnal_data.kib_peer_hash_size; i++) {
1850                         LASSERT (list_empty (&kibnal_data.kib_peers[i]));
1851                 }
1852                 LASSERT (atomic_read (&kibnal_data.kib_nconns) == 0);
1853                 LASSERT (list_empty (&kibnal_data.kib_sched_rxq));
1854                 LASSERT (list_empty (&kibnal_data.kib_sched_txq));
1855                 LASSERT (list_empty (&kibnal_data.kib_reaper_conns));
1856                 LASSERT (list_empty (&kibnal_data.kib_connd_peers));
1857                 LASSERT (list_empty (&kibnal_data.kib_connd_acceptq));
1858
1859                 /* flag threads to terminate; wake and wait for them to die */
1860                 kibnal_data.kib_shutdown = 1;
1861                 wake_up_all (&kibnal_data.kib_sched_waitq);
1862                 wake_up_all (&kibnal_data.kib_reaper_waitq);
1863                 wake_up_all (&kibnal_data.kib_connd_waitq);
1864
1865                 i = 2;
1866                 while (atomic_read (&kibnal_data.kib_nthreads) != 0) {
1867                         i++;
1868                         CDEBUG(((i & (-i)) == i) ? D_WARNING : D_NET, /* power of 2? */
1869                                "Waiting for %d threads to terminate\n",
1870                                atomic_read (&kibnal_data.kib_nthreads));
1871                         set_current_state (TASK_INTERRUPTIBLE);
1872                         schedule_timeout (HZ);
1873                 }
1874                 /* fall through */
1875
1876         case IBNAL_INIT_NOTHING:
1877                 break;
1878         }
1879
1880         if (kibnal_data.kib_tx_descs != NULL)
1881                 PORTAL_FREE (kibnal_data.kib_tx_descs,
1882                              IBNAL_TX_MSGS * sizeof(kib_tx_t));
1883
1884         if (kibnal_data.kib_peers != NULL)
1885                 PORTAL_FREE (kibnal_data.kib_peers,
1886                              sizeof (struct list_head) *
1887                              kibnal_data.kib_peer_hash_size);
1888
1889         CDEBUG(D_MALLOC, "after NAL cleanup: kmem %d\n",
1890                atomic_read (&portal_kmemory));
1891         printk(KERN_INFO "Lustre: OpenIB NAL unloaded (final mem %d)\n",
1892                atomic_read(&portal_kmemory));
1893
1894         kibnal_data.kib_init = IBNAL_INIT_NOTHING;
1895 }
1896
1897 int
1898 kibnal_api_startup (nal_t *nal, ptl_pid_t requested_pid,
1899                      ptl_ni_limits_t *requested_limits,
1900                      ptl_ni_limits_t *actual_limits)
1901 {
1902         struct timeval    tv;
1903         ptl_process_id_t  process_id;
1904         int               pkmem = atomic_read(&portal_kmemory);
1905         int               rc;
1906         int               i;
1907
1908         LASSERT (nal == &kibnal_api);
1909
1910         if (nal->nal_refct != 0) {
1911                 if (actual_limits != NULL)
1912                         *actual_limits = kibnal_lib.libnal_ni.ni_actual_limits;
1913                 /* This module got the first ref */
1914                 PORTAL_MODULE_USE;
1915                 return (PTL_OK);
1916         }
1917
1918         LASSERT (kibnal_data.kib_init == IBNAL_INIT_NOTHING);
1919
1920         memset (&kibnal_data, 0, sizeof (kibnal_data)); /* zero pointers, flags etc */
1921
1922         do_gettimeofday(&tv);
1923         kibnal_data.kib_incarnation = (((__u64)tv.tv_sec) * 1000000) + tv.tv_usec;
1924
1925         init_MUTEX (&kibnal_data.kib_nid_mutex);
1926         init_MUTEX_LOCKED (&kibnal_data.kib_listener_signal);
1927
1928         rwlock_init(&kibnal_data.kib_global_lock);
1929
1930         kibnal_data.kib_peer_hash_size = IBNAL_PEER_HASH_SIZE;
1931         PORTAL_ALLOC (kibnal_data.kib_peers,
1932                       sizeof (struct list_head) * kibnal_data.kib_peer_hash_size);
1933         if (kibnal_data.kib_peers == NULL) {
1934                 goto failed;
1935         }
1936         for (i = 0; i < kibnal_data.kib_peer_hash_size; i++)
1937                 INIT_LIST_HEAD(&kibnal_data.kib_peers[i]);
1938
1939         spin_lock_init (&kibnal_data.kib_reaper_lock);
1940         INIT_LIST_HEAD (&kibnal_data.kib_reaper_conns);
1941         init_waitqueue_head (&kibnal_data.kib_reaper_waitq);
1942
1943         spin_lock_init (&kibnal_data.kib_connd_lock);
1944         INIT_LIST_HEAD (&kibnal_data.kib_connd_acceptq);
1945         INIT_LIST_HEAD (&kibnal_data.kib_connd_peers);
1946         init_waitqueue_head (&kibnal_data.kib_connd_waitq);
1947
1948         spin_lock_init (&kibnal_data.kib_sched_lock);
1949         INIT_LIST_HEAD (&kibnal_data.kib_sched_txq);
1950         INIT_LIST_HEAD (&kibnal_data.kib_sched_rxq);
1951         init_waitqueue_head (&kibnal_data.kib_sched_waitq);
1952
1953         spin_lock_init (&kibnal_data.kib_tx_lock);
1954         INIT_LIST_HEAD (&kibnal_data.kib_idle_txs);
1955         INIT_LIST_HEAD (&kibnal_data.kib_idle_nblk_txs);
1956         init_waitqueue_head(&kibnal_data.kib_idle_tx_waitq);
1957
1958         PORTAL_ALLOC (kibnal_data.kib_tx_descs,
1959                       IBNAL_TX_MSGS * sizeof(kib_tx_t));
1960         if (kibnal_data.kib_tx_descs == NULL) {
1961                 CERROR ("Can't allocate tx descs\n");
1962                 goto failed;
1963         }
1964
1965         /* lists/ptrs/locks initialised */
1966         kibnal_data.kib_init = IBNAL_INIT_DATA;
1967         /*****************************************************/
1968
1969
1970         process_id.pid = requested_pid;
1971         process_id.nid = PTL_NID_ANY;           /* don't know my NID yet */
1972
1973         rc = lib_init(&kibnal_lib, nal, process_id,
1974                       requested_limits, actual_limits);
1975         if (rc != PTL_OK) {
1976                 CERROR("lib_init failed: error %d\n", rc);
1977                 goto failed;
1978         }
1979
1980         /* lib interface initialised */
1981         kibnal_data.kib_init = IBNAL_INIT_LIB;
1982         /*****************************************************/
1983
1984         for (i = 0; i < IBNAL_N_SCHED; i++) {
1985                 rc = kibnal_thread_start (kibnal_scheduler,
1986                                           (void *)((unsigned long)i));
1987                 if (rc != 0) {
1988                         CERROR("Can't spawn openibnal scheduler[%d]: %d\n",
1989                                i, rc);
1990                         goto failed;
1991                 }
1992         }
1993
1994         for (i = 0; i < IBNAL_N_CONND; i++) {
1995                 rc = kibnal_thread_start (kibnal_connd,
1996                                           (void *)((unsigned long)i));
1997                 if (rc != 0) {
1998                         CERROR("Can't spawn openibnal connd[%d]: %d\n",
1999                                i, rc);
2000                         goto failed;
2001                 }
2002         }
2003
2004         rc = kibnal_thread_start (kibnal_reaper, NULL);
2005         if (rc != 0) {
2006                 CERROR ("Can't spawn openibnal reaper: %d\n", rc);
2007                 goto failed;
2008         }
2009
2010         kibnal_data.kib_device = ib_device_get_by_index(0);
2011         if (kibnal_data.kib_device == NULL) {
2012                 CERROR ("Can't open ib device 0\n");
2013                 goto failed;
2014         }
2015
2016         rc = ib_device_properties_get(kibnal_data.kib_device,
2017                                       &kibnal_data.kib_device_props);
2018         if (rc != 0) {
2019                 CERROR ("Can't get device props: %d\n", rc);
2020                 goto failed;
2021         }
2022
2023         CDEBUG(D_NET, "Max Initiator: %d Max Responder %d\n",
2024                kibnal_data.kib_device_props.max_initiator_per_qp,
2025                kibnal_data.kib_device_props.max_responder_per_qp);
2026
2027         kibnal_data.kib_port = 0;
2028         for (i = 1; i <= 2; i++) {
2029                 rc = ib_port_properties_get(kibnal_data.kib_device, i,
2030                                             &kibnal_data.kib_port_props);
2031                 if (rc == 0) {
2032                         kibnal_data.kib_port = i;
2033                         break;
2034                 }
2035         }
2036         if (kibnal_data.kib_port == 0) {
2037                 CERROR ("Can't find a port\n");
2038                 goto failed;
2039         }
2040
2041         rc = ib_pd_create(kibnal_data.kib_device,
2042                           NULL, &kibnal_data.kib_pd);
2043         if (rc != 0) {
2044                 CERROR ("Can't create PD: %d\n", rc);
2045                 goto failed;
2046         }
2047
2048         /* flag PD initialised */
2049         kibnal_data.kib_init = IBNAL_INIT_PD;
2050         /*****************************************************/
2051 #if IBNAL_FMR
2052         {
2053                 const int pool_size = IBNAL_NTX + IBNAL_NTX_NBLK;
2054                 struct ib_fmr_pool_param params = {
2055                         .max_pages_per_fmr = PTL_MTU/PAGE_SIZE,
2056                         .access            = (IB_ACCESS_LOCAL_WRITE |
2057                                               IB_ACCESS_REMOTE_WRITE |
2058                                               IB_ACCESS_REMOTE_READ),
2059                         .pool_size         = pool_size,
2060                         .dirty_watermark   = (pool_size * 3)/4,
2061                         .flush_function    = NULL,
2062                         .flush_arg         = NULL,
2063                         .cache             = 1,
2064                 };
2065                 rc = ib_fmr_pool_create(kibnal_data.kib_pd, &params,
2066                                         &kibnal_data.kib_fmr_pool);
2067                 if (rc != 0) {
2068                         CERROR ("Can't create FMR pool size %d: %d\n",
2069                                 pool_size, rc);
2070                         goto failed;
2071                 }
2072         }
2073
2074         /* flag FMR pool initialised */
2075         kibnal_data.kib_init = IBNAL_INIT_FMR;
2076 #endif
2077         /*****************************************************/
2078
2079         rc = kibnal_setup_tx_descs();
2080         if (rc != 0) {
2081                 CERROR ("Can't register tx descs: %d\n", rc);
2082                 goto failed;
2083         }
2084
2085         /* flag TX descs initialised */
2086         kibnal_data.kib_init = IBNAL_INIT_TXD;
2087         /*****************************************************/
2088
2089         {
2090                 struct ib_cq_callback callback = {
2091                         .context        = IBNAL_CALLBACK_CTXT,
2092                         .policy         = IB_CQ_PROVIDER_REARM,
2093                         .function       = {
2094                                 .entry  = kibnal_callback,
2095                         },
2096                         .arg            = NULL,
2097                 };
2098                 int  nentries = IBNAL_CQ_ENTRIES;
2099
2100                 rc = ib_cq_create (kibnal_data.kib_device,
2101                                    &nentries, &callback, NULL,
2102                                    &kibnal_data.kib_cq);
2103                 if (rc != 0) {
2104                         CERROR ("Can't create CQ: %d\n", rc);
2105                         goto failed;
2106                 }
2107
2108                 /* I only want solicited events */
2109                 rc = ib_cq_request_notification(kibnal_data.kib_cq, 1);
2110                 LASSERT (rc == 0);
2111         }
2112
2113         /* flag CQ initialised */
2114         kibnal_data.kib_init = IBNAL_INIT_CQ;
2115         /*****************************************************/
2116
2117         rc = libcfs_nal_cmd_register(OPENIBNAL, &kibnal_cmd, NULL);
2118         if (rc != 0) {
2119                 CERROR ("Can't initialise command interface (rc = %d)\n", rc);
2120                 goto failed;
2121         }
2122
2123         /* flag everything initialised */
2124         kibnal_data.kib_init = IBNAL_INIT_ALL;
2125         /*****************************************************/
2126
2127         printk(KERN_INFO "Lustre: OpenIB NAL loaded "
2128                "(initial mem %d)\n", pkmem);
2129
2130         return (PTL_OK);
2131
2132  failed:
2133         kibnal_api_shutdown (&kibnal_api);
2134         return (PTL_FAIL);
2135 }
2136
2137 void __exit
2138 kibnal_module_fini (void)
2139 {
2140         if (kibnal_tunables.kib_sysctl != NULL)
2141                 unregister_sysctl_table (kibnal_tunables.kib_sysctl);
2142         PtlNIFini(kibnal_ni);
2143
2144         ptl_unregister_nal(OPENIBNAL);
2145 }
2146
2147 int __init
2148 kibnal_module_init (void)
2149 {
2150         int    rc;
2151
2152         /* the following must be sizeof(int) for proc_dointvec() */
2153         LASSERT (sizeof(kibnal_tunables.kib_io_timeout) == sizeof(int));
2154         LASSERT (sizeof(kibnal_tunables.kib_listener_timeout) == sizeof(int));
2155         LASSERT (sizeof(kibnal_tunables.kib_backlog) == sizeof(int));
2156         LASSERT (sizeof(kibnal_tunables.kib_port) == sizeof(int));
2157
2158         kibnal_api.nal_ni_init = kibnal_api_startup;
2159         kibnal_api.nal_ni_fini = kibnal_api_shutdown;
2160
2161         /* Initialise dynamic tunables to defaults once only */
2162         kibnal_tunables.kib_io_timeout = IBNAL_IO_TIMEOUT;
2163         kibnal_tunables.kib_listener_timeout = IBNAL_LISTENER_TIMEOUT;
2164         kibnal_tunables.kib_backlog = IBNAL_BACKLOG;
2165         kibnal_tunables.kib_port = IBNAL_PORT;
2166
2167         rc = ptl_register_nal(OPENIBNAL, &kibnal_api);
2168         if (rc != PTL_OK) {
2169                 CERROR("Can't register IBNAL: %d\n", rc);
2170                 return (-ENOMEM);               /* or something... */
2171         }
2172
2173         /* Pure gateways want the NAL started up at module load time... */
2174         rc = PtlNIInit(OPENIBNAL, LUSTRE_SRV_PTL_PID, NULL, NULL, &kibnal_ni);
2175         if (rc != PTL_OK && rc != PTL_IFACE_DUP) {
2176                 ptl_unregister_nal(OPENIBNAL);
2177                 return (-ENODEV);
2178         }
2179
2180         kibnal_tunables.kib_sysctl =
2181                 register_sysctl_table (kibnal_top_ctl_table, 0);
2182         if (kibnal_tunables.kib_sysctl == NULL) {
2183                 CERROR("Can't register sysctl table\n");
2184                 PtlNIFini(kibnal_ni);
2185                 ptl_unregister_nal(OPENIBNAL);
2186                 return (-ENOMEM);
2187         }
2188
2189         return (0);
2190 }
2191
2192 MODULE_AUTHOR("Cluster File Systems, Inc. <info@clusterfs.com>");
2193 MODULE_DESCRIPTION("Kernel OpenIB NAL v0.01");
2194 MODULE_LICENSE("GPL");
2195
2196 module_init(kibnal_module_init);
2197 module_exit(kibnal_module_fini);
2198