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