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 #ifdef CONFIG_SYSCTL
  32 #define IBNAL_SYSCTL             202
  33
  34 #define IBNAL_SYSCTL_TIMEOUT     1
  35
  36 static ctl_table kibnal_ctl_table[] = {
  37         {IBNAL_SYSCTL_TIMEOUT, "timeout",
  38          &kibnal_tunables.kib_io_timeout, sizeof (int),
  39          0644, NULL, &proc_dointvec},
  40         { 0 }
  41 };
  42
  43 static ctl_table kibnal_top_ctl_table[] = {
  44         {IBNAL_SYSCTL, "openibnal", NULL, 0, 0555, kibnal_ctl_table},
  45         { 0 }
  46 };
  47 #endif
  48
  49 void
  50 print_service(struct ib_common_attrib_service *service, char *tag, int rc)
  51 {
  52         char name[32];
  53
  54         if (service == NULL)
  55         {
  56                 CWARN("tag       : %s\n"
  57                       "status    : %d (NULL)\n", tag, rc);
  58                 return;
  59         }
  60         strncpy (name, service->service_name, sizeof(name)-1);
  61         name[sizeof(name)-1] = 0;
  62
  63         CWARN("tag       : %s\n"
  64               "status    : %d\n"
  65               "service id: "LPX64"\n"
  66               "name      : %s\n"
  67               "NID       : "LPX64"\n", tag, rc,
  68               service->service_id, name,
  69               *kibnal_service_nid_field(service));
  70 }
  71
  72 void
  73 kibnal_service_setunset_done (tTS_IB_CLIENT_QUERY_TID tid, int status,
  74                                struct ib_common_attrib_service *service, void *arg)
  75 {
  76         *(int *)arg = status;
  77         up (&kibnal_data.kib_nid_signal);
  78 }
  79
  80 #if IBNAL_CHECK_ADVERT
  81 void
  82 kibnal_check_advert (void)
  83 {
  84         struct ib_common_attrib_service *svc;
  85         __u64   tid;
  86         int     rc;
  87         int     rc2;
  88
  89         PORTAL_ALLOC(svc, sizeof(*svc));
  90         if (svc == NULL)
  91                 return;
  92
  93         memset (svc, 0, sizeof (*svc));
  94         kibnal_set_service_keys(svc, kibnal_data.kib_nid);
  95
  96         rc = ib_service_get (kibnal_data.kib_device,
  97                              kibnal_data.kib_port,
  98                              svc,
  99                              KIBNAL_SERVICE_KEY_MASK,
 100                              kibnal_tunables.kib_io_timeout * HZ,
 101                              kibnal_service_setunset_done, &rc2,
 102                              &tid);
 103
 104         if (rc != 0) {
 105                 CERROR ("Immediate error %d checking SM service\n", rc);
 106         } else {
 107                 down (&kibnal_data.kib_nid_signal);
 108                 rc = rc2;
 109
 110                 if (rc != 0)
 111                         CERROR ("Error %d checking SM service\n", rc);
 112         }
 113
 114         PORTAL_FREE(svc, sizeof(*svc));
 115 }
 116 #endif
 117
 118 int
 119 kibnal_advertise (void)
 120 {
 121         struct ib_common_attrib_service *svc;
 122         __u64   tid;
 123         int     rc;
 124         int     rc2;
 125
 126         LASSERT (kibnal_data.kib_nid != PTL_NID_ANY);
 127
 128         PORTAL_ALLOC(svc, sizeof(*svc));
 129         if (svc == NULL)
 130                 return (-ENOMEM);
 131
 132         memset (svc, 0, sizeof (*svc));
 133
 134         svc->service_id = kibnal_data.kib_service_id;
 135
 136         rc = ib_cached_gid_get(kibnal_data.kib_device,
 137                                kibnal_data.kib_port,
 138                                0,
 139                                svc->service_gid);
 140         if (rc != 0) {
 141                 CERROR ("Can't get port %d GID: %d\n",
 142                         kibnal_data.kib_port, rc);
 143                 goto out;
 144         }
 145
 146         rc = ib_cached_pkey_get(kibnal_data.kib_device,
 147                                 kibnal_data.kib_port,
 148                                 0,
 149                                 &svc->service_pkey);
 150         if (rc != 0) {
 151                 CERROR ("Can't get port %d PKEY: %d\n",
 152                         kibnal_data.kib_port, rc);
 153                 goto out;
 154         }
 155
 156         svc->service_lease = 0xffffffff;
 157
 158         kibnal_set_service_keys(svc, kibnal_data.kib_nid);
 159
 160         CDEBUG(D_NET, "Advertising service id "LPX64" %s:"LPX64"\n",
 161                svc->service_id,
 162                svc->service_name, *kibnal_service_nid_field(svc));
 163
 164         rc = ib_service_set (kibnal_data.kib_device,
 165                              kibnal_data.kib_port,
 166                              svc,
 167                              IB_SA_SERVICE_COMP_MASK_ID |
 168                              IB_SA_SERVICE_COMP_MASK_GID |
 169                              IB_SA_SERVICE_COMP_MASK_PKEY |
 170                              IB_SA_SERVICE_COMP_MASK_LEASE |
 171                              KIBNAL_SERVICE_KEY_MASK,
 172                              kibnal_tunables.kib_io_timeout * HZ,
 173                              kibnal_service_setunset_done, &rc2, &tid);
 174
 175         if (rc != 0) {
 176                 CERROR ("Immediate error %d advertising NID "LPX64"\n",
 177                         rc, kibnal_data.kib_nid);
 178                 goto out;
 179         }
 180
 181         down (&kibnal_data.kib_nid_signal);
 182
 183         rc = rc2;
 184         if (rc != 0)
 185                 CERROR ("Error %d advertising NID "LPX64"\n",
 186                         rc, kibnal_data.kib_nid);
 187  out:
 188         PORTAL_FREE(svc, sizeof(*svc));
 189         return (rc);
 190 }
 191
 192 void
 193 kibnal_unadvertise (int expect_success)
 194 {
 195         struct ib_common_attrib_service *svc;
 196         __u64   tid;
 197         int     rc;
 198         int     rc2;
 199
 200         LASSERT (kibnal_data.kib_nid != PTL_NID_ANY);
 201
 202         PORTAL_ALLOC(svc, sizeof(*svc));
 203         if (svc == NULL)
 204                 return;
 205
 206         memset (svc, 0, sizeof(*svc));
 207
 208         kibnal_set_service_keys(svc, kibnal_data.kib_nid);
 209
 210         CDEBUG(D_NET, "Unadvertising service %s:"LPX64"\n",
 211                svc->service_name, *kibnal_service_nid_field(svc));
 212
 213         rc = ib_service_delete (kibnal_data.kib_device,
 214                                 kibnal_data.kib_port,
 215                                 svc,
 216                                 KIBNAL_SERVICE_KEY_MASK,
 217                                 kibnal_tunables.kib_io_timeout * HZ,
 218                                 kibnal_service_setunset_done, &rc2, &tid);
 219         if (rc != 0) {
 220                 CERROR ("Immediate error %d unadvertising NID "LPX64"\n",
 221                         rc, kibnal_data.kib_nid);
 222                 goto out;
 223         }
 224
 225         down (&kibnal_data.kib_nid_signal);
 226
 227         if ((rc2 == 0) == !!expect_success)
 228                 goto out;                       /* success: rc == 0 */
 229
 230         if (expect_success)
 231                 CERROR("Error %d unadvertising NID "LPX64"\n",
 232                        rc, kibnal_data.kib_nid);
 233         else
 234                 CWARN("Removed conflicting NID "LPX64"\n",
 235                       kibnal_data.kib_nid);
 236  out:
 237         PORTAL_FREE(svc, sizeof(*svc));
 238 }
 239
 240 int
 241 kibnal_set_mynid(ptl_nid_t nid)
 242 {
 243         struct timeval tv;
 244         lib_ni_t      *ni = &kibnal_lib.libnal_ni;
 245         int            rc;
 246
 247         CDEBUG(D_IOCTL, "setting mynid to "LPX64" (old nid="LPX64")\n",
 248                nid, ni->ni_pid.nid);
 249
 250         do_gettimeofday(&tv);
 251
 252         down (&kibnal_data.kib_nid_mutex);
 253
 254         if (nid == kibnal_data.kib_nid) {
 255                 /* no change of NID */
 256                 up (&kibnal_data.kib_nid_mutex);
 257                 return (0);
 258         }
 259
 260         CDEBUG(D_NET, "NID "LPX64"("LPX64")\n",
 261                kibnal_data.kib_nid, nid);
 262
 263         if (kibnal_data.kib_nid != PTL_NID_ANY) {
 264
 265                 kibnal_unadvertise (1);
 266
 267                 rc = ib_cm_listen_stop (kibnal_data.kib_listen_handle);
 268                 if (rc != 0)
 269                         CERROR ("Error %d stopping listener\n", rc);
 270         }
 271
 272         kibnal_data.kib_nid = ni->ni_pid.nid = nid;
 273         kibnal_data.kib_incarnation = (((__u64)tv.tv_sec) * 1000000) + tv.tv_usec;
 274
 275         /* Delete all existing peers and their connections after new
 276          * NID/incarnation set to ensure no old connections in our brave
 277          * new world. */
 278         kibnal_del_peer (PTL_NID_ANY, 0);
 279
 280         if (kibnal_data.kib_nid == PTL_NID_ANY) {
 281                 /* No new NID to install */
 282                 up (&kibnal_data.kib_nid_mutex);
 283                 return (0);
 284         }
 285
 286         /* remove any previous advert (crashed node etc) */
 287         kibnal_unadvertise(0);
 288
 289         /* Assign new service number */
 290         kibnal_data.kib_service_id = ib_cm_service_assign();
 291         CDEBUG(D_NET, "service_id "LPX64"\n", kibnal_data.kib_service_id);
 292
 293         rc = ib_cm_listen(kibnal_data.kib_service_id,
 294                           TS_IB_CM_SERVICE_EXACT_MASK,
 295                           kibnal_passive_conn_callback, NULL,
 296                           &kibnal_data.kib_listen_handle);
 297         if (rc == 0) {
 298                 rc = kibnal_advertise();
 299                 if (rc == 0) {
 300 #if IBNAL_CHECK_ADVERT
 301                         kibnal_check_advert();
 302 #endif
 303                         up (&kibnal_data.kib_nid_mutex);
 304                         return (0);
 305                 }
 306
 307                 ib_cm_listen_stop(kibnal_data.kib_listen_handle);
 308                 /* remove any peers that sprung up while I failed to
 309                  * advertise myself */
 310                 kibnal_del_peer (PTL_NID_ANY, 0);
 311         }
 312
 313         kibnal_data.kib_nid = PTL_NID_ANY;
 314         up (&kibnal_data.kib_nid_mutex);
 315         return (rc);
 316 }
 317
 318 kib_peer_t *
 319 kibnal_create_peer (ptl_nid_t nid)
 320 {
 321         kib_peer_t *peer;
 322
 323         LASSERT (nid != PTL_NID_ANY);
 324
 325         PORTAL_ALLOC (peer, sizeof (*peer));
 326         if (peer == NULL)
 327                 return (NULL);
 328
 329         memset(peer, 0, sizeof(*peer));         /* zero flags etc */
 330
 331         peer->ibp_nid = nid;
 332         atomic_set (&peer->ibp_refcount, 1);    /* 1 ref for caller */
 333
 334         INIT_LIST_HEAD (&peer->ibp_list);       /* not in the peer table yet */
 335         INIT_LIST_HEAD (&peer->ibp_conns);
 336         INIT_LIST_HEAD (&peer->ibp_tx_queue);
 337
 338         peer->ibp_reconnect_time = jiffies;
 339         peer->ibp_reconnect_interval = IBNAL_MIN_RECONNECT_INTERVAL;
 340
 341         atomic_inc (&kibnal_data.kib_npeers);
 342         return (peer);
 343 }
 344
 345 void
 346 kibnal_destroy_peer (kib_peer_t *peer)
 347 {
 348         CDEBUG (D_NET, "peer "LPX64" %p deleted\n", peer->ibp_nid, peer);
 349
 350         LASSERT (atomic_read (&peer->ibp_refcount) == 0);
 351         LASSERT (peer->ibp_persistence == 0);
 352         LASSERT (!kibnal_peer_active(peer));
 353         LASSERT (peer->ibp_connecting == 0);
 354         LASSERT (list_empty (&peer->ibp_conns));
 355         LASSERT (list_empty (&peer->ibp_tx_queue));
 356
 357         PORTAL_FREE (peer, sizeof (*peer));
 358
 359         /* NB a peer's connections keep a reference on their peer until
 360          * they are destroyed, so we can be assured that _all_ state to do
 361          * with this peer has been cleaned up when its refcount drops to
 362          * zero. */
 363         atomic_dec (&kibnal_data.kib_npeers);
 364 }
 365
 366 void
 367 kibnal_put_peer (kib_peer_t *peer)
 368 {
 369         CDEBUG (D_OTHER, "putting peer[%p] -> "LPX64" (%d)\n",
 370                 peer, peer->ibp_nid,
 371                 atomic_read (&peer->ibp_refcount));
 372
 373         LASSERT (atomic_read (&peer->ibp_refcount) > 0);
 374         if (!atomic_dec_and_test (&peer->ibp_refcount))
 375                 return;
 376
 377         kibnal_destroy_peer (peer);
 378 }
 379
 380 kib_peer_t *
 381 kibnal_find_peer_locked (ptl_nid_t nid)
 382 {
 383         struct list_head *peer_list = kibnal_nid2peerlist (nid);
 384         struct list_head *tmp;
 385         kib_peer_t       *peer;
 386
 387         list_for_each (tmp, peer_list) {
 388
 389                 peer = list_entry (tmp, kib_peer_t, ibp_list);
 390
 391                 LASSERT (peer->ibp_persistence != 0 || /* persistent peer */
 392                          peer->ibp_connecting != 0 || /* creating conns */
 393                          !list_empty (&peer->ibp_conns));  /* active conn */
 394
 395                 if (peer->ibp_nid != nid)
 396                         continue;
 397
 398                 CDEBUG(D_NET, "got peer [%p] -> "LPX64" (%d)\n",
 399                        peer, nid, atomic_read (&peer->ibp_refcount));
 400                 return (peer);
 401         }
 402         return (NULL);
 403 }
 404
 405 kib_peer_t *
 406 kibnal_get_peer (ptl_nid_t nid)
 407 {
 408         kib_peer_t     *peer;
 409
 410         read_lock (&kibnal_data.kib_global_lock);
 411         peer = kibnal_find_peer_locked (nid);
 412         if (peer != NULL)                       /* +1 ref for caller? */
 413                 atomic_inc (&peer->ibp_refcount);
 414         read_unlock (&kibnal_data.kib_global_lock);
 415
 416         return (peer);
 417 }
 418
 419 void
 420 kibnal_unlink_peer_locked (kib_peer_t *peer)
 421 {
 422         LASSERT (peer->ibp_persistence == 0);
 423         LASSERT (list_empty(&peer->ibp_conns));
 424
 425         LASSERT (kibnal_peer_active(peer));
 426         list_del_init (&peer->ibp_list);
 427         /* lose peerlist's ref */
 428         kibnal_put_peer (peer);
 429 }
 430
 431 int
 432 kibnal_get_peer_info (int index, ptl_nid_t *nidp, int *persistencep)
 433 {
 434         kib_peer_t        *peer;
 435         struct list_head  *ptmp;
 436         int                i;
 437
 438         read_lock (&kibnal_data.kib_global_lock);
 439
 440         for (i = 0; i < kibnal_data.kib_peer_hash_size; i++) {
 441
 442                 list_for_each (ptmp, &kibnal_data.kib_peers[i]) {
 443
 444                         peer = list_entry (ptmp, kib_peer_t, ibp_list);
 445                         LASSERT (peer->ibp_persistence != 0 ||
 446                                  peer->ibp_connecting != 0 ||
 447                                  !list_empty (&peer->ibp_conns));
 448
 449                         if (index-- > 0)
 450                                 continue;
 451
 452                         *nidp = peer->ibp_nid;
 453                         *persistencep = peer->ibp_persistence;
 454
 455                         read_unlock (&kibnal_data.kib_global_lock);
 456                         return (0);
 457                 }
 458         }
 459
 460         read_unlock (&kibnal_data.kib_global_lock);
 461         return (-ENOENT);
 462 }
 463
 464 int
 465 kibnal_add_persistent_peer (ptl_nid_t nid)
 466 {
 467         unsigned long      flags;
 468         kib_peer_t        *peer;
 469         kib_peer_t        *peer2;
 470
 471         if (nid == PTL_NID_ANY)
 472                 return (-EINVAL);
 473
 474         peer = kibnal_create_peer (nid);
 475         if (peer == NULL)
 476                 return (-ENOMEM);
 477
 478         write_lock_irqsave (&kibnal_data.kib_global_lock, flags);
 479
 480         peer2 = kibnal_find_peer_locked (nid);
 481         if (peer2 != NULL) {
 482                 kibnal_put_peer (peer);
 483                 peer = peer2;
 484         } else {
 485                 /* peer table takes existing ref on peer */
 486                 list_add_tail (&peer->ibp_list,
 487                                kibnal_nid2peerlist (nid));
 488         }
 489
 490         peer->ibp_persistence++;
 491
 492         write_unlock_irqrestore (&kibnal_data.kib_global_lock, flags);
 493         return (0);
 494 }
 495
 496 void
 497 kibnal_del_peer_locked (kib_peer_t *peer, int single_share)
 498 {
 499         struct list_head *ctmp;
 500         struct list_head *cnxt;
 501         kib_conn_t       *conn;
 502
 503         if (!single_share)
 504                 peer->ibp_persistence = 0;
 505         else if (peer->ibp_persistence > 0)
 506                 peer->ibp_persistence--;
 507
 508         if (peer->ibp_persistence != 0)
 509                 return;
 510
 511         list_for_each_safe (ctmp, cnxt, &peer->ibp_conns) {
 512                 conn = list_entry(ctmp, kib_conn_t, ibc_list);
 513
 514                 kibnal_close_conn_locked (conn, 0);
 515         }
 516
 517         /* NB peer unlinks itself when last conn is closed */
 518 }
 519
 520 int
 521 kibnal_del_peer (ptl_nid_t nid, int single_share)
 522 {
 523         unsigned long      flags;
 524         struct list_head  *ptmp;
 525         struct list_head  *pnxt;
 526         kib_peer_t        *peer;
 527         int                lo;
 528         int                hi;
 529         int                i;
 530         int                rc = -ENOENT;
 531
 532         write_lock_irqsave (&kibnal_data.kib_global_lock, flags);
 533
 534         if (nid != PTL_NID_ANY)
 535                 lo = hi = kibnal_nid2peerlist(nid) - kibnal_data.kib_peers;
 536         else {
 537                 lo = 0;
 538                 hi = kibnal_data.kib_peer_hash_size - 1;
 539         }
 540
 541         for (i = lo; i <= hi; i++) {
 542                 list_for_each_safe (ptmp, pnxt, &kibnal_data.kib_peers[i]) {
 543                         peer = list_entry (ptmp, kib_peer_t, ibp_list);
 544                         LASSERT (peer->ibp_persistence != 0 ||
 545                                  peer->ibp_connecting != 0 ||
 546                                  !list_empty (&peer->ibp_conns));
 547
 548                         if (!(nid == PTL_NID_ANY || peer->ibp_nid == nid))
 549                                 continue;
 550
 551                         kibnal_del_peer_locked (peer, single_share);
 552                         rc = 0;         /* matched something */
 553
 554                         if (single_share)
 555                                 goto out;
 556                 }
 557         }
 558  out:
 559         write_unlock_irqrestore (&kibnal_data.kib_global_lock, flags);
 560
 561         return (rc);
 562 }
 563
 564 kib_conn_t *
 565 kibnal_get_conn_by_idx (int index)
 566 {
 567         kib_peer_t        *peer;
 568         struct list_head  *ptmp;
 569         kib_conn_t        *conn;
 570         struct list_head  *ctmp;
 571         int                i;
 572
 573         read_lock (&kibnal_data.kib_global_lock);
 574
 575         for (i = 0; i < kibnal_data.kib_peer_hash_size; i++) {
 576                 list_for_each (ptmp, &kibnal_data.kib_peers[i]) {
 577
 578                         peer = list_entry (ptmp, kib_peer_t, ibp_list);
 579                         LASSERT (peer->ibp_persistence > 0 ||
 580                                  peer->ibp_connecting != 0 ||
 581                                  !list_empty (&peer->ibp_conns));
 582
 583                         list_for_each (ctmp, &peer->ibp_conns) {
 584                                 if (index-- > 0)
 585                                         continue;
 586
 587                                 conn = list_entry (ctmp, kib_conn_t, ibc_list);
 588                                 CDEBUG(D_NET, "++conn[%p] state %d -> "LPX64" (%d)\n",
 589                                        conn, conn->ibc_state, conn->ibc_peer->ibp_nid,
 590                                        atomic_read (&conn->ibc_refcount));
 591                                 atomic_inc (&conn->ibc_refcount);
 592                                 read_unlock (&kibnal_data.kib_global_lock);
 593                                 return (conn);
 594                         }
 595                 }
 596         }
 597
 598         read_unlock (&kibnal_data.kib_global_lock);
 599         return (NULL);
 600 }
 601
 602 kib_conn_t *
 603 kibnal_create_conn (void)
 604 {
 605         kib_conn_t  *conn;
 606         int          i;
 607         __u64        vaddr = 0;
 608         __u64        vaddr_base;
 609         int          page_offset;
 610         int          ipage;
 611         int          rc;
 612         union {
 613                 struct ib_qp_create_param  qp_create;
 614                 struct ib_qp_attribute     qp_attr;
 615         } params;
 616
 617         PORTAL_ALLOC (conn, sizeof (*conn));
 618         if (conn == NULL) {
 619                 CERROR ("Can't allocate connection\n");
 620                 return (NULL);
 621         }
 622
 623         /* zero flags, NULL pointers etc... */
 624         memset (conn, 0, sizeof (*conn));
 625
 626         INIT_LIST_HEAD (&conn->ibc_tx_queue);
 627         INIT_LIST_HEAD (&conn->ibc_active_txs);
 628         spin_lock_init (&conn->ibc_lock);
 629
 630         atomic_inc (&kibnal_data.kib_nconns);
 631         /* well not really, but I call destroy() on failure, which decrements */
 632
 633         PORTAL_ALLOC (conn->ibc_rxs, IBNAL_RX_MSGS * sizeof (kib_rx_t));
 634         if (conn->ibc_rxs == NULL)
 635                 goto failed;
 636         memset (conn->ibc_rxs, 0, IBNAL_RX_MSGS * sizeof(kib_rx_t));
 637
 638         rc = kibnal_alloc_pages(&conn->ibc_rx_pages,
 639                                 IBNAL_RX_MSG_PAGES,
 640                                 IB_ACCESS_LOCAL_WRITE);
 641         if (rc != 0)
 642                 goto failed;
 643
 644         vaddr_base = vaddr = conn->ibc_rx_pages->ibp_vaddr;
 645
 646         for (i = ipage = page_offset = 0; i < IBNAL_RX_MSGS; i++) {
 647                 struct page *page = conn->ibc_rx_pages->ibp_pages[ipage];
 648                 kib_rx_t   *rx = &conn->ibc_rxs[i];
 649
 650                 rx->rx_conn = conn;
 651                 rx->rx_vaddr = vaddr;
 652                 rx->rx_msg = (kib_msg_t *)(((char *)page_address(page)) + page_offset);
 653
 654                 vaddr += IBNAL_MSG_SIZE;
 655                 LASSERT (vaddr <= vaddr_base + IBNAL_RX_MSG_BYTES);
 656
 657                 page_offset += IBNAL_MSG_SIZE;
 658                 LASSERT (page_offset <= PAGE_SIZE);
 659
 660                 if (page_offset == PAGE_SIZE) {
 661                         page_offset = 0;
 662                         ipage++;
 663                         LASSERT (ipage <= IBNAL_RX_MSG_PAGES);
 664                 }
 665         }
 666
 667         params.qp_create = (struct ib_qp_create_param) {
 668                 .limit = {
 669                         /* Sends have an optional RDMA */
 670                         .max_outstanding_send_request    = 2 * IBNAL_MSG_QUEUE_SIZE,
 671                         .max_outstanding_receive_request = IBNAL_MSG_QUEUE_SIZE,
 672                         .max_send_gather_element         = 1,
 673                         .max_receive_scatter_element     = 1,
 674                 },
 675                 .pd              = kibnal_data.kib_pd,
 676                 .send_queue      = kibnal_data.kib_cq,
 677                 .receive_queue   = kibnal_data.kib_cq,
 678                 .send_policy     = IB_WQ_SIGNAL_SELECTABLE,
 679                 .receive_policy  = IB_WQ_SIGNAL_SELECTABLE,
 680                 .rd_domain       = 0,
 681                 .transport       = IB_TRANSPORT_RC,
 682                 .device_specific = NULL,
 683         };
 684
 685         rc = ib_qp_create (&params.qp_create, &conn->ibc_qp, &conn->ibc_qpn);
 686         if (rc != 0) {
 687                 CERROR ("Failed to create queue pair: %d\n", rc);
 688                 goto failed;
 689         }
 690
 691         /* Mark QP created */
 692         conn->ibc_state = IBNAL_CONN_INIT_QP;
 693
 694         params.qp_attr = (struct ib_qp_attribute) {
 695                 .state             = IB_QP_STATE_INIT,
 696                 .port              = kibnal_data.kib_port,
 697                 .enable_rdma_read  = 1,
 698                 .enable_rdma_write = 1,
 699                 .valid_fields      = (IB_QP_ATTRIBUTE_STATE |
 700                                       IB_QP_ATTRIBUTE_PORT |
 701                                       IB_QP_ATTRIBUTE_PKEY_INDEX |
 702                                       IB_QP_ATTRIBUTE_RDMA_ATOMIC_ENABLE),
 703         };
 704         rc = ib_qp_modify(conn->ibc_qp, &params.qp_attr);
 705         if (rc != 0) {
 706                 CERROR ("Failed to modify queue pair: %d\n", rc);
 707                 goto failed;
 708         }
 709
 710         /* 1 ref for caller */
 711         atomic_set (&conn->ibc_refcount, 1);
 712         return (conn);
 713
 714  failed:
 715         kibnal_destroy_conn (conn);
 716         return (NULL);
 717 }
 718
 719 void
 720 kibnal_destroy_conn (kib_conn_t *conn)
 721 {
 722         int    rc;
 723
 724         CDEBUG (D_NET, "connection %p\n", conn);
 725
 726         LASSERT (atomic_read (&conn->ibc_refcount) == 0);
 727         LASSERT (list_empty(&conn->ibc_tx_queue));
 728         LASSERT (list_empty(&conn->ibc_active_txs));
 729         LASSERT (conn->ibc_nsends_posted == 0);
 730         LASSERT (conn->ibc_connreq == NULL);
 731
 732         switch (conn->ibc_state) {
 733         case IBNAL_CONN_ZOMBIE:
 734                 /* called after connection sequence initiated */
 735
 736         case IBNAL_CONN_INIT_QP:
 737                 rc = ib_qp_destroy(conn->ibc_qp);
 738                 if (rc != 0)
 739                         CERROR("Can't destroy QP: %d\n", rc);
 740                 /* fall through */
 741
 742         case IBNAL_CONN_INIT_NOTHING:
 743                 break;
 744
 745         default:
 746                 LASSERT (0);
 747         }
 748
 749         if (conn->ibc_rx_pages != NULL)
 750                 kibnal_free_pages(conn->ibc_rx_pages);
 751
 752         if (conn->ibc_rxs != NULL)
 753                 PORTAL_FREE(conn->ibc_rxs,
 754                             IBNAL_RX_MSGS * sizeof(kib_rx_t));
 755
 756         if (conn->ibc_peer != NULL)
 757                 kibnal_put_peer(conn->ibc_peer);
 758
 759         PORTAL_FREE(conn, sizeof (*conn));
 760
 761         atomic_dec(&kibnal_data.kib_nconns);
 762
 763         if (atomic_read (&kibnal_data.kib_nconns) == 0 &&
 764             kibnal_data.kib_shutdown) {
 765                 /* I just nuked the last connection on shutdown; wake up
 766                  * everyone so they can exit. */
 767                 wake_up_all(&kibnal_data.kib_sched_waitq);
 768                 wake_up_all(&kibnal_data.kib_connd_waitq);
 769         }
 770 }
 771
 772 void
 773 kibnal_put_conn (kib_conn_t *conn)
 774 {
 775         unsigned long flags;
 776
 777         CDEBUG (D_NET, "putting conn[%p] state %d -> "LPX64" (%d)\n",
 778                 conn, conn->ibc_state, conn->ibc_peer->ibp_nid,
 779                 atomic_read (&conn->ibc_refcount));
 780
 781         LASSERT (atomic_read (&conn->ibc_refcount) > 0);
 782         if (!atomic_dec_and_test (&conn->ibc_refcount))
 783                 return;
 784
 785         /* last ref only goes on zombies */
 786         LASSERT (conn->ibc_state == IBNAL_CONN_ZOMBIE);
 787
 788         spin_lock_irqsave (&kibnal_data.kib_connd_lock, flags);
 789
 790         list_add (&conn->ibc_list, &kibnal_data.kib_connd_conns);
 791         wake_up (&kibnal_data.kib_connd_waitq);
 792
 793         spin_unlock_irqrestore (&kibnal_data.kib_connd_lock, flags);
 794 }
 795
 796 int
 797 kibnal_close_peer_conns_locked (kib_peer_t *peer, int why)
 798 {
 799         kib_conn_t         *conn;
 800         struct list_head   *ctmp;
 801         struct list_head   *cnxt;
 802         int                 count = 0;
 803
 804         list_for_each_safe (ctmp, cnxt, &peer->ibp_conns) {
 805                 conn = list_entry (ctmp, kib_conn_t, ibc_list);
 806
 807                 count++;
 808                 kibnal_close_conn_locked (conn, why);
 809         }
 810
 811         return (count);
 812 }
 813
 814 int
 815 kibnal_close_stale_conns_locked (kib_peer_t *peer, __u64 incarnation)
 816 {
 817         kib_conn_t         *conn;
 818         struct list_head   *ctmp;
 819         struct list_head   *cnxt;
 820         int                 count = 0;
 821
 822         list_for_each_safe (ctmp, cnxt, &peer->ibp_conns) {
 823                 conn = list_entry (ctmp, kib_conn_t, ibc_list);
 824
 825                 if (conn->ibc_incarnation == incarnation)
 826                         continue;
 827
 828                 CDEBUG(D_NET, "Closing stale conn nid:"LPX64" incarnation:"LPX64"("LPX64")\n",
 829                        peer->ibp_nid, conn->ibc_incarnation, incarnation);
 830
 831                 count++;
 832                 kibnal_close_conn_locked (conn, -ESTALE);
 833         }
 834
 835         return (count);
 836 }
 837
 838 int
 839 kibnal_close_matching_conns (ptl_nid_t nid)
 840 {
 841         unsigned long       flags;
 842         kib_peer_t         *peer;
 843         struct list_head   *ptmp;
 844         struct list_head   *pnxt;
 845         int                 lo;
 846         int                 hi;
 847         int                 i;
 848         int                 count = 0;
 849
 850         write_lock_irqsave (&kibnal_data.kib_global_lock, flags);
 851
 852         if (nid != PTL_NID_ANY)
 853                 lo = hi = kibnal_nid2peerlist(nid) - kibnal_data.kib_peers;
 854         else {
 855                 lo = 0;
 856                 hi = kibnal_data.kib_peer_hash_size - 1;
 857         }
 858
 859         for (i = lo; i <= hi; i++) {
 860                 list_for_each_safe (ptmp, pnxt, &kibnal_data.kib_peers[i]) {
 861
 862                         peer = list_entry (ptmp, kib_peer_t, ibp_list);
 863                         LASSERT (peer->ibp_persistence != 0 ||
 864                                  peer->ibp_connecting != 0 ||
 865                                  !list_empty (&peer->ibp_conns));
 866
 867                         if (!(nid == PTL_NID_ANY || nid == peer->ibp_nid))
 868                                 continue;
 869
 870                         count += kibnal_close_peer_conns_locked (peer, 0);
 871                 }
 872         }
 873
 874         write_unlock_irqrestore (&kibnal_data.kib_global_lock, flags);
 875
 876         /* wildcards always succeed */
 877         if (nid == PTL_NID_ANY)
 878                 return (0);
 879
 880         return (count == 0 ? -ENOENT : 0);
 881 }
 882
 883 int
 884 kibnal_cmd(struct portals_cfg *pcfg, void * private)
 885 {
 886         int rc = -EINVAL;
 887
 888         LASSERT (pcfg != NULL);
 889
 890         switch(pcfg->pcfg_command) {
 891         case NAL_CMD_GET_PEER: {
 892                 ptl_nid_t   nid = 0;
 893                 int         share_count = 0;
 894
 895                 rc = kibnal_get_peer_info(pcfg->pcfg_count,
 896                                           &nid, &share_count);
 897                 pcfg->pcfg_nid   = nid;
 898                 pcfg->pcfg_size  = 0;
 899                 pcfg->pcfg_id    = 0;
 900                 pcfg->pcfg_misc  = 0;
 901                 pcfg->pcfg_count = 0;
 902                 pcfg->pcfg_wait  = share_count;
 903                 break;
 904         }
 905         case NAL_CMD_ADD_PEER: {
 906                 rc = kibnal_add_persistent_peer (pcfg->pcfg_nid);
 907                 break;
 908         }
 909         case NAL_CMD_DEL_PEER: {
 910                 rc = kibnal_del_peer (pcfg->pcfg_nid,
 911                                        /* flags == single_share */
 912                                        pcfg->pcfg_flags != 0);
 913                 break;
 914         }
 915         case NAL_CMD_GET_CONN: {
 916                 kib_conn_t *conn = kibnal_get_conn_by_idx (pcfg->pcfg_count);
 917
 918                 if (conn == NULL)
 919                         rc = -ENOENT;
 920                 else {
 921                         rc = 0;
 922                         pcfg->pcfg_nid   = conn->ibc_peer->ibp_nid;
 923                         pcfg->pcfg_id    = 0;
 924                         pcfg->pcfg_misc  = 0;
 925                         pcfg->pcfg_flags = 0;
 926                         kibnal_put_conn (conn);
 927                 }
 928                 break;
 929         }
 930         case NAL_CMD_CLOSE_CONNECTION: {
 931                 rc = kibnal_close_matching_conns (pcfg->pcfg_nid);
 932                 break;
 933         }
 934         case NAL_CMD_REGISTER_MYNID: {
 935                 if (pcfg->pcfg_nid == PTL_NID_ANY)
 936                         rc = -EINVAL;
 937                 else
 938                         rc = kibnal_set_mynid (pcfg->pcfg_nid);
 939                 break;
 940         }
 941         }
 942
 943         return rc;
 944 }
 945
 946 void
 947 kibnal_free_pages (kib_pages_t *p)
 948 {
 949         int     npages = p->ibp_npages;
 950         int     rc;
 951         int     i;
 952
 953         if (p->ibp_mapped) {
 954                 rc = ib_memory_deregister(p->ibp_handle);
 955                 if (rc != 0)
 956                         CERROR ("Deregister error: %d\n", rc);
 957         }
 958
 959         for (i = 0; i < npages; i++)
 960                 if (p->ibp_pages[i] != NULL)
 961                         __free_page(p->ibp_pages[i]);
 962
 963         PORTAL_FREE (p, offsetof(kib_pages_t, ibp_pages[npages]));
 964 }
 965
 966 int
 967 kibnal_alloc_pages (kib_pages_t **pp, int npages, int access)
 968 {
 969         kib_pages_t                *p;
 970         struct ib_physical_buffer  *phys_pages;
 971         int                         i;
 972         int                         rc;
 973
 974         PORTAL_ALLOC(p, offsetof(kib_pages_t, ibp_pages[npages]));
 975         if (p == NULL) {
 976                 CERROR ("Can't allocate buffer %d\n", npages);
 977                 return (-ENOMEM);
 978         }
 979
 980         memset (p, 0, offsetof(kib_pages_t, ibp_pages[npages]));
 981         p->ibp_npages = npages;
 982
 983         for (i = 0; i < npages; i++) {
 984                 p->ibp_pages[i] = alloc_page (GFP_KERNEL);
 985                 if (p->ibp_pages[i] == NULL) {
 986                         CERROR ("Can't allocate page %d of %d\n", i, npages);
 987                         kibnal_free_pages(p);
 988                         return (-ENOMEM);
 989                 }
 990         }
 991
 992         PORTAL_ALLOC(phys_pages, npages * sizeof(*phys_pages));
 993         if (phys_pages == NULL) {
 994                 CERROR ("Can't allocate physarray for %d pages\n", npages);
 995                 kibnal_free_pages(p);
 996                 return (-ENOMEM);
 997         }
 998
 999         for (i = 0; i < npages; i++) {
1000                 phys_pages[i].size = PAGE_SIZE;
1001                 phys_pages[i].address =
1002                         kibnal_page2phys(p->ibp_pages[i]);
1003         }
1004
1005         p->ibp_vaddr = 0;
1006         rc = ib_memory_register_physical(kibnal_data.kib_pd,
1007                                          phys_pages, npages,
1008                                          &p->ibp_vaddr,
1009                                          npages * PAGE_SIZE, 0,
1010                                          access,
1011                                          &p->ibp_handle,
1012                                          &p->ibp_lkey,
1013                                          &p->ibp_rkey);
1014
1015         PORTAL_FREE(phys_pages, npages * sizeof(*phys_pages));
1016
1017         if (rc != 0) {
1018                 CERROR ("Error %d mapping %d pages\n", rc, npages);
1019                 kibnal_free_pages(p);
1020                 return (rc);
1021         }
1022
1023         p->ibp_mapped = 1;
1024         *pp = p;
1025         return (0);
1026 }
1027
1028 int
1029 kibnal_setup_tx_descs (void)
1030 {
1031         int           ipage = 0;
1032         int           page_offset = 0;
1033         __u64         vaddr;
1034         __u64         vaddr_base;
1035         struct page  *page;
1036         kib_tx_t     *tx;
1037         int           i;
1038         int           rc;
1039
1040         /* pre-mapped messages are not bigger than 1 page */
1041         LASSERT (IBNAL_MSG_SIZE <= PAGE_SIZE);
1042
1043         /* No fancy arithmetic when we do the buffer calculations */
1044         LASSERT (PAGE_SIZE % IBNAL_MSG_SIZE == 0);
1045
1046         rc = kibnal_alloc_pages(&kibnal_data.kib_tx_pages,
1047                                 IBNAL_TX_MSG_PAGES,
1048                                 0);            /* local read access only */
1049         if (rc != 0)
1050                 return (rc);
1051
1052         vaddr = vaddr_base = kibnal_data.kib_tx_pages->ibp_vaddr;
1053
1054         for (i = 0; i < IBNAL_TX_MSGS; i++) {
1055                 page = kibnal_data.kib_tx_pages->ibp_pages[ipage];
1056                 tx = &kibnal_data.kib_tx_descs[i];
1057
1058                 memset (tx, 0, sizeof(*tx));    /* zero flags etc */
1059
1060                 tx->tx_msg = (kib_msg_t *)(((char *)page_address(page)) + page_offset);
1061                 tx->tx_vaddr = vaddr;
1062                 tx->tx_isnblk = (i >= IBNAL_NTX);
1063                 tx->tx_mapped = KIB_TX_UNMAPPED;
1064
1065                 CDEBUG(D_NET, "Tx[%d] %p->%p - "LPX64"\n",
1066                        i, tx, tx->tx_msg, tx->tx_vaddr);
1067
1068                 if (tx->tx_isnblk)
1069                         list_add (&tx->tx_list,
1070                                   &kibnal_data.kib_idle_nblk_txs);
1071                 else
1072                         list_add (&tx->tx_list,
1073                                   &kibnal_data.kib_idle_txs);
1074
1075                 vaddr += IBNAL_MSG_SIZE;
1076                 LASSERT (vaddr <= vaddr_base + IBNAL_TX_MSG_BYTES);
1077
1078                 page_offset += IBNAL_MSG_SIZE;
1079                 LASSERT (page_offset <= PAGE_SIZE);
1080
1081                 if (page_offset == PAGE_SIZE) {
1082                         page_offset = 0;
1083                         ipage++;
1084                         LASSERT (ipage <= IBNAL_TX_MSG_PAGES);
1085                 }
1086         }
1087
1088         return (0);
1089 }
1090
1091 void
1092 kibnal_api_shutdown (nal_t *nal)
1093 {
1094         int   i;
1095         int   rc;
1096
1097         if (nal->nal_refct != 0) {
1098                 /* This module got the first ref */
1099                 PORTAL_MODULE_UNUSE;
1100                 return;
1101         }
1102
1103         CDEBUG(D_MALLOC, "before NAL cleanup: kmem %d\n",
1104                atomic_read (&portal_kmemory));
1105
1106         LASSERT(nal == &kibnal_api);
1107
1108         switch (kibnal_data.kib_init) {
1109         default:
1110                 CERROR ("Unexpected state %d\n", kibnal_data.kib_init);
1111                 LBUG();
1112
1113         case IBNAL_INIT_ALL:
1114                 /* stop calls to nal_cmd */
1115                 libcfs_nal_cmd_unregister(OPENIBNAL);
1116                 /* No new peers */
1117
1118                 /* resetting my NID unadvertises me, removes my
1119                  * listener and nukes all current peers */
1120                 kibnal_set_mynid (PTL_NID_ANY);
1121
1122                 /* Wait for all peer state to clean up */
1123                 i = 2;
1124                 while (atomic_read (&kibnal_data.kib_npeers) != 0) {
1125                         i++;
1126                         CDEBUG(((i & (-i)) == i) ? D_WARNING : D_NET, /* power of 2? */
1127                                "waiting for %d peers to close down\n",
1128                                atomic_read (&kibnal_data.kib_npeers));
1129                         set_current_state (TASK_INTERRUPTIBLE);
1130                         schedule_timeout (HZ);
1131                 }
1132                 /* fall through */
1133
1134         case IBNAL_INIT_CQ:
1135                 rc = ib_cq_destroy (kibnal_data.kib_cq);
1136                 if (rc != 0)
1137                         CERROR ("Destroy CQ error: %d\n", rc);
1138                 /* fall through */
1139
1140         case IBNAL_INIT_TXD:
1141                 kibnal_free_pages (kibnal_data.kib_tx_pages);
1142                 /* fall through */
1143 #if IBNAL_FMR
1144         case IBNAL_INIT_FMR:
1145                 rc = ib_fmr_pool_destroy (kibnal_data.kib_fmr_pool);
1146                 if (rc != 0)
1147                         CERROR ("Destroy FMR pool error: %d\n", rc);
1148                 /* fall through */
1149 #endif
1150         case IBNAL_INIT_PD:
1151                 rc = ib_pd_destroy(kibnal_data.kib_pd);
1152                 if (rc != 0)
1153                         CERROR ("Destroy PD error: %d\n", rc);
1154                 /* fall through */
1155
1156         case IBNAL_INIT_LIB:
1157                 lib_fini(&kibnal_lib);
1158                 /* fall through */
1159
1160         case IBNAL_INIT_DATA:
1161                 /* Module refcount only gets to zero when all peers
1162                  * have been closed so all lists must be empty */
1163                 LASSERT (atomic_read (&kibnal_data.kib_npeers) == 0);
1164                 LASSERT (kibnal_data.kib_peers != NULL);
1165                 for (i = 0; i < kibnal_data.kib_peer_hash_size; i++) {
1166                         LASSERT (list_empty (&kibnal_data.kib_peers[i]));
1167                 }
1168                 LASSERT (atomic_read (&kibnal_data.kib_nconns) == 0);
1169                 LASSERT (list_empty (&kibnal_data.kib_sched_rxq));
1170                 LASSERT (list_empty (&kibnal_data.kib_sched_txq));
1171                 LASSERT (list_empty (&kibnal_data.kib_connd_conns));
1172                 LASSERT (list_empty (&kibnal_data.kib_connd_peers));
1173
1174                 /* flag threads to terminate; wake and wait for them to die */
1175                 kibnal_data.kib_shutdown = 1;
1176                 wake_up_all (&kibnal_data.kib_sched_waitq);
1177                 wake_up_all (&kibnal_data.kib_connd_waitq);
1178
1179                 i = 2;
1180                 while (atomic_read (&kibnal_data.kib_nthreads) != 0) {
1181                         i++;
1182                         CDEBUG(((i & (-i)) == i) ? D_WARNING : D_NET, /* power of 2? */
1183                                "Waiting for %d threads to terminate\n",
1184                                atomic_read (&kibnal_data.kib_nthreads));
1185                         set_current_state (TASK_INTERRUPTIBLE);
1186                         schedule_timeout (HZ);
1187                 }
1188                 /* fall through */
1189
1190         case IBNAL_INIT_NOTHING:
1191                 break;
1192         }
1193
1194         if (kibnal_data.kib_tx_descs != NULL)
1195                 PORTAL_FREE (kibnal_data.kib_tx_descs,
1196                              IBNAL_TX_MSGS * sizeof(kib_tx_t));
1197
1198         if (kibnal_data.kib_peers != NULL)
1199                 PORTAL_FREE (kibnal_data.kib_peers,
1200                              sizeof (struct list_head) *
1201                              kibnal_data.kib_peer_hash_size);
1202
1203         CDEBUG(D_MALLOC, "after NAL cleanup: kmem %d\n",
1204                atomic_read (&portal_kmemory));
1205         printk(KERN_INFO "Lustre: OpenIB NAL unloaded (final mem %d)\n",
1206                atomic_read(&portal_kmemory));
1207
1208         kibnal_data.kib_init = IBNAL_INIT_NOTHING;
1209 }
1210
1211 int
1212 kibnal_api_startup (nal_t *nal, ptl_pid_t requested_pid,
1213                      ptl_ni_limits_t *requested_limits,
1214                      ptl_ni_limits_t *actual_limits)
1215 {
1216         ptl_process_id_t  process_id;
1217         int               pkmem = atomic_read(&portal_kmemory);
1218         int               rc;
1219         int               i;
1220
1221         LASSERT (nal == &kibnal_api);
1222
1223         if (nal->nal_refct != 0) {
1224                 if (actual_limits != NULL)
1225                         *actual_limits = kibnal_lib.libnal_ni.ni_actual_limits;
1226                 /* This module got the first ref */
1227                 PORTAL_MODULE_USE;
1228                 return (PTL_OK);
1229         }
1230
1231         LASSERT (kibnal_data.kib_init == IBNAL_INIT_NOTHING);
1232
1233         memset (&kibnal_data, 0, sizeof (kibnal_data)); /* zero pointers, flags etc */
1234
1235         init_MUTEX (&kibnal_data.kib_nid_mutex);
1236         init_MUTEX_LOCKED (&kibnal_data.kib_nid_signal);
1237         kibnal_data.kib_nid = PTL_NID_ANY;
1238
1239         rwlock_init(&kibnal_data.kib_global_lock);
1240
1241         kibnal_data.kib_peer_hash_size = IBNAL_PEER_HASH_SIZE;
1242         PORTAL_ALLOC (kibnal_data.kib_peers,
1243                       sizeof (struct list_head) * kibnal_data.kib_peer_hash_size);
1244         if (kibnal_data.kib_peers == NULL) {
1245                 goto failed;
1246         }
1247         for (i = 0; i < kibnal_data.kib_peer_hash_size; i++)
1248                 INIT_LIST_HEAD(&kibnal_data.kib_peers[i]);
1249
1250         spin_lock_init (&kibnal_data.kib_connd_lock);
1251         INIT_LIST_HEAD (&kibnal_data.kib_connd_peers);
1252         INIT_LIST_HEAD (&kibnal_data.kib_connd_conns);
1253         init_waitqueue_head (&kibnal_data.kib_connd_waitq);
1254
1255         spin_lock_init (&kibnal_data.kib_sched_lock);
1256         INIT_LIST_HEAD (&kibnal_data.kib_sched_txq);
1257         INIT_LIST_HEAD (&kibnal_data.kib_sched_rxq);
1258         init_waitqueue_head (&kibnal_data.kib_sched_waitq);
1259
1260         spin_lock_init (&kibnal_data.kib_tx_lock);
1261         INIT_LIST_HEAD (&kibnal_data.kib_idle_txs);
1262         INIT_LIST_HEAD (&kibnal_data.kib_idle_nblk_txs);
1263         init_waitqueue_head(&kibnal_data.kib_idle_tx_waitq);
1264
1265         PORTAL_ALLOC (kibnal_data.kib_tx_descs,
1266                       IBNAL_TX_MSGS * sizeof(kib_tx_t));
1267         if (kibnal_data.kib_tx_descs == NULL) {
1268                 CERROR ("Can't allocate tx descs\n");
1269                 goto failed;
1270         }
1271
1272         /* lists/ptrs/locks initialised */
1273         kibnal_data.kib_init = IBNAL_INIT_DATA;
1274         /*****************************************************/
1275
1276
1277         process_id.pid = requested_pid;
1278         process_id.nid = kibnal_data.kib_nid;
1279
1280         rc = lib_init(&kibnal_lib, nal, process_id,
1281                       requested_limits, actual_limits);
1282         if (rc != PTL_OK) {
1283                 CERROR("lib_init failed: error %d\n", rc);
1284                 goto failed;
1285         }
1286
1287         /* lib interface initialised */
1288         kibnal_data.kib_init = IBNAL_INIT_LIB;
1289         /*****************************************************/
1290
1291         for (i = 0; i < IBNAL_N_SCHED; i++) {
1292                 rc = kibnal_thread_start (kibnal_scheduler,
1293                                           (void *)((unsigned long)i));
1294                 if (rc != 0) {
1295                         CERROR("Can't spawn openibnal scheduler[%d]: %d\n",
1296                                i, rc);
1297                         goto failed;
1298                 }
1299         }
1300
1301         rc = kibnal_thread_start (kibnal_connd, NULL);
1302         if (rc != 0) {
1303                 CERROR ("Can't spawn openibnal connd: %d\n", rc);
1304                 goto failed;
1305         }
1306
1307         kibnal_data.kib_device = ib_device_get_by_index(0);
1308         if (kibnal_data.kib_device == NULL) {
1309                 CERROR ("Can't open ib device 0\n");
1310                 goto failed;
1311         }
1312
1313         rc = ib_device_properties_get(kibnal_data.kib_device,
1314                                       &kibnal_data.kib_device_props);
1315         if (rc != 0) {
1316                 CERROR ("Can't get device props: %d\n", rc);
1317                 goto failed;
1318         }
1319
1320         CDEBUG(D_NET, "Max Initiator: %d Max Responder %d\n",
1321                kibnal_data.kib_device_props.max_initiator_per_qp,
1322                kibnal_data.kib_device_props.max_responder_per_qp);
1323
1324         kibnal_data.kib_port = 0;
1325         for (i = 1; i <= 2; i++) {
1326                 rc = ib_port_properties_get(kibnal_data.kib_device, i,
1327                                             &kibnal_data.kib_port_props);
1328                 if (rc == 0) {
1329                         kibnal_data.kib_port = i;
1330                         break;
1331                 }
1332         }
1333         if (kibnal_data.kib_port == 0) {
1334                 CERROR ("Can't find a port\n");
1335                 goto failed;
1336         }
1337
1338         rc = ib_pd_create(kibnal_data.kib_device,
1339                           NULL, &kibnal_data.kib_pd);
1340         if (rc != 0) {
1341                 CERROR ("Can't create PD: %d\n", rc);
1342                 goto failed;
1343         }
1344
1345         /* flag PD initialised */
1346         kibnal_data.kib_init = IBNAL_INIT_PD;
1347         /*****************************************************/
1348 #if IBNAL_FMR
1349         {
1350                 const int pool_size = IBNAL_NTX + IBNAL_NTX_NBLK;
1351                 struct ib_fmr_pool_param params = {
1352                         .max_pages_per_fmr = PTL_MTU/PAGE_SIZE,
1353                         .access            = (IB_ACCESS_LOCAL_WRITE |
1354                                               IB_ACCESS_REMOTE_WRITE |
1355                                               IB_ACCESS_REMOTE_READ),
1356                         .pool_size         = pool_size,
1357                         .dirty_watermark   = (pool_size * 3)/4,
1358                         .flush_function    = NULL,
1359                         .flush_arg         = NULL,
1360                         .cache             = 1,
1361                 };
1362                 rc = ib_fmr_pool_create(kibnal_data.kib_pd, &params,
1363                                         &kibnal_data.kib_fmr_pool);
1364                 if (rc != 0) {
1365                         CERROR ("Can't create FMR pool size %d: %d\n",
1366                                 pool_size, rc);
1367                         goto failed;
1368                 }
1369         }
1370
1371         /* flag FMR pool initialised */
1372         kibnal_data.kib_init = IBNAL_INIT_FMR;
1373 #endif
1374         /*****************************************************/
1375
1376         rc = kibnal_setup_tx_descs();
1377         if (rc != 0) {
1378                 CERROR ("Can't register tx descs: %d\n", rc);
1379                 goto failed;
1380         }
1381
1382         /* flag TX descs initialised */
1383         kibnal_data.kib_init = IBNAL_INIT_TXD;
1384         /*****************************************************/
1385
1386         {
1387                 struct ib_cq_callback callback = {
1388                         .context        = IBNAL_CALLBACK_CTXT,
1389                         .policy         = IB_CQ_PROVIDER_REARM,
1390                         .function       = {
1391                                 .entry  = kibnal_callback,
1392                         },
1393                         .arg            = NULL,
1394                 };
1395                 int  nentries = IBNAL_CQ_ENTRIES;
1396
1397                 rc = ib_cq_create (kibnal_data.kib_device,
1398                                    &nentries, &callback, NULL,
1399                                    &kibnal_data.kib_cq);
1400                 if (rc != 0) {
1401                         CERROR ("Can't create CQ: %d\n", rc);
1402                         goto failed;
1403                 }
1404
1405                 /* I only want solicited events */
1406                 rc = ib_cq_request_notification(kibnal_data.kib_cq, 1);
1407                 LASSERT (rc == 0);
1408         }
1409
1410         /* flag CQ initialised */
1411         kibnal_data.kib_init = IBNAL_INIT_CQ;
1412         /*****************************************************/
1413
1414         rc = libcfs_nal_cmd_register(OPENIBNAL, &kibnal_cmd, NULL);
1415         if (rc != 0) {
1416                 CERROR ("Can't initialise command interface (rc = %d)\n", rc);
1417                 goto failed;
1418         }
1419
1420         /* flag everything initialised */
1421         kibnal_data.kib_init = IBNAL_INIT_ALL;
1422         /*****************************************************/
1423
1424         printk(KERN_INFO "Lustre: OpenIB NAL loaded "
1425                "(initial mem %d)\n", pkmem);
1426
1427         return (PTL_OK);
1428
1429  failed:
1430         kibnal_api_shutdown (&kibnal_api);
1431         return (PTL_FAIL);
1432 }
1433
1434 void __exit
1435 kibnal_module_fini (void)
1436 {
1437 #ifdef CONFIG_SYSCTL
1438         if (kibnal_tunables.kib_sysctl != NULL)
1439                 unregister_sysctl_table (kibnal_tunables.kib_sysctl);
1440 #endif
1441         PtlNIFini(kibnal_ni);
1442
1443         ptl_unregister_nal(OPENIBNAL);
1444 }
1445
1446 int __init
1447 kibnal_module_init (void)
1448 {
1449         int    rc;
1450
1451         /* the following must be sizeof(int) for proc_dointvec() */
1452         LASSERT(sizeof (kibnal_tunables.kib_io_timeout) == sizeof (int));
1453
1454         kibnal_api.nal_ni_init = kibnal_api_startup;
1455         kibnal_api.nal_ni_fini = kibnal_api_shutdown;
1456
1457         /* Initialise dynamic tunables to defaults once only */
1458         kibnal_tunables.kib_io_timeout = IBNAL_IO_TIMEOUT;
1459
1460         rc = ptl_register_nal(OPENIBNAL, &kibnal_api);
1461         if (rc != PTL_OK) {
1462                 CERROR("Can't register IBNAL: %d\n", rc);
1463                 return (-ENOMEM);               /* or something... */
1464         }
1465
1466         /* Pure gateways want the NAL started up at module load time... */
1467         rc = PtlNIInit(OPENIBNAL, LUSTRE_SRV_PTL_PID, NULL, NULL, &kibnal_ni);
1468         if (rc != PTL_OK && rc != PTL_IFACE_DUP) {
1469                 ptl_unregister_nal(OPENIBNAL);
1470                 return (-ENODEV);
1471         }
1472
1473 #ifdef CONFIG_SYSCTL
1474         /* Press on regardless even if registering sysctl doesn't work */
1475         kibnal_tunables.kib_sysctl =
1476                 register_sysctl_table (kibnal_top_ctl_table, 0);
1477 #endif
1478         return (0);
1479 }
1480
1481 MODULE_AUTHOR("Cluster File Systems, Inc. <info@clusterfs.com>");
1482 MODULE_DESCRIPTION("Kernel OpenIB NAL v0.01");
1483 MODULE_LICENSE("GPL");
1484
1485 module_init(kibnal_module_init);
1486 module_exit(kibnal_module_fini);
1487