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