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         unsigned int        idx;
 696
 697         /* spread connections over all devices using both peer NIDs to ensure
 698          * all nids use all devices */
 699         idx = peer->rap_nid + kranal_lib.libnal_ni.ni_pid.nid;
 700         dev = &kranal_data.kra_devices[idx % kranal_data.kra_ndevs];
 701
 702         rc = kranal_create_conn(&conn, dev);
 703         if (rc != 0)
 704                 return rc;
 705
 706         kranal_pack_connreq(&connreq, conn);
 707
 708         rc = ranal_connect_sock(peer, &sock);
 709         if (rc != 0)
 710                 goto failed_0;
 711
 712         /* CAVEAT EMPTOR: the passive side receives with a SHORT rx timeout
 713          * immediately after accepting a connection, so we connect and then
 714          * send immediately. */
 715
 716         rc = kranal_sock_write(sock, &connreq, sizeof(connreq));
 717         if (rc != 0) {
 718                 CERROR("Can't tx connreq to %u.%u.%u.%u/%d: %d\n",
 719                        HIPQUAD(peer->rap_ip), peer->rap_port, rc);
 720                 goto failed_1;
 721         }
 722
 723         rc = kranal_recv_connreq(sock, &connreq, kranal_tunables.kra_timeout);
 724         if (rc != 0) {
 725                 CERROR("Can't rx connreq from %u.%u.%u.%u/%d: %d\n",
 726                        HIPQUAD(peer->rap_ip), peer->rap_port, rc);
 727                 goto failed_1;
 728         }
 729
 730         sock_release(sock);
 731         rc = -EPROTO;
 732
 733         if (connreq.racr_nid != peer->rap_nid) {
 734                 CERROR("Unexpected nid from %u.%u.%u.%u/%d: "
 735                        "received "LPX64" expected "LPX64"\n",
 736                        HIPQUAD(peer->rap_ip), peer->rap_port,
 737                        connreq.racr_nid, peer->rap_nid);
 738                 goto failed_0;
 739         }
 740
 741         if (connreq.racr_devid != dev->rad_id) {
 742                 CERROR("Unexpected device id from %u.%u.%u.%u/%d: "
 743                        "received %d expected %d\n",
 744                        HIPQUAD(peer->rap_ip), peer->rap_port,
 745                        connreq.racr_devid, dev->rad_id);
 746                 goto failed_0;
 747         }
 748
 749         rc = kranal_set_conn_params(conn, &connreq,
 750                                     peer->rap_ip, peer->rap_port);
 751         if (rc != 0)
 752                 goto failed_0;
 753
 754         *connp = conn;
 755         return 0;
 756
 757  failed_1:
 758         sock_release(sock);
 759  failed_0:
 760         kranal_conn_decref(conn);
 761         return rc;
 762 }
 763
 764 int
 765 kranal_conn_handshake (struct socket *sock, kra_peer_t *peer)
 766 {
 767         kra_peer_t        *peer2;
 768         kra_tx_t          *tx;
 769         ptl_nid_t          peer_nid;
 770         unsigned long      flags;
 771         kra_conn_t        *conn;
 772         int                rc;
 773         int                nstale;
 774
 775         if (sock == NULL) {
 776                 /* active: connd wants to connect to 'peer' */
 777                 LASSERT (peer != NULL);
 778                 LASSERT (peer->rap_connecting);
 779
 780                 rc = kranal_active_conn_handshake(peer, &conn);
 781                 if (rc != 0)
 782                         return rc;
 783
 784                 write_lock_irqsave(&kranal_data.kra_global_lock, flags);
 785
 786                 if (!kranal_peer_active(peer)) {
 787                         /* raced with peer getting unlinked */
 788                         write_unlock_irqrestore(&kranal_data.kra_global_lock,
 789                                                 flags);
 790                         kranal_conn_decref(conn);
 791                         return ESTALE;
 792                 }
 793
 794                 peer_nid = peer->rap_nid;
 795
 796         } else {
 797                 /* passive: listener accepted 'sock' */
 798                 LASSERT (peer == NULL);
 799
 800                 rc = kranal_passive_conn_handshake(sock, &peer_nid, &conn);
 801                 if (rc != 0)
 802                         return rc;
 803
 804                 /* assume this is a new peer */
 805                 peer = kranal_create_peer(peer_nid);
 806                 if (peer == NULL) {
 807                         CERROR("Can't allocate peer for "LPX64"\n", peer_nid);
 808                         kranal_conn_decref(conn);
 809                         return -ENOMEM;
 810                 }
 811
 812                 write_lock_irqsave(&kranal_data.kra_global_lock, flags);
 813
 814                 peer2 = kranal_find_peer_locked(peer_nid);
 815                 if (peer2 == NULL) {
 816                         /* peer table takes my initial ref on peer */
 817                         list_add_tail(&peer->rap_list,
 818                                       kranal_nid2peerlist(peer_nid));
 819                 } else {
 820                         /* peer_nid already in the peer table */
 821                         kranal_peer_decref(peer);
 822                         peer = peer2;
 823                 }
 824                 /* NB I may now have a non-persistent peer in the peer
 825                  * table with no connections: I can't drop the global lock
 826                  * until I've given it a connection or removed it, and when
 827                  * I do 'peer' can disappear under me. */
 828         }
 829
 830         LASSERT (kranal_peer_active(peer));     /* peer is in the peer table */
 831
 832         /* Refuse to duplicate an existing connection (both sides might try
 833          * to connect at once).  NB we return success!  We _do_ have a
 834          * connection (so we don't need to remove the peer from the peer
 835          * table) and we _don't_ have any blocked txs to complete */
 836         rc = kranal_conn_isdup_locked(peer, conn);
 837         if (rc != 0) {
 838                 LASSERT (!list_empty(&peer->rap_conns));
 839                 LASSERT (list_empty(&peer->rap_tx_queue));
 840                 write_unlock_irqrestore(&kranal_data.kra_global_lock, flags);
 841                 CWARN("Not creating duplicate connection to "LPX64": %d\n",
 842                       peer_nid, rc);
 843                 kranal_conn_decref(conn);
 844                 return 0;
 845         }
 846
 847         kranal_peer_addref(peer);               /* +1 ref for conn */
 848         conn->rac_peer = peer;
 849         list_add_tail(&conn->rac_list, &peer->rap_conns);
 850
 851         kranal_conn_addref(conn);               /* +1 ref for conn table */
 852         list_add_tail(&conn->rac_hashlist,
 853                       kranal_cqid2connlist(conn->rac_cqid));
 854
 855         /* Schedule all packets blocking for a connection */
 856         while (!list_empty(&peer->rap_tx_queue)) {
 857                 tx = list_entry(&peer->rap_tx_queue.next,
 858                                 kra_tx_t, tx_list);
 859
 860                 list_del(&tx->tx_list);
 861                 kranal_post_fma(conn, tx);
 862         }
 863
 864         nstale = kranal_close_stale_conns_locked(peer, conn);
 865
 866         write_unlock_irqrestore(&kranal_data.kra_global_lock, flags);
 867
 868         /* CAVEAT EMPTOR: passive peer can disappear NOW */
 869
 870         if (nstale != 0)
 871                 CWARN("Closed %d stale conns to "LPX64"\n", nstale, peer_nid);
 872
 873         /* Ensure conn gets checked.  Transmits may have been queued and an
 874          * FMA event may have happened before it got in the cq hash table */
 875         kranal_schedule_conn(conn);
 876         return 0;
 877 }
 878
 879 void
 880 kranal_connect (kra_peer_t *peer)
 881 {
 882         kra_tx_t          *tx;
 883         unsigned long      flags;
 884         struct list_head   zombies;
 885         int                rc;
 886
 887         LASSERT (peer->rap_connecting);
 888
 889         rc = kranal_conn_handshake(NULL, peer);
 890
 891         write_lock_irqsave(&kranal_data.kra_global_lock, flags);
 892
 893         LASSERT (peer->rap_connecting);
 894         peer->rap_connecting = 0;
 895
 896         if (rc == 0) {
 897                 /* kranal_conn_handshake() queues blocked txs immediately on
 898                  * success to avoid messages jumping the queue */
 899                 LASSERT (list_empty(&peer->rap_tx_queue));
 900
 901                 /* reset reconnection timeouts */
 902                 peer->rap_reconnect_interval = RANAL_MIN_RECONNECT_INTERVAL;
 903                 peer->rap_reconnect_time = CURRENT_TIME;
 904
 905                 write_unlock_irqrestore(&kranal_data.kra_global_lock, flags);
 906                 return;
 907         }
 908
 909         LASSERT (peer->rap_reconnect_interval != 0);
 910         peer->rap_reconnect_time = CURRENT_TIME + peer->rap_reconnect_interval;
 911         peer->rap_reconnect_interval = MAX(RANAL_MAX_RECONNECT_INTERVAL,
 912                                            1 * peer->rap_reconnect_interval);
 913
 914         /* Grab all blocked packets while we have the global lock */
 915         list_add(&zombies, &peer->rap_tx_queue);
 916         list_del_init(&peer->rap_tx_queue);
 917
 918         write_unlock_irqrestore(&kranal_data.kra_global_lock, flags);
 919
 920         if (list_empty(&zombies))
 921                 return;
 922
 923         CWARN("Dropping packets for "LPX64": connection failed\n",
 924               peer->rap_nid);
 925
 926         do {
 927                 tx = list_entry(zombies.next, kra_tx_t, tx_list);
 928
 929                 list_del(&tx->tx_list);
 930                 kranal_tx_done(tx, -EHOSTUNREACH);
 931
 932         } while (!list_empty(&zombies));
 933 }
 934
 935 int
 936 kranal_listener(void *arg)
 937 {
 938         struct sockaddr_in addr;
 939         wait_queue_t       wait;
 940         struct socket     *sock;
 941         struct socket     *newsock;
 942         int                port;
 943         kra_connreq_t     *connreqs;
 944         char               name[16];
 945         int                rc;
 946
 947         /* Parent thread holds kra_nid_mutex, and is, or is about to
 948          * block on kra_listener_signal */
 949
 950         port = kranal_tunables.kra_port;
 951         snprintf(name, sizeof(name), "kranal_lstn%03d", port);
 952         kportal_daemonize(name);
 953         kportal_blockallsigs();
 954
 955         init_waitqueue_entry(&wait, current);
 956
 957         rc = -ENOMEM;
 958         PORTAL_ALLOC(connreqs, 2 * sizeof(*connreqs));
 959         if (connreqs == NULL)
 960                 goto out_0;
 961
 962         rc = kranal_create_sock(&sock);
 963         if (rc != 0)
 964                 goto out_1;
 965
 966         memset(&addr, 0, sizeof(addr));
 967         addr.sin_family      = AF_INET;
 968         addr.sin_port        = htons(port);
 969         addr.sin_addr.s_addr = INADDR_ANY;
 970
 971         rc = sock->ops->bind(sock, (struct sockaddr *)&addr, sizeof(addr));
 972         if (rc != 0) {
 973                 CERROR("Can't bind to port %d\n", port);
 974                 goto out_2;
 975         }
 976
 977         rc = sock->ops->listen(sock, kranal_tunables.kra_backlog);
 978         if (rc != 0) {
 979                 CERROR("Can't set listen backlog %d: %d\n",
 980                        kranal_tunables.kra_backlog, rc);
 981                 goto out_2;
 982         }
 983
 984         LASSERT (kranal_data.kra_listener_sock == NULL);
 985         kranal_data.kra_listener_sock = sock;
 986
 987         /* unblock waiting parent */
 988         LASSERT (kranal_data.kra_listener_shutdown == 0);
 989         up(&kranal_data.kra_listener_signal);
 990
 991         /* Wake me any time something happens on my socket */
 992         add_wait_queue(sock->sk->sk_sleep, &wait);
 993
 994         while (kranal_data.kra_listener_shutdown == 0) {
 995
 996                 newsock = sock_alloc();
 997                 if (newsock == NULL) {
 998                         CERROR("Can't allocate new socket for accept\n");
 999                         kranal_pause(HZ);
1000                         continue;
1001                 }
1002
1003                 set_current_state(TASK_INTERRUPTIBLE);
1004
1005                 rc = sock->ops->accept(sock, newsock, O_NONBLOCK);
1006
1007                 if (rc == -EAGAIN &&
1008                     kranal_data.kra_listener_shutdown == 0)
1009                         schedule();
1010
1011                 set_current_state(TASK_RUNNING);
1012
1013                 if (rc != 0) {
1014                         sock_release(newsock);
1015                         if (rc != -EAGAIN) {
1016                                 CERROR("Accept failed: %d\n", rc);
1017                                 kranal_pause(HZ);
1018                         }
1019                         continue;
1020                 }
1021
1022                 kranal_conn_handshake(newsock, NULL);
1023                 sock_release(newsock);
1024         }
1025
1026         rc = 0;
1027         remove_wait_queue(sock->sk->sk_sleep, &wait);
1028  out_2:
1029         sock_release(sock);
1030         kranal_data.kra_listener_sock = NULL;
1031  out_1:
1032         PORTAL_FREE(connreqs, 2 * sizeof(*connreqs));
1033  out_0:
1034         /* set completion status and unblock thread waiting for me
1035          * (parent on startup failure, executioner on normal shutdown) */
1036         kranal_data.kra_listener_shutdown = rc;
1037         up(&kranal_data.kra_listener_signal);
1038
1039         return 0;
1040 }
1041
1042 int
1043 kranal_start_listener (void)
1044 {
1045         long           pid;
1046         int            rc;
1047
1048         CDEBUG(D_WARNING, "Starting listener\n");
1049
1050         /* Called holding kra_nid_mutex: listener stopped */
1051         LASSERT (kranal_data.kra_listener_sock == NULL);
1052
1053         kranal_data.kra_listener_shutdown = 0;
1054         pid = kernel_thread(kranal_listener, NULL, 0);
1055         if (pid < 0) {
1056                 CERROR("Can't spawn listener: %ld\n", pid);
1057                 return (int)pid;
1058         }
1059
1060         /* Block until listener has started up. */
1061         down(&kranal_data.kra_listener_signal);
1062
1063         rc = kranal_data.kra_listener_shutdown;
1064         LASSERT ((rc != 0) == (kranal_data.kra_listener_sock == NULL));
1065
1066         CDEBUG(D_WARNING, "Listener %ld started OK\n", pid);
1067         return rc;
1068 }
1069
1070 void
1071 kranal_stop_listener(void)
1072 {
1073         CDEBUG(D_WARNING, "Stopping listener\n");
1074
1075         /* Called holding kra_nid_mutex: listener running */
1076         LASSERT (kranal_data.kra_listener_sock != NULL);
1077
1078         kranal_data.kra_listener_shutdown = 1;
1079         wake_up_all(kranal_data.kra_listener_sock->sk->sk_sleep);
1080
1081         /* Block until listener has torn down. */
1082         down(&kranal_data.kra_listener_signal);
1083
1084         LASSERT (kranal_data.kra_listener_sock == NULL);
1085         CDEBUG(D_WARNING, "Listener stopped\n");
1086 }
1087
1088 int
1089 kranal_listener_procint(ctl_table *table, int write, struct file *filp,
1090                         void *buffer, size_t *lenp)
1091 {
1092         int   *tunable = (int *)table->data;
1093         int    old_val;
1094         int    rc;
1095
1096         /* No race with nal initialisation since the nal is setup all the time
1097          * it's loaded.  When that changes, change this! */
1098         LASSERT (kranal_data.kra_init == RANAL_INIT_ALL);
1099
1100         down(&kranal_data.kra_nid_mutex);
1101
1102         LASSERT (tunable == &kranal_tunables.kra_port ||
1103                  tunable == &kranal_tunables.kra_backlog);
1104         old_val = *tunable;
1105
1106         rc = proc_dointvec(table, write, filp, buffer, lenp);
1107
1108         if (write &&
1109             (*tunable != old_val ||
1110              kranal_data.kra_listener_sock == NULL)) {
1111
1112                 if (kranal_data.kra_listener_sock != NULL)
1113                         kranal_stop_listener();
1114
1115                 rc = kranal_start_listener();
1116
1117                 if (rc != 0) {
1118                         CWARN("Unable to start listener with new tunable:"
1119                               " reverting to old value\n");
1120                         *tunable = old_val;
1121                         kranal_start_listener();
1122                 }
1123         }
1124
1125         up(&kranal_data.kra_nid_mutex);
1126
1127         LASSERT (kranal_data.kra_init == RANAL_INIT_ALL);
1128         return rc;
1129 }
1130
1131 int
1132 kranal_set_mynid(ptl_nid_t nid)
1133 {
1134         unsigned long  flags;
1135         lib_ni_t      *ni = &kranal_lib.libnal_ni;
1136         int            rc = 0;
1137
1138         CDEBUG(D_NET, "setting mynid to "LPX64" (old nid="LPX64")\n",
1139                nid, ni->ni_pid.nid);
1140
1141         down(&kranal_data.kra_nid_mutex);
1142
1143         if (nid == ni->ni_pid.nid) {
1144                 /* no change of NID */
1145                 up(&kranal_data.kra_nid_mutex);
1146                 return 0;
1147         }
1148
1149         if (kranal_data.kra_listener_sock != NULL)
1150                 kranal_stop_listener();
1151
1152         write_lock_irqsave(&kranal_data.kra_global_lock, flags);
1153         kranal_data.kra_peerstamp++;
1154         write_unlock_irqrestore(&kranal_data.kra_global_lock, flags);
1155
1156         ni->ni_pid.nid = nid;
1157
1158         /* Delete all existing peers and their connections after new
1159          * NID/connstamp set to ensure no old connections in our brave
1160          * new world. */
1161         kranal_del_peer(PTL_NID_ANY, 0);
1162
1163         if (nid != PTL_NID_ANY)
1164                 rc = kranal_start_listener();
1165
1166         up(&kranal_data.kra_nid_mutex);
1167         return rc;
1168 }
1169
1170 kra_peer_t *
1171 kranal_create_peer (ptl_nid_t nid)
1172 {
1173         kra_peer_t *peer;
1174
1175         LASSERT (nid != PTL_NID_ANY);
1176
1177         PORTAL_ALLOC(peer, sizeof(*peer));
1178         if (peer == NULL)
1179                 return NULL;
1180
1181         memset(peer, 0, sizeof(*peer));         /* zero flags etc */
1182
1183         peer->rap_nid = nid;
1184         atomic_set(&peer->rap_refcount, 1);     /* 1 ref for caller */
1185
1186         INIT_LIST_HEAD(&peer->rap_list);
1187         INIT_LIST_HEAD(&peer->rap_connd_list);
1188         INIT_LIST_HEAD(&peer->rap_conns);
1189         INIT_LIST_HEAD(&peer->rap_tx_queue);
1190
1191         peer->rap_reconnect_time = CURRENT_TIME;
1192         peer->rap_reconnect_interval = RANAL_MIN_RECONNECT_INTERVAL;
1193
1194         atomic_inc(&kranal_data.kra_npeers);
1195         return peer;
1196 }
1197
1198 void
1199 kranal_destroy_peer (kra_peer_t *peer)
1200 {
1201         CDEBUG(D_NET, "peer "LPX64" %p deleted\n", peer->rap_nid, peer);
1202
1203         LASSERT (atomic_read(&peer->rap_refcount) == 0);
1204         LASSERT (peer->rap_persistence == 0);
1205         LASSERT (!kranal_peer_active(peer));
1206         LASSERT (!peer->rap_connecting);
1207         LASSERT (list_empty(&peer->rap_conns));
1208         LASSERT (list_empty(&peer->rap_tx_queue));
1209         LASSERT (list_empty(&peer->rap_connd_list));
1210
1211         PORTAL_FREE(peer, sizeof(*peer));
1212
1213         /* NB a peer's connections keep a reference on their peer until
1214          * they are destroyed, so we can be assured that _all_ state to do
1215          * with this peer has been cleaned up when its refcount drops to
1216          * zero. */
1217         atomic_dec(&kranal_data.kra_npeers);
1218 }
1219
1220 kra_peer_t *
1221 kranal_find_peer_locked (ptl_nid_t nid)
1222 {
1223         struct list_head *peer_list = kranal_nid2peerlist(nid);
1224         struct list_head *tmp;
1225         kra_peer_t       *peer;
1226
1227         list_for_each (tmp, peer_list) {
1228
1229                 peer = list_entry(tmp, kra_peer_t, rap_list);
1230
1231                 LASSERT (peer->rap_persistence > 0 ||     /* persistent peer */
1232                          !list_empty(&peer->rap_conns));  /* active conn */
1233
1234                 if (peer->rap_nid != nid)
1235                         continue;
1236
1237                 CDEBUG(D_NET, "got peer [%p] -> "LPX64" (%d)\n",
1238                        peer, nid, atomic_read(&peer->rap_refcount));
1239                 return peer;
1240         }
1241         return NULL;
1242 }
1243
1244 kra_peer_t *
1245 kranal_find_peer (ptl_nid_t nid)
1246 {
1247         kra_peer_t     *peer;
1248
1249         read_lock(&kranal_data.kra_global_lock);
1250         peer = kranal_find_peer_locked(nid);
1251         if (peer != NULL)                       /* +1 ref for caller? */
1252                 kranal_peer_addref(peer);
1253         read_unlock(&kranal_data.kra_global_lock);
1254
1255         return peer;
1256 }
1257
1258 void
1259 kranal_unlink_peer_locked (kra_peer_t *peer)
1260 {
1261         LASSERT (peer->rap_persistence == 0);
1262         LASSERT (list_empty(&peer->rap_conns));
1263
1264         LASSERT (kranal_peer_active(peer));
1265         list_del_init(&peer->rap_list);
1266
1267         /* lose peerlist's ref */
1268         kranal_peer_decref(peer);
1269 }
1270
1271 int
1272 kranal_get_peer_info (int index, ptl_nid_t *nidp, __u32 *ipp, int *portp,
1273                       int *persistencep)
1274 {
1275         kra_peer_t        *peer;
1276         struct list_head  *ptmp;
1277         int                i;
1278
1279         read_lock(&kranal_data.kra_global_lock);
1280
1281         for (i = 0; i < kranal_data.kra_peer_hash_size; i++) {
1282
1283                 list_for_each(ptmp, &kranal_data.kra_peers[i]) {
1284
1285                         peer = list_entry(ptmp, kra_peer_t, rap_list);
1286                         LASSERT (peer->rap_persistence > 0 ||
1287                                  !list_empty(&peer->rap_conns));
1288
1289                         if (index-- > 0)
1290                                 continue;
1291
1292                         *nidp = peer->rap_nid;
1293                         *ipp = peer->rap_ip;
1294                         *portp = peer->rap_port;
1295                         *persistencep = peer->rap_persistence;
1296
1297                         read_unlock(&kranal_data.kra_global_lock);
1298                         return 0;
1299                 }
1300         }
1301
1302         read_unlock(&kranal_data.kra_global_lock);
1303         return -ENOENT;
1304 }
1305
1306 int
1307 kranal_add_persistent_peer (ptl_nid_t nid, __u32 ip, int port)
1308 {
1309         unsigned long      flags;
1310         kra_peer_t        *peer;
1311         kra_peer_t        *peer2;
1312
1313         if (nid == PTL_NID_ANY)
1314                 return -EINVAL;
1315
1316         peer = kranal_create_peer(nid);
1317         if (peer == NULL)
1318                 return -ENOMEM;
1319
1320         write_lock_irqsave(&kranal_data.kra_global_lock, flags);
1321
1322         peer2 = kranal_find_peer_locked(nid);
1323         if (peer2 != NULL) {
1324                 kranal_peer_decref(peer);
1325                 peer = peer2;
1326         } else {
1327                 /* peer table takes existing ref on peer */
1328                 list_add_tail(&peer->rap_list,
1329                               kranal_nid2peerlist(nid));
1330         }
1331
1332         peer->rap_ip = ip;
1333         peer->rap_port = port;
1334         peer->rap_persistence++;
1335
1336         write_unlock_irqrestore(&kranal_data.kra_global_lock, flags);
1337         return 0;
1338 }
1339
1340 void
1341 kranal_del_peer_locked (kra_peer_t *peer, int single_share)
1342 {
1343         struct list_head *ctmp;
1344         struct list_head *cnxt;
1345         kra_conn_t       *conn;
1346
1347         if (!single_share)
1348                 peer->rap_persistence = 0;
1349         else if (peer->rap_persistence > 0)
1350                 peer->rap_persistence--;
1351
1352         if (peer->rap_persistence != 0)
1353                 return;
1354
1355         if (list_empty(&peer->rap_conns)) {
1356                 kranal_unlink_peer_locked(peer);
1357         } else {
1358                 list_for_each_safe(ctmp, cnxt, &peer->rap_conns) {
1359                         conn = list_entry(ctmp, kra_conn_t, rac_list);
1360
1361                         kranal_close_conn_locked(conn, 0);
1362                 }
1363                 /* peer unlinks itself when last conn is closed */
1364         }
1365 }
1366
1367 int
1368 kranal_del_peer (ptl_nid_t nid, int single_share)
1369 {
1370         unsigned long      flags;
1371         struct list_head  *ptmp;
1372         struct list_head  *pnxt;
1373         kra_peer_t        *peer;
1374         int                lo;
1375         int                hi;
1376         int                i;
1377         int                rc = -ENOENT;
1378
1379         write_lock_irqsave(&kranal_data.kra_global_lock, flags);
1380
1381         if (nid != PTL_NID_ANY)
1382                 lo = hi = kranal_nid2peerlist(nid) - kranal_data.kra_peers;
1383         else {
1384                 lo = 0;
1385                 hi = kranal_data.kra_peer_hash_size - 1;
1386         }
1387
1388         for (i = lo; i <= hi; i++) {
1389                 list_for_each_safe (ptmp, pnxt, &kranal_data.kra_peers[i]) {
1390                         peer = list_entry(ptmp, kra_peer_t, rap_list);
1391                         LASSERT (peer->rap_persistence > 0 ||
1392                                  !list_empty(&peer->rap_conns));
1393
1394                         if (!(nid == PTL_NID_ANY || peer->rap_nid == nid))
1395                                 continue;
1396
1397                         kranal_del_peer_locked(peer, single_share);
1398                         rc = 0;         /* matched something */
1399
1400                         if (single_share)
1401                                 goto out;
1402                 }
1403         }
1404  out:
1405         write_unlock_irqrestore(&kranal_data.kra_global_lock, flags);
1406
1407         return rc;
1408 }
1409
1410 kra_conn_t *
1411 kranal_get_conn_by_idx (int index)
1412 {
1413         kra_peer_t        *peer;
1414         struct list_head  *ptmp;
1415         kra_conn_t        *conn;
1416         struct list_head  *ctmp;
1417         int                i;
1418
1419         read_lock (&kranal_data.kra_global_lock);
1420
1421         for (i = 0; i < kranal_data.kra_peer_hash_size; i++) {
1422                 list_for_each (ptmp, &kranal_data.kra_peers[i]) {
1423
1424                         peer = list_entry(ptmp, kra_peer_t, rap_list);
1425                         LASSERT (peer->rap_persistence > 0 ||
1426                                  !list_empty(&peer->rap_conns));
1427
1428                         list_for_each (ctmp, &peer->rap_conns) {
1429                                 if (index-- > 0)
1430                                         continue;
1431
1432                                 conn = list_entry(ctmp, kra_conn_t, rac_list);
1433                                 CDEBUG(D_NET, "++conn[%p] -> "LPX64" (%d)\n",
1434                                        conn, conn->rac_peer->rap_nid,
1435                                        atomic_read(&conn->rac_refcount));
1436                                 atomic_inc(&conn->rac_refcount);
1437                                 read_unlock(&kranal_data.kra_global_lock);
1438                                 return conn;
1439                         }
1440                 }
1441         }
1442
1443         read_unlock(&kranal_data.kra_global_lock);
1444         return NULL;
1445 }
1446
1447 int
1448 kranal_close_peer_conns_locked (kra_peer_t *peer, int why)
1449 {
1450         kra_conn_t         *conn;
1451         struct list_head   *ctmp;
1452         struct list_head   *cnxt;
1453         int                 count = 0;
1454
1455         list_for_each_safe (ctmp, cnxt, &peer->rap_conns) {
1456                 conn = list_entry(ctmp, kra_conn_t, rac_list);
1457
1458                 count++;
1459                 kranal_close_conn_locked(conn, why);
1460         }
1461
1462         return count;
1463 }
1464
1465 int
1466 kranal_close_matching_conns (ptl_nid_t nid)
1467 {
1468         unsigned long       flags;
1469         kra_peer_t         *peer;
1470         struct list_head   *ptmp;
1471         struct list_head   *pnxt;
1472         int                 lo;
1473         int                 hi;
1474         int                 i;
1475         int                 count = 0;
1476
1477         write_lock_irqsave(&kranal_data.kra_global_lock, flags);
1478
1479         if (nid != PTL_NID_ANY)
1480                 lo = hi = kranal_nid2peerlist(nid) - kranal_data.kra_peers;
1481         else {
1482                 lo = 0;
1483                 hi = kranal_data.kra_peer_hash_size - 1;
1484         }
1485
1486         for (i = lo; i <= hi; i++) {
1487                 list_for_each_safe (ptmp, pnxt, &kranal_data.kra_peers[i]) {
1488
1489                         peer = list_entry(ptmp, kra_peer_t, rap_list);
1490                         LASSERT (peer->rap_persistence > 0 ||
1491                                  !list_empty(&peer->rap_conns));
1492
1493                         if (!(nid == PTL_NID_ANY || nid == peer->rap_nid))
1494                                 continue;
1495
1496                         count += kranal_close_peer_conns_locked(peer, 0);
1497                 }
1498         }
1499
1500         write_unlock_irqrestore(&kranal_data.kra_global_lock, flags);
1501
1502         /* wildcards always succeed */
1503         if (nid == PTL_NID_ANY)
1504                 return 0;
1505
1506         return (count == 0) ? -ENOENT : 0;
1507 }
1508
1509 int
1510 kranal_cmd(struct portals_cfg *pcfg, void * private)
1511 {
1512         int rc = -EINVAL;
1513
1514         LASSERT (pcfg != NULL);
1515
1516         switch(pcfg->pcfg_command) {
1517         case NAL_CMD_GET_PEER: {
1518                 ptl_nid_t   nid = 0;
1519                 __u32       ip = 0;
1520                 int         port = 0;
1521                 int         share_count = 0;
1522
1523                 rc = kranal_get_peer_info(pcfg->pcfg_count,
1524                                           &nid, &ip, &port, &share_count);
1525                 pcfg->pcfg_nid   = nid;
1526                 pcfg->pcfg_size  = 0;
1527                 pcfg->pcfg_id    = ip;
1528                 pcfg->pcfg_misc  = port;
1529                 pcfg->pcfg_count = 0;
1530                 pcfg->pcfg_wait  = share_count;
1531                 break;
1532         }
1533         case NAL_CMD_ADD_PEER: {
1534                 rc = kranal_add_persistent_peer(pcfg->pcfg_nid,
1535                                                 pcfg->pcfg_id, /* IP */
1536                                                 pcfg->pcfg_misc); /* port */
1537                 break;
1538         }
1539         case NAL_CMD_DEL_PEER: {
1540                 rc = kranal_del_peer(pcfg->pcfg_nid,
1541                                      /* flags == single_share */
1542                                      pcfg->pcfg_flags != 0);
1543                 break;
1544         }
1545         case NAL_CMD_GET_CONN: {
1546                 kra_conn_t *conn = kranal_get_conn_by_idx(pcfg->pcfg_count);
1547
1548                 if (conn == NULL)
1549                         rc = -ENOENT;
1550                 else {
1551                         rc = 0;
1552                         pcfg->pcfg_nid   = conn->rac_peer->rap_nid;
1553                         pcfg->pcfg_id    = conn->rac_device->rad_id;
1554                         pcfg->pcfg_misc  = 0;
1555                         pcfg->pcfg_flags = 0;
1556                         kranal_conn_decref(conn);
1557                 }
1558                 break;
1559         }
1560         case NAL_CMD_CLOSE_CONNECTION: {
1561                 rc = kranal_close_matching_conns(pcfg->pcfg_nid);
1562                 break;
1563         }
1564         case NAL_CMD_REGISTER_MYNID: {
1565                 if (pcfg->pcfg_nid == PTL_NID_ANY)
1566                         rc = -EINVAL;
1567                 else
1568                         rc = kranal_set_mynid(pcfg->pcfg_nid);
1569                 break;
1570         }
1571         }
1572
1573         return rc;
1574 }
1575
1576 void
1577 kranal_free_txdescs(struct list_head *freelist)
1578 {
1579         kra_tx_t    *tx;
1580
1581         while (!list_empty(freelist)) {
1582                 tx = list_entry(freelist->next, kra_tx_t, tx_list);
1583
1584                 list_del(&tx->tx_list);
1585                 PORTAL_FREE(tx->tx_phys, PTL_MD_MAX_IOV * sizeof(*tx->tx_phys));
1586                 PORTAL_FREE(tx, sizeof(*tx));
1587         }
1588 }
1589
1590 int
1591 kranal_alloc_txdescs(struct list_head *freelist, int n)
1592 {
1593         int            isnblk = (freelist == &kranal_data.kra_idle_nblk_txs);
1594         int            i;
1595         kra_tx_t      *tx;
1596
1597         LASSERT (freelist == &kranal_data.kra_idle_txs ||
1598                  freelist == &kranal_data.kra_idle_nblk_txs);
1599         LASSERT (list_empty(freelist));
1600
1601         for (i = 0; i < n; i++) {
1602
1603                 PORTAL_ALLOC(tx, sizeof(*tx));
1604                 if (tx == NULL) {
1605                         CERROR("Can't allocate %stx[%d]\n",
1606                                isnblk ? "nblk " : "", i);
1607                         kranal_free_txdescs(freelist);
1608                         return -ENOMEM;
1609                 }
1610
1611                 PORTAL_ALLOC(tx->tx_phys,
1612                              PTL_MD_MAX_IOV * sizeof(*tx->tx_phys));
1613                 if (tx->tx_phys == NULL) {
1614                         CERROR("Can't allocate %stx[%d]->tx_phys\n",
1615                                isnblk ? "nblk " : "", i);
1616
1617                         PORTAL_FREE(tx, sizeof(*tx));
1618                         kranal_free_txdescs(freelist);
1619                         return -ENOMEM;
1620                 }
1621
1622                 tx->tx_isnblk = isnblk;
1623                 tx->tx_buftype = RANAL_BUF_NONE;
1624                 tx->tx_msg.ram_type = RANAL_MSG_NONE;
1625
1626                 list_add(&tx->tx_list, freelist);
1627         }
1628
1629         return 0;
1630 }
1631
1632 int
1633 kranal_device_init(int id, kra_device_t *dev)
1634 {
1635         const int         total_ntx = RANAL_NTX + RANAL_NTX_NBLK;
1636         RAP_RETURN        rrc;
1637
1638         dev->rad_id = id;
1639         rrc = RapkGetDeviceByIndex(id, kranal_device_callback,
1640                                    &dev->rad_handle);
1641         if (rrc != RAP_SUCCESS) {
1642                 CERROR("Can't get Rapidarray Device %d: %d\n", id, rrc);
1643                 goto failed_0;
1644         }
1645
1646         rrc = RapkReserveRdma(dev->rad_handle, total_ntx);
1647         if (rrc != RAP_SUCCESS) {
1648                 CERROR("Can't reserve %d RDMA descriptors"
1649                        " for device %d: %d\n", total_ntx, id, rrc);
1650                 goto failed_1;
1651         }
1652
1653         rrc = RapkCreatePtag(dev->rad_handle,
1654                              &dev->rad_ptag);
1655         if (rrc != RAP_SUCCESS) {
1656                 CERROR("Can't create ptag"
1657                        " for device %d: %d\n", id, rrc);
1658                 goto failed_1;
1659         }
1660
1661         rrc = RapkCreateCQ(dev->rad_handle, total_ntx, dev->rad_ptag,
1662                            &dev->rad_rdma_cq);
1663         if (rrc != RAP_SUCCESS) {
1664                 CERROR("Can't create rdma cq size %d"
1665                        " for device %d: %d\n", total_ntx, id, rrc);
1666                 goto failed_2;
1667         }
1668
1669         rrc = RapkCreateCQ(dev->rad_handle, RANAL_FMA_CQ_SIZE,
1670                            dev->rad_ptag, &dev->rad_fma_cq);
1671         if (rrc != RAP_SUCCESS) {
1672                 CERROR("Can't create fma cq size %d"
1673                        " for device %d: %d\n", RANAL_FMA_CQ_SIZE, id, rrc);
1674                 goto failed_3;
1675         }
1676
1677         return 0;
1678
1679  failed_3:
1680         RapkDestroyCQ(dev->rad_handle, dev->rad_rdma_cq, dev->rad_ptag);
1681  failed_2:
1682         RapkDestroyPtag(dev->rad_handle, dev->rad_ptag);
1683  failed_1:
1684         RapkReleaseDevice(dev->rad_handle);
1685  failed_0:
1686         return -ENODEV;
1687 }
1688
1689 void
1690 kranal_device_fini(kra_device_t *dev)
1691 {
1692         LASSERT(dev->rad_scheduler == NULL);
1693         RapkDestroyCQ(dev->rad_handle, dev->rad_fma_cq, dev->rad_ptag);
1694         RapkDestroyCQ(dev->rad_handle, dev->rad_rdma_cq, dev->rad_ptag);
1695         RapkDestroyPtag(dev->rad_handle, dev->rad_ptag);
1696         RapkReleaseDevice(dev->rad_handle);
1697 }
1698
1699 void
1700 kranal_api_shutdown (nal_t *nal)
1701 {
1702         int           i;
1703         unsigned long flags;
1704
1705         if (nal->nal_refct != 0) {
1706                 /* This module got the first ref */
1707                 PORTAL_MODULE_UNUSE;
1708                 return;
1709         }
1710
1711         CDEBUG(D_MALLOC, "before NAL cleanup: kmem %d\n",
1712                atomic_read(&portal_kmemory));
1713
1714         LASSERT (nal == &kranal_api);
1715
1716         switch (kranal_data.kra_init) {
1717         default:
1718                 CERROR("Unexpected state %d\n", kranal_data.kra_init);
1719                 LBUG();
1720
1721         case RANAL_INIT_ALL:
1722                 /* stop calls to nal_cmd */
1723                 libcfs_nal_cmd_unregister(RANAL);
1724                 /* No new persistent peers */
1725
1726                 /* resetting my NID to unadvertises me, removes my
1727                  * listener and nukes all current peers */
1728                 kranal_set_mynid(PTL_NID_ANY);
1729                 /* no new peers or conns */
1730
1731                 /* Wait for all peer/conn state to clean up */
1732                 i = 2;
1733                 while (atomic_read(&kranal_data.kra_nconns) != 0 ||
1734                        atomic_read(&kranal_data.kra_npeers) != 0) {
1735                         i++;
1736                         CDEBUG(((i & (-i)) == i) ? D_WARNING : D_NET, /* power of 2? */
1737                                "waiting for %d peers and %d conns to close down\n",
1738                                atomic_read(&kranal_data.kra_npeers),
1739                                atomic_read(&kranal_data.kra_nconns));
1740                         kranal_pause(HZ);
1741                 }
1742                 /* fall through */
1743
1744         case RANAL_INIT_LIB:
1745                 lib_fini(&kranal_lib);
1746                 /* fall through */
1747
1748         case RANAL_INIT_DATA:
1749                 break;
1750         }
1751
1752         /* flag threads to terminate; wake and wait for them to die */
1753         kranal_data.kra_shutdown = 1;
1754
1755         for (i = 0; i < kranal_data.kra_ndevs; i++) {
1756                 kra_device_t *dev = &kranal_data.kra_devices[i];
1757
1758                 LASSERT (list_empty(&dev->rad_connq));
1759
1760                 spin_lock_irqsave(&dev->rad_lock, flags);
1761                 wake_up(&dev->rad_waitq);
1762                 spin_unlock_irqrestore(&dev->rad_lock, flags);
1763         }
1764
1765         spin_lock_irqsave(&kranal_data.kra_reaper_lock, flags);
1766         wake_up_all(&kranal_data.kra_reaper_waitq);
1767         spin_unlock_irqrestore(&kranal_data.kra_reaper_lock, flags);
1768
1769         LASSERT (list_empty(&kranal_data.kra_connd_peers));
1770         spin_lock_irqsave(&kranal_data.kra_connd_lock, flags);
1771         wake_up_all(&kranal_data.kra_connd_waitq);
1772         spin_unlock_irqrestore(&kranal_data.kra_connd_lock, flags);
1773
1774         i = 2;
1775         while (atomic_read(&kranal_data.kra_nthreads) != 0) {
1776                 i++;
1777                 CDEBUG(((i & (-i)) == i) ? D_WARNING : D_NET, /* power of 2? */
1778                        "Waiting for %d threads to terminate\n",
1779                        atomic_read(&kranal_data.kra_nthreads));
1780                 kranal_pause(HZ);
1781         }
1782
1783         LASSERT (atomic_read(&kranal_data.kra_npeers) == 0);
1784         if (kranal_data.kra_peers != NULL) {
1785                 for (i = 0; i < kranal_data.kra_peer_hash_size; i++)
1786                         LASSERT (list_empty(&kranal_data.kra_peers[i]));
1787
1788                 PORTAL_FREE(kranal_data.kra_peers,
1789                             sizeof (struct list_head) *
1790                             kranal_data.kra_peer_hash_size);
1791         }
1792
1793         LASSERT (atomic_read(&kranal_data.kra_nconns) == 0);
1794         if (kranal_data.kra_conns != NULL) {
1795                 for (i = 0; i < kranal_data.kra_conn_hash_size; i++)
1796                         LASSERT (list_empty(&kranal_data.kra_conns[i]));
1797
1798                 PORTAL_FREE(kranal_data.kra_conns,
1799                             sizeof (struct list_head) *
1800                             kranal_data.kra_conn_hash_size);
1801         }
1802
1803         for (i = 0; i < kranal_data.kra_ndevs; i++)
1804                 kranal_device_fini(&kranal_data.kra_devices[i]);
1805
1806         kranal_free_txdescs(&kranal_data.kra_idle_txs);
1807         kranal_free_txdescs(&kranal_data.kra_idle_nblk_txs);
1808
1809         CDEBUG(D_MALLOC, "after NAL cleanup: kmem %d\n",
1810                atomic_read(&portal_kmemory));
1811         printk(KERN_INFO "Lustre: RapidArray NAL unloaded (final mem %d)\n",
1812                atomic_read(&portal_kmemory));
1813
1814         kranal_data.kra_init = RANAL_INIT_NOTHING;
1815 }
1816
1817 int
1818 kranal_api_startup (nal_t *nal, ptl_pid_t requested_pid,
1819                     ptl_ni_limits_t *requested_limits,
1820                     ptl_ni_limits_t *actual_limits)
1821 {
1822         static int        device_ids[] = {RAPK_MAIN_DEVICE_ID,
1823                                           RAPK_EXPANSION_DEVICE_ID};
1824         struct timeval    tv;
1825         ptl_process_id_t  process_id;
1826         int               pkmem = atomic_read(&portal_kmemory);
1827         int               rc;
1828         int               i;
1829         kra_device_t     *dev;
1830
1831         LASSERT (nal == &kranal_api);
1832
1833         if (nal->nal_refct != 0) {
1834                 if (actual_limits != NULL)
1835                         *actual_limits = kranal_lib.libnal_ni.ni_actual_limits;
1836                 /* This module got the first ref */
1837                 PORTAL_MODULE_USE;
1838                 return PTL_OK;
1839         }
1840
1841         LASSERT (kranal_data.kra_init == RANAL_INIT_NOTHING);
1842
1843         memset(&kranal_data, 0, sizeof(kranal_data)); /* zero pointers, flags etc */
1844
1845         /* CAVEAT EMPTOR: Every 'Fma' message includes the sender's NID and
1846          * a unique (for all time) connstamp so we can uniquely identify
1847          * the sender.  The connstamp is an incrementing counter
1848          * initialised with seconds + microseconds at startup time.  So we
1849          * rely on NOT creating connections more frequently on average than
1850          * 1MHz to ensure we don't use old connstamps when we reboot. */
1851         do_gettimeofday(&tv);
1852         kranal_data.kra_connstamp =
1853         kranal_data.kra_peerstamp = (((__u64)tv.tv_sec) * 1000000) + tv.tv_usec;
1854
1855         init_MUTEX(&kranal_data.kra_nid_mutex);
1856         init_MUTEX_LOCKED(&kranal_data.kra_listener_signal);
1857
1858         rwlock_init(&kranal_data.kra_global_lock);
1859
1860         for (i = 0; i < RANAL_MAXDEVS; i++ ) {
1861                 kra_device_t  *dev = &kranal_data.kra_devices[i];
1862
1863                 dev->rad_idx = i;
1864                 INIT_LIST_HEAD(&dev->rad_connq);
1865                 init_waitqueue_head(&dev->rad_waitq);
1866                 spin_lock_init(&dev->rad_lock);
1867         }
1868
1869         kranal_data.kra_new_min_timeout = MAX_SCHEDULE_TIMEOUT;
1870         init_waitqueue_head(&kranal_data.kra_reaper_waitq);
1871         spin_lock_init(&kranal_data.kra_reaper_lock);
1872
1873         INIT_LIST_HEAD(&kranal_data.kra_connd_peers);
1874         init_waitqueue_head(&kranal_data.kra_connd_waitq);
1875         spin_lock_init(&kranal_data.kra_connd_lock);
1876
1877         INIT_LIST_HEAD(&kranal_data.kra_idle_txs);
1878         INIT_LIST_HEAD(&kranal_data.kra_idle_nblk_txs);
1879         init_waitqueue_head(&kranal_data.kra_idle_tx_waitq);
1880         spin_lock_init(&kranal_data.kra_tx_lock);
1881
1882         /* OK to call kranal_api_shutdown() to cleanup now */
1883         kranal_data.kra_init = RANAL_INIT_DATA;
1884
1885         kranal_data.kra_peer_hash_size = RANAL_PEER_HASH_SIZE;
1886         PORTAL_ALLOC(kranal_data.kra_peers,
1887                      sizeof(struct list_head) * kranal_data.kra_peer_hash_size);
1888         if (kranal_data.kra_peers == NULL)
1889                 goto failed;
1890
1891         for (i = 0; i < kranal_data.kra_peer_hash_size; i++)
1892                 INIT_LIST_HEAD(&kranal_data.kra_peers[i]);
1893
1894         kranal_data.kra_conn_hash_size = RANAL_PEER_HASH_SIZE;
1895         PORTAL_ALLOC(kranal_data.kra_conns,
1896                      sizeof(struct list_head) * kranal_data.kra_conn_hash_size);
1897         if (kranal_data.kra_conns == NULL)
1898                 goto failed;
1899
1900         for (i = 0; i < kranal_data.kra_conn_hash_size; i++)
1901                 INIT_LIST_HEAD(&kranal_data.kra_conns[i]);
1902
1903         rc = kranal_alloc_txdescs(&kranal_data.kra_idle_txs, RANAL_NTX);
1904         if (rc != 0)
1905                 goto failed;
1906
1907         rc = kranal_alloc_txdescs(&kranal_data.kra_idle_nblk_txs,RANAL_NTX_NBLK);
1908         if (rc != 0)
1909                 goto failed;
1910
1911         process_id.pid = requested_pid;
1912         process_id.nid = PTL_NID_ANY;           /* don't know my NID yet */
1913
1914         rc = lib_init(&kranal_lib, nal, process_id,
1915                       requested_limits, actual_limits);
1916         if (rc != PTL_OK) {
1917                 CERROR("lib_init failed: error %d\n", rc);
1918                 goto failed;
1919         }
1920
1921         /* lib interface initialised */
1922         kranal_data.kra_init = RANAL_INIT_LIB;
1923         /*****************************************************/
1924
1925         rc = kranal_thread_start(kranal_reaper, NULL);
1926         if (rc != 0) {
1927                 CERROR("Can't spawn ranal reaper: %d\n", rc);
1928                 goto failed;
1929         }
1930
1931         for (i = 0; i < RANAL_N_CONND; i++) {
1932                 rc = kranal_thread_start(kranal_connd, (void *)i);
1933                 if (rc != 0) {
1934                         CERROR("Can't spawn ranal connd[%d]: %d\n",
1935                                i, rc);
1936                         goto failed;
1937                 }
1938         }
1939
1940         LASSERT(kranal_data.kra_ndevs == 0);
1941         for (i = 0; i < sizeof(device_ids)/sizeof(device_ids[0]); i++) {
1942                 dev = &kranal_data.kra_devices[kranal_data.kra_ndevs];
1943
1944                 rc = kranal_device_init(device_ids[i], dev);
1945                 if (rc == 0)
1946                         kranal_data.kra_ndevs++;
1947
1948                 rc = kranal_thread_start(kranal_scheduler, dev);
1949                 if (rc != 0) {
1950                         CERROR("Can't spawn ranal scheduler[%d]: %d\n",
1951                                i, rc);
1952                         goto failed;
1953                 }
1954         }
1955
1956         if (kranal_data.kra_ndevs == 0)
1957                 goto failed;
1958
1959         rc = libcfs_nal_cmd_register(RANAL, &kranal_cmd, NULL);
1960         if (rc != 0) {
1961                 CERROR("Can't initialise command interface (rc = %d)\n", rc);
1962                 goto failed;
1963         }
1964
1965         /* flag everything initialised */
1966         kranal_data.kra_init = RANAL_INIT_ALL;
1967         /*****************************************************/
1968
1969         CDEBUG(D_MALLOC, "initial kmem %d\n", atomic_read(&portal_kmemory));
1970         printk(KERN_INFO "Lustre: RapidArray NAL loaded "
1971                "(initial mem %d)\n", pkmem);
1972
1973         return PTL_OK;
1974
1975  failed:
1976         kranal_api_shutdown(&kranal_api);
1977         return PTL_FAIL;
1978 }
1979
1980 void __exit
1981 kranal_module_fini (void)
1982 {
1983 #ifdef CONFIG_SYSCTL
1984         if (kranal_tunables.kra_sysctl != NULL)
1985                 unregister_sysctl_table(kranal_tunables.kra_sysctl);
1986 #endif
1987         PtlNIFini(kranal_ni);
1988
1989         ptl_unregister_nal(RANAL);
1990 }
1991
1992 int __init
1993 kranal_module_init (void)
1994 {
1995         int    rc;
1996
1997         /* the following must be sizeof(int) for
1998          * proc_dointvec/kranal_listener_procint() */
1999         LASSERT (sizeof(kranal_tunables.kra_timeout) == sizeof(int));
2000         LASSERT (sizeof(kranal_tunables.kra_listener_timeout) == sizeof(int));
2001         LASSERT (sizeof(kranal_tunables.kra_backlog) == sizeof(int));
2002         LASSERT (sizeof(kranal_tunables.kra_port) == sizeof(int));
2003         LASSERT (sizeof(kranal_tunables.kra_max_immediate) == sizeof(int));
2004
2005         kranal_api.nal_ni_init = kranal_api_startup;
2006         kranal_api.nal_ni_fini = kranal_api_shutdown;
2007
2008         /* Initialise dynamic tunables to defaults once only */
2009         kranal_tunables.kra_timeout = RANAL_TIMEOUT;
2010
2011         rc = ptl_register_nal(RANAL, &kranal_api);
2012         if (rc != PTL_OK) {
2013                 CERROR("Can't register RANAL: %d\n", rc);
2014                 return -ENOMEM;               /* or something... */
2015         }
2016
2017         /* Pure gateways want the NAL started up at module load time... */
2018         rc = PtlNIInit(RANAL, LUSTRE_SRV_PTL_PID, NULL, NULL, &kranal_ni);
2019         if (rc != PTL_OK && rc != PTL_IFACE_DUP) {
2020                 ptl_unregister_nal(RANAL);
2021                 return -ENODEV;
2022         }
2023
2024 #ifdef CONFIG_SYSCTL
2025         /* Press on regardless even if registering sysctl doesn't work */
2026         kranal_tunables.kra_sysctl =
2027                 register_sysctl_table(kranal_top_ctl_table, 0);
2028 #endif
2029         return 0;
2030 }
2031
2032 MODULE_AUTHOR("Cluster File Systems, Inc. <info@clusterfs.com>");
2033 MODULE_DESCRIPTION("Kernel RapidArray NAL v0.01");
2034 MODULE_LICENSE("GPL");
2035
2036 module_init(kranal_module_init);
2037 module_exit(kranal_module_fini);