lnet/klnds/socklnd/socklnd.c

   1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
   2  * vim:expandtab:shiftwidth=8:tabstop=8:
   3  *
   4  * Copyright (C) 2001, 2002 Cluster File Systems, Inc.
   5  *   Author: Zach Brown <zab@zabbo.net>
   6  *   Author: Peter J. Braam <braam@clusterfs.com>
   7  *   Author: Phil Schwan <phil@clusterfs.com>
   8  *   Author: Eric Barton <eric@bartonsoftware.com>
   9  *
  10  *   This file is part of Portals, http://www.sf.net/projects/sandiaportals/
  11  *
  12  *   Portals is free software; you can redistribute it and/or
  13  *   modify it under the terms of version 2 of the GNU General Public
  14  *   License as published by the Free Software Foundation.
  15  *
  16  *   Portals is distributed in the hope that it will be useful,
  17  *   but WITHOUT ANY WARRANTY; without even the implied warranty of
  18  *   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  19  *   GNU General Public License for more details.
  20  *
  21  *   You should have received a copy of the GNU General Public License
  22  *   along with Portals; if not, write to the Free Software
  23  *   Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
  24  */
  25
  26 #include "socknal.h"
  27
  28 ptl_handle_ni_t         ksocknal_ni;
  29 static nal_t            ksocknal_api;
  30 #if (LINUX_VERSION_CODE >= KERNEL_VERSION(2,5,0))
  31 ksock_nal_data_t ksocknal_data;
  32 #else
  33 static ksock_nal_data_t ksocknal_data;
  34 #endif
  35
  36 kpr_nal_interface_t ksocknal_router_interface = {
  37         kprni_nalid:      SOCKNAL,
  38         kprni_arg:        &ksocknal_data,
  39         kprni_fwd:        ksocknal_fwd_packet,
  40         kprni_notify:     ksocknal_notify,
  41 };
  42
  43 #define SOCKNAL_SYSCTL  200
  44
  45 #define SOCKNAL_SYSCTL_TIMEOUT     1
  46 #define SOCKNAL_SYSCTL_EAGER_ACK   2
  47 #define SOCKNAL_SYSCTL_ZERO_COPY   3
  48 #define SOCKNAL_SYSCTL_TYPED       4
  49 #define SOCKNAL_SYSCTL_MIN_BULK    5
  50
  51 static ctl_table ksocknal_ctl_table[] = {
  52         {SOCKNAL_SYSCTL_TIMEOUT, "timeout",
  53          &ksocknal_data.ksnd_io_timeout, sizeof (int),
  54          0644, NULL, &proc_dointvec},
  55         {SOCKNAL_SYSCTL_EAGER_ACK, "eager_ack",
  56          &ksocknal_data.ksnd_eager_ack, sizeof (int),
  57          0644, NULL, &proc_dointvec},
  58 #if SOCKNAL_ZC
  59         {SOCKNAL_SYSCTL_ZERO_COPY, "zero_copy",
  60          &ksocknal_data.ksnd_zc_min_frag, sizeof (int),
  61          0644, NULL, &proc_dointvec},
  62 #endif
  63         {SOCKNAL_SYSCTL_TYPED, "typed",
  64          &ksocknal_data.ksnd_typed_conns, sizeof (int),
  65          0644, NULL, &proc_dointvec},
  66         {SOCKNAL_SYSCTL_MIN_BULK, "min_bulk",
  67          &ksocknal_data.ksnd_min_bulk, sizeof (int),
  68          0644, NULL, &proc_dointvec},
  69         { 0 }
  70 };
  71
  72 static ctl_table ksocknal_top_ctl_table[] = {
  73         {SOCKNAL_SYSCTL, "socknal", NULL, 0, 0555, ksocknal_ctl_table},
  74         { 0 }
  75 };
  76
  77 int
  78 ksocknal_api_forward(nal_t *nal, int id, void *args, size_t args_len,
  79                        void *ret, size_t ret_len)
  80 {
  81         ksock_nal_data_t *k;
  82         nal_cb_t *nal_cb;
  83
  84         k = nal->nal_data;
  85         nal_cb = k->ksnd_nal_cb;
  86
  87         lib_dispatch(nal_cb, k, id, args, ret); /* ksocknal_send needs k */
  88         return PTL_OK;
  89 }
  90
  91 int
  92 ksocknal_api_shutdown(nal_t *nal, int ni)
  93 {
  94         return PTL_OK;
  95 }
  96
  97 void
  98 ksocknal_api_yield(nal_t *nal)
  99 {
 100         our_cond_resched();
 101         return;
 102 }
 103
 104 void
 105 ksocknal_api_lock(nal_t *nal, unsigned long *flags)
 106 {
 107         ksock_nal_data_t *k;
 108         nal_cb_t *nal_cb;
 109
 110         k = nal->nal_data;
 111         nal_cb = k->ksnd_nal_cb;
 112         nal_cb->cb_cli(nal_cb,flags);
 113 }
 114
 115 void
 116 ksocknal_api_unlock(nal_t *nal, unsigned long *flags)
 117 {
 118         ksock_nal_data_t *k;
 119         nal_cb_t *nal_cb;
 120
 121         k = nal->nal_data;
 122         nal_cb = k->ksnd_nal_cb;
 123         nal_cb->cb_sti(nal_cb,flags);
 124 }
 125
 126 nal_t *
 127 ksocknal_init(int interface, ptl_pt_index_t ptl_size,
 128               ptl_ac_index_t ac_size, ptl_pid_t requested_pid)
 129 {
 130         CDEBUG(D_NET, "calling lib_init with nid "LPX64"\n", (ptl_nid_t)0);
 131         lib_init(&ksocknal_lib, (ptl_nid_t)0, 0, 10, ptl_size, ac_size);
 132         return (&ksocknal_api);
 133 }
 134
 135 /*
 136  *  EXTRA functions follow
 137  */
 138
 139 int
 140 ksocknal_set_mynid(ptl_nid_t nid)
 141 {
 142         lib_ni_t *ni = &ksocknal_lib.ni;
 143
 144         /* FIXME: we have to do this because we call lib_init() at module
 145          * insertion time, which is before we have 'mynid' available.  lib_init
 146          * sets the NAL's nid, which it uses to tell other nodes where packets
 147          * are coming from.  This is not a very graceful solution to this
 148          * problem. */
 149
 150         CDEBUG(D_IOCTL, "setting mynid to "LPX64" (old nid="LPX64")\n",
 151                nid, ni->nid);
 152
 153         ni->nid = nid;
 154         return (0);
 155 }
 156
 157 void
 158 ksocknal_bind_irq (unsigned int irq)
 159 {
 160 #if (defined(CONFIG_SMP) && CPU_AFFINITY)
 161         int              bind;
 162         unsigned long    flags;
 163         char             cmdline[64];
 164         ksock_irqinfo_t *info;
 165         char            *argv[] = {"/bin/sh",
 166                                    "-c",
 167                                    cmdline,
 168                                    NULL};
 169         char            *envp[] = {"HOME=/",
 170                                    "PATH=/sbin:/bin:/usr/sbin:/usr/bin",
 171                                    NULL};
 172
 173         LASSERT (irq < NR_IRQS);
 174         if (irq == 0)                           /* software NIC */
 175                 return;
 176
 177         info = &ksocknal_data.ksnd_irqinfo[irq];
 178
 179         write_lock_irqsave (&ksocknal_data.ksnd_global_lock, flags);
 180
 181         LASSERT (info->ksni_valid);
 182         bind = !info->ksni_bound;
 183         info->ksni_bound = 1;
 184
 185         write_unlock_irqrestore (&ksocknal_data.ksnd_global_lock, flags);
 186
 187         if (!bind)                              /* bound already */
 188                 return;
 189
 190         snprintf (cmdline, sizeof (cmdline),
 191                   "echo %d > /proc/irq/%u/smp_affinity", 1 << info->ksni_sched, irq);
 192
 193         printk (KERN_INFO "Lustre: Binding irq %u to CPU %d with cmd: %s\n",
 194                 irq, info->ksni_sched, cmdline);
 195
 196         /* FIXME: Find a better method of setting IRQ affinity...
 197          */
 198
 199         call_usermodehelper (argv[0], argv, envp);
 200 #endif
 201 }
 202
 203 ksock_route_t *
 204 ksocknal_create_route (__u32 ipaddr, int port, int buffer_size,
 205                        int irq_affinity, int eager)
 206 {
 207         ksock_route_t *route;
 208
 209         PORTAL_ALLOC (route, sizeof (*route));
 210         if (route == NULL)
 211                 return (NULL);
 212
 213         atomic_set (&route->ksnr_refcount, 1);
 214         route->ksnr_sharecount = 0;
 215         route->ksnr_peer = NULL;
 216         route->ksnr_timeout = jiffies;
 217         route->ksnr_retry_interval = SOCKNAL_MIN_RECONNECT_INTERVAL;
 218         route->ksnr_ipaddr = ipaddr;
 219         route->ksnr_port = port;
 220         route->ksnr_buffer_size = buffer_size;
 221         route->ksnr_irq_affinity = irq_affinity;
 222         route->ksnr_eager = eager;
 223         route->ksnr_connecting = 0;
 224         route->ksnr_connected = 0;
 225         route->ksnr_deleted = 0;
 226         route->ksnr_conn_count = 0;
 227
 228         return (route);
 229 }
 230
 231 void
 232 ksocknal_destroy_route (ksock_route_t *route)
 233 {
 234         LASSERT (route->ksnr_sharecount == 0);
 235
 236         if (route->ksnr_peer != NULL)
 237                 ksocknal_put_peer (route->ksnr_peer);
 238
 239         PORTAL_FREE (route, sizeof (*route));
 240 }
 241
 242 void
 243 ksocknal_put_route (ksock_route_t *route)
 244 {
 245         CDEBUG (D_OTHER, "putting route[%p] (%d)\n",
 246                 route, atomic_read (&route->ksnr_refcount));
 247
 248         LASSERT (atomic_read (&route->ksnr_refcount) > 0);
 249         if (!atomic_dec_and_test (&route->ksnr_refcount))
 250              return;
 251
 252         ksocknal_destroy_route (route);
 253 }
 254
 255 ksock_peer_t *
 256 ksocknal_create_peer (ptl_nid_t nid)
 257 {
 258         ksock_peer_t *peer;
 259
 260         LASSERT (nid != PTL_NID_ANY);
 261
 262         PORTAL_ALLOC (peer, sizeof (*peer));
 263         if (peer == NULL)
 264                 return (NULL);
 265
 266         memset (peer, 0, sizeof (*peer));
 267
 268         peer->ksnp_nid = nid;
 269         atomic_set (&peer->ksnp_refcount, 1);   /* 1 ref for caller */
 270         peer->ksnp_closing = 0;
 271         INIT_LIST_HEAD (&peer->ksnp_conns);
 272         INIT_LIST_HEAD (&peer->ksnp_routes);
 273         INIT_LIST_HEAD (&peer->ksnp_tx_queue);
 274
 275         atomic_inc (&ksocknal_data.ksnd_npeers);
 276         return (peer);
 277 }
 278
 279 void
 280 ksocknal_destroy_peer (ksock_peer_t *peer)
 281 {
 282         CDEBUG (D_NET, "peer "LPX64" %p deleted\n", peer->ksnp_nid, peer);
 283
 284         LASSERT (atomic_read (&peer->ksnp_refcount) == 0);
 285         LASSERT (list_empty (&peer->ksnp_conns));
 286         LASSERT (list_empty (&peer->ksnp_routes));
 287         LASSERT (list_empty (&peer->ksnp_tx_queue));
 288
 289         PORTAL_FREE (peer, sizeof (*peer));
 290
 291         /* NB a peer's connections and autoconnect routes keep a reference
 292          * on their peer until they are destroyed, so we can be assured
 293          * that _all_ state to do with this peer has been cleaned up when
 294          * its refcount drops to zero. */
 295         atomic_dec (&ksocknal_data.ksnd_npeers);
 296 }
 297
 298 void
 299 ksocknal_put_peer (ksock_peer_t *peer)
 300 {
 301         CDEBUG (D_OTHER, "putting peer[%p] -> "LPX64" (%d)\n",
 302                 peer, peer->ksnp_nid,
 303                 atomic_read (&peer->ksnp_refcount));
 304
 305         LASSERT (atomic_read (&peer->ksnp_refcount) > 0);
 306         if (!atomic_dec_and_test (&peer->ksnp_refcount))
 307                 return;
 308
 309         ksocknal_destroy_peer (peer);
 310 }
 311
 312 ksock_peer_t *
 313 ksocknal_find_peer_locked (ptl_nid_t nid)
 314 {
 315         struct list_head *peer_list = ksocknal_nid2peerlist (nid);
 316         struct list_head *tmp;
 317         ksock_peer_t     *peer;
 318
 319         list_for_each (tmp, peer_list) {
 320
 321                 peer = list_entry (tmp, ksock_peer_t, ksnp_list);
 322
 323                 LASSERT (!peer->ksnp_closing);
 324                 LASSERT (!(list_empty (&peer->ksnp_routes) &&
 325                            list_empty (&peer->ksnp_conns)));
 326
 327                 if (peer->ksnp_nid != nid)
 328                         continue;
 329
 330                 CDEBUG(D_NET, "got peer [%p] -> "LPX64" (%d)\n",
 331                        peer, nid, atomic_read (&peer->ksnp_refcount));
 332                 return (peer);
 333         }
 334         return (NULL);
 335 }
 336
 337 ksock_peer_t *
 338 ksocknal_get_peer (ptl_nid_t nid)
 339 {
 340         ksock_peer_t     *peer;
 341
 342         read_lock (&ksocknal_data.ksnd_global_lock);
 343         peer = ksocknal_find_peer_locked (nid);
 344         if (peer != NULL)                       /* +1 ref for caller? */
 345                 atomic_inc (&peer->ksnp_refcount);
 346         read_unlock (&ksocknal_data.ksnd_global_lock);
 347
 348         return (peer);
 349 }
 350
 351 void
 352 ksocknal_unlink_peer_locked (ksock_peer_t *peer)
 353 {
 354         LASSERT (!peer->ksnp_closing);
 355         peer->ksnp_closing = 1;
 356         list_del (&peer->ksnp_list);
 357         /* lose peerlist's ref */
 358         ksocknal_put_peer (peer);
 359 }
 360
 361 ksock_route_t *
 362 ksocknal_get_route_by_idx (int index)
 363 {
 364         ksock_peer_t      *peer;
 365         struct list_head  *ptmp;
 366         ksock_route_t     *route;
 367         struct list_head  *rtmp;
 368         int                i;
 369
 370         read_lock (&ksocknal_data.ksnd_global_lock);
 371
 372         for (i = 0; i < ksocknal_data.ksnd_peer_hash_size; i++) {
 373                 list_for_each (ptmp, &ksocknal_data.ksnd_peers[i]) {
 374                         peer = list_entry (ptmp, ksock_peer_t, ksnp_list);
 375
 376                         LASSERT (!(list_empty (&peer->ksnp_routes) &&
 377                                    list_empty (&peer->ksnp_conns)));
 378
 379                         list_for_each (rtmp, &peer->ksnp_routes) {
 380                                 if (index-- > 0)
 381                                         continue;
 382
 383                                 route = list_entry (rtmp, ksock_route_t, ksnr_list);
 384                                 atomic_inc (&route->ksnr_refcount);
 385                                 read_unlock (&ksocknal_data.ksnd_global_lock);
 386                                 return (route);
 387                         }
 388                 }
 389         }
 390
 391         read_unlock (&ksocknal_data.ksnd_global_lock);
 392         return (NULL);
 393 }
 394
 395 int
 396 ksocknal_add_route (ptl_nid_t nid, __u32 ipaddr, int port, int bufnob,
 397                     int bind_irq, int share, int eager)
 398 {
 399         unsigned long      flags;
 400         ksock_peer_t      *peer;
 401         ksock_peer_t      *peer2;
 402         ksock_route_t     *route;
 403         struct list_head  *rtmp;
 404         ksock_route_t     *route2;
 405
 406         if (nid == PTL_NID_ANY)
 407                 return (-EINVAL);
 408
 409         /* Have a brand new peer ready... */
 410         peer = ksocknal_create_peer (nid);
 411         if (peer == NULL)
 412                 return (-ENOMEM);
 413
 414         route = ksocknal_create_route (ipaddr, port, bufnob,
 415                                        bind_irq, eager);
 416         if (route == NULL) {
 417                 ksocknal_put_peer (peer);
 418                 return (-ENOMEM);
 419         }
 420
 421         write_lock_irqsave (&ksocknal_data.ksnd_global_lock, flags);
 422
 423         peer2 = ksocknal_find_peer_locked (nid);
 424         if (peer2 != NULL) {
 425                 ksocknal_put_peer (peer);
 426                 peer = peer2;
 427         } else {
 428                 /* peer table takes existing ref on peer */
 429                 list_add (&peer->ksnp_list,
 430                           ksocknal_nid2peerlist (nid));
 431         }
 432
 433         route2 = NULL;
 434         if (share) {
 435                 /* check for existing route to this NID via this ipaddr */
 436                 list_for_each (rtmp, &peer->ksnp_routes) {
 437                         route2 = list_entry (rtmp, ksock_route_t, ksnr_list);
 438
 439                         if (route2->ksnr_ipaddr == ipaddr)
 440                                 break;
 441
 442                         route2 = NULL;
 443                 }
 444         }
 445
 446         if (route2 != NULL) {
 447                 ksocknal_put_route (route);
 448                 route = route2;
 449         } else {
 450                 /* route takes a ref on peer */
 451                 route->ksnr_peer = peer;
 452                 atomic_inc (&peer->ksnp_refcount);
 453                 /* peer's route list takes existing ref on route */
 454                 list_add_tail (&route->ksnr_list, &peer->ksnp_routes);
 455         }
 456
 457         route->ksnr_sharecount++;
 458
 459         write_unlock_irqrestore (&ksocknal_data.ksnd_global_lock, flags);
 460
 461         return (0);
 462 }
 463
 464 void
 465 ksocknal_del_route_locked (ksock_route_t *route, int share, int keep_conn)
 466 {
 467         ksock_peer_t     *peer = route->ksnr_peer;
 468         ksock_conn_t     *conn;
 469         struct list_head *ctmp;
 470         struct list_head *cnxt;
 471
 472         if (!share)
 473                 route->ksnr_sharecount = 0;
 474         else {
 475                 route->ksnr_sharecount--;
 476                 if (route->ksnr_sharecount != 0)
 477                         return;
 478         }
 479
 480         list_for_each_safe (ctmp, cnxt, &peer->ksnp_conns) {
 481                 conn = list_entry(ctmp, ksock_conn_t, ksnc_list);
 482
 483                 if (conn->ksnc_route != route)
 484                         continue;
 485
 486                 if (!keep_conn) {
 487                         ksocknal_close_conn_locked (conn, 0);
 488                         continue;
 489                 }
 490
 491                 /* keeping the conn; just dissociate it and route... */
 492                 conn->ksnc_route = NULL;
 493                 ksocknal_put_route (route); /* drop conn's ref on route */
 494         }
 495
 496         route->ksnr_deleted = 1;
 497         list_del (&route->ksnr_list);
 498         ksocknal_put_route (route);             /* drop peer's ref */
 499
 500         if (list_empty (&peer->ksnp_routes) &&
 501             list_empty (&peer->ksnp_conns)) {
 502                 /* I've just removed the last autoconnect route of a peer
 503                  * with no active connections */
 504                 ksocknal_unlink_peer_locked (peer);
 505         }
 506 }
 507
 508 int
 509 ksocknal_del_route (ptl_nid_t nid, __u32 ipaddr, int share, int keep_conn)
 510 {
 511         unsigned long      flags;
 512         struct list_head  *ptmp;
 513         struct list_head  *pnxt;
 514         ksock_peer_t      *peer;
 515         struct list_head  *rtmp;
 516         struct list_head  *rnxt;
 517         ksock_route_t     *route;
 518         int                lo;
 519         int                hi;
 520         int                i;
 521         int                rc = -ENOENT;
 522
 523         write_lock_irqsave (&ksocknal_data.ksnd_global_lock, flags);
 524
 525         if (nid != PTL_NID_ANY)
 526                 lo = hi = ksocknal_nid2peerlist(nid) - ksocknal_data.ksnd_peers;
 527         else {
 528                 lo = 0;
 529                 hi = ksocknal_data.ksnd_peer_hash_size - 1;
 530         }
 531
 532         for (i = lo; i <= hi; i++) {
 533                 list_for_each_safe (ptmp, pnxt, &ksocknal_data.ksnd_peers[i]) {
 534                         peer = list_entry (ptmp, ksock_peer_t, ksnp_list);
 535
 536                         if (!(nid == PTL_NID_ANY || peer->ksnp_nid == nid))
 537                                 continue;
 538
 539                         list_for_each_safe (rtmp, rnxt, &peer->ksnp_routes) {
 540                                 route = list_entry (rtmp, ksock_route_t,
 541                                                     ksnr_list);
 542
 543                                 if (!(ipaddr == 0 ||
 544                                       route->ksnr_ipaddr == ipaddr))
 545                                         continue;
 546
 547                                 ksocknal_del_route_locked (route, share, keep_conn);
 548                                 rc = 0;         /* matched something */
 549                                 if (share)
 550                                         goto out;
 551                         }
 552                 }
 553         }
 554  out:
 555         write_unlock_irqrestore (&ksocknal_data.ksnd_global_lock, flags);
 556
 557         return (rc);
 558 }
 559
 560 ksock_conn_t *
 561 ksocknal_get_conn_by_idx (int index)
 562 {
 563         ksock_peer_t      *peer;
 564         struct list_head  *ptmp;
 565         ksock_conn_t      *conn;
 566         struct list_head  *ctmp;
 567         int                i;
 568
 569         read_lock (&ksocknal_data.ksnd_global_lock);
 570
 571         for (i = 0; i < ksocknal_data.ksnd_peer_hash_size; i++) {
 572                 list_for_each (ptmp, &ksocknal_data.ksnd_peers[i]) {
 573                         peer = list_entry (ptmp, ksock_peer_t, ksnp_list);
 574
 575                         LASSERT (!(list_empty (&peer->ksnp_routes) &&
 576                                    list_empty (&peer->ksnp_conns)));
 577
 578                         list_for_each (ctmp, &peer->ksnp_conns) {
 579                                 if (index-- > 0)
 580                                         continue;
 581
 582                                 conn = list_entry (ctmp, ksock_conn_t, ksnc_list);
 583                                 atomic_inc (&conn->ksnc_refcount);
 584                                 read_unlock (&ksocknal_data.ksnd_global_lock);
 585                                 return (conn);
 586                         }
 587                 }
 588         }
 589
 590         read_unlock (&ksocknal_data.ksnd_global_lock);
 591         return (NULL);
 592 }
 593
 594 void
 595 ksocknal_get_peer_addr (ksock_conn_t *conn)
 596 {
 597         struct sockaddr_in sin;
 598         int                len = sizeof (sin);
 599         int                rc;
 600
 601         rc = conn->ksnc_sock->ops->getname (conn->ksnc_sock,
 602                                             (struct sockaddr *)&sin, &len, 2);
 603         /* Didn't need the {get,put}connsock dance to deref ksnc_sock... */
 604         LASSERT (!conn->ksnc_closing);
 605         LASSERT (len <= sizeof (sin));
 606
 607         if (rc != 0) {
 608                 CERROR ("Error %d getting sock peer IP\n", rc);
 609                 return;
 610         }
 611
 612         conn->ksnc_ipaddr = ntohl (sin.sin_addr.s_addr);
 613         conn->ksnc_port   = ntohs (sin.sin_port);
 614 }
 615
 616 unsigned int
 617 ksocknal_conn_irq (ksock_conn_t *conn)
 618 {
 619         int                irq = 0;
 620         struct dst_entry  *dst;
 621
 622         dst = sk_dst_get (conn->ksnc_sock->sk);
 623         if (dst != NULL) {
 624                 if (dst->dev != NULL) {
 625                         irq = dst->dev->irq;
 626                         if (irq >= NR_IRQS) {
 627                                 CERROR ("Unexpected IRQ %x\n", irq);
 628                                 irq = 0;
 629                         }
 630                 }
 631                 dst_release (dst);
 632         }
 633
 634         /* Didn't need the {get,put}connsock dance to deref ksnc_sock... */
 635         LASSERT (!conn->ksnc_closing);
 636         return (irq);
 637 }
 638
 639 ksock_sched_t *
 640 ksocknal_choose_scheduler_locked (unsigned int irq)
 641 {
 642         ksock_sched_t    *sched;
 643         ksock_irqinfo_t  *info;
 644         int               i;
 645
 646         LASSERT (irq < NR_IRQS);
 647         info = &ksocknal_data.ksnd_irqinfo[irq];
 648
 649         if (irq != 0 &&                         /* hardware NIC */
 650             info->ksni_valid) {                 /* already set up */
 651                 return (&ksocknal_data.ksnd_schedulers[info->ksni_sched]);
 652         }
 653
 654         /* software NIC (irq == 0) || not associated with a scheduler yet.
 655          * Choose the CPU with the fewest connections... */
 656         sched = &ksocknal_data.ksnd_schedulers[0];
 657         for (i = 1; i < SOCKNAL_N_SCHED; i++)
 658                 if (sched->kss_nconns >
 659                     ksocknal_data.ksnd_schedulers[i].kss_nconns)
 660                         sched = &ksocknal_data.ksnd_schedulers[i];
 661
 662         if (irq != 0) {                         /* Hardware NIC */
 663                 info->ksni_valid = 1;
 664                 info->ksni_sched = sched - ksocknal_data.ksnd_schedulers;
 665
 666                 /* no overflow... */
 667                 LASSERT (info->ksni_sched == sched - ksocknal_data.ksnd_schedulers);
 668         }
 669
 670         return (sched);
 671 }
 672
 673 int
 674 ksocknal_create_conn (ksock_route_t *route, struct socket *sock,
 675                       int bind_irq, int type)
 676 {
 677         ptl_nid_t          nid;
 678         __u64              incarnation;
 679         unsigned long      flags;
 680         ksock_conn_t      *conn;
 681         ksock_peer_t      *peer;
 682         ksock_peer_t      *peer2;
 683         ksock_sched_t     *sched;
 684         unsigned int       irq;
 685         ksock_tx_t        *tx;
 686         int                rc;
 687
 688         /* NB, sock has an associated file since (a) this connection might
 689          * have been created in userland and (b) we need to refcount the
 690          * socket so that we don't close it while I/O is being done on
 691          * it, and sock->file has that pre-cooked... */
 692         LASSERT (sock->file != NULL);
 693         LASSERT (file_count(sock->file) > 0);
 694
 695         rc = ksocknal_setup_sock (sock);
 696         if (rc != 0)
 697                 return (rc);
 698
 699         if (route == NULL) {
 700                 /* acceptor or explicit connect */
 701                 nid = PTL_NID_ANY;
 702         } else {
 703                 LASSERT (type != SOCKNAL_CONN_NONE);
 704                 /* autoconnect: expect this nid on exchange */
 705                 nid = route->ksnr_peer->ksnp_nid;
 706         }
 707
 708         rc = ksocknal_hello (sock, &nid, &type, &incarnation);
 709         if (rc != 0)
 710                 return (rc);
 711
 712         peer = NULL;
 713         if (route == NULL) {                    /* not autoconnect */
 714                 /* Assume this socket connects to a brand new peer */
 715                 peer = ksocknal_create_peer (nid);
 716                 if (peer == NULL)
 717                         return (-ENOMEM);
 718         }
 719
 720         PORTAL_ALLOC(conn, sizeof(*conn));
 721         if (conn == NULL) {
 722                 if (peer != NULL)
 723                         ksocknal_put_peer (peer);
 724                 return (-ENOMEM);
 725         }
 726
 727         memset (conn, 0, sizeof (*conn));
 728         conn->ksnc_peer = NULL;
 729         conn->ksnc_route = NULL;
 730         conn->ksnc_sock = sock;
 731         conn->ksnc_type = type;
 732         conn->ksnc_incarnation = incarnation;
 733         conn->ksnc_saved_data_ready = sock->sk->sk_data_ready;
 734         conn->ksnc_saved_write_space = sock->sk->sk_write_space;
 735         atomic_set (&conn->ksnc_refcount, 1);    /* 1 ref for me */
 736
 737         conn->ksnc_rx_ready = 0;
 738         conn->ksnc_rx_scheduled = 0;
 739         ksocknal_new_packet (conn, 0);
 740
 741         INIT_LIST_HEAD (&conn->ksnc_tx_queue);
 742         conn->ksnc_tx_ready = 0;
 743         conn->ksnc_tx_scheduled = 0;
 744         atomic_set (&conn->ksnc_tx_nob, 0);
 745
 746         ksocknal_get_peer_addr (conn);
 747
 748         irq = ksocknal_conn_irq (conn);
 749
 750         write_lock_irqsave (&ksocknal_data.ksnd_global_lock, flags);
 751
 752         if (route != NULL) {
 753                 /* Autoconnected! */
 754                 LASSERT ((route->ksnr_connected & (1 << type)) == 0);
 755                 LASSERT ((route->ksnr_connecting & (1 << type)) != 0);
 756
 757                 if (route->ksnr_deleted) {
 758                         /* This conn was autoconnected, but the autoconnect
 759                          * route got deleted while it was being
 760                          * established! */
 761                         write_unlock_irqrestore (&ksocknal_data.ksnd_global_lock,
 762                                                  flags);
 763                         PORTAL_FREE (conn, sizeof (*conn));
 764                         return (-ESTALE);
 765                 }
 766
 767
 768                 /* associate conn/route */
 769                 conn->ksnc_route = route;
 770                 atomic_inc (&route->ksnr_refcount);
 771
 772                 route->ksnr_connecting &= ~(1 << type);
 773                 route->ksnr_connected  |= (1 << type);
 774                 route->ksnr_conn_count++;
 775                 route->ksnr_retry_interval = SOCKNAL_MIN_RECONNECT_INTERVAL;
 776
 777                 peer = route->ksnr_peer;
 778         } else {
 779                 /* Not an autoconnected connection; see if there is an
 780                  * existing peer for this NID */
 781                 peer2 = ksocknal_find_peer_locked (nid);
 782                 if (peer2 != NULL) {
 783                         ksocknal_put_peer (peer);
 784                         peer = peer2;
 785                 } else {
 786                         list_add (&peer->ksnp_list,
 787                                   ksocknal_nid2peerlist (nid));
 788                         /* peer list takes over existing ref */
 789                 }
 790         }
 791
 792         LASSERT (!peer->ksnp_closing);
 793
 794         conn->ksnc_peer = peer;
 795         atomic_inc (&peer->ksnp_refcount);
 796         peer->ksnp_last_alive = jiffies;
 797         peer->ksnp_error = 0;
 798
 799         /* Set the deadline for the outgoing HELLO to drain */
 800         conn->ksnc_tx_deadline = jiffies +
 801                                  ksocknal_data.ksnd_io_timeout * HZ;
 802
 803         list_add (&conn->ksnc_list, &peer->ksnp_conns);
 804         atomic_inc (&conn->ksnc_refcount);
 805
 806         sched = ksocknal_choose_scheduler_locked (irq);
 807         sched->kss_nconns++;
 808         conn->ksnc_scheduler = sched;
 809
 810         /* NB my callbacks block while I hold ksnd_global_lock */
 811         sock->sk->sk_user_data = conn;
 812         sock->sk->sk_data_ready = ksocknal_data_ready;
 813         sock->sk->sk_write_space = ksocknal_write_space;
 814
 815         /* Take all the packets blocking for a connection.
 816          * NB, it might be nicer to share these blocked packets among any
 817          * other connections that are becoming established, however that
 818          * confuses the normal packet launching operation, which selects a
 819          * connection and queues the packet on it without needing an
 820          * exclusive lock on ksnd_global_lock. */
 821         while (!list_empty (&peer->ksnp_tx_queue)) {
 822                 tx = list_entry (peer->ksnp_tx_queue.next,
 823                                  ksock_tx_t, tx_list);
 824
 825                 list_del (&tx->tx_list);
 826                 ksocknal_queue_tx_locked (tx, conn);
 827         }
 828
 829         rc = ksocknal_close_stale_conns_locked (peer, incarnation);
 830
 831         write_unlock_irqrestore (&ksocknal_data.ksnd_global_lock, flags);
 832
 833         if (rc != 0)
 834                 CERROR ("Closed %d stale conns to nid "LPX64" ip %d.%d.%d.%d\n",
 835                         rc, conn->ksnc_peer->ksnp_nid,
 836                         HIPQUAD(conn->ksnc_ipaddr));
 837
 838         if (bind_irq)                           /* irq binding required */
 839                 ksocknal_bind_irq (irq);
 840
 841         /* Call the callbacks right now to get things going. */
 842         ksocknal_data_ready (sock->sk, 0);
 843         ksocknal_write_space (sock->sk);
 844
 845         CDEBUG(D_IOCTL, "conn [%p] registered for nid "LPX64" ip %d.%d.%d.%d\n",
 846                conn, conn->ksnc_peer->ksnp_nid, HIPQUAD(conn->ksnc_ipaddr));
 847
 848         ksocknal_put_conn (conn);
 849         return (0);
 850 }
 851
 852 void
 853 ksocknal_close_conn_locked (ksock_conn_t *conn, int error)
 854 {
 855         /* This just does the immmediate housekeeping, and queues the
 856          * connection for the reaper to terminate.
 857          * Caller holds ksnd_global_lock exclusively in irq context */
 858         ksock_peer_t   *peer = conn->ksnc_peer;
 859         ksock_route_t  *route;
 860
 861         LASSERT (peer->ksnp_error == 0);
 862         LASSERT (!conn->ksnc_closing);
 863         conn->ksnc_closing = 1;
 864         atomic_inc (&ksocknal_data.ksnd_nclosing_conns);
 865
 866         route = conn->ksnc_route;
 867         if (route != NULL) {
 868                 /* dissociate conn from route... */
 869                 LASSERT (!route->ksnr_deleted);
 870                 LASSERT ((route->ksnr_connecting & (1 << conn->ksnc_type)) == 0);
 871                 LASSERT ((route->ksnr_connected & (1 << conn->ksnc_type)) != 0);
 872
 873                 route->ksnr_connected &= ~(1 << conn->ksnc_type);
 874                 conn->ksnc_route = NULL;
 875
 876                 list_del (&route->ksnr_list);   /* make route least favourite */
 877                 list_add_tail (&route->ksnr_list, &peer->ksnp_routes);
 878
 879                 ksocknal_put_route (route);     /* drop conn's ref on route */
 880         }
 881
 882         /* ksnd_deathrow_conns takes over peer's ref */
 883         list_del (&conn->ksnc_list);
 884
 885         if (list_empty (&peer->ksnp_conns)) {
 886                 /* No more connections to this peer */
 887
 888                 peer->ksnp_error = error;       /* stash last conn close reason */
 889
 890                 if (list_empty (&peer->ksnp_routes)) {
 891                         /* I've just closed last conn belonging to a
 892                          * non-autoconnecting peer */
 893                         ksocknal_unlink_peer_locked (peer);
 894                 }
 895         }
 896
 897         spin_lock (&ksocknal_data.ksnd_reaper_lock);
 898
 899         list_add_tail (&conn->ksnc_list, &ksocknal_data.ksnd_deathrow_conns);
 900         wake_up (&ksocknal_data.ksnd_reaper_waitq);
 901
 902         spin_unlock (&ksocknal_data.ksnd_reaper_lock);
 903 }
 904
 905 void
 906 ksocknal_terminate_conn (ksock_conn_t *conn)
 907 {
 908         /* This gets called by the reaper (guaranteed thread context) to
 909          * disengage the socket from its callbacks and close it.
 910          * ksnc_refcount will eventually hit zero, and then the reaper will
 911          * destroy it. */
 912         unsigned long   flags;
 913         ksock_peer_t   *peer = conn->ksnc_peer;
 914         ksock_sched_t  *sched = conn->ksnc_scheduler;
 915         struct timeval  now;
 916         time_t          then = 0;
 917         int             notify = 0;
 918
 919         LASSERT(conn->ksnc_closing);
 920
 921         /* wake up the scheduler to "send" all remaining packets to /dev/null */
 922         spin_lock_irqsave(&sched->kss_lock, flags);
 923
 924         if (!conn->ksnc_tx_scheduled &&
 925             !list_empty(&conn->ksnc_tx_queue)){
 926                 list_add_tail (&conn->ksnc_tx_list,
 927                                &sched->kss_tx_conns);
 928                 /* a closing conn is always ready to tx */
 929                 conn->ksnc_tx_ready = 1;
 930                 conn->ksnc_tx_scheduled = 1;
 931                 /* extra ref for scheduler */
 932                 atomic_inc (&conn->ksnc_refcount);
 933
 934                 wake_up (&sched->kss_waitq);
 935         }
 936
 937         spin_unlock_irqrestore (&sched->kss_lock, flags);
 938
 939         /* serialise with callbacks */
 940         write_lock_irqsave (&ksocknal_data.ksnd_global_lock, flags);
 941
 942         /* Remove conn's network callbacks.
 943          * NB I _have_ to restore the callback, rather than storing a noop,
 944          * since the socket could survive past this module being unloaded!! */
 945         conn->ksnc_sock->sk->sk_data_ready = conn->ksnc_saved_data_ready;
 946         conn->ksnc_sock->sk->sk_write_space = conn->ksnc_saved_write_space;
 947
 948         /* A callback could be in progress already; they hold a read lock
 949          * on ksnd_global_lock (to serialise with me) and NOOP if
 950          * sk_user_data is NULL. */
 951         conn->ksnc_sock->sk->sk_user_data = NULL;
 952
 953         /* OK, so this conn may not be completely disengaged from its
 954          * scheduler yet, but it _has_ committed to terminate... */
 955         conn->ksnc_scheduler->kss_nconns--;
 956
 957         if (peer->ksnp_error != 0) {
 958                 /* peer's last conn closed in error */
 959                 LASSERT (list_empty (&peer->ksnp_conns));
 960
 961                 /* convert peer's last-known-alive timestamp from jiffies */
 962                 do_gettimeofday (&now);
 963                 then = now.tv_sec - (jiffies - peer->ksnp_last_alive)/HZ;
 964                 notify = 1;
 965         }
 966
 967         write_unlock_irqrestore (&ksocknal_data.ksnd_global_lock, flags);
 968
 969         /* The socket is closed on the final put; either here, or in
 970          * ksocknal_{send,recv}msg().  Since we set up the linger2 option
 971          * when the connection was established, this will close the socket
 972          * immediately, aborting anything buffered in it. Any hung
 973          * zero-copy transmits will therefore complete in finite time. */
 974         ksocknal_putconnsock (conn);
 975
 976         if (notify)
 977                 kpr_notify (&ksocknal_data.ksnd_router, peer->ksnp_nid,
 978                             0, then);
 979 }
 980
 981 void
 982 ksocknal_destroy_conn (ksock_conn_t *conn)
 983 {
 984         /* Final coup-de-grace of the reaper */
 985         CDEBUG (D_NET, "connection %p\n", conn);
 986
 987         LASSERT (atomic_read (&conn->ksnc_refcount) == 0);
 988         LASSERT (conn->ksnc_route == NULL);
 989         LASSERT (!conn->ksnc_tx_scheduled);
 990         LASSERT (!conn->ksnc_rx_scheduled);
 991         LASSERT (list_empty(&conn->ksnc_tx_queue));
 992
 993         /* complete current receive if any */
 994         switch (conn->ksnc_rx_state) {
 995         case SOCKNAL_RX_BODY:
 996                 CERROR("Completing partial receive from "LPX64
 997                        ", ip %d.%d.%d.%d:%d, with error\n",
 998                        conn->ksnc_peer->ksnp_nid,
 999                        HIPQUAD(conn->ksnc_ipaddr), conn->ksnc_port);
1000                 lib_finalize (&ksocknal_lib, NULL, conn->ksnc_cookie, PTL_FAIL);
1001                 break;
1002         case SOCKNAL_RX_BODY_FWD:
1003                 ksocknal_fmb_callback (conn->ksnc_cookie, -ECONNABORTED);
1004                 break;
1005         case SOCKNAL_RX_HEADER:
1006         case SOCKNAL_RX_SLOP:
1007                 break;
1008         default:
1009                 LBUG ();
1010                 break;
1011         }
1012
1013         ksocknal_put_peer (conn->ksnc_peer);
1014
1015         PORTAL_FREE (conn, sizeof (*conn));
1016         atomic_dec (&ksocknal_data.ksnd_nclosing_conns);
1017 }
1018
1019 void
1020 ksocknal_put_conn (ksock_conn_t *conn)
1021 {
1022         unsigned long flags;
1023
1024         CDEBUG (D_OTHER, "putting conn[%p] -> "LPX64" (%d)\n",
1025                 conn, conn->ksnc_peer->ksnp_nid,
1026                 atomic_read (&conn->ksnc_refcount));
1027
1028         LASSERT (atomic_read (&conn->ksnc_refcount) > 0);
1029         if (!atomic_dec_and_test (&conn->ksnc_refcount))
1030                 return;
1031
1032         spin_lock_irqsave (&ksocknal_data.ksnd_reaper_lock, flags);
1033
1034         list_add (&conn->ksnc_list, &ksocknal_data.ksnd_zombie_conns);
1035         wake_up (&ksocknal_data.ksnd_reaper_waitq);
1036
1037         spin_unlock_irqrestore (&ksocknal_data.ksnd_reaper_lock, flags);
1038 }
1039
1040 int
1041 ksocknal_close_peer_conns_locked (ksock_peer_t *peer, __u32 ipaddr, int why)
1042 {
1043         ksock_conn_t       *conn;
1044         struct list_head   *ctmp;
1045         struct list_head   *cnxt;
1046         int                 count = 0;
1047
1048         list_for_each_safe (ctmp, cnxt, &peer->ksnp_conns) {
1049                 conn = list_entry (ctmp, ksock_conn_t, ksnc_list);
1050
1051                 if (ipaddr == 0 ||
1052                     conn->ksnc_ipaddr == ipaddr) {
1053                         count++;
1054                         ksocknal_close_conn_locked (conn, why);
1055                 }
1056         }
1057
1058         return (count);
1059 }
1060
1061 int
1062 ksocknal_close_stale_conns_locked (ksock_peer_t *peer, __u64 incarnation)
1063 {
1064         ksock_conn_t       *conn;
1065         struct list_head   *ctmp;
1066         struct list_head   *cnxt;
1067         int                 count = 0;
1068
1069         list_for_each_safe (ctmp, cnxt, &peer->ksnp_conns) {
1070                 conn = list_entry (ctmp, ksock_conn_t, ksnc_list);
1071
1072                 if (conn->ksnc_incarnation == incarnation)
1073                         continue;
1074
1075                 count++;
1076                 ksocknal_close_conn_locked (conn, -ESTALE);
1077         }
1078
1079         return (count);
1080 }
1081
1082 int
1083 ksocknal_close_conn_and_siblings (ksock_conn_t *conn, int why)
1084 {
1085         ksock_peer_t     *peer = conn->ksnc_peer;
1086         __u32             ipaddr = conn->ksnc_ipaddr;
1087         unsigned long     flags;
1088         int               count;
1089
1090         write_lock_irqsave (&ksocknal_data.ksnd_global_lock, flags);
1091
1092         count = ksocknal_close_peer_conns_locked (peer, ipaddr, why);
1093
1094         write_unlock_irqrestore (&ksocknal_data.ksnd_global_lock, flags);
1095
1096         return (count);
1097 }
1098
1099 int
1100 ksocknal_close_matching_conns (ptl_nid_t nid, __u32 ipaddr)
1101 {
1102         unsigned long       flags;
1103         ksock_peer_t       *peer;
1104         struct list_head   *ptmp;
1105         struct list_head   *pnxt;
1106         int                 lo;
1107         int                 hi;
1108         int                 i;
1109         int                 count = 0;
1110
1111         write_lock_irqsave (&ksocknal_data.ksnd_global_lock, flags);
1112
1113         if (nid != PTL_NID_ANY)
1114                 lo = hi = ksocknal_nid2peerlist(nid) - ksocknal_data.ksnd_peers;
1115         else {
1116                 lo = 0;
1117                 hi = ksocknal_data.ksnd_peer_hash_size - 1;
1118         }
1119
1120         for (i = lo; i <= hi; i++) {
1121                 list_for_each_safe (ptmp, pnxt, &ksocknal_data.ksnd_peers[i]) {
1122
1123                         peer = list_entry (ptmp, ksock_peer_t, ksnp_list);
1124
1125                         if (!(nid == PTL_NID_ANY || nid == peer->ksnp_nid))
1126                                 continue;
1127
1128                         count += ksocknal_close_peer_conns_locked (peer, ipaddr, 0);
1129                 }
1130         }
1131
1132         write_unlock_irqrestore (&ksocknal_data.ksnd_global_lock, flags);
1133
1134         /* wildcards always succeed */
1135         if (nid == PTL_NID_ANY || ipaddr == 0)
1136                 return (0);
1137
1138         return (count == 0 ? -ENOENT : 0);
1139 }
1140
1141 void
1142 ksocknal_notify (void *arg, ptl_nid_t gw_nid, int alive)
1143 {
1144         /* The router is telling me she's been notified of a change in
1145          * gateway state.... */
1146
1147         CDEBUG (D_NET, "gw "LPX64" %s\n", gw_nid, alive ? "up" : "down");
1148
1149         if (!alive) {
1150                 /* If the gateway crashed, close all open connections... */
1151                 ksocknal_close_matching_conns (gw_nid, 0);
1152                 return;
1153         }
1154
1155         /* ...otherwise do nothing.  We can only establish new connections
1156          * if we have autroutes, and these connect on demand. */
1157 }
1158
1159 #if (LINUX_VERSION_CODE < KERNEL_VERSION(2,5,0))
1160 struct tcp_opt *sock2tcp_opt(struct sock *sk)
1161 {
1162         return &(sk->tp_pinfo.af_tcp);
1163 }
1164 #else
1165 struct tcp_opt *sock2tcp_opt(struct sock *sk)
1166 {
1167         struct tcp_sock *s = (struct tcp_sock *)sk;
1168         return &s->tcp;
1169 }
1170 #endif
1171
1172 void
1173 ksocknal_push_conn (ksock_conn_t *conn)
1174 {
1175         struct sock    *sk;
1176         struct tcp_opt *tp;
1177         int             nonagle;
1178         int             val = 1;
1179         int             rc;
1180         mm_segment_t    oldmm;
1181
1182         rc = ksocknal_getconnsock (conn);
1183         if (rc != 0)                            /* being shut down */
1184                 return;
1185
1186         sk = conn->ksnc_sock->sk;
1187         tp = sock2tcp_opt(sk);
1188
1189         lock_sock (sk);
1190         nonagle = tp->nonagle;
1191         tp->nonagle = 1;
1192         release_sock (sk);
1193
1194         oldmm = get_fs ();
1195         set_fs (KERNEL_DS);
1196
1197         rc = sk->sk_prot->setsockopt (sk, SOL_TCP, TCP_NODELAY,
1198                                       (char *)&val, sizeof (val));
1199         LASSERT (rc == 0);
1200
1201         set_fs (oldmm);
1202
1203         lock_sock (sk);
1204         tp->nonagle = nonagle;
1205         release_sock (sk);
1206
1207         ksocknal_putconnsock (conn);
1208 }
1209
1210 void
1211 ksocknal_push_peer (ksock_peer_t *peer)
1212 {
1213         int               index;
1214         int               i;
1215         struct list_head *tmp;
1216         ksock_conn_t     *conn;
1217
1218         for (index = 0; ; index++) {
1219                 read_lock (&ksocknal_data.ksnd_global_lock);
1220
1221                 i = 0;
1222                 conn = NULL;
1223
1224                 list_for_each (tmp, &peer->ksnp_conns) {
1225                         if (i++ == index) {
1226                                 conn = list_entry (tmp, ksock_conn_t, ksnc_list);
1227                                 atomic_inc (&conn->ksnc_refcount);
1228                                 break;
1229                         }
1230                 }
1231
1232                 read_unlock (&ksocknal_data.ksnd_global_lock);
1233
1234                 if (conn == NULL)
1235                         break;
1236
1237                 ksocknal_push_conn (conn);
1238                 ksocknal_put_conn (conn);
1239         }
1240 }
1241
1242 int
1243 ksocknal_push (ptl_nid_t nid)
1244 {
1245         ksock_peer_t      *peer;
1246         struct list_head  *tmp;
1247         int                index;
1248         int                i;
1249         int                j;
1250         int                rc = -ENOENT;
1251
1252         if (nid != PTL_NID_ANY) {
1253                 peer = ksocknal_get_peer (nid);
1254
1255                 if (peer != NULL) {
1256                         rc = 0;
1257                         ksocknal_push_peer (peer);
1258                         ksocknal_put_peer (peer);
1259                 }
1260                 return (rc);
1261         }
1262
1263         for (i = 0; i < ksocknal_data.ksnd_peer_hash_size; i++) {
1264                 for (j = 0; ; j++) {
1265                         read_lock (&ksocknal_data.ksnd_global_lock);
1266
1267                         index = 0;
1268                         peer = NULL;
1269
1270                         list_for_each (tmp, &ksocknal_data.ksnd_peers[i]) {
1271                                 if (index++ == j) {
1272                                         peer = list_entry(tmp, ksock_peer_t,
1273                                                           ksnp_list);
1274                                         atomic_inc (&peer->ksnp_refcount);
1275                                         break;
1276                                 }
1277                         }
1278
1279                         read_unlock (&ksocknal_data.ksnd_global_lock);
1280
1281                         if (peer != NULL) {
1282                                 rc = 0;
1283                                 ksocknal_push_peer (peer);
1284                                 ksocknal_put_peer (peer);
1285                         }
1286                 }
1287
1288         }
1289
1290         return (rc);
1291 }
1292
1293 int
1294 ksocknal_cmd(struct portals_cfg *pcfg, void * private)
1295 {
1296         int rc = -EINVAL;
1297
1298         LASSERT (pcfg != NULL);
1299
1300         switch(pcfg->pcfg_command) {
1301         case NAL_CMD_GET_AUTOCONN: {
1302                 ksock_route_t *route = ksocknal_get_route_by_idx (pcfg->pcfg_count);
1303
1304                 if (route == NULL)
1305                         rc = -ENOENT;
1306                 else {
1307                         rc = 0;
1308                         pcfg->pcfg_nid   = route->ksnr_peer->ksnp_nid;
1309                         pcfg->pcfg_id    = route->ksnr_ipaddr;
1310                         pcfg->pcfg_misc  = route->ksnr_port;
1311                         pcfg->pcfg_count = route->ksnr_conn_count;
1312                         pcfg->pcfg_size  = route->ksnr_buffer_size;
1313                         pcfg->pcfg_wait  = route->ksnr_sharecount;
1314                         pcfg->pcfg_flags = (route->ksnr_irq_affinity ? 2 : 0) |
1315                                            (route->ksnr_eager        ? 4 : 0);
1316                         ksocknal_put_route (route);
1317                 }
1318                 break;
1319         }
1320         case NAL_CMD_ADD_AUTOCONN: {
1321                 rc = ksocknal_add_route (pcfg->pcfg_nid, pcfg->pcfg_id,
1322                                          pcfg->pcfg_misc, pcfg->pcfg_size,
1323                                          (pcfg->pcfg_flags & 0x02) != 0,
1324                                          (pcfg->pcfg_flags & 0x04) != 0,
1325                                          (pcfg->pcfg_flags & 0x08) != 0);
1326                 break;
1327         }
1328         case NAL_CMD_DEL_AUTOCONN: {
1329                 rc = ksocknal_del_route (pcfg->pcfg_nid, pcfg->pcfg_id,
1330                                          (pcfg->pcfg_flags & 1) != 0,
1331                                          (pcfg->pcfg_flags & 2) != 0);
1332                 break;
1333         }
1334         case NAL_CMD_GET_CONN: {
1335                 ksock_conn_t *conn = ksocknal_get_conn_by_idx (pcfg->pcfg_count);
1336
1337                 if (conn == NULL)
1338                         rc = -ENOENT;
1339                 else {
1340                         rc = 0;
1341                         pcfg->pcfg_nid   = conn->ksnc_peer->ksnp_nid;
1342                         pcfg->pcfg_id    = conn->ksnc_ipaddr;
1343                         pcfg->pcfg_misc  = conn->ksnc_port;
1344                         pcfg->pcfg_flags = conn->ksnc_type;
1345                         ksocknal_put_conn (conn);
1346                 }
1347                 break;
1348         }
1349         case NAL_CMD_REGISTER_PEER_FD: {
1350                 struct socket *sock = sockfd_lookup (pcfg->pcfg_fd, &rc);
1351                 int            type = pcfg->pcfg_misc;
1352
1353                 if (sock == NULL)
1354                         break;
1355
1356                 switch (type) {
1357                 case SOCKNAL_CONN_NONE:
1358                 case SOCKNAL_CONN_ANY:
1359                 case SOCKNAL_CONN_CONTROL:
1360                 case SOCKNAL_CONN_BULK_IN:
1361                 case SOCKNAL_CONN_BULK_OUT:
1362                         rc = ksocknal_create_conn(NULL, sock, pcfg->pcfg_flags, type);
1363                 default:
1364                         break;
1365                 }
1366                 if (rc != 0)
1367                         fput (sock->file);
1368                 break;
1369         }
1370         case NAL_CMD_CLOSE_CONNECTION: {
1371                 rc = ksocknal_close_matching_conns (pcfg->pcfg_nid,
1372                                                     pcfg->pcfg_id);
1373                 break;
1374         }
1375         case NAL_CMD_REGISTER_MYNID: {
1376                 rc = ksocknal_set_mynid (pcfg->pcfg_nid);
1377                 break;
1378         }
1379         case NAL_CMD_PUSH_CONNECTION: {
1380                 rc = ksocknal_push (pcfg->pcfg_nid);
1381                 break;
1382         }
1383         }
1384
1385         return rc;
1386 }
1387
1388 void
1389 ksocknal_free_fmbs (ksock_fmb_pool_t *p)
1390 {
1391         ksock_fmb_t *fmb;
1392         int          i;
1393
1394         LASSERT (list_empty(&p->fmp_blocked_conns));
1395         LASSERT (p->fmp_nactive_fmbs == 0);
1396
1397         while (!list_empty(&p->fmp_idle_fmbs)) {
1398
1399                 fmb = list_entry(p->fmp_idle_fmbs.next,
1400                                  ksock_fmb_t, fmb_list);
1401
1402                 for (i = 0; i < fmb->fmb_npages; i++)
1403                         if (fmb->fmb_pages[i] != NULL)
1404                                 __free_page(fmb->fmb_pages[i]);
1405
1406                 list_del(&fmb->fmb_list);
1407                 PORTAL_FREE(fmb, sizeof(*fmb));
1408         }
1409 }
1410
1411 void
1412 ksocknal_free_buffers (void)
1413 {
1414         ksocknal_free_fmbs(&ksocknal_data.ksnd_small_fmp);
1415         ksocknal_free_fmbs(&ksocknal_data.ksnd_large_fmp);
1416
1417         LASSERT (atomic_read(&ksocknal_data.ksnd_nactive_ltxs) == 0);
1418
1419         if (ksocknal_data.ksnd_schedulers != NULL)
1420                 PORTAL_FREE (ksocknal_data.ksnd_schedulers,
1421                              sizeof (ksock_sched_t) * SOCKNAL_N_SCHED);
1422
1423         PORTAL_FREE (ksocknal_data.ksnd_peers,
1424                      sizeof (struct list_head) *
1425                      ksocknal_data.ksnd_peer_hash_size);
1426 }
1427
1428 void
1429 ksocknal_module_fini (void)
1430 {
1431         int   i;
1432
1433         CDEBUG(D_MALLOC, "before NAL cleanup: kmem %d\n",
1434                atomic_read (&portal_kmemory));
1435
1436         switch (ksocknal_data.ksnd_init) {
1437         default:
1438                 LASSERT (0);
1439
1440         case SOCKNAL_INIT_ALL:
1441 #if CONFIG_SYSCTL
1442                 if (ksocknal_data.ksnd_sysctl != NULL)
1443                         unregister_sysctl_table (ksocknal_data.ksnd_sysctl);
1444 #endif
1445                 kportal_nal_unregister(SOCKNAL);
1446                 PORTAL_SYMBOL_UNREGISTER (ksocknal_ni);
1447                 /* fall through */
1448
1449         case SOCKNAL_INIT_PTL:
1450                 /* No more calls to ksocknal_cmd() to create new
1451                  * autoroutes/connections since we're being unloaded. */
1452                 PtlNIFini(ksocknal_ni);
1453
1454                 /* Delete all autoroute entries */
1455                 ksocknal_del_route(PTL_NID_ANY, 0, 0, 0);
1456
1457                 /* Delete all connections */
1458                 ksocknal_close_matching_conns (PTL_NID_ANY, 0);
1459
1460                 /* Wait for all peer state to clean up */
1461                 i = 2;
1462                 while (atomic_read (&ksocknal_data.ksnd_npeers) != 0) {
1463                         i++;
1464                         CDEBUG(((i & (-i)) == i) ? D_WARNING : D_NET, /* power of 2? */
1465                                "waiting for %d peers to disconnect\n",
1466                                atomic_read (&ksocknal_data.ksnd_npeers));
1467                         set_current_state (TASK_UNINTERRUPTIBLE);
1468                         schedule_timeout (HZ);
1469                 }
1470
1471                 /* Tell lib we've stopped calling into her. */
1472                 lib_fini(&ksocknal_lib);
1473                 /* fall through */
1474
1475         case SOCKNAL_INIT_DATA:
1476                 /* Module refcount only gets to zero when all peers
1477                  * have been closed so all lists must be empty */
1478                 LASSERT (atomic_read (&ksocknal_data.ksnd_npeers) == 0);
1479                 LASSERT (ksocknal_data.ksnd_peers != NULL);
1480                 for (i = 0; i < ksocknal_data.ksnd_peer_hash_size; i++) {
1481                         LASSERT (list_empty (&ksocknal_data.ksnd_peers[i]));
1482                 }
1483                 LASSERT (list_empty (&ksocknal_data.ksnd_enomem_conns));
1484                 LASSERT (list_empty (&ksocknal_data.ksnd_zombie_conns));
1485                 LASSERT (list_empty (&ksocknal_data.ksnd_autoconnectd_routes));
1486                 LASSERT (list_empty (&ksocknal_data.ksnd_small_fmp.fmp_blocked_conns));
1487                 LASSERT (list_empty (&ksocknal_data.ksnd_large_fmp.fmp_blocked_conns));
1488
1489                 if (ksocknal_data.ksnd_schedulers != NULL)
1490                         for (i = 0; i < SOCKNAL_N_SCHED; i++) {
1491                                 ksock_sched_t *kss =
1492                                         &ksocknal_data.ksnd_schedulers[i];
1493
1494                                 LASSERT (list_empty (&kss->kss_tx_conns));
1495                                 LASSERT (list_empty (&kss->kss_rx_conns));
1496                                 LASSERT (kss->kss_nconns == 0);
1497                         }
1498
1499                 /* stop router calling me */
1500                 kpr_shutdown (&ksocknal_data.ksnd_router);
1501
1502                 /* flag threads to terminate; wake and wait for them to die */
1503                 ksocknal_data.ksnd_shuttingdown = 1;
1504                 wake_up_all (&ksocknal_data.ksnd_autoconnectd_waitq);
1505                 wake_up_all (&ksocknal_data.ksnd_reaper_waitq);
1506
1507                 for (i = 0; i < SOCKNAL_N_SCHED; i++)
1508                        wake_up_all(&ksocknal_data.ksnd_schedulers[i].kss_waitq);
1509
1510                 while (atomic_read (&ksocknal_data.ksnd_nthreads) != 0) {
1511                         CDEBUG (D_NET, "waitinf for %d threads to terminate\n",
1512                                 atomic_read (&ksocknal_data.ksnd_nthreads));
1513                         set_current_state (TASK_UNINTERRUPTIBLE);
1514                         schedule_timeout (HZ);
1515                 }
1516
1517                 kpr_deregister (&ksocknal_data.ksnd_router);
1518
1519                 ksocknal_free_buffers();
1520                 /* fall through */
1521
1522         case SOCKNAL_INIT_NOTHING:
1523                 break;
1524         }
1525
1526         CDEBUG(D_MALLOC, "after NAL cleanup: kmem %d\n",
1527                atomic_read (&portal_kmemory));
1528
1529         printk(KERN_INFO "Lustre: Routing socket NAL unloaded (final mem %d)\n",
1530                atomic_read(&portal_kmemory));
1531 }
1532
1533
1534 void __init
1535 ksocknal_init_incarnation (void)
1536 {
1537         struct timeval tv;
1538
1539         /* The incarnation number is the time this module loaded and it
1540          * identifies this particular instance of the socknal.  Hopefully
1541          * we won't be able to reboot more frequently than 1MHz for the
1542          * forseeable future :) */
1543
1544         do_gettimeofday(&tv);
1545
1546         ksocknal_data.ksnd_incarnation =
1547                 (((__u64)tv.tv_sec) * 1000000) + tv.tv_usec;
1548 }
1549
1550 int __init
1551 ksocknal_module_init (void)
1552 {
1553         int   pkmem = atomic_read(&portal_kmemory);
1554         int   rc;
1555         int   i;
1556         int   j;
1557
1558         /* packet descriptor must fit in a router descriptor's scratchpad */
1559         LASSERT(sizeof (ksock_tx_t) <= sizeof (kprfd_scratch_t));
1560         /* the following must be sizeof(int) for proc_dointvec() */
1561         LASSERT(sizeof (ksocknal_data.ksnd_io_timeout) == sizeof (int));
1562         LASSERT(sizeof (ksocknal_data.ksnd_eager_ack) == sizeof (int));
1563         /* check ksnr_connected/connecting field large enough */
1564         LASSERT(SOCKNAL_CONN_NTYPES <= 4);
1565
1566         LASSERT (ksocknal_data.ksnd_init == SOCKNAL_INIT_NOTHING);
1567
1568         ksocknal_api.forward  = ksocknal_api_forward;
1569         ksocknal_api.shutdown = ksocknal_api_shutdown;
1570         ksocknal_api.yield    = ksocknal_api_yield;
1571         ksocknal_api.validate = NULL;           /* our api validate is a NOOP */
1572         ksocknal_api.lock     = ksocknal_api_lock;
1573         ksocknal_api.unlock   = ksocknal_api_unlock;
1574         ksocknal_api.nal_data = &ksocknal_data;
1575
1576         ksocknal_lib.nal_data = &ksocknal_data;
1577
1578         memset (&ksocknal_data, 0, sizeof (ksocknal_data)); /* zero pointers */
1579
1580         ksocknal_data.ksnd_io_timeout = SOCKNAL_IO_TIMEOUT;
1581         ksocknal_data.ksnd_eager_ack  = SOCKNAL_EAGER_ACK;
1582         ksocknal_data.ksnd_typed_conns = SOCKNAL_TYPED_CONNS;
1583         ksocknal_data.ksnd_min_bulk   = SOCKNAL_MIN_BULK;
1584 #if SOCKNAL_ZC
1585         ksocknal_data.ksnd_zc_min_frag = SOCKNAL_ZC_MIN_FRAG;
1586 #endif
1587         ksocknal_init_incarnation();
1588
1589         ksocknal_data.ksnd_peer_hash_size = SOCKNAL_PEER_HASH_SIZE;
1590         PORTAL_ALLOC (ksocknal_data.ksnd_peers,
1591                       sizeof (struct list_head) * ksocknal_data.ksnd_peer_hash_size);
1592         if (ksocknal_data.ksnd_peers == NULL)
1593                 return (-ENOMEM);
1594
1595         for (i = 0; i < ksocknal_data.ksnd_peer_hash_size; i++)
1596                 INIT_LIST_HEAD(&ksocknal_data.ksnd_peers[i]);
1597
1598         rwlock_init(&ksocknal_data.ksnd_global_lock);
1599
1600         ksocknal_data.ksnd_nal_cb = &ksocknal_lib;
1601         spin_lock_init (&ksocknal_data.ksnd_nal_cb_lock);
1602
1603         spin_lock_init(&ksocknal_data.ksnd_small_fmp.fmp_lock);
1604         INIT_LIST_HEAD(&ksocknal_data.ksnd_small_fmp.fmp_idle_fmbs);
1605         INIT_LIST_HEAD(&ksocknal_data.ksnd_small_fmp.fmp_blocked_conns);
1606
1607         spin_lock_init(&ksocknal_data.ksnd_large_fmp.fmp_lock);
1608         INIT_LIST_HEAD(&ksocknal_data.ksnd_large_fmp.fmp_idle_fmbs);
1609         INIT_LIST_HEAD(&ksocknal_data.ksnd_large_fmp.fmp_blocked_conns);
1610
1611         spin_lock_init (&ksocknal_data.ksnd_reaper_lock);
1612         INIT_LIST_HEAD (&ksocknal_data.ksnd_enomem_conns);
1613         INIT_LIST_HEAD (&ksocknal_data.ksnd_zombie_conns);
1614         INIT_LIST_HEAD (&ksocknal_data.ksnd_deathrow_conns);
1615         init_waitqueue_head(&ksocknal_data.ksnd_reaper_waitq);
1616
1617         spin_lock_init (&ksocknal_data.ksnd_autoconnectd_lock);
1618         INIT_LIST_HEAD (&ksocknal_data.ksnd_autoconnectd_routes);
1619         init_waitqueue_head(&ksocknal_data.ksnd_autoconnectd_waitq);
1620
1621         /* NB memset above zeros whole of ksocknal_data, including
1622          * ksocknal_data.ksnd_irqinfo[all].ksni_valid */
1623
1624         /* flag lists/ptrs/locks initialised */
1625         ksocknal_data.ksnd_init = SOCKNAL_INIT_DATA;
1626
1627         PORTAL_ALLOC(ksocknal_data.ksnd_schedulers,
1628                      sizeof(ksock_sched_t) * SOCKNAL_N_SCHED);
1629         if (ksocknal_data.ksnd_schedulers == NULL) {
1630                 ksocknal_module_fini ();
1631                 return (-ENOMEM);
1632         }
1633
1634         for (i = 0; i < SOCKNAL_N_SCHED; i++) {
1635                 ksock_sched_t *kss = &ksocknal_data.ksnd_schedulers[i];
1636
1637                 spin_lock_init (&kss->kss_lock);
1638                 INIT_LIST_HEAD (&kss->kss_rx_conns);
1639                 INIT_LIST_HEAD (&kss->kss_tx_conns);
1640 #if SOCKNAL_ZC
1641                 INIT_LIST_HEAD (&kss->kss_zctxdone_list);
1642 #endif
1643                 init_waitqueue_head (&kss->kss_waitq);
1644         }
1645
1646         rc = PtlNIInit(ksocknal_init, 32, 4, 0, &ksocknal_ni);
1647         if (rc != 0) {
1648                 CERROR("ksocknal: PtlNIInit failed: error %d\n", rc);
1649                 ksocknal_module_fini ();
1650                 return (rc);
1651         }
1652         PtlNIDebug(ksocknal_ni, ~0);
1653
1654         ksocknal_data.ksnd_init = SOCKNAL_INIT_PTL; // flag PtlNIInit() called
1655
1656         for (i = 0; i < SOCKNAL_N_SCHED; i++) {
1657                 rc = ksocknal_thread_start (ksocknal_scheduler,
1658                                             &ksocknal_data.ksnd_schedulers[i]);
1659                 if (rc != 0) {
1660                         CERROR("Can't spawn socknal scheduler[%d]: %d\n",
1661                                i, rc);
1662                         ksocknal_module_fini ();
1663                         return (rc);
1664                 }
1665         }
1666
1667         for (i = 0; i < SOCKNAL_N_AUTOCONNECTD; i++) {
1668                 rc = ksocknal_thread_start (ksocknal_autoconnectd, (void *)((long)i));
1669                 if (rc != 0) {
1670                         CERROR("Can't spawn socknal autoconnectd: %d\n", rc);
1671                         ksocknal_module_fini ();
1672                         return (rc);
1673                 }
1674         }
1675
1676         rc = ksocknal_thread_start (ksocknal_reaper, NULL);
1677         if (rc != 0) {
1678                 CERROR ("Can't spawn socknal reaper: %d\n", rc);
1679                 ksocknal_module_fini ();
1680                 return (rc);
1681         }
1682
1683         rc = kpr_register(&ksocknal_data.ksnd_router,
1684                           &ksocknal_router_interface);
1685         if (rc != 0) {
1686                 CDEBUG(D_NET, "Can't initialise routing interface "
1687                        "(rc = %d): not routing\n", rc);
1688         } else {
1689                 /* Only allocate forwarding buffers if I'm on a gateway */
1690
1691                 for (i = 0; i < (SOCKNAL_SMALL_FWD_NMSGS +
1692                                  SOCKNAL_LARGE_FWD_NMSGS); i++) {
1693                         ksock_fmb_t *fmb;
1694
1695                         PORTAL_ALLOC(fmb, sizeof(*fmb));
1696                         if (fmb == NULL) {
1697                                 ksocknal_module_fini();
1698                                 return (-ENOMEM);
1699                         }
1700
1701                         if (i < SOCKNAL_SMALL_FWD_NMSGS) {
1702                                 fmb->fmb_npages = SOCKNAL_SMALL_FWD_PAGES;
1703                                 fmb->fmb_pool = &ksocknal_data.ksnd_small_fmp;
1704                         } else {
1705                                 fmb->fmb_npages = SOCKNAL_LARGE_FWD_PAGES;
1706                                 fmb->fmb_pool = &ksocknal_data.ksnd_large_fmp;
1707                         }
1708
1709                         for (j = 0; j < fmb->fmb_npages; j++) {
1710                                 fmb->fmb_pages[j] = alloc_page(GFP_KERNEL);
1711
1712                                 if (fmb->fmb_pages[j] == NULL) {
1713                                         ksocknal_module_fini ();
1714                                         return (-ENOMEM);
1715                                 }
1716
1717                                 LASSERT(page_address(fmb->fmb_pages[j]) != NULL);
1718                         }
1719
1720                         list_add(&fmb->fmb_list, &fmb->fmb_pool->fmp_idle_fmbs);
1721                 }
1722         }
1723
1724         rc = kportal_nal_register(SOCKNAL, &ksocknal_cmd, NULL);
1725         if (rc != 0) {
1726                 CERROR ("Can't initialise command interface (rc = %d)\n", rc);
1727                 ksocknal_module_fini ();
1728                 return (rc);
1729         }
1730
1731         PORTAL_SYMBOL_REGISTER(ksocknal_ni);
1732
1733 #ifdef CONFIG_SYSCTL
1734         /* Press on regardless even if registering sysctl doesn't work */
1735         ksocknal_data.ksnd_sysctl = register_sysctl_table (ksocknal_top_ctl_table, 0);
1736 #endif
1737         /* flag everything initialised */
1738         ksocknal_data.ksnd_init = SOCKNAL_INIT_ALL;
1739
1740         printk(KERN_INFO "Lustre: Routing socket NAL loaded "
1741                "(Routing %s, initial mem %d)\n",
1742                kpr_routing (&ksocknal_data.ksnd_router) ?
1743                "enabled" : "disabled", pkmem);
1744
1745         return (0);
1746 }
1747
1748 MODULE_AUTHOR("Cluster File Systems, Inc. <info@clusterfs.com>");
1749 MODULE_DESCRIPTION("Kernel TCP Socket NAL v0.01");
1750 MODULE_LICENSE("GPL");
1751
1752 module_init(ksocknal_module_init);
1753 module_exit(ksocknal_module_fini);
1754
1755 EXPORT_SYMBOL (ksocknal_ni);