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