lustre/portals/knals/ranal/ranal.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 #include "ranal.h"
  24
  25
  26 nal_t                   kranal_api;
  27 ptl_handle_ni_t         kranal_ni;
  28 kra_data_t              kranal_data;
  29 kra_tunables_t          kranal_tunables;
  30
  31 #ifdef CONFIG_SYSCTL
  32 #define RANAL_SYSCTL_TIMEOUT           1
  33 #define RANAL_SYSCTL_LISTENER_TIMEOUT  2
  34 #define RANAL_SYSCTL_BACKLOG           3
  35 #define RANAL_SYSCTL_PORT              4
  36 #define RANAL_SYSCTL_MAX_IMMEDIATE     5
  37
  38 #define RANAL_SYSCTL                   202
  39
  40 static ctl_table kranal_ctl_table[] = {
  41         {RANAL_SYSCTL_TIMEOUT, "timeout",
  42          &kranal_tunables.kra_timeout, sizeof(int),
  43          0644, NULL, &proc_dointvec},
  44         {RANAL_SYSCTL_LISTENER_TIMEOUT, "listener_timeout",
  45          &kranal_tunables.kra_listener_timeout, sizeof(int),
  46          0644, NULL, &proc_dointvec},
  47         {RANAL_SYSCTL_BACKLOG, "backlog",
  48          &kranal_tunables.kra_backlog, sizeof(int),
  49          0644, NULL, kranal_listener_procint},
  50         {RANAL_SYSCTL_PORT, "port",
  51          &kranal_tunables.kra_port, sizeof(int),
  52          0644, NULL, kranal_listener_procint},
  53         {RANAL_SYSCTL_MAX_IMMEDIATE, "max_immediate",
  54          &kranal_tunables.kra_max_immediate, sizeof(int),
  55          0644, NULL, &proc_dointvec},
  56         { 0 }
  57 };
  58
  59 static ctl_table kranal_top_ctl_table[] = {
  60         {RANAL_SYSCTL, "ranal", NULL, 0, 0555, kranal_ctl_table},
  61         { 0 }
  62 };
  63 #endif
  64
  65 int
  66 kranal_sock_write (struct socket *sock, void *buffer, int nob)
  67 {
  68         int           rc;
  69         mm_segment_t  oldmm = get_fs();
  70         struct iovec  iov = {
  71                 .iov_base = buffer,
  72                 .iov_len  = nob
  73         };
  74         struct msghdr msg = {
  75                 .msg_name       = NULL,
  76                 .msg_namelen    = 0,
  77                 .msg_iov        = &iov,
  78                 .msg_iovlen     = 1,
  79                 .msg_control    = NULL,
  80                 .msg_controllen = 0,
  81                 .msg_flags      = MSG_DONTWAIT
  82         };
  83
  84         /* We've set up the socket's send buffer to be large enough for
  85          * everything we send, so a single non-blocking send should
  86          * complete without error. */
  87
  88         set_fs(KERNEL_DS);
  89         rc = sock_sendmsg(sock, &msg, iov.iov_len);
  90         set_fs(oldmm);
  91
  92         return rc;
  93 }
  94
  95 int
  96 kranal_sock_read (struct socket *sock, void *buffer, int nob, int timeout)
  97 {
  98         int            rc;
  99         mm_segment_t   oldmm = get_fs();
 100         long           ticks = timeout * HZ;
 101         unsigned long  then;
 102         struct timeval tv;
 103
 104         LASSERT (nob > 0);
 105         LASSERT (ticks > 0);
 106
 107         for (;;) {
 108                 struct iovec  iov = {
 109                         .iov_base = buffer,
 110                         .iov_len  = nob
 111                 };
 112                 struct msghdr msg = {
 113                         .msg_name       = NULL,
 114                         .msg_namelen    = 0,
 115                         .msg_iov        = &iov,
 116                         .msg_iovlen     = 1,
 117                         .msg_control    = NULL,
 118                         .msg_controllen = 0,
 119                         .msg_flags      = 0
 120                 };
 121
 122                 /* Set receive timeout to remaining time */
 123                 tv = (struct timeval) {
 124                         .tv_sec = ticks / HZ,
 125                         .tv_usec = ((ticks % HZ) * 1000000) / HZ
 126                 };
 127                 set_fs(KERNEL_DS);
 128                 rc = sock_setsockopt(sock, SOL_SOCKET, SO_RCVTIMEO,
 129                                      (char *)&tv, sizeof(tv));
 130                 set_fs(oldmm);
 131                 if (rc != 0) {
 132                         CERROR("Can't set socket recv timeout %d: %d\n",
 133                                timeout, rc);
 134                         return rc;
 135                 }
 136
 137                 set_fs(KERNEL_DS);
 138                 then = jiffies;
 139                 rc = sock_recvmsg(sock, &msg, iov.iov_len, 0);
 140                 ticks -= jiffies - then;
 141                 set_fs(oldmm);
 142
 143                 if (rc < 0)
 144                         return rc;
 145
 146                 if (rc == 0)
 147                         return -ECONNABORTED;
 148
 149                 buffer = ((char *)buffer) + rc;
 150                 nob -= rc;
 151
 152                 if (nob == 0)
 153                         return 0;
 154
 155                 if (ticks <= 0)
 156                         return -ETIMEDOUT;
 157         }
 158 }
 159
 160 int
 161 kranal_create_sock(struct socket **sockp)
 162 {
 163         struct socket       *sock;
 164         int                  rc;
 165         struct timeval       tv;
 166         int                  option;
 167         mm_segment_t         oldmm = get_fs();
 168
 169         rc = sock_create(PF_INET, SOCK_STREAM, 0, &sock);
 170         if (rc != 0) {
 171                 CERROR("Can't create socket: %d\n", rc);
 172                 return rc;
 173         }
 174
 175         /* Ensure sending connection info doesn't block */
 176         option = 2 * sizeof(kra_connreq_t);
 177         set_fs(KERNEL_DS);
 178         rc = sock_setsockopt(sock, SOL_SOCKET, SO_SNDBUF,
 179                              (char *)&option, sizeof(option));
 180         set_fs(oldmm);
 181         if (rc != 0) {
 182                 CERROR("Can't set send buffer %d: %d\n", option, rc);
 183                 goto failed;
 184         }
 185
 186         option = 1;
 187         set_fs(KERNEL_DS);
 188         rc = sock_setsockopt(sock, SOL_SOCKET, SO_REUSEADDR,
 189                              (char *)&option, sizeof(option));
 190         set_fs(oldmm);
 191         if (rc != 0) {
 192                 CERROR("Can't set SO_REUSEADDR: %d\n", rc);
 193                 goto failed;
 194         }
 195
 196         *sockp = sock;
 197         return 0;
 198
 199  failed:
 200         sock_release(sock);
 201         return rc;
 202 }
 203
 204 void
 205 kranal_pause(int ticks)
 206 {
 207         set_current_state(TASK_UNINTERRUPTIBLE);
 208         schedule_timeout(ticks);
 209 }
 210
 211 void
 212 kranal_pack_connreq(kra_connreq_t *connreq, kra_conn_t *conn)
 213 {
 214         RAP_RETURN   rrc;
 215
 216         memset(connreq, 0, sizeof(*connreq));
 217
 218         connreq->racr_magic       = RANAL_MSG_MAGIC;
 219         connreq->racr_version     = RANAL_MSG_VERSION;
 220         connreq->racr_devid       = conn->rac_device->rad_id;
 221         connreq->racr_nid         = kranal_lib.libnal_ni.ni_pid.nid;
 222         connreq->racr_timeout     = conn->rac_timeout;
 223         connreq->racr_incarnation = conn->rac_my_incarnation;
 224
 225         rrc = RapkGetRiParams(conn->rac_rihandle, &connreq->racr_riparams);
 226         LASSERT(rrc == RAP_SUCCESS);
 227 }
 228
 229 int
 230 kranal_recv_connreq(struct socket *sock, kra_connreq_t *connreq, int timeout)
 231 {
 232         int         i;
 233         int         rc;
 234
 235         rc = kranal_sock_read(sock, connreq, sizeof(*connreq), timeout);
 236         if (rc != 0) {
 237                 CERROR("Read failed: %d\n", rc);
 238                 return rc;
 239         }
 240
 241         if (connreq->racr_magic != RANAL_MSG_MAGIC) {
 242                 if (__swab32(connreq->racr_magic) != RANAL_MSG_MAGIC) {
 243                         CERROR("Unexpected magic %08x\n", connreq->racr_magic);
 244                         return -EPROTO;
 245                 }
 246
 247                 __swab32s(&connreq->racr_magic);
 248                 __swab16s(&connreq->racr_version);
 249                 __swab16s(&connreq->racr_devid);
 250                 __swab64s(&connreq->racr_nid);
 251                 __swab64s(&connreq->racr_incarnation);
 252                 __swab32s(&connreq->racr_timeout);
 253
 254                 __swab32s(&connreq->racr_riparams.FmaDomainHndl);
 255                 __swab32s(&connreq->racr_riparams.RcvCqHndl);
 256                 __swab32s(&connreq->racr_riparams.PTag);
 257                 __swab32s(&connreq->racr_riparams.CompletionCookie);
 258         }
 259
 260         if (connreq->racr_version != RANAL_MSG_VERSION) {
 261                 CERROR("Unexpected version %d\n", connreq->racr_version);
 262                 return -EPROTO;
 263         }
 264
 265         if (connreq->racr_nid == PTL_NID_ANY) {
 266                 CERROR("Received PTL_NID_ANY\n");
 267                 return -EPROTO;
 268         }
 269
 270         if (connreq->racr_timeout < RANAL_MIN_TIMEOUT) {
 271                 CERROR("Received timeout %d < MIN %d\n",
 272                        connreq->racr_timeout, RANAL_MIN_TIMEOUT);
 273                 return -EPROTO;
 274         }
 275
 276         for (i = 0; i < kranal_data.kra_ndevs; i++)
 277                 if (connreq->racr_devid ==
 278                     kranal_data.kra_devices[i].rad_id)
 279                         break;
 280
 281         if (i == kranal_data.kra_ndevs) {
 282                 CERROR("Can't match device %d\n", connreq->racr_devid);
 283                 return -ENODEV;
 284         }
 285
 286         return 0;
 287 }
 288
 289 int
 290 kranal_conn_isdup_locked(kra_peer_t *peer, __u64 incarnation)
 291 {
 292         kra_conn_t       *conn;
 293         struct list_head *tmp;
 294         int               loopback = 0;
 295
 296         list_for_each(tmp, &peer->rap_conns) {
 297                 conn = list_entry(tmp, kra_conn_t, rac_list);
 298
 299                 if (conn->rac_peer_incarnation < incarnation) {
 300                         /* Conns with an older incarnation get culled later */
 301                         continue;
 302                 }
 303
 304                 if (!loopback &&
 305                     conn->rac_peer_incarnation == incarnation &&
 306                     peer->rap_nid == kranal_lib.libnal_ni.ni_pid.nid) {
 307                         /* loopback creates 2 conns */
 308                         loopback = 1;
 309                         continue;
 310                 }
 311
 312                 return 1;
 313         }
 314
 315         return 0;
 316 }
 317
 318 void
 319 kranal_set_conn_uniqueness (kra_conn_t *conn)
 320 {
 321         unsigned long  flags;
 322
 323         write_lock_irqsave(&kranal_data.kra_global_lock, flags);
 324
 325         conn->rac_my_incarnation = kranal_data.kra_next_incarnation++;
 326
 327         do {    /* allocate a unique cqid */
 328                 conn->rac_cqid = kranal_data.kra_next_cqid++;
 329         } while (kranal_cqid2conn_locked(conn->rac_cqid) != NULL);
 330
 331
 332         write_unlock_irqrestore(&kranal_data.kra_global_lock, flags);
 333 }
 334
 335 int
 336 kranal_alloc_conn(kra_conn_t **connp, kra_device_t *dev)
 337 {
 338         kra_conn_t    *conn;
 339         RAP_RETURN     rrc;
 340
 341         LASSERT (!in_interrupt());
 342         PORTAL_ALLOC(conn, sizeof(*conn));
 343
 344         if (conn == NULL)
 345                 return -ENOMEM;
 346
 347         memset(conn, 0, sizeof(*conn));
 348         atomic_set(&conn->rac_refcount, 1);
 349         INIT_LIST_HEAD(&conn->rac_list);
 350         INIT_LIST_HEAD(&conn->rac_hashlist);
 351         INIT_LIST_HEAD(&conn->rac_fmaq);
 352         INIT_LIST_HEAD(&conn->rac_rdmaq);
 353         INIT_LIST_HEAD(&conn->rac_replyq);
 354         spin_lock_init(&conn->rac_lock);
 355
 356         kranal_set_conn_uniqueness(conn);
 357
 358         conn->rac_timeout = MAX(kranal_tunables.kra_timeout, RANAL_MIN_TIMEOUT);
 359         kranal_update_reaper_timeout(conn->rac_timeout);
 360
 361         rrc = RapkCreateRi(dev->rad_handle, conn->rac_cqid,
 362                            dev->rad_ptag,
 363                            dev->rad_rdma_cq, dev->rad_fma_cq,
 364                            &conn->rac_rihandle);
 365         if (rrc != RAP_SUCCESS) {
 366                 CERROR("RapkCreateRi failed: %d\n", rrc);
 367                 PORTAL_FREE(conn, sizeof(*conn));
 368                 return -ENETDOWN;
 369         }
 370
 371         atomic_inc(&kranal_data.kra_nconns);
 372         *connp = conn;
 373         return 0;
 374 }
 375
 376 void
 377 __kranal_conn_decref(kra_conn_t *conn)
 378 {
 379         kra_tx_t          *tx;
 380         RAP_RETURN         rrc;
 381
 382         LASSERT (!in_interrupt());
 383         LASSERT (!conn->rac_scheduled);
 384         LASSERT (list_empty(&conn->rac_list));
 385         LASSERT (list_empty(&conn->rac_hashlist));
 386         LASSERT (atomic_read(&conn->rac_refcount) == 0);
 387
 388         while (!list_empty(&conn->rac_fmaq)) {
 389                 tx = list_entry(conn->rac_fmaq.next, kra_tx_t, tx_list);
 390
 391                 list_del(&tx->tx_list);
 392                 kranal_tx_done(tx, -ECONNABORTED);
 393         }
 394
 395         /* We may not destroy this connection while it has RDMAs outstanding */
 396         LASSERT (list_empty(&conn->rac_rdmaq));
 397
 398         while (!list_empty(&conn->rac_replyq)) {
 399                 tx = list_entry(conn->rac_replyq.next, kra_tx_t, tx_list);
 400
 401                 list_del(&tx->tx_list);
 402                 kranal_tx_done(tx, -ECONNABORTED);
 403         }
 404
 405         rrc = RapkDestroyRi(conn->rac_device->rad_handle,
 406                             conn->rac_rihandle);
 407         LASSERT (rrc == RAP_SUCCESS);
 408
 409         if (conn->rac_peer != NULL)
 410                 kranal_peer_decref(conn->rac_peer);
 411
 412         PORTAL_FREE(conn, sizeof(*conn));
 413         atomic_dec(&kranal_data.kra_nconns);
 414 }
 415
 416 void
 417 kranal_terminate_conn_locked (kra_conn_t *conn)
 418 {
 419         kra_peer_t *peer = conn->rac_peer;
 420
 421         LASSERT (!in_interrupt());
 422         LASSERT (conn->rac_closing);
 423         LASSERT (!list_empty(&conn->rac_hashlist));
 424         LASSERT (list_empty(&conn->rac_list));
 425
 426         /* Remove from conn hash table (no new callbacks) */
 427         list_del_init(&conn->rac_hashlist);
 428         kranal_conn_decref(conn);
 429
 430         /* Conn is now just waiting for remaining refs to go */
 431 }
 432
 433 void
 434 kranal_close_conn_locked (kra_conn_t *conn, int error)
 435 {
 436         kra_peer_t        *peer = conn->rac_peer;
 437
 438         CDEBUG(error == 0 ? D_NET : D_ERROR,
 439                "closing conn to "LPX64": error %d\n", peer->rap_nid, error);
 440
 441         LASSERT (!in_interrupt());
 442         LASSERT (!conn->rac_closing);
 443         LASSERT (!list_empty(&conn->rac_hashlist));
 444         LASSERT (!list_empty(&conn->rac_list));
 445
 446         list_del_init(&conn->rac_list);
 447
 448         if (list_empty(&peer->rap_conns) &&
 449             peer->rap_persistence == 0) {
 450                 /* Non-persistent peer with no more conns... */
 451                 kranal_unlink_peer_locked(peer);
 452         }
 453
 454         conn->rac_closing = 1;
 455         kranal_schedule_conn(conn);
 456
 457         kranal_conn_decref(conn);               /* lose peer's ref */
 458 }
 459
 460 void
 461 kranal_close_conn (kra_conn_t *conn, int error)
 462 {
 463         unsigned long    flags;
 464
 465
 466         write_lock_irqsave(&kranal_data.kra_global_lock, flags);
 467
 468         if (!conn->rac_closing)
 469                 kranal_close_conn_locked(conn, error);
 470
 471         write_unlock_irqrestore(&kranal_data.kra_global_lock, flags);
 472 }
 473
 474 int
 475 kranal_passive_conn_handshake (struct socket *sock,
 476                                ptl_nid_t *peer_nidp, kra_conn_t **connp)
 477 {
 478         struct sockaddr_in   addr;
 479         __u32                peer_ip;
 480         unsigned int         peer_port;
 481         kra_connreq_t        connreq;
 482         ptl_nid_t            peer_nid;
 483         kra_conn_t          *conn;
 484         kra_device_t        *dev;
 485         RAP_RETURN           rrc;
 486         int                  rc;
 487         int                  len;
 488         int                  i;
 489
 490         len = sizeof(addr);
 491         rc = sock->ops->getname(sock, (struct sockaddr *)&addr, &len, 2);
 492         if (rc != 0) {
 493                 CERROR("Can't get peer's IP: %d\n", rc);
 494                 return rc;
 495         }
 496
 497         peer_ip = ntohl(addr.sin_addr.s_addr);
 498         peer_port = ntohs(addr.sin_port);
 499
 500         if (peer_port >= 1024) {
 501                 CERROR("Refusing unprivileged connection from %u.%u.%u.%u/%d\n",
 502                        HIPQUAD(peer_ip), peer_port);
 503                 return -ECONNREFUSED;
 504         }
 505
 506         rc = kranal_recv_connreq(sock, &connreq,
 507                                  kranal_tunables.kra_listener_timeout);
 508         if (rc != 0) {
 509                 CERROR("Can't rx connreq from %u.%u.%u.%u/%d: %d\n",
 510                        HIPQUAD(peer_ip), peer_port, rc);
 511                 return rc;
 512         }
 513
 514         peer_nid = connreq.racr_nid;
 515         LASSERT (peer_nid != PTL_NID_ANY);
 516
 517         for (i = 0;;i++) {
 518                 LASSERT(i < kranal_data.kra_ndevs);
 519                 dev = &kranal_data.kra_devices[i];
 520                 if (dev->rad_id == connreq.racr_devid)
 521                         break;
 522         }
 523
 524         rc = kranal_alloc_conn(&conn, dev);
 525         if (rc != 0)
 526                 return rc;
 527
 528         conn->rac_peer_incarnation = connreq.racr_incarnation;
 529         conn->rac_keepalive = RANAL_TIMEOUT2KEEPALIVE(connreq.racr_timeout);
 530         kranal_update_reaper_timeout(conn->rac_keepalive);
 531
 532         rrc = RapkSetRiParams(conn->rac_rihandle, &connreq.racr_riparams);
 533         if (rrc != RAP_SUCCESS) {
 534                 CERROR("Can't set riparams for "LPX64": %d\n", peer_nid, rrc);
 535                 kranal_conn_decref(conn);
 536                 return -EPROTO;
 537         }
 538
 539         kranal_pack_connreq(&connreq, conn);
 540
 541         rc = kranal_sock_write(sock, &connreq, sizeof(connreq));
 542         if (rc != 0) {
 543                 CERROR("Can't tx connreq to %u.%u.%u.%u/%d: %d\n",
 544                        HIPQUAD(peer_ip), peer_port, rc);
 545                 kranal_conn_decref(conn);
 546                 return rc;
 547         }
 548
 549         *connp = conn;
 550         *peer_nidp = peer_nid;
 551         return 0;
 552 }
 553
 554 int
 555 ranal_connect_sock(kra_peer_t *peer, struct socket **sockp)
 556 {
 557         struct sockaddr_in  locaddr;
 558         struct sockaddr_in  srvaddr;
 559         struct socket      *sock;
 560         unsigned int        port;
 561         int                 rc;
 562         int                 option;
 563         mm_segment_t        oldmm = get_fs();
 564         struct timeval      tv;
 565
 566         for (port = 1023; port >= 512; port--) {
 567
 568                 memset(&locaddr, 0, sizeof(locaddr));
 569                 locaddr.sin_family      = AF_INET;
 570                 locaddr.sin_port        = htons(port);
 571                 locaddr.sin_addr.s_addr = htonl(INADDR_ANY);
 572
 573                 memset (&srvaddr, 0, sizeof (srvaddr));
 574                 srvaddr.sin_family      = AF_INET;
 575                 srvaddr.sin_port        = htons (peer->rap_port);
 576                 srvaddr.sin_addr.s_addr = htonl (peer->rap_ip);
 577
 578                 rc = kranal_create_sock(&sock);
 579                 if (rc != 0)
 580                         return rc;
 581
 582                 rc = sock->ops->bind(sock,
 583                                      (struct sockaddr *)&locaddr, sizeof(locaddr));
 584                 if (rc != 0) {
 585                         sock_release(sock);
 586
 587                         if (rc == -EADDRINUSE) {
 588                                 CDEBUG(D_NET, "Port %d already in use\n", port);
 589                                 continue;
 590                         }
 591
 592                         CERROR("Can't bind to reserved port %d: %d\n", port, rc);
 593                         return rc;
 594                 }
 595
 596                 rc = sock->ops->connect(sock,
 597                                         (struct sockaddr *)&srvaddr, sizeof(srvaddr),
 598                                         0);
 599                 if (rc == 0) {
 600                         *sockp = sock;
 601                         return 0;
 602                 }
 603
 604                 sock_release(sock);
 605
 606                 if (rc != -EADDRNOTAVAIL) {
 607                         CERROR("Can't connect port %d to %u.%u.%u.%u/%d: %d\n",
 608                                port, HIPQUAD(peer->rap_ip), peer->rap_port, rc);
 609                         return rc;
 610                 }
 611
 612                 CDEBUG(D_NET, "Port %d not available for %u.%u.%u.%u/%d\n",
 613                        port, HIPQUAD(peer->rap_ip), peer->rap_port);
 614         }
 615
 616         /* all ports busy */
 617         return -EHOSTUNREACH;
 618 }
 619
 620
 621 int
 622 kranal_active_conn_handshake(kra_peer_t *peer, kra_conn_t **connp)
 623 {
 624         struct sockaddr_in  dstaddr;
 625         kra_connreq_t       connreq;
 626         kra_conn_t         *conn;
 627         kra_device_t       *dev;
 628         struct socket      *sock;
 629         RAP_RETURN          rrc;
 630         int                 rc;
 631         int                 idx;
 632
 633         idx = peer->rap_nid & 0x7fffffff;
 634         dev = &kranal_data.kra_devices[idx % kranal_data.kra_ndevs];
 635
 636         rc = kranal_alloc_conn(&conn, dev);
 637         if (rc != 0)
 638                 return rc;
 639
 640         kranal_pack_connreq(&connreq, conn);
 641
 642         rc = ranal_connect_sock(peer, &sock);
 643         if (rc != 0)
 644                 goto failed_0;
 645
 646         /* CAVEAT EMPTOR: the passive side receives with a SHORT rx timeout
 647          * immediately after accepting a connection, so we connect and then
 648          * send immediately. */
 649
 650         rc = kranal_sock_write(sock, &connreq, sizeof(connreq));
 651         if (rc != 0) {
 652                 CERROR("Can't tx connreq to %u.%u.%u.%u/%d: %d\n",
 653                        HIPQUAD(peer->rap_ip), peer->rap_port, rc);
 654                 goto failed_1;
 655         }
 656
 657         rc = kranal_recv_connreq(sock, &connreq, kranal_tunables.kra_timeout);
 658         if (rc != 0) {
 659                 CERROR("Can't rx connreq from %u.%u.%u.%u/%d: %d\n",
 660                        HIPQUAD(peer->rap_ip), peer->rap_port, rc);
 661                 goto failed_1;
 662         }
 663
 664         sock_release(sock);
 665         rc = -EPROTO;
 666
 667         if (connreq.racr_nid != peer->rap_nid) {
 668                 CERROR("Unexpected nid from %u.%u.%u.%u/%d: "
 669                        "received "LPX64" expected "LPX64"\n",
 670                        HIPQUAD(peer->rap_ip), peer->rap_port,
 671                        connreq.racr_nid, peer->rap_nid);
 672                 goto failed_0;
 673         }
 674
 675         if (connreq.racr_devid != dev->rad_id) {
 676                 CERROR("Unexpected device id from %u.%u.%u.%u/%d: "
 677                        "received %d expected %d\n",
 678                        HIPQUAD(peer->rap_ip), peer->rap_port,
 679                        connreq.racr_devid, dev->rad_id);
 680                 goto failed_0;
 681         }
 682
 683         conn->rac_peer_incarnation = connreq.racr_incarnation;
 684         conn->rac_keepalive = RANAL_TIMEOUT2KEEPALIVE(connreq.racr_timeout);
 685         kranal_update_reaper_timeout(conn->rac_keepalive);
 686
 687         rc = -ENETDOWN;
 688         rrc = RapkSetRiParams(conn->rac_rihandle, &connreq.racr_riparams);
 689         if (rrc != RAP_SUCCESS) {
 690                 CERROR("Can't set riparams for "LPX64": %d\n",
 691                        peer->rap_nid, rrc);
 692                 goto failed_0;
 693         }
 694
 695         *connp = conn;
 696         return 0;
 697
 698  failed_1:
 699         sock_release(sock);
 700  failed_0:
 701         kranal_conn_decref(conn);
 702         return rc;
 703 }
 704
 705 int
 706 kranal_conn_handshake (struct socket *sock, kra_peer_t *peer)
 707 {
 708         kra_peer_t        *peer2;
 709         kra_tx_t          *tx;
 710         ptl_nid_t          peer_nid;
 711         unsigned long      flags;
 712         unsigned long      timeout;
 713         kra_conn_t        *conn;
 714         int                rc;
 715         int                nstale;
 716
 717         if (sock != NULL) {
 718                 /* passive: listener accepted sock */
 719                 LASSERT (peer == NULL);
 720
 721                 rc = kranal_passive_conn_handshake(sock, &peer_nid, &conn);
 722                 if (rc != 0)
 723                         return rc;
 724
 725                 /* assume this is a new peer */
 726                 peer = kranal_create_peer(peer_nid);
 727                 if (peer == NULL) {
 728                         CERROR("Can't allocate peer for "LPX64"\n", peer_nid);
 729                         kranal_conn_decref(conn);
 730                         return -ENOMEM;
 731                 }
 732
 733                 write_lock_irqsave(&kranal_data.kra_global_lock, flags);
 734
 735                 peer2 = kranal_find_peer_locked(peer_nid);
 736                 if (peer2 == NULL) {
 737                         /* peer table takes my initial ref on peer */
 738                         list_add_tail(&peer->rap_list,
 739                                       kranal_nid2peerlist(peer_nid));
 740                 } else {
 741                         /* peer_nid already in the peer table */
 742                         kranal_peer_decref(peer);
 743                         peer = peer2;
 744                 }
 745                 /* NB I may now have a non-persistent peer in the peer
 746                  * table with no connections: I can't drop the global lock
 747                  * until I've given it a connection or removed it, and when
 748                  * I do 'peer' can disappear under me. */
 749         } else {
 750                 /* active: connd wants to connect to peer */
 751                 LASSERT (peer != NULL);
 752                 LASSERT (peer->rap_connecting);
 753
 754                 rc = kranal_active_conn_handshake(peer, &conn);
 755                 if (rc != 0)
 756                         return rc;
 757
 758                 write_lock_irqsave(&kranal_data.kra_global_lock, flags);
 759
 760                 if (!kranal_peer_active(peer)) {
 761                         /* raced with peer getting unlinked */
 762                         write_unlock_irqrestore(&kranal_data.kra_global_lock,
 763                                                 flags);
 764                         kranal_conn_decref(conn);
 765                         return ESTALE;
 766                 }
 767         }
 768
 769         LASSERT (kranal_peer_active(peer));     /* peer is in the peer table */
 770         peer_nid = peer->rap_nid;
 771
 772         /* Refuse to duplicate an existing connection (both sides might try
 773          * to connect at once).  NB we return success!  We _do_ have a
 774          * connection (so we don't need to remove the peer from the peer
 775          * table) and we _don't_ have any blocked txs to complete */
 776         if (kranal_conn_isdup_locked(peer, conn->rac_peer_incarnation)) {
 777                 LASSERT (!list_empty(&peer->rap_conns));
 778                 LASSERT (list_empty(&peer->rap_tx_queue));
 779                 write_unlock_irqrestore(&kranal_data.kra_global_lock, flags);
 780                 CWARN("Not creating duplicate connection to "LPX64"\n",
 781                       peer_nid);
 782                 kranal_conn_decref(conn);
 783                 return 0;
 784         }
 785
 786         kranal_peer_addref(peer);               /* +1 ref for conn */
 787         conn->rac_peer = peer;
 788         list_add_tail(&conn->rac_list, &peer->rap_conns);
 789
 790         kranal_conn_addref(conn);               /* +1 ref for conn table */
 791         list_add_tail(&conn->rac_hashlist,
 792                       kranal_cqid2connlist(conn->rac_cqid));
 793
 794         /* Schedule all packets blocking for a connection */
 795         while (!list_empty(&peer->rap_tx_queue)) {
 796                 tx = list_entry(&peer->rap_tx_queue.next,
 797                                 kra_tx_t, tx_list);
 798
 799                 list_del(&tx->tx_list);
 800                 kranal_post_fma(conn, tx);
 801         }
 802
 803         nstale = kranal_close_stale_conns_locked(peer, conn->rac_peer_incarnation);
 804
 805         write_unlock_irqrestore(&kranal_data.kra_global_lock, flags);
 806
 807         /* CAVEAT EMPTOR: passive peer can disappear NOW */
 808
 809         if (nstale != 0)
 810                 CWARN("Closed %d stale conns to "LPX64"\n", nstale, peer_nid);
 811
 812         /* Ensure conn gets checked.  Transmits may have been queued and an
 813          * FMA event may have happened before it got in the cq hash table */
 814         kranal_schedule_conn(conn);
 815         return 0;
 816 }
 817
 818 void
 819 kranal_connect (kra_peer_t *peer)
 820 {
 821         kra_tx_t          *tx;
 822         unsigned long      flags;
 823         struct list_head   zombies;
 824         int                rc;
 825
 826         LASSERT (peer->rap_connecting);
 827
 828         rc = kranal_conn_handshake(NULL, peer);
 829
 830         write_lock_irqsave(&kranal_data.kra_global_lock, flags);
 831
 832         LASSERT (peer->rap_connecting);
 833         peer->rap_connecting = 0;
 834
 835         if (rc == 0) {
 836                 /* kranal_conn_handshake() queues blocked txs immediately on
 837                  * success to avoid messages jumping the queue */
 838                 LASSERT (list_empty(&peer->rap_tx_queue));
 839
 840                 /* reset reconnection timeouts */
 841                 peer->rap_reconnect_interval = RANAL_MIN_RECONNECT_INTERVAL;
 842                 peer->rap_reconnect_time = CURRENT_TIME;
 843
 844                 write_unlock_irqrestore(&kranal_data.kra_global_lock, flags);
 845                 return;
 846         }
 847
 848         LASSERT (peer->rap_reconnect_interval != 0);
 849         peer->rap_reconnect_time = CURRENT_TIME + peer->rap_reconnect_interval;
 850         peer->rap_reconnect_interval = MAX(RANAL_MAX_RECONNECT_INTERVAL,
 851                                            1 * peer->rap_reconnect_interval);
 852
 853         /* Grab all blocked packets while we have the global lock */
 854         list_add(&zombies, &peer->rap_tx_queue);
 855         list_del_init(&peer->rap_tx_queue);
 856
 857         write_unlock_irqrestore(&kranal_data.kra_global_lock, flags);
 858
 859         if (list_empty(&zombies))
 860                 return;
 861
 862         CWARN("Dropping packets for "LPX64": connection failed\n",
 863               peer->rap_nid);
 864
 865         do {
 866                 tx = list_entry(zombies.next, kra_tx_t, tx_list);
 867
 868                 list_del(&tx->tx_list);
 869                 kranal_tx_done(tx, -EHOSTUNREACH);
 870
 871         } while (!list_empty(&zombies));
 872 }
 873
 874 int
 875 kranal_listener(void *arg)
 876 {
 877         struct sockaddr_in addr;
 878         wait_queue_t       wait;
 879         struct socket     *sock;
 880         struct socket     *newsock;
 881         int                port;
 882         kra_connreq_t     *connreqs;
 883         char               name[16];
 884         int                rc;
 885
 886         /* Parent thread holds kra_nid_mutex, and is, or is about to
 887          * block on kra_listener_signal */
 888
 889         port = kranal_tunables.kra_port;
 890         snprintf(name, sizeof(name), "kranal_lstn%03d", port);
 891         kportal_daemonize(name);
 892         kportal_blockallsigs();
 893
 894         init_waitqueue_entry(&wait, current);
 895
 896         rc = -ENOMEM;
 897         PORTAL_ALLOC(connreqs, 2 * sizeof(*connreqs));
 898         if (connreqs == NULL)
 899                 goto out_0;
 900
 901         rc = kranal_create_sock(&sock);
 902         if (rc != 0)
 903                 goto out_1;
 904
 905         memset(&addr, 0, sizeof(addr));
 906         addr.sin_family      = AF_INET;
 907         addr.sin_port        = htons(port);
 908         addr.sin_addr.s_addr = INADDR_ANY;
 909
 910         rc = sock->ops->bind(sock, (struct sockaddr *)&addr, sizeof(addr));
 911         if (rc != 0) {
 912                 CERROR("Can't bind to port %d\n", port);
 913                 goto out_2;
 914         }
 915
 916         rc = sock->ops->listen(sock, kranal_tunables.kra_backlog);
 917         if (rc != 0) {
 918                 CERROR("Can't set listen backlog %d: %d\n",
 919                        kranal_tunables.kra_backlog, rc);
 920                 goto out_2;
 921         }
 922
 923         LASSERT (kranal_data.kra_listener_sock == NULL);
 924         kranal_data.kra_listener_sock = sock;
 925
 926         /* unblock waiting parent */
 927         LASSERT (kranal_data.kra_listener_shutdown == 0);
 928         up(&kranal_data.kra_listener_signal);
 929
 930         /* Wake me any time something happens on my socket */
 931         add_wait_queue(sock->sk->sk_sleep, &wait);
 932
 933         while (kranal_data.kra_listener_shutdown == 0) {
 934
 935                 newsock = sock_alloc();
 936                 if (newsock == NULL) {
 937                         CERROR("Can't allocate new socket for accept\n");
 938                         kranal_pause(HZ);
 939                         continue;
 940                 }
 941
 942                 set_current_state(TASK_INTERRUPTIBLE);
 943
 944                 rc = sock->ops->accept(sock, newsock, O_NONBLOCK);
 945
 946                 if (rc == -EAGAIN &&
 947                     kranal_data.kra_listener_shutdown == 0)
 948                         schedule();
 949
 950                 set_current_state(TASK_RUNNING);
 951
 952                 if (rc != 0) {
 953                         sock_release(newsock);
 954                         if (rc != -EAGAIN) {
 955                                 CERROR("Accept failed: %d\n", rc);
 956                                 kranal_pause(HZ);
 957                         }
 958                         continue;
 959                 }
 960
 961                 kranal_conn_handshake(newsock, NULL);
 962                 sock_release(newsock);
 963         }
 964
 965         rc = 0;
 966         remove_wait_queue(sock->sk->sk_sleep, &wait);
 967  out_2:
 968         sock_release(sock);
 969         kranal_data.kra_listener_sock = NULL;
 970  out_1:
 971         PORTAL_FREE(connreqs, 2 * sizeof(*connreqs));
 972  out_0:
 973         /* set completion status and unblock thread waiting for me
 974          * (parent on startup failure, executioner on normal shutdown) */
 975         kranal_data.kra_listener_shutdown = rc;
 976         up(&kranal_data.kra_listener_signal);
 977
 978         return 0;
 979 }
 980
 981 int
 982 kranal_start_listener (void)
 983 {
 984         long           pid;
 985         int            rc;
 986
 987         CDEBUG(D_WARNING, "Starting listener\n");
 988
 989         /* Called holding kra_nid_mutex: listener stopped */
 990         LASSERT (kranal_data.kra_listener_sock == NULL);
 991
 992         kranal_data.kra_listener_shutdown == 0;
 993         pid = kernel_thread(kranal_listener, NULL, 0);
 994         if (pid < 0) {
 995                 CERROR("Can't spawn listener: %ld\n", pid);
 996                 return (int)pid;
 997         }
 998
 999         /* Block until listener has started up. */
1000         down(&kranal_data.kra_listener_signal);
1001
1002         rc = kranal_data.kra_listener_shutdown;
1003         LASSERT ((rc != 0) == (kranal_data.kra_listener_sock == NULL));
1004
1005         CDEBUG(D_WARNING, "Listener %ld started OK\n", pid);
1006         return rc;
1007 }
1008
1009 void
1010 kranal_stop_listener(void)
1011 {
1012         CDEBUG(D_WARNING, "Stopping listener\n");
1013
1014         /* Called holding kra_nid_mutex: listener running */
1015         LASSERT (kranal_data.kra_listener_sock != NULL);
1016
1017         kranal_data.kra_listener_shutdown = 1;
1018         wake_up_all(kranal_data.kra_listener_sock->sk->sk_sleep);
1019
1020         /* Block until listener has torn down. */
1021         down(&kranal_data.kra_listener_signal);
1022
1023         LASSERT (kranal_data.kra_listener_sock == NULL);
1024         CDEBUG(D_WARNING, "Listener stopped\n");
1025 }
1026
1027 int
1028 kranal_listener_procint(ctl_table *table, int write, struct file *filp,
1029                         void *buffer, size_t *lenp)
1030 {
1031         int   *tunable = (int *)table->data;
1032         int    old_val;
1033         int    rc;
1034
1035         down(&kranal_data.kra_nid_mutex);
1036
1037         LASSERT (tunable == &kranal_tunables.kra_port ||
1038                  tunable == &kranal_tunables.kra_backlog);
1039         old_val = *tunable;
1040
1041         rc = proc_dointvec(table, write, filp, buffer, lenp);
1042
1043         if (write &&
1044             (*tunable != old_val ||
1045              kranal_data.kra_listener_sock == NULL)) {
1046
1047                 if (kranal_data.kra_listener_sock != NULL)
1048                         kranal_stop_listener();
1049
1050                 rc = kranal_start_listener();
1051
1052                 if (rc != 0) {
1053                         *tunable = old_val;
1054                         kranal_start_listener();
1055                 }
1056         }
1057
1058         up(&kranal_data.kra_nid_mutex);
1059         return rc;
1060 }
1061
1062 int
1063 kranal_set_mynid(ptl_nid_t nid)
1064 {
1065         lib_ni_t      *ni = &kranal_lib.libnal_ni;
1066         int            rc = 0;
1067
1068         CDEBUG(D_NET, "setting mynid to "LPX64" (old nid="LPX64")\n",
1069                nid, ni->ni_pid.nid);
1070
1071         down(&kranal_data.kra_nid_mutex);
1072
1073         if (nid == ni->ni_pid.nid) {
1074                 /* no change of NID */
1075                 up(&kranal_data.kra_nid_mutex);
1076                 return 0;
1077         }
1078
1079         if (kranal_data.kra_listener_sock != NULL)
1080                 kranal_stop_listener();
1081
1082         ni->ni_pid.nid = nid;
1083
1084         /* Delete all existing peers and their connections after new
1085          * NID/incarnation set to ensure no old connections in our brave
1086          * new world. */
1087         kranal_del_peer(PTL_NID_ANY, 0);
1088
1089         if (nid != PTL_NID_ANY)
1090                 rc = kranal_start_listener();
1091
1092         up(&kranal_data.kra_nid_mutex);
1093         return rc;
1094 }
1095
1096 kra_peer_t *
1097 kranal_create_peer (ptl_nid_t nid)
1098 {
1099         kra_peer_t *peer;
1100
1101         LASSERT (nid != PTL_NID_ANY);
1102
1103         PORTAL_ALLOC(peer, sizeof(*peer));
1104         if (peer == NULL)
1105                 return NULL;
1106
1107         memset(peer, 0, sizeof(*peer));         /* zero flags etc */
1108
1109         peer->rap_nid = nid;
1110         atomic_set(&peer->rap_refcount, 1);     /* 1 ref for caller */
1111
1112         INIT_LIST_HEAD(&peer->rap_list);        /* not in the peer table yet */
1113         INIT_LIST_HEAD(&peer->rap_conns);
1114         INIT_LIST_HEAD(&peer->rap_tx_queue);
1115
1116         peer->rap_reconnect_time = CURRENT_TIME;
1117         peer->rap_reconnect_interval = RANAL_MIN_RECONNECT_INTERVAL;
1118
1119         atomic_inc(&kranal_data.kra_npeers);
1120         return peer;
1121 }
1122
1123 void
1124 __kranal_peer_decref (kra_peer_t *peer)
1125 {
1126         CDEBUG(D_NET, "peer "LPX64" %p deleted\n", peer->rap_nid, peer);
1127
1128         LASSERT (atomic_read(&peer->rap_refcount) == 0);
1129         LASSERT (peer->rap_persistence == 0);
1130         LASSERT (!kranal_peer_active(peer));
1131         LASSERT (peer->rap_connecting == 0);
1132         LASSERT (list_empty(&peer->rap_conns));
1133         LASSERT (list_empty(&peer->rap_tx_queue));
1134
1135         PORTAL_FREE(peer, sizeof(*peer));
1136
1137         /* NB a peer's connections keep a reference on their peer until
1138          * they are destroyed, so we can be assured that _all_ state to do
1139          * with this peer has been cleaned up when its refcount drops to
1140          * zero. */
1141         atomic_dec(&kranal_data.kra_npeers);
1142 }
1143
1144 kra_peer_t *
1145 kranal_find_peer_locked (ptl_nid_t nid)
1146 {
1147         struct list_head *peer_list = kranal_nid2peerlist(nid);
1148         struct list_head *tmp;
1149         kra_peer_t       *peer;
1150
1151         list_for_each (tmp, peer_list) {
1152
1153                 peer = list_entry(tmp, kra_peer_t, rap_list);
1154
1155                 LASSERT (peer->rap_persistence > 0 ||     /* persistent peer */
1156                          !list_empty(&peer->rap_conns));  /* active conn */
1157
1158                 if (peer->rap_nid != nid)
1159                         continue;
1160
1161                 CDEBUG(D_NET, "got peer [%p] -> "LPX64" (%d)\n",
1162                        peer, nid, atomic_read(&peer->rap_refcount));
1163                 return peer;
1164         }
1165         return NULL;
1166 }
1167
1168 kra_peer_t *
1169 kranal_find_peer (ptl_nid_t nid)
1170 {
1171         kra_peer_t     *peer;
1172
1173         read_lock(&kranal_data.kra_global_lock);
1174         peer = kranal_find_peer_locked(nid);
1175         if (peer != NULL)                       /* +1 ref for caller? */
1176                 kranal_peer_addref(peer);
1177         read_unlock(&kranal_data.kra_global_lock);
1178
1179         return peer;
1180 }
1181
1182 void
1183 kranal_unlink_peer_locked (kra_peer_t *peer)
1184 {
1185         LASSERT (peer->rap_persistence == 0);
1186         LASSERT (list_empty(&peer->rap_conns));
1187
1188         LASSERT (kranal_peer_active(peer));
1189         list_del_init(&peer->rap_list);
1190
1191         /* lose peerlist's ref */
1192         kranal_peer_decref(peer);
1193 }
1194
1195 int
1196 kranal_get_peer_info (int index, ptl_nid_t *nidp, __u32 *ipp, int *portp,
1197                       int *persistencep)
1198 {
1199         kra_peer_t        *peer;
1200         struct list_head  *ptmp;
1201         int                i;
1202
1203         read_lock(&kranal_data.kra_global_lock);
1204
1205         for (i = 0; i < kranal_data.kra_peer_hash_size; i++) {
1206
1207                 list_for_each(ptmp, &kranal_data.kra_peers[i]) {
1208
1209                         peer = list_entry(ptmp, kra_peer_t, rap_list);
1210                         LASSERT (peer->rap_persistence > 0 ||
1211                                  !list_empty(&peer->rap_conns));
1212
1213                         if (index-- > 0)
1214                                 continue;
1215
1216                         *nidp = peer->rap_nid;
1217                         *ipp = peer->rap_ip;
1218                         *portp = peer->rap_port;
1219                         *persistencep = peer->rap_persistence;
1220
1221                         read_unlock(&kranal_data.kra_global_lock);
1222                         return 0;
1223                 }
1224         }
1225
1226         read_unlock(&kranal_data.kra_global_lock);
1227         return -ENOENT;
1228 }
1229
1230 int
1231 kranal_add_persistent_peer (ptl_nid_t nid, __u32 ip, int port)
1232 {
1233         unsigned long      flags;
1234         kra_peer_t        *peer;
1235         kra_peer_t        *peer2;
1236
1237         if (nid == PTL_NID_ANY)
1238                 return -EINVAL;
1239
1240         peer = kranal_create_peer(nid);
1241         if (peer == NULL)
1242                 return -ENOMEM;
1243
1244         write_lock_irqsave(&kranal_data.kra_global_lock, flags);
1245
1246         peer2 = kranal_find_peer_locked(nid);
1247         if (peer2 != NULL) {
1248                 kranal_peer_decref(peer);
1249                 peer = peer2;
1250         } else {
1251                 /* peer table takes existing ref on peer */
1252                 list_add_tail(&peer->rap_list,
1253                               kranal_nid2peerlist(nid));
1254         }
1255
1256         peer->rap_ip = ip;
1257         peer->rap_port = port;
1258         peer->rap_persistence++;
1259
1260         write_unlock_irqrestore(&kranal_data.kra_global_lock, flags);
1261         return 0;
1262 }
1263
1264 void
1265 kranal_del_peer_locked (kra_peer_t *peer, int single_share)
1266 {
1267         struct list_head *ctmp;
1268         struct list_head *cnxt;
1269         kra_conn_t       *conn;
1270
1271         if (!single_share)
1272                 peer->rap_persistence = 0;
1273         else if (peer->rap_persistence > 0)
1274                 peer->rap_persistence--;
1275
1276         if (peer->rap_persistence != 0)
1277                 return;
1278
1279         if (list_empty(&peer->rap_conns)) {
1280                 kranal_unlink_peer_locked(peer);
1281         } else {
1282                 list_for_each_safe(ctmp, cnxt, &peer->rap_conns) {
1283                         conn = list_entry(ctmp, kra_conn_t, rac_list);
1284
1285                         kranal_close_conn_locked(conn, 0);
1286                 }
1287                 /* peer unlinks itself when last conn is closed */
1288         }
1289 }
1290
1291 int
1292 kranal_del_peer (ptl_nid_t nid, int single_share)
1293 {
1294         unsigned long      flags;
1295         struct list_head  *ptmp;
1296         struct list_head  *pnxt;
1297         kra_peer_t        *peer;
1298         int                lo;
1299         int                hi;
1300         int                i;
1301         int                rc = -ENOENT;
1302
1303         write_lock_irqsave(&kranal_data.kra_global_lock, flags);
1304
1305         if (nid != PTL_NID_ANY)
1306                 lo = hi = kranal_nid2peerlist(nid) - kranal_data.kra_peers;
1307         else {
1308                 lo = 0;
1309                 hi = kranal_data.kra_peer_hash_size - 1;
1310         }
1311
1312         for (i = lo; i <= hi; i++) {
1313                 list_for_each_safe (ptmp, pnxt, &kranal_data.kra_peers[i]) {
1314                         peer = list_entry(ptmp, kra_peer_t, rap_list);
1315                         LASSERT (peer->rap_persistence > 0 ||
1316                                  !list_empty(&peer->rap_conns));
1317
1318                         if (!(nid == PTL_NID_ANY || peer->rap_nid == nid))
1319                                 continue;
1320
1321                         kranal_del_peer_locked(peer, single_share);
1322                         rc = 0;         /* matched something */
1323
1324                         if (single_share)
1325                                 goto out;
1326                 }
1327         }
1328  out:
1329         write_unlock_irqrestore(&kranal_data.kra_global_lock, flags);
1330
1331         return rc;
1332 }
1333
1334 kra_conn_t *
1335 kranal_get_conn_by_idx (int index)
1336 {
1337         kra_peer_t        *peer;
1338         struct list_head  *ptmp;
1339         kra_conn_t        *conn;
1340         struct list_head  *ctmp;
1341         int                i;
1342
1343         read_lock (&kranal_data.kra_global_lock);
1344
1345         for (i = 0; i < kranal_data.kra_peer_hash_size; i++) {
1346                 list_for_each (ptmp, &kranal_data.kra_peers[i]) {
1347
1348                         peer = list_entry(ptmp, kra_peer_t, rap_list);
1349                         LASSERT (peer->rap_persistence > 0 ||
1350                                  !list_empty(&peer->rap_conns));
1351
1352                         list_for_each (ctmp, &peer->rap_conns) {
1353                                 if (index-- > 0)
1354                                         continue;
1355
1356                                 conn = list_entry(ctmp, kra_conn_t, rac_list);
1357                                 CDEBUG(D_NET, "++conn[%p] -> "LPX64" (%d)\n",
1358                                        conn, conn->rac_peer->rap_nid,
1359                                        atomic_read(&conn->rac_refcount));
1360                                 atomic_inc(&conn->rac_refcount);
1361                                 read_unlock(&kranal_data.kra_global_lock);
1362                                 return conn;
1363                         }
1364                 }
1365         }
1366
1367         read_unlock(&kranal_data.kra_global_lock);
1368         return NULL;
1369 }
1370
1371 int
1372 kranal_close_peer_conns_locked (kra_peer_t *peer, int why)
1373 {
1374         kra_conn_t         *conn;
1375         struct list_head   *ctmp;
1376         struct list_head   *cnxt;
1377         int                 count = 0;
1378
1379         list_for_each_safe (ctmp, cnxt, &peer->rap_conns) {
1380                 conn = list_entry(ctmp, kra_conn_t, rac_list);
1381
1382                 count++;
1383                 kranal_close_conn_locked(conn, why);
1384         }
1385
1386         return count;
1387 }
1388
1389 int
1390 kranal_close_stale_conns_locked (kra_peer_t *peer, __u64 incarnation)
1391 {
1392         kra_conn_t         *conn;
1393         struct list_head   *ctmp;
1394         struct list_head   *cnxt;
1395         int                 count = 0;
1396
1397         list_for_each_safe (ctmp, cnxt, &peer->rap_conns) {
1398                 conn = list_entry(ctmp, kra_conn_t, rac_list);
1399
1400                 if (conn->rac_peer_incarnation == incarnation)
1401                         continue;
1402
1403                 CDEBUG(D_NET, "Closing stale conn nid:"LPX64" incarnation:"LPX64"("LPX64")\n",
1404                        peer->rap_nid, conn->rac_peer_incarnation, incarnation);
1405                 LASSERT (conn->rac_peer_incarnation < incarnation);
1406
1407                 count++;
1408                 kranal_close_conn_locked(conn, -ESTALE);
1409         }
1410
1411         return count;
1412 }
1413
1414 int
1415 kranal_close_matching_conns (ptl_nid_t nid)
1416 {
1417         unsigned long       flags;
1418         kra_peer_t         *peer;
1419         struct list_head   *ptmp;
1420         struct list_head   *pnxt;
1421         int                 lo;
1422         int                 hi;
1423         int                 i;
1424         int                 count = 0;
1425
1426         write_lock_irqsave(&kranal_data.kra_global_lock, flags);
1427
1428         if (nid != PTL_NID_ANY)
1429                 lo = hi = kranal_nid2peerlist(nid) - kranal_data.kra_peers;
1430         else {
1431                 lo = 0;
1432                 hi = kranal_data.kra_peer_hash_size - 1;
1433         }
1434
1435         for (i = lo; i <= hi; i++) {
1436                 list_for_each_safe (ptmp, pnxt, &kranal_data.kra_peers[i]) {
1437
1438                         peer = list_entry(ptmp, kra_peer_t, rap_list);
1439                         LASSERT (peer->rap_persistence > 0 ||
1440                                  !list_empty(&peer->rap_conns));
1441
1442                         if (!(nid == PTL_NID_ANY || nid == peer->rap_nid))
1443                                 continue;
1444
1445                         count += kranal_close_peer_conns_locked(peer, 0);
1446                 }
1447         }
1448
1449         write_unlock_irqrestore(&kranal_data.kra_global_lock, flags);
1450
1451         /* wildcards always succeed */
1452         if (nid == PTL_NID_ANY)
1453                 return 0;
1454
1455         return (count == 0) ? -ENOENT : 0;
1456 }
1457
1458 int
1459 kranal_cmd(struct portals_cfg *pcfg, void * private)
1460 {
1461         int rc = -EINVAL;
1462
1463         LASSERT (pcfg != NULL);
1464
1465         switch(pcfg->pcfg_command) {
1466         case NAL_CMD_GET_PEER: {
1467                 ptl_nid_t   nid = 0;
1468                 __u32       ip = 0;
1469                 int         port = 0;
1470                 int         share_count = 0;
1471
1472                 rc = kranal_get_peer_info(pcfg->pcfg_count,
1473                                           &nid, &ip, &port, &share_count);
1474                 pcfg->pcfg_nid   = nid;
1475                 pcfg->pcfg_size  = 0;
1476                 pcfg->pcfg_id    = ip;
1477                 pcfg->pcfg_misc  = port;
1478                 pcfg->pcfg_count = 0;
1479                 pcfg->pcfg_wait  = share_count;
1480                 break;
1481         }
1482         case NAL_CMD_ADD_PEER: {
1483                 rc = kranal_add_persistent_peer(pcfg->pcfg_nid,
1484                                                 pcfg->pcfg_id, /* IP */
1485                                                 pcfg->pcfg_misc); /* port */
1486                 break;
1487         }
1488         case NAL_CMD_DEL_PEER: {
1489                 rc = kranal_del_peer(pcfg->pcfg_nid,
1490                                      /* flags == single_share */
1491                                      pcfg->pcfg_flags != 0);
1492                 break;
1493         }
1494         case NAL_CMD_GET_CONN: {
1495                 kra_conn_t *conn = kranal_get_conn_by_idx(pcfg->pcfg_count);
1496
1497                 if (conn == NULL)
1498                         rc = -ENOENT;
1499                 else {
1500                         rc = 0;
1501                         pcfg->pcfg_nid   = conn->rac_peer->rap_nid;
1502                         pcfg->pcfg_id    = 0;
1503                         pcfg->pcfg_misc  = 0;
1504                         pcfg->pcfg_flags = 0;
1505                         kranal_conn_decref(conn);
1506                 }
1507                 break;
1508         }
1509         case NAL_CMD_CLOSE_CONNECTION: {
1510                 rc = kranal_close_matching_conns(pcfg->pcfg_nid);
1511                 break;
1512         }
1513         case NAL_CMD_REGISTER_MYNID: {
1514                 if (pcfg->pcfg_nid == PTL_NID_ANY)
1515                         rc = -EINVAL;
1516                 else
1517                         rc = kranal_set_mynid(pcfg->pcfg_nid);
1518                 break;
1519         }
1520         }
1521
1522         return rc;
1523 }
1524
1525 void
1526 kranal_free_txdescs(struct list_head *freelist)
1527 {
1528         kra_tx_t    *tx;
1529
1530         while (!list_empty(freelist)) {
1531                 tx = list_entry(freelist->next, kra_tx_t, tx_list);
1532
1533                 list_del(&tx->tx_list);
1534                 PORTAL_FREE(tx->tx_phys, PTL_MD_MAX_IOV * sizeof(*tx->tx_phys));
1535                 PORTAL_FREE(tx, sizeof(*tx));
1536         }
1537 }
1538
1539 int
1540 kranal_alloc_txdescs(struct list_head *freelist, int n)
1541 {
1542         int            isnblk = (freelist == &kranal_data.kra_idle_nblk_txs);
1543         int            i;
1544         kra_tx_t      *tx;
1545
1546         LASSERT (freelist == &kranal_data.kra_idle_txs ||
1547                  freelist == &kranal_data.kra_idle_nblk_txs);
1548         LASSERT (list_empty(freelist));
1549
1550         for (i = 0; i < n; i++) {
1551
1552                 PORTAL_ALLOC(tx, sizeof(*tx));
1553                 if (tx == NULL) {
1554                         CERROR("Can't allocate %stx[%d]\n",
1555                                isnblk ? "nblk " : "", i);
1556                         kranal_free_txdescs(freelist);
1557                         return -ENOMEM;
1558                 }
1559
1560                 PORTAL_ALLOC(tx->tx_phys,
1561                              PTL_MD_MAX_IOV * sizeof(*tx->tx_phys));
1562                 if (tx->tx_phys == NULL) {
1563                         CERROR("Can't allocate %stx[%d]->tx_phys\n",
1564                                isnblk ? "nblk " : "", i);
1565
1566                         PORTAL_FREE(tx, sizeof(*tx));
1567                         kranal_free_txdescs(freelist);
1568                         return -ENOMEM;
1569                 }
1570
1571                 tx->tx_isnblk = isnblk;
1572                 tx->tx_buftype = RANAL_BUF_NONE;
1573
1574                 list_add(&tx->tx_list, freelist);
1575         }
1576
1577         return 0;
1578 }
1579
1580 int
1581 kranal_device_init(int id, kra_device_t *dev)
1582 {
1583         const int         total_ntx = RANAL_NTX + RANAL_NTX_NBLK;
1584         RAP_RETURN        rrc;
1585
1586         dev->rad_id = id;
1587         rrc = RapkGetDeviceByIndex(id, kranal_device_callback,
1588                                    &dev->rad_handle);
1589         if (rrc != RAP_SUCCESS) {
1590                 CERROR("Can't get Rapidarray Device %d: %d\n", id, rrc);
1591                 goto failed_0;
1592         }
1593
1594         rrc = RapkReserveRdma(dev->rad_handle, total_ntx);
1595         if (rrc != RAP_SUCCESS) {
1596                 CERROR("Can't reserve %d RDMA descriptors"
1597                        " for device %d: %d\n", total_ntx, id, rrc);
1598                 goto failed_1;
1599         }
1600
1601         rrc = RapkCreatePtag(dev->rad_handle,
1602                              &dev->rad_ptag);
1603         if (rrc != RAP_SUCCESS) {
1604                 CERROR("Can't create ptag"
1605                        " for device %d: %d\n", id, rrc);
1606                 goto failed_1;
1607         }
1608
1609         rrc = RapkCreateCQ(dev->rad_handle, total_ntx, dev->rad_ptag,
1610                            &dev->rad_rdma_cq);
1611         if (rrc != RAP_SUCCESS) {
1612                 CERROR("Can't create rdma cq size %d"
1613                        " for device %d: %d\n", total_ntx, id, rrc);
1614                 goto failed_2;
1615         }
1616
1617         rrc = RapkCreateCQ(dev->rad_handle, RANAL_FMA_CQ_SIZE,
1618                            dev->rad_ptag, &dev->rad_fma_cq);
1619         if (rrc != RAP_SUCCESS) {
1620                 CERROR("Can't create fma cq size %d"
1621                        " for device %d: %d\n", RANAL_FMA_CQ_SIZE, id, rrc);
1622                 goto failed_3;
1623         }
1624
1625         return 0;
1626
1627  failed_3:
1628         RapkDestroyCQ(dev->rad_handle, dev->rad_rdma_cq, dev->rad_ptag);
1629  failed_2:
1630         RapkDestroyPtag(dev->rad_handle, dev->rad_ptag);
1631  failed_1:
1632         RapkReleaseDevice(dev->rad_handle);
1633  failed_0:
1634         return -ENODEV;
1635 }
1636
1637 void
1638 kranal_device_fini(kra_device_t *dev)
1639 {
1640         RapkDestroyCQ(dev->rad_handle, dev->rad_fma_cq, dev->rad_ptag);
1641         RapkDestroyCQ(dev->rad_handle, dev->rad_rdma_cq, dev->rad_ptag);
1642         RapkDestroyPtag(dev->rad_handle, dev->rad_ptag);
1643         RapkReleaseDevice(dev->rad_handle);
1644 }
1645
1646 void
1647 kranal_api_shutdown (nal_t *nal)
1648 {
1649         int           i;
1650         int           rc;
1651         unsigned long flags;
1652
1653         if (nal->nal_refct != 0) {
1654                 /* This module got the first ref */
1655                 PORTAL_MODULE_UNUSE;
1656                 return;
1657         }
1658
1659         CDEBUG(D_MALLOC, "before NAL cleanup: kmem %d\n",
1660                atomic_read(&portal_kmemory));
1661
1662         LASSERT (nal == &kranal_api);
1663
1664         switch (kranal_data.kra_init) {
1665         default:
1666                 CERROR("Unexpected state %d\n", kranal_data.kra_init);
1667                 LBUG();
1668
1669         case RANAL_INIT_ALL:
1670                 /* stop calls to nal_cmd */
1671                 libcfs_nal_cmd_unregister(RANAL);
1672                 /* No new persistent peers */
1673
1674                 /* resetting my NID to unadvertises me, removes my
1675                  * listener and nukes all current peers */
1676                 kranal_set_mynid(PTL_NID_ANY);
1677                 /* no new peers or conns */
1678
1679                 /* Wait for all peer/conn state to clean up */
1680                 i = 2;
1681                 while (atomic_read(&kranal_data.kra_nconns) != 0 ||
1682                        atomic_read(&kranal_data.kra_npeers) != 0) {
1683                         i++;
1684                         CDEBUG(((i & (-i)) == i) ? D_WARNING : D_NET, /* power of 2? */
1685                                "waiting for %d peers and %d conns to close down\n",
1686                                atomic_read(&kranal_data.kra_npeers),
1687                                atomic_read(&kranal_data.kra_nconns));
1688                         kranal_pause(HZ);
1689                 }
1690                 /* fall through */
1691
1692         case RANAL_INIT_LIB:
1693                 lib_fini(&kranal_lib);
1694                 /* fall through */
1695
1696         case RANAL_INIT_DATA:
1697                 break;
1698         }
1699
1700         /* flag threads to terminate; wake and wait for them to die */
1701         kranal_data.kra_shutdown = 1;
1702
1703         for (i = 0; i < kranal_data.kra_ndevs; i++) {
1704                 kra_device_t *dev = &kranal_data.kra_devices[i];
1705
1706                 LASSERT (list_empty(&dev->rad_connq));
1707
1708                 spin_lock_irqsave(&dev->rad_lock, flags);
1709                 wake_up(&dev->rad_waitq);
1710                 spin_unlock_irqrestore(&dev->rad_lock, flags);
1711         }
1712
1713         spin_lock_irqsave(&kranal_data.kra_reaper_lock, flags);
1714         wake_up_all(&kranal_data.kra_reaper_waitq);
1715         spin_unlock_irqrestore(&kranal_data.kra_reaper_lock, flags);
1716
1717         LASSERT (list_empty(&kranal_data.kra_connd_peers));
1718         spin_lock_irqsave(&kranal_data.kra_connd_lock, flags);
1719         wake_up_all(&kranal_data.kra_connd_waitq);
1720         spin_unlock_irqrestore(&kranal_data.kra_connd_lock, flags);
1721
1722         i = 2;
1723         while (atomic_read(&kranal_data.kra_nthreads) != 0) {
1724                 i++;
1725                 CDEBUG(((i & (-i)) == i) ? D_WARNING : D_NET, /* power of 2? */
1726                        "Waiting for %d threads to terminate\n",
1727                        atomic_read(&kranal_data.kra_nthreads));
1728                 kranal_pause(HZ);
1729         }
1730
1731         LASSERT (atomic_read(&kranal_data.kra_npeers) == 0);
1732         if (kranal_data.kra_peers != NULL) {
1733                 for (i = 0; i < kranal_data.kra_peer_hash_size; i++)
1734                         LASSERT (list_empty(&kranal_data.kra_peers[i]));
1735
1736                 PORTAL_FREE(kranal_data.kra_peers,
1737                             sizeof (struct list_head) *
1738                             kranal_data.kra_peer_hash_size);
1739         }
1740
1741         LASSERT (atomic_read(&kranal_data.kra_nconns) == 0);
1742         if (kranal_data.kra_conns != NULL) {
1743                 for (i = 0; i < kranal_data.kra_conn_hash_size; i++)
1744                         LASSERT (list_empty(&kranal_data.kra_conns[i]));
1745
1746                 PORTAL_FREE(kranal_data.kra_conns,
1747                             sizeof (struct list_head) *
1748                             kranal_data.kra_conn_hash_size);
1749         }
1750
1751         for (i = 0; i < kranal_data.kra_ndevs; i++)
1752                 kranal_device_fini(&kranal_data.kra_devices[i]);
1753
1754         kranal_free_txdescs(&kranal_data.kra_idle_txs);
1755         kranal_free_txdescs(&kranal_data.kra_idle_nblk_txs);
1756
1757         CDEBUG(D_MALLOC, "after NAL cleanup: kmem %d\n",
1758                atomic_read(&portal_kmemory));
1759         printk(KERN_INFO "Lustre: RapidArray NAL unloaded (final mem %d)\n",
1760                atomic_read(&portal_kmemory));
1761
1762         kranal_data.kra_init = RANAL_INIT_NOTHING;
1763 }
1764
1765 int
1766 kranal_api_startup (nal_t *nal, ptl_pid_t requested_pid,
1767                     ptl_ni_limits_t *requested_limits,
1768                     ptl_ni_limits_t *actual_limits)
1769 {
1770         static int        device_ids[] = {RAPK_MAIN_DEVICE_ID,
1771                                           RAPK_EXPANSION_DEVICE_ID};
1772         struct timeval    tv;
1773         ptl_process_id_t  process_id;
1774         int               pkmem = atomic_read(&portal_kmemory);
1775         int               rc;
1776         int               i;
1777         kra_device_t     *dev;
1778
1779         LASSERT (nal == &kranal_api);
1780
1781         if (nal->nal_refct != 0) {
1782                 if (actual_limits != NULL)
1783                         *actual_limits = kranal_lib.libnal_ni.ni_actual_limits;
1784                 /* This module got the first ref */
1785                 PORTAL_MODULE_USE;
1786                 return PTL_OK;
1787         }
1788
1789         LASSERT (kranal_data.kra_init == RANAL_INIT_NOTHING);
1790
1791         memset(&kranal_data, 0, sizeof(kranal_data)); /* zero pointers, flags etc */
1792
1793         /* CAVEAT EMPTOR: Every 'Fma' message includes the sender's NID and
1794          * a unique (for all time) incarnation so we can uniquely identify
1795          * the sender.  The incarnation is an incrementing counter
1796          * initialised with seconds + microseconds at startup time.  So we
1797          * rely on NOT creating connections more frequently on average than
1798          * 1MHz to ensure we don't use old incarnations when we reboot. */
1799         do_gettimeofday(&tv);
1800         kranal_data.kra_next_incarnation = (((__u64)tv.tv_sec) * 1000000) + tv.tv_usec;
1801
1802         init_MUTEX(&kranal_data.kra_nid_mutex);
1803         init_MUTEX_LOCKED(&kranal_data.kra_listener_signal);
1804
1805         rwlock_init(&kranal_data.kra_global_lock);
1806
1807         for (i = 0; i < RANAL_MAXDEVS; i++ ) {
1808                 kra_device_t  *dev = &kranal_data.kra_devices[i];
1809
1810                 dev->rad_idx = i;
1811                 INIT_LIST_HEAD(&dev->rad_connq);
1812                 init_waitqueue_head(&dev->rad_waitq);
1813                 spin_lock_init(&dev->rad_lock);
1814         }
1815
1816         init_waitqueue_head(&kranal_data.kra_reaper_waitq);
1817         spin_lock_init(&kranal_data.kra_reaper_lock);
1818
1819         INIT_LIST_HEAD(&kranal_data.kra_connd_peers);
1820         init_waitqueue_head(&kranal_data.kra_connd_waitq);
1821         spin_lock_init(&kranal_data.kra_connd_lock);
1822
1823         INIT_LIST_HEAD(&kranal_data.kra_idle_txs);
1824         INIT_LIST_HEAD(&kranal_data.kra_idle_nblk_txs);
1825         init_waitqueue_head(&kranal_data.kra_idle_tx_waitq);
1826         spin_lock_init(&kranal_data.kra_tx_lock);
1827
1828         /* OK to call kranal_api_shutdown() to cleanup now */
1829         kranal_data.kra_init = RANAL_INIT_DATA;
1830
1831         kranal_data.kra_peer_hash_size = RANAL_PEER_HASH_SIZE;
1832         PORTAL_ALLOC(kranal_data.kra_peers,
1833                      sizeof(struct list_head) * kranal_data.kra_peer_hash_size);
1834         if (kranal_data.kra_peers == NULL)
1835                 goto failed;
1836
1837         for (i = 0; i < kranal_data.kra_peer_hash_size; i++)
1838                 INIT_LIST_HEAD(&kranal_data.kra_peers[i]);
1839
1840         kranal_data.kra_conn_hash_size = RANAL_PEER_HASH_SIZE;
1841         PORTAL_ALLOC(kranal_data.kra_conns,
1842                      sizeof(struct list_head) * kranal_data.kra_conn_hash_size);
1843         if (kranal_data.kra_conns == NULL)
1844                 goto failed;
1845
1846         for (i = 0; i < kranal_data.kra_conn_hash_size; i++)
1847                 INIT_LIST_HEAD(&kranal_data.kra_conns[i]);
1848
1849         rc = kranal_alloc_txdescs(&kranal_data.kra_idle_txs, RANAL_NTX);
1850         if (rc != 0)
1851                 goto failed;
1852
1853         rc = kranal_alloc_txdescs(&kranal_data.kra_idle_nblk_txs,RANAL_NTX_NBLK);
1854         if (rc != 0)
1855                 goto failed;
1856
1857         process_id.pid = requested_pid;
1858         process_id.nid = PTL_NID_ANY;           /* don't know my NID yet */
1859
1860         rc = lib_init(&kranal_lib, nal, process_id,
1861                       requested_limits, actual_limits);
1862         if (rc != PTL_OK) {
1863                 CERROR("lib_init failed: error %d\n", rc);
1864                 goto failed;
1865         }
1866
1867         /* lib interface initialised */
1868         kranal_data.kra_init = RANAL_INIT_LIB;
1869         /*****************************************************/
1870
1871         rc = kranal_thread_start(kranal_reaper, NULL);
1872         if (rc != 0) {
1873                 CERROR("Can't spawn ranal reaper: %d\n", rc);
1874                 goto failed;
1875         }
1876
1877         for (i = 0; i < RANAL_N_CONND; i++) {
1878                 rc = kranal_thread_start(kranal_connd, (void *)i);
1879                 if (rc != 0) {
1880                         CERROR("Can't spawn ranal connd[%d]: %d\n",
1881                                i, rc);
1882                         goto failed;
1883                 }
1884         }
1885
1886         LASSERT(kranal_data.kra_ndevs == 0);
1887         for (i = 0; i < sizeof(device_ids)/sizeof(device_ids[0]); i++) {
1888                 dev = &kranal_data.kra_devices[kranal_data.kra_ndevs];
1889
1890                 rc = kranal_device_init(device_ids[i], dev);
1891                 if (rc == 0)
1892                         kranal_data.kra_ndevs++;
1893
1894                 rc = kranal_thread_start(kranal_scheduler, dev);
1895                 if (rc != 0) {
1896                         CERROR("Can't spawn ranal scheduler[%d]: %d\n",
1897                                i, rc);
1898                         goto failed;
1899                 }
1900         }
1901
1902         if (kranal_data.kra_ndevs == 0)
1903                 goto failed;
1904
1905         rc = libcfs_nal_cmd_register(RANAL, &kranal_cmd, NULL);
1906         if (rc != 0) {
1907                 CERROR("Can't initialise command interface (rc = %d)\n", rc);
1908                 goto failed;
1909         }
1910
1911         /* flag everything initialised */
1912         kranal_data.kra_init = RANAL_INIT_ALL;
1913         /*****************************************************/
1914
1915         CDEBUG(D_MALLOC, "initial kmem %d\n", atomic_read(&portal_kmemory));
1916         printk(KERN_INFO "Lustre: RapidArray NAL loaded "
1917                "(initial mem %d)\n", pkmem);
1918
1919         return PTL_OK;
1920
1921  failed:
1922         kranal_api_shutdown(&kranal_api);
1923         return PTL_FAIL;
1924 }
1925
1926 void __exit
1927 kranal_module_fini (void)
1928 {
1929 #ifdef CONFIG_SYSCTL
1930         if (kranal_tunables.kra_sysctl != NULL)
1931                 unregister_sysctl_table(kranal_tunables.kra_sysctl);
1932 #endif
1933         PtlNIFini(kranal_ni);
1934
1935         ptl_unregister_nal(RANAL);
1936 }
1937
1938 int __init
1939 kranal_module_init (void)
1940 {
1941         int    rc;
1942
1943         /* the following must be sizeof(int) for
1944          * proc_dointvec/kranal_listener_procint() */
1945         LASSERT (sizeof(kranal_tunables.kra_timeout) == sizeof(int));
1946         LASSERT (sizeof(kranal_tunables.kra_listener_timeout) == sizeof(int));
1947         LASSERT (sizeof(kranal_tunables.kra_backlog) == sizeof(int));
1948         LASSERT (sizeof(kranal_tunables.kra_port) == sizeof(int));
1949         LASSERT (sizeof(kranal_tunables.kra_max_immediate) == sizeof(int));
1950
1951         kranal_api.nal_ni_init = kranal_api_startup;
1952         kranal_api.nal_ni_fini = kranal_api_shutdown;
1953
1954         /* Initialise dynamic tunables to defaults once only */
1955         kranal_tunables.kra_timeout = RANAL_TIMEOUT;
1956
1957         rc = ptl_register_nal(RANAL, &kranal_api);
1958         if (rc != PTL_OK) {
1959                 CERROR("Can't register RANAL: %d\n", rc);
1960                 return -ENOMEM;               /* or something... */
1961         }
1962
1963         /* Pure gateways want the NAL started up at module load time... */
1964         rc = PtlNIInit(RANAL, LUSTRE_SRV_PTL_PID, NULL, NULL, &kranal_ni);
1965         if (rc != PTL_OK && rc != PTL_IFACE_DUP) {
1966                 ptl_unregister_nal(RANAL);
1967                 return -ENODEV;
1968         }
1969
1970 #ifdef CONFIG_SYSCTL
1971         /* Press on regardless even if registering sysctl doesn't work */
1972         kranal_tunables.kra_sysctl =
1973                 register_sysctl_table(kranal_top_ctl_table, 0);
1974 #endif
1975         return 0;
1976 }
1977
1978 MODULE_AUTHOR("Cluster File Systems, Inc. <info@clusterfs.com>");
1979 MODULE_DESCRIPTION("Kernel RapidArray NAL v0.01");
1980 MODULE_LICENSE("GPL");
1981
1982 module_init(kranal_module_init);
1983 module_exit(kranal_module_fini);