lnet/klnds/socklnd/socklnd_cb.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 #if (LINUX_VERSION_CODE > KERNEL_VERSION(2,5,0))
  28 # include <linux/syscalls.h>
  29 #endif
  30
  31 /*
  32  *  LIB functions follow
  33  *
  34  */
  35 int
  36 ksocknal_dist(lib_nal_t *nal, ptl_nid_t nid, unsigned long *dist)
  37 {
  38         /* I would guess that if ksocknal_get_peer (nid) == NULL,
  39            and we're not routing, then 'nid' is very distant :) */
  40         if (nal->libnal_ni.ni_pid.nid == nid) {
  41                 *dist = 0;
  42         } else {
  43                 *dist = 1;
  44         }
  45
  46         return 0;
  47 }
  48
  49 void
  50 ksocknal_free_ltx (ksock_ltx_t *ltx)
  51 {
  52         atomic_dec(&ksocknal_data.ksnd_nactive_ltxs);
  53         PORTAL_FREE(ltx, ltx->ltx_desc_size);
  54 }
  55
  56 #if (SOCKNAL_ZC && SOCKNAL_VADDR_ZC)
  57 struct page *
  58 ksocknal_kvaddr_to_page (unsigned long vaddr)
  59 {
  60         struct page *page;
  61
  62         if (vaddr >= VMALLOC_START &&
  63             vaddr < VMALLOC_END)
  64                 page = vmalloc_to_page ((void *)vaddr);
  65 #if CONFIG_HIGHMEM
  66         else if (vaddr >= PKMAP_BASE &&
  67                  vaddr < (PKMAP_BASE + LAST_PKMAP * PAGE_SIZE))
  68                 page = vmalloc_to_page ((void *)vaddr);
  69                 /* in 2.4 ^ just walks the page tables */
  70 #endif
  71         else
  72                 page = virt_to_page (vaddr);
  73
  74         if (page == NULL ||
  75             !VALID_PAGE (page))
  76                 return (NULL);
  77
  78         return (page);
  79 }
  80 #endif
  81
  82 int
  83 ksocknal_send_iov (ksock_conn_t *conn, ksock_tx_t *tx)
  84 {
  85         struct socket *sock = conn->ksnc_sock;
  86         struct iovec  *iov = tx->tx_iov;
  87         int            fragsize = iov->iov_len;
  88         unsigned long  vaddr = (unsigned long)iov->iov_base;
  89         int            more = (tx->tx_niov > 1) ||
  90                               (tx->tx_nkiov > 0) ||
  91                               (!list_empty (&conn->ksnc_tx_queue));
  92 #if (SOCKNAL_ZC && SOCKNAL_VADDR_ZC)
  93         int            offset = vaddr & (PAGE_SIZE - 1);
  94         int            zcsize = MIN (fragsize, PAGE_SIZE - offset);
  95         struct page   *page;
  96 #endif
  97         int            rc;
  98
  99         /* NB we can't trust socket ops to either consume our iovs
 100          * or leave them alone, so we only send 1 frag at a time. */
 101         LASSERT (fragsize <= tx->tx_resid);
 102         LASSERT (tx->tx_niov > 0);
 103
 104 #if (SOCKNAL_ZC && SOCKNAL_VADDR_ZC)
 105         if (zcsize >= ksocknal_data.ksnd_zc_min_frag &&
 106             (sock->sk->route_caps & NETIF_F_SG) &&
 107             (sock->sk->route_caps & (NETIF_F_IP_CSUM | NETIF_F_NO_CSUM | NETIF_F_HW_CSUM)) &&
 108             (page = ksocknal_kvaddr_to_page (vaddr)) != NULL) {
 109
 110                 CDEBUG(D_NET, "vaddr %p, page %p->%p + offset %x for %d\n",
 111                        (void *)vaddr, page, page_address(page), offset, zcsize);
 112
 113                 if (fragsize > zcsize) {
 114                         more = 1;
 115                         fragsize = zcsize;
 116                 }
 117
 118                 rc = tcp_sendpage_zccd(sock, page, offset, zcsize,
 119                                        more ? (MSG_DONTWAIT | MSG_MORE) : MSG_DONTWAIT,
 120                                        &tx->tx_zccd);
 121         } else
 122 #endif
 123         {
 124                 /* NB don't pass tx's iov; sendmsg may or may not update it */
 125                 struct iovec fragiov = { .iov_base = (void *)vaddr,
 126                                          .iov_len  = fragsize};
 127                 struct msghdr msg = {
 128                         .msg_name       = NULL,
 129                         .msg_namelen    = 0,
 130                         .msg_iov        = &fragiov,
 131                         .msg_iovlen     = 1,
 132                         .msg_control    = NULL,
 133                         .msg_controllen = 0,
 134                         .msg_flags      = more ? (MSG_DONTWAIT | MSG_MORE) : MSG_DONTWAIT
 135                 };
 136                 mm_segment_t oldmm = get_fs();
 137
 138                 set_fs (KERNEL_DS);
 139                 rc = sock_sendmsg(sock, &msg, fragsize);
 140                 set_fs (oldmm);
 141         }
 142
 143         if (rc > 0) {
 144                 tx->tx_resid -= rc;
 145
 146                 if (rc < iov->iov_len) {
 147                         /* didn't send whole iov entry... */
 148                         iov->iov_base = (void *)(vaddr + rc);
 149                         iov->iov_len -= rc;
 150                 } else {
 151                         tx->tx_iov++;
 152                         tx->tx_niov--;
 153                 }
 154         }
 155
 156         return (rc);
 157 }
 158
 159 int
 160 ksocknal_send_kiov (ksock_conn_t *conn, ksock_tx_t *tx)
 161 {
 162         struct socket *sock = conn->ksnc_sock;
 163         ptl_kiov_t    *kiov = tx->tx_kiov;
 164         int            fragsize = kiov->kiov_len;
 165         struct page   *page = kiov->kiov_page;
 166         int            offset = kiov->kiov_offset;
 167         int            more = (tx->tx_nkiov > 1) ||
 168                               (!list_empty (&conn->ksnc_tx_queue));
 169         int            rc;
 170
 171         /* NB we can't trust socket ops to either consume our iovs
 172          * or leave them alone, so we only send 1 frag at a time. */
 173         LASSERT (fragsize <= tx->tx_resid);
 174         LASSERT (offset + fragsize <= PAGE_SIZE);
 175         LASSERT (tx->tx_niov == 0);
 176         LASSERT (tx->tx_nkiov > 0);
 177
 178 #if SOCKNAL_ZC
 179         if (fragsize >= ksocknal_tunables.ksnd_zc_min_frag &&
 180             (sock->sk->route_caps & NETIF_F_SG) &&
 181             (sock->sk->route_caps & (NETIF_F_IP_CSUM | NETIF_F_NO_CSUM | NETIF_F_HW_CSUM))) {
 182
 183                 CDEBUG(D_NET, "page %p + offset %x for %d\n",
 184                                page, offset, fragsize);
 185
 186                 rc = tcp_sendpage_zccd(sock, page, offset, fragsize,
 187                                        more ? (MSG_DONTWAIT | MSG_MORE) : MSG_DONTWAIT,
 188                                        &tx->tx_zccd);
 189         } else
 190 #endif
 191         {
 192                 char *addr = ((char *)kmap (page)) + offset;
 193                 struct iovec fragiov = {.iov_base = addr,
 194                                         .iov_len  = fragsize};
 195                 struct msghdr msg = {
 196                         .msg_name       = NULL,
 197                         .msg_namelen    = 0,
 198                         .msg_iov        = &fragiov,
 199                         .msg_iovlen     = 1,
 200                         .msg_control    = NULL,
 201                         .msg_controllen = 0,
 202                         .msg_flags      = more ? (MSG_DONTWAIT | MSG_MORE) : MSG_DONTWAIT
 203                 };
 204                 mm_segment_t  oldmm = get_fs();
 205
 206                 set_fs (KERNEL_DS);
 207                 rc = sock_sendmsg(sock, &msg, fragsize);
 208                 set_fs (oldmm);
 209
 210                 kunmap (page);
 211         }
 212
 213         if (rc > 0) {
 214                 tx->tx_resid -= rc;
 215
 216                 if (rc < fragsize) {
 217                         kiov->kiov_offset = offset + rc;
 218                         kiov->kiov_len    = fragsize - rc;
 219                 } else {
 220                         tx->tx_kiov++;
 221                         tx->tx_nkiov--;
 222                 }
 223         }
 224
 225         return (rc);
 226 }
 227
 228 int
 229 ksocknal_transmit (ksock_conn_t *conn, ksock_tx_t *tx)
 230 {
 231         int      rc;
 232         int      bufnob;
 233
 234         if (ksocknal_data.ksnd_stall_tx != 0) {
 235                 set_current_state (TASK_UNINTERRUPTIBLE);
 236                 schedule_timeout (ksocknal_data.ksnd_stall_tx * HZ);
 237         }
 238
 239         LASSERT (tx->tx_resid != 0);
 240
 241         rc = ksocknal_getconnsock (conn);
 242         if (rc != 0) {
 243                 LASSERT (conn->ksnc_closing);
 244                 return (-ESHUTDOWN);
 245         }
 246
 247         do {
 248                 if (ksocknal_data.ksnd_enomem_tx > 0) {
 249                         /* testing... */
 250                         ksocknal_data.ksnd_enomem_tx--;
 251                         rc = -EAGAIN;
 252                 } else if (tx->tx_niov != 0) {
 253                         rc = ksocknal_send_iov (conn, tx);
 254                 } else {
 255                         rc = ksocknal_send_kiov (conn, tx);
 256                 }
 257
 258                 bufnob = conn->ksnc_sock->sk->sk_wmem_queued;
 259                 if (rc > 0)                     /* sent something? */
 260                         conn->ksnc_tx_bufnob += rc; /* account it */
 261
 262                 if (bufnob < conn->ksnc_tx_bufnob) {
 263                         /* allocated send buffer bytes < computed; infer
 264                          * something got ACKed */
 265                         conn->ksnc_tx_deadline = jiffies +
 266                                                  ksocknal_tunables.ksnd_io_timeout * HZ;
 267                         conn->ksnc_peer->ksnp_last_alive = jiffies;
 268                         conn->ksnc_tx_bufnob = bufnob;
 269                         mb();
 270                 }
 271
 272                 if (rc <= 0) {
 273                         /* Didn't write anything.
 274                          *
 275                          * NB: rc == 0 and rc == -EAGAIN both mean try
 276                          * again later (linux stack returns -EAGAIN for
 277                          * this, but Adaptech TOE returns 0).
 278                          *
 279                          * Also, sends never fail with -ENOMEM, just
 280                          * -EAGAIN, but with the added bonus that we can't
 281                          * expect write_space() to call us back to tell us
 282                          * when to try sending again.  We use the
 283                          * SOCK_NOSPACE flag to diagnose...  */
 284
 285                         LASSERT(rc != -ENOMEM);
 286
 287                         if (rc == 0 || rc == -EAGAIN) {
 288                                 if (test_bit(SOCK_NOSPACE,
 289                                              &conn->ksnc_sock->flags)) {
 290                                         rc = -EAGAIN;
 291                                 } else {
 292                                         static int counter;
 293
 294                                         counter++;
 295                                         if ((counter & (-counter)) == counter)
 296                                                 CWARN("%d ENOMEM tx %p\n",
 297                                                       counter, conn);
 298                                         rc = -ENOMEM;
 299                                 }
 300                         }
 301                         break;
 302                 }
 303
 304                 /* socket's wmem_queued now includes 'rc' bytes */
 305                 atomic_sub (rc, &conn->ksnc_tx_nob);
 306                 rc = 0;
 307
 308         } while (tx->tx_resid != 0);
 309
 310         ksocknal_putconnsock (conn);
 311         return (rc);
 312 }
 313
 314 void
 315 ksocknal_eager_ack (ksock_conn_t *conn)
 316 {
 317         int            opt = 1;
 318         mm_segment_t   oldmm = get_fs();
 319         struct socket *sock = conn->ksnc_sock;
 320
 321         /* Remind the socket to ACK eagerly.  If I don't, the socket might
 322          * think I'm about to send something it could piggy-back the ACK
 323          * on, introducing delay in completing zero-copy sends in my
 324          * peer. */
 325
 326         set_fs(KERNEL_DS);
 327         sock->ops->setsockopt (sock, SOL_TCP, TCP_QUICKACK,
 328                                (char *)&opt, sizeof (opt));
 329         set_fs(oldmm);
 330 }
 331
 332 int
 333 ksocknal_recv_iov (ksock_conn_t *conn)
 334 {
 335         struct iovec *iov = conn->ksnc_rx_iov;
 336         int           fragsize  = iov->iov_len;
 337         unsigned long vaddr = (unsigned long)iov->iov_base;
 338         struct iovec  fragiov = { .iov_base = (void *)vaddr,
 339                                   .iov_len  = fragsize};
 340         struct msghdr msg = {
 341                 .msg_name       = NULL,
 342                 .msg_namelen    = 0,
 343                 .msg_iov        = &fragiov,
 344                 .msg_iovlen     = 1,
 345                 .msg_control    = NULL,
 346                 .msg_controllen = 0,
 347                 .msg_flags      = 0
 348         };
 349         mm_segment_t oldmm = get_fs();
 350         int          rc;
 351
 352         /* NB we can't trust socket ops to either consume our iovs
 353          * or leave them alone, so we only receive 1 frag at a time. */
 354         LASSERT (conn->ksnc_rx_niov > 0);
 355         LASSERT (fragsize <= conn->ksnc_rx_nob_wanted);
 356
 357         set_fs (KERNEL_DS);
 358         rc = sock_recvmsg (conn->ksnc_sock, &msg, fragsize, MSG_DONTWAIT);
 359         /* NB this is just a boolean............................^ */
 360         set_fs (oldmm);
 361
 362         if (rc <= 0)
 363                 return (rc);
 364
 365         /* received something... */
 366         conn->ksnc_peer->ksnp_last_alive = jiffies;
 367         conn->ksnc_rx_deadline = jiffies +
 368                                  ksocknal_tunables.ksnd_io_timeout * HZ;
 369         mb();                           /* order with setting rx_started */
 370         conn->ksnc_rx_started = 1;
 371
 372         conn->ksnc_rx_nob_wanted -= rc;
 373         conn->ksnc_rx_nob_left -= rc;
 374
 375         if (rc < fragsize) {
 376                 iov->iov_base = (void *)(vaddr + rc);
 377                 iov->iov_len = fragsize - rc;
 378                 return (-EAGAIN);
 379         }
 380
 381         conn->ksnc_rx_iov++;
 382         conn->ksnc_rx_niov--;
 383         return (1);
 384 }
 385
 386 int
 387 ksocknal_recv_kiov (ksock_conn_t *conn)
 388 {
 389         ptl_kiov_t   *kiov = conn->ksnc_rx_kiov;
 390         struct page  *page = kiov->kiov_page;
 391         int           offset = kiov->kiov_offset;
 392         int           fragsize = kiov->kiov_len;
 393         unsigned long vaddr = ((unsigned long)kmap (page)) + offset;
 394         struct iovec  fragiov = { .iov_base = (void *)vaddr,
 395                                   .iov_len  = fragsize};
 396         struct msghdr msg = {
 397                 .msg_name       = NULL,
 398                 .msg_namelen    = 0,
 399                 .msg_iov        = &fragiov,
 400                 .msg_iovlen     = 1,
 401                 .msg_control    = NULL,
 402                 .msg_controllen = 0,
 403                 .msg_flags      = 0
 404         };
 405         mm_segment_t oldmm = get_fs();
 406         int          rc;
 407
 408         /* NB we can't trust socket ops to either consume our iovs
 409          * or leave them alone, so we only receive 1 frag at a time. */
 410         LASSERT (fragsize <= conn->ksnc_rx_nob_wanted);
 411         LASSERT (conn->ksnc_rx_nkiov > 0);
 412         LASSERT (offset + fragsize <= PAGE_SIZE);
 413
 414         set_fs (KERNEL_DS);
 415         rc = sock_recvmsg (conn->ksnc_sock, &msg, fragsize, MSG_DONTWAIT);
 416         /* NB this is just a boolean............................^ */
 417         set_fs (oldmm);
 418
 419         kunmap (page);
 420
 421         if (rc <= 0)
 422                 return (rc);
 423
 424         /* received something... */
 425         conn->ksnc_peer->ksnp_last_alive = jiffies;
 426         conn->ksnc_rx_deadline = jiffies +
 427                                  ksocknal_tunables.ksnd_io_timeout * HZ;
 428         mb();                           /* order with setting rx_started */
 429         conn->ksnc_rx_started = 1;
 430
 431         conn->ksnc_rx_nob_wanted -= rc;
 432         conn->ksnc_rx_nob_left -= rc;
 433
 434         if (rc < fragsize) {
 435                 kiov->kiov_offset = offset + rc;
 436                 kiov->kiov_len = fragsize - rc;
 437                 return (-EAGAIN);
 438         }
 439
 440         conn->ksnc_rx_kiov++;
 441         conn->ksnc_rx_nkiov--;
 442         return (1);
 443 }
 444
 445 int
 446 ksocknal_receive (ksock_conn_t *conn)
 447 {
 448         /* Return 1 on success, 0 on EOF, < 0 on error.
 449          * Caller checks ksnc_rx_nob_wanted to determine
 450          * progress/completion. */
 451         int     rc;
 452         ENTRY;
 453
 454         if (ksocknal_data.ksnd_stall_rx != 0) {
 455                 set_current_state (TASK_UNINTERRUPTIBLE);
 456                 schedule_timeout (ksocknal_data.ksnd_stall_rx * HZ);
 457         }
 458
 459         rc = ksocknal_getconnsock (conn);
 460         if (rc != 0) {
 461                 LASSERT (conn->ksnc_closing);
 462                 return (-ESHUTDOWN);
 463         }
 464
 465         for (;;) {
 466                 if (conn->ksnc_rx_niov != 0)
 467                         rc = ksocknal_recv_iov (conn);
 468                 else
 469                         rc = ksocknal_recv_kiov (conn);
 470
 471                 if (rc <= 0) {
 472                         /* error/EOF or partial receive */
 473                         if (rc == -EAGAIN) {
 474                                 rc = 1;
 475                         } else if (rc == 0 && conn->ksnc_rx_started) {
 476                                 /* EOF in the middle of a message */
 477                                 rc = -EPROTO;
 478                         }
 479                         break;
 480                 }
 481
 482                 /* Completed a fragment */
 483
 484                 if (conn->ksnc_rx_nob_wanted == 0) {
 485                         /* Completed a message segment (header or payload) */
 486                         if ((ksocknal_tunables.ksnd_eager_ack & conn->ksnc_type) != 0 &&
 487                             (conn->ksnc_rx_state ==  SOCKNAL_RX_BODY ||
 488                              conn->ksnc_rx_state == SOCKNAL_RX_BODY_FWD)) {
 489                                 /* Remind the socket to ack eagerly... */
 490                                 ksocknal_eager_ack(conn);
 491                         }
 492                         rc = 1;
 493                         break;
 494                 }
 495         }
 496
 497         ksocknal_putconnsock (conn);
 498         RETURN (rc);
 499 }
 500
 501 #if SOCKNAL_ZC
 502 void
 503 ksocknal_zc_callback (zccd_t *zcd)
 504 {
 505         ksock_tx_t    *tx = KSOCK_ZCCD_2_TX(zcd);
 506         ksock_sched_t *sched = tx->tx_conn->ksnc_scheduler;
 507         unsigned long  flags;
 508         ENTRY;
 509
 510         /* Schedule tx for cleanup (can't do it now due to lock conflicts) */
 511
 512         spin_lock_irqsave (&sched->kss_lock, flags);
 513
 514         list_add_tail (&tx->tx_list, &sched->kss_zctxdone_list);
 515         wake_up (&sched->kss_waitq);
 516
 517         spin_unlock_irqrestore (&sched->kss_lock, flags);
 518         EXIT;
 519 }
 520 #endif
 521
 522 void
 523 ksocknal_tx_done (ksock_tx_t *tx, int asynch)
 524 {
 525         ksock_ltx_t   *ltx;
 526         ENTRY;
 527
 528         if (tx->tx_conn != NULL) {
 529 #if SOCKNAL_ZC
 530                 /* zero copy completion isn't always from
 531                  * process_transmit() so it needs to keep a ref on
 532                  * tx_conn... */
 533                 if (asynch)
 534                         ksocknal_put_conn (tx->tx_conn);
 535 #else
 536                 LASSERT (!asynch);
 537 #endif
 538         }
 539
 540         if (tx->tx_isfwd) {             /* was a forwarded packet? */
 541                 kpr_fwd_done (&ksocknal_data.ksnd_router,
 542                               KSOCK_TX_2_KPR_FWD_DESC (tx),
 543                               (tx->tx_resid == 0) ? 0 : -ECONNABORTED);
 544                 EXIT;
 545                 return;
 546         }
 547
 548         /* local send */
 549         ltx = KSOCK_TX_2_KSOCK_LTX (tx);
 550
 551         lib_finalize (&ksocknal_lib, ltx->ltx_private, ltx->ltx_cookie,
 552                       (tx->tx_resid == 0) ? PTL_OK : PTL_FAIL);
 553
 554         ksocknal_free_ltx (ltx);
 555         EXIT;
 556 }
 557
 558 void
 559 ksocknal_tx_launched (ksock_tx_t *tx)
 560 {
 561 #if SOCKNAL_ZC
 562         if (atomic_read (&tx->tx_zccd.zccd_count) != 1) {
 563                 ksock_conn_t  *conn = tx->tx_conn;
 564
 565                 /* zccd skbufs are still in-flight.  First take a ref on
 566                  * conn, so it hangs about for ksocknal_tx_done... */
 567                 atomic_inc (&conn->ksnc_refcount);
 568
 569                 /* ...then drop the initial ref on zccd, so the zero copy
 570                  * callback can occur */
 571                 zccd_put (&tx->tx_zccd);
 572                 return;
 573         }
 574 #endif
 575         /* Any zero-copy-ness (if any) has completed; I can complete the
 576          * transmit now, avoiding an extra schedule */
 577         ksocknal_tx_done (tx, 0);
 578 }
 579
 580 int
 581 ksocknal_process_transmit (ksock_conn_t *conn, ksock_tx_t *tx)
 582 {
 583         unsigned long  flags;
 584         int            rc;
 585
 586         rc = ksocknal_transmit (conn, tx);
 587
 588         CDEBUG (D_NET, "send(%d) %d\n", tx->tx_resid, rc);
 589
 590         if (tx->tx_resid == 0) {
 591                 /* Sent everything OK */
 592                 LASSERT (rc == 0);
 593
 594                 ksocknal_tx_launched (tx);
 595                 return (0);
 596         }
 597
 598         if (rc == -EAGAIN)
 599                 return (rc);
 600
 601         if (rc == -ENOMEM) {
 602                 /* Queue on ksnd_enomem_conns for retry after a timeout */
 603                 spin_lock_irqsave(&ksocknal_data.ksnd_reaper_lock, flags);
 604
 605                 /* enomem list takes over scheduler's ref... */
 606                 LASSERT (conn->ksnc_tx_scheduled);
 607                 list_add_tail(&conn->ksnc_tx_list,
 608                               &ksocknal_data.ksnd_enomem_conns);
 609                 if (!time_after_eq(jiffies + SOCKNAL_ENOMEM_RETRY,
 610                                    ksocknal_data.ksnd_reaper_waketime))
 611                         wake_up (&ksocknal_data.ksnd_reaper_waitq);
 612
 613                 spin_unlock_irqrestore(&ksocknal_data.ksnd_reaper_lock, flags);
 614                 return (rc);
 615         }
 616
 617         /* Actual error */
 618         LASSERT (rc < 0);
 619
 620         if (!conn->ksnc_closing)
 621                 CERROR("[%p] Error %d on write to "LPX64
 622                        " ip %d.%d.%d.%d:%d\n", conn, rc,
 623                        conn->ksnc_peer->ksnp_nid,
 624                        HIPQUAD(conn->ksnc_ipaddr),
 625                        conn->ksnc_port);
 626
 627         ksocknal_close_conn_and_siblings (conn, rc);
 628         ksocknal_tx_launched (tx);
 629
 630         return (rc);
 631 }
 632
 633 void
 634 ksocknal_launch_autoconnect_locked (ksock_route_t *route)
 635 {
 636         unsigned long     flags;
 637
 638         /* called holding write lock on ksnd_global_lock */
 639
 640         LASSERT (!route->ksnr_deleted);
 641         LASSERT ((route->ksnr_connected & (1 << SOCKNAL_CONN_ANY)) == 0);
 642         LASSERT ((route->ksnr_connected & KSNR_TYPED_ROUTES) != KSNR_TYPED_ROUTES);
 643         LASSERT (route->ksnr_connecting == 0);
 644
 645         if (ksocknal_tunables.ksnd_typed_conns)
 646                 route->ksnr_connecting =
 647                         KSNR_TYPED_ROUTES & ~route->ksnr_connected;
 648         else
 649                 route->ksnr_connecting = (1 << SOCKNAL_CONN_ANY);
 650
 651         atomic_inc (&route->ksnr_refcount);     /* extra ref for asynchd */
 652
 653         spin_lock_irqsave (&ksocknal_data.ksnd_autoconnectd_lock, flags);
 654
 655         list_add_tail (&route->ksnr_connect_list,
 656                        &ksocknal_data.ksnd_autoconnectd_routes);
 657         wake_up (&ksocknal_data.ksnd_autoconnectd_waitq);
 658
 659         spin_unlock_irqrestore (&ksocknal_data.ksnd_autoconnectd_lock, flags);
 660 }
 661
 662 ksock_peer_t *
 663 ksocknal_find_target_peer_locked (ksock_tx_t *tx, ptl_nid_t nid)
 664 {
 665         char          ipbuf[PTL_NALFMT_SIZE];
 666         ptl_nid_t     target_nid;
 667         int           rc;
 668         ksock_peer_t *peer = ksocknal_find_peer_locked (nid);
 669
 670         if (peer != NULL)
 671                 return (peer);
 672
 673         if (tx->tx_isfwd) {
 674                 CERROR ("Can't send packet to "LPX64
 675                        " %s: routed target is not a peer\n",
 676                         nid, portals_nid2str(SOCKNAL, nid, ipbuf));
 677                 return (NULL);
 678         }
 679
 680         rc = kpr_lookup (&ksocknal_data.ksnd_router, nid, tx->tx_nob,
 681                          &target_nid);
 682         if (rc != 0) {
 683                 CERROR ("Can't route to "LPX64" %s: router error %d\n",
 684                         nid, portals_nid2str(SOCKNAL, nid, ipbuf), rc);
 685                 return (NULL);
 686         }
 687
 688         peer = ksocknal_find_peer_locked (target_nid);
 689         if (peer != NULL)
 690                 return (peer);
 691
 692         CERROR ("Can't send packet to "LPX64" %s: no peer entry\n",
 693                 target_nid, portals_nid2str(SOCKNAL, target_nid, ipbuf));
 694         return (NULL);
 695 }
 696
 697 ksock_conn_t *
 698 ksocknal_find_conn_locked (ksock_tx_t *tx, ksock_peer_t *peer)
 699 {
 700         struct list_head *tmp;
 701         ksock_conn_t     *typed = NULL;
 702         int               tnob  = 0;
 703         ksock_conn_t     *fallback = NULL;
 704         int               fnob     = 0;
 705         ksock_conn_t     *conn;
 706
 707         list_for_each (tmp, &peer->ksnp_conns) {
 708                 ksock_conn_t *c = list_entry(tmp, ksock_conn_t, ksnc_list);
 709 #if SOCKNAL_ROUND_ROBIN
 710                 const int     nob = 0;
 711 #else
 712                 int           nob = atomic_read(&c->ksnc_tx_nob) +
 713                                         c->ksnc_sock->sk->sk_wmem_queued;
 714 #endif
 715                 LASSERT (!c->ksnc_closing);
 716
 717                 if (fallback == NULL || nob < fnob) {
 718                         fallback = c;
 719                         fnob     = nob;
 720                 }
 721
 722                 if (!ksocknal_tunables.ksnd_typed_conns)
 723                         continue;
 724
 725                 switch (c->ksnc_type) {
 726                 default:
 727                         LBUG();
 728                 case SOCKNAL_CONN_ANY:
 729                         break;
 730                 case SOCKNAL_CONN_BULK_IN:
 731                         continue;
 732                 case SOCKNAL_CONN_BULK_OUT:
 733                         if (tx->tx_nob < ksocknal_tunables.ksnd_min_bulk)
 734                                 continue;
 735                         break;
 736                 case SOCKNAL_CONN_CONTROL:
 737                         if (tx->tx_nob >= ksocknal_tunables.ksnd_min_bulk)
 738                                 continue;
 739                         break;
 740                 }
 741
 742                 if (typed == NULL || nob < tnob) {
 743                         typed = c;
 744                         tnob  = nob;
 745                 }
 746         }
 747
 748         /* prefer the typed selection */
 749         conn = (typed != NULL) ? typed : fallback;
 750
 751 #if SOCKNAL_ROUND_ROBIN
 752         if (conn != NULL) {
 753                 /* round-robin all else being equal */
 754                 list_del (&conn->ksnc_list);
 755                 list_add_tail (&conn->ksnc_list, &peer->ksnp_conns);
 756         }
 757 #endif
 758         return conn;
 759 }
 760
 761 void
 762 ksocknal_queue_tx_locked (ksock_tx_t *tx, ksock_conn_t *conn)
 763 {
 764         unsigned long  flags;
 765         ksock_sched_t *sched = conn->ksnc_scheduler;
 766
 767         /* called holding global lock (read or irq-write) and caller may
 768          * not have dropped this lock between finding conn and calling me,
 769          * so we don't need the {get,put}connsock dance to deref
 770          * ksnc_sock... */
 771         LASSERT(!conn->ksnc_closing);
 772         LASSERT(tx->tx_resid == tx->tx_nob);
 773
 774         CDEBUG (D_NET, "Sending to "LPX64" ip %d.%d.%d.%d:%d\n",
 775                 conn->ksnc_peer->ksnp_nid,
 776                 HIPQUAD(conn->ksnc_ipaddr),
 777                 conn->ksnc_port);
 778
 779         atomic_add (tx->tx_nob, &conn->ksnc_tx_nob);
 780         tx->tx_conn = conn;
 781
 782 #if SOCKNAL_ZC
 783         zccd_init (&tx->tx_zccd, ksocknal_zc_callback);
 784         /* NB this sets 1 ref on zccd, so the callback can only occur after
 785          * I've released this ref. */
 786 #endif
 787         spin_lock_irqsave (&sched->kss_lock, flags);
 788
 789         if (list_empty(&conn->ksnc_tx_queue) &&
 790             conn->ksnc_sock->sk->sk_wmem_queued == 0) {
 791                 /* First packet starts the timeout */
 792                 conn->ksnc_tx_deadline = jiffies +
 793                                          ksocknal_tunables.ksnd_io_timeout * HZ;
 794                 conn->ksnc_tx_bufnob = 0;
 795                 mb();    /* order with adding to tx_queue */
 796         }
 797
 798         list_add_tail (&tx->tx_list, &conn->ksnc_tx_queue);
 799
 800         if (conn->ksnc_tx_ready &&      /* able to send */
 801             !conn->ksnc_tx_scheduled) { /* not scheduled to send */
 802                 /* +1 ref for scheduler */
 803                 atomic_inc (&conn->ksnc_refcount);
 804                 list_add_tail (&conn->ksnc_tx_list,
 805                                &sched->kss_tx_conns);
 806                 conn->ksnc_tx_scheduled = 1;
 807                 wake_up (&sched->kss_waitq);
 808         }
 809
 810         spin_unlock_irqrestore (&sched->kss_lock, flags);
 811 }
 812
 813 ksock_route_t *
 814 ksocknal_find_connectable_route_locked (ksock_peer_t *peer)
 815 {
 816         struct list_head  *tmp;
 817         ksock_route_t     *route;
 818         int                bits;
 819
 820         list_for_each (tmp, &peer->ksnp_routes) {
 821                 route = list_entry (tmp, ksock_route_t, ksnr_list);
 822                 bits  = route->ksnr_connected;
 823
 824                 /* All typed connections established? */
 825                 if ((bits & KSNR_TYPED_ROUTES) == KSNR_TYPED_ROUTES)
 826                         continue;
 827
 828                 /* Untyped connection established? */
 829                 if ((bits & (1 << SOCKNAL_CONN_ANY)) != 0)
 830                         continue;
 831
 832                 /* connection being established? */
 833                 if (route->ksnr_connecting != 0)
 834                         continue;
 835
 836                 /* too soon to retry this guy? */
 837                 if (!time_after_eq (jiffies, route->ksnr_timeout))
 838                         continue;
 839
 840                 return (route);
 841         }
 842
 843         return (NULL);
 844 }
 845
 846 ksock_route_t *
 847 ksocknal_find_connecting_route_locked (ksock_peer_t *peer)
 848 {
 849         struct list_head  *tmp;
 850         ksock_route_t     *route;
 851
 852         list_for_each (tmp, &peer->ksnp_routes) {
 853                 route = list_entry (tmp, ksock_route_t, ksnr_list);
 854
 855                 if (route->ksnr_connecting != 0)
 856                         return (route);
 857         }
 858
 859         return (NULL);
 860 }
 861
 862 int
 863 ksocknal_launch_packet (ksock_tx_t *tx, ptl_nid_t nid)
 864 {
 865         unsigned long     flags;
 866         ksock_peer_t     *peer;
 867         ksock_conn_t     *conn;
 868         ksock_route_t    *route;
 869         rwlock_t         *g_lock;
 870
 871         /* Ensure the frags we've been given EXACTLY match the number of
 872          * bytes we want to send.  Many TCP/IP stacks disregard any total
 873          * size parameters passed to them and just look at the frags.
 874          *
 875          * We always expect at least 1 mapped fragment containing the
 876          * complete portals header. */
 877         LASSERT (lib_iov_nob (tx->tx_niov, tx->tx_iov) +
 878                  lib_kiov_nob (tx->tx_nkiov, tx->tx_kiov) == tx->tx_nob);
 879         LASSERT (tx->tx_niov >= 1);
 880         LASSERT (tx->tx_iov[0].iov_len >= sizeof (ptl_hdr_t));
 881
 882         CDEBUG (D_NET, "packet %p type %d, nob %d niov %d nkiov %d\n",
 883                 tx, ((ptl_hdr_t *)tx->tx_iov[0].iov_base)->type,
 884                 tx->tx_nob, tx->tx_niov, tx->tx_nkiov);
 885
 886         tx->tx_conn = NULL;                     /* only set when assigned a conn */
 887         tx->tx_resid = tx->tx_nob;
 888         tx->tx_hdr = (ptl_hdr_t *)tx->tx_iov[0].iov_base;
 889
 890         g_lock = &ksocknal_data.ksnd_global_lock;
 891 #if !SOCKNAL_ROUND_ROBIN
 892         read_lock (g_lock);
 893
 894         peer = ksocknal_find_target_peer_locked (tx, nid);
 895         if (peer == NULL) {
 896                 read_unlock (g_lock);
 897                 return (-EHOSTUNREACH);
 898         }
 899
 900         if (ksocknal_find_connectable_route_locked(peer) == NULL) {
 901                 conn = ksocknal_find_conn_locked (tx, peer);
 902                 if (conn != NULL) {
 903                         /* I've got no autoconnect routes that need to be
 904                          * connecting and I do have an actual connection... */
 905                         ksocknal_queue_tx_locked (tx, conn);
 906                         read_unlock (g_lock);
 907                         return (0);
 908                 }
 909         }
 910
 911         /* I'll need a write lock... */
 912         read_unlock (g_lock);
 913 #endif
 914         write_lock_irqsave(g_lock, flags);
 915
 916         peer = ksocknal_find_target_peer_locked (tx, nid);
 917         if (peer == NULL) {
 918                 write_unlock_irqrestore(g_lock, flags);
 919                 return (-EHOSTUNREACH);
 920         }
 921
 922         for (;;) {
 923                 /* launch any/all autoconnections that need it */
 924                 route = ksocknal_find_connectable_route_locked (peer);
 925                 if (route == NULL)
 926                         break;
 927
 928                 ksocknal_launch_autoconnect_locked (route);
 929         }
 930
 931         conn = ksocknal_find_conn_locked (tx, peer);
 932         if (conn != NULL) {
 933                 /* Connection exists; queue message on it */
 934                 ksocknal_queue_tx_locked (tx, conn);
 935                 write_unlock_irqrestore (g_lock, flags);
 936                 return (0);
 937         }
 938
 939         route = ksocknal_find_connecting_route_locked (peer);
 940         if (route != NULL) {
 941                 /* At least 1 connection is being established; queue the
 942                  * message... */
 943                 list_add_tail (&tx->tx_list, &peer->ksnp_tx_queue);
 944                 write_unlock_irqrestore (g_lock, flags);
 945                 return (0);
 946         }
 947
 948         write_unlock_irqrestore (g_lock, flags);
 949         return (-EHOSTUNREACH);
 950 }
 951
 952 ptl_err_t
 953 ksocknal_sendmsg(lib_nal_t     *nal,
 954                  void         *private,
 955                  lib_msg_t    *cookie,
 956                  ptl_hdr_t    *hdr,
 957                  int           type,
 958                  ptl_nid_t     nid,
 959                  ptl_pid_t     pid,
 960                  unsigned int  payload_niov,
 961                  struct iovec *payload_iov,
 962                  ptl_kiov_t   *payload_kiov,
 963                  size_t        payload_offset,
 964                  size_t        payload_nob)
 965 {
 966         ksock_ltx_t  *ltx;
 967         int           desc_size;
 968         int           rc;
 969
 970         /* NB 'private' is different depending on what we're sending.
 971          * Just ignore it... */
 972
 973         CDEBUG(D_NET, "sending "LPSZ" bytes in %d frags to nid:"LPX64
 974                " pid %d\n", payload_nob, payload_niov, nid , pid);
 975
 976         LASSERT (payload_nob == 0 || payload_niov > 0);
 977         LASSERT (payload_niov <= PTL_MD_MAX_IOV);
 978
 979         /* It must be OK to kmap() if required */
 980         LASSERT (payload_kiov == NULL || !in_interrupt ());
 981         /* payload is either all vaddrs or all pages */
 982         LASSERT (!(payload_kiov != NULL && payload_iov != NULL));
 983
 984         if (payload_iov != NULL)
 985                 desc_size = offsetof(ksock_ltx_t, ltx_iov[1 + payload_niov]);
 986         else
 987                 desc_size = offsetof(ksock_ltx_t, ltx_kiov[payload_niov]);
 988
 989         if (in_interrupt() ||
 990             type == PTL_MSG_ACK ||
 991             type == PTL_MSG_REPLY) {
 992                 /* Can't block if in interrupt or responding to an incoming
 993                  * message */
 994                 PORTAL_ALLOC_ATOMIC(ltx, desc_size);
 995         } else {
 996                 PORTAL_ALLOC(ltx, desc_size);
 997         }
 998
 999         if (ltx == NULL) {
1000                 CERROR("Can't allocate tx desc type %d size %d %s\n",
1001                        type, desc_size, in_interrupt() ? "(intr)" : "");
1002                 return (PTL_NO_SPACE);
1003         }
1004
1005         atomic_inc(&ksocknal_data.ksnd_nactive_ltxs);
1006
1007         ltx->ltx_desc_size = desc_size;
1008
1009         /* We always have 1 mapped frag for the header */
1010         ltx->ltx_tx.tx_iov = ltx->ltx_iov;
1011         ltx->ltx_iov[0].iov_base = &ltx->ltx_hdr;
1012         ltx->ltx_iov[0].iov_len = sizeof(*hdr);
1013         ltx->ltx_hdr = *hdr;
1014
1015         ltx->ltx_private = private;
1016         ltx->ltx_cookie = cookie;
1017
1018         ltx->ltx_tx.tx_isfwd = 0;
1019         ltx->ltx_tx.tx_nob = sizeof (*hdr) + payload_nob;
1020
1021         if (payload_iov != NULL) {
1022                 /* payload is all mapped */
1023                 ltx->ltx_tx.tx_kiov  = NULL;
1024                 ltx->ltx_tx.tx_nkiov = 0;
1025
1026                 ltx->ltx_tx.tx_niov =
1027                         1 + lib_extract_iov(payload_niov, &ltx->ltx_iov[1],
1028                                             payload_niov, payload_iov,
1029                                             payload_offset, payload_nob);
1030         } else {
1031                 /* payload is all pages */
1032                 ltx->ltx_tx.tx_niov = 1;
1033
1034                 ltx->ltx_tx.tx_kiov = ltx->ltx_kiov;
1035                 ltx->ltx_tx.tx_nkiov =
1036                         lib_extract_kiov(payload_niov, ltx->ltx_kiov,
1037                                          payload_niov, payload_kiov,
1038                                          payload_offset, payload_nob);
1039         }
1040
1041         rc = ksocknal_launch_packet(&ltx->ltx_tx, nid);
1042         if (rc == 0)
1043                 return (PTL_OK);
1044
1045         ksocknal_free_ltx(ltx);
1046         return (PTL_FAIL);
1047 }
1048
1049 ptl_err_t
1050 ksocknal_send (lib_nal_t *nal, void *private, lib_msg_t *cookie,
1051                ptl_hdr_t *hdr, int type, ptl_nid_t nid, ptl_pid_t pid,
1052                unsigned int payload_niov, struct iovec *payload_iov,
1053                size_t payload_offset, size_t payload_len)
1054 {
1055         return (ksocknal_sendmsg(nal, private, cookie,
1056                                  hdr, type, nid, pid,
1057                                  payload_niov, payload_iov, NULL,
1058                                  payload_offset, payload_len));
1059 }
1060
1061 ptl_err_t
1062 ksocknal_send_pages (lib_nal_t *nal, void *private, lib_msg_t *cookie,
1063                      ptl_hdr_t *hdr, int type, ptl_nid_t nid, ptl_pid_t pid,
1064                      unsigned int payload_niov, ptl_kiov_t *payload_kiov,
1065                      size_t payload_offset, size_t payload_len)
1066 {
1067         return (ksocknal_sendmsg(nal, private, cookie,
1068                                  hdr, type, nid, pid,
1069                                  payload_niov, NULL, payload_kiov,
1070                                  payload_offset, payload_len));
1071 }
1072
1073 void
1074 ksocknal_fwd_packet (void *arg, kpr_fwd_desc_t *fwd)
1075 {
1076         ptl_nid_t     nid = fwd->kprfd_gateway_nid;
1077         ksock_ftx_t  *ftx = (ksock_ftx_t *)&fwd->kprfd_scratch;
1078         int           rc;
1079
1080         CDEBUG (D_NET, "Forwarding [%p] -> "LPX64" ("LPX64"))\n", fwd,
1081                 fwd->kprfd_gateway_nid, fwd->kprfd_target_nid);
1082
1083         /* I'm the gateway; must be the last hop */
1084         if (nid == ksocknal_lib.libnal_ni.ni_pid.nid)
1085                 nid = fwd->kprfd_target_nid;
1086
1087         /* setup iov for hdr */
1088         ftx->ftx_iov.iov_base = fwd->kprfd_hdr;
1089         ftx->ftx_iov.iov_len = sizeof(ptl_hdr_t);
1090
1091         ftx->ftx_tx.tx_isfwd = 1;                  /* This is a forwarding packet */
1092         ftx->ftx_tx.tx_nob   = sizeof(ptl_hdr_t) + fwd->kprfd_nob;
1093         ftx->ftx_tx.tx_niov  = 1;
1094         ftx->ftx_tx.tx_iov   = &ftx->ftx_iov;
1095         ftx->ftx_tx.tx_nkiov = fwd->kprfd_niov;
1096         ftx->ftx_tx.tx_kiov  = fwd->kprfd_kiov;
1097
1098         rc = ksocknal_launch_packet (&ftx->ftx_tx, nid);
1099         if (rc != 0)
1100                 kpr_fwd_done (&ksocknal_data.ksnd_router, fwd, rc);
1101 }
1102
1103 int
1104 ksocknal_thread_start (int (*fn)(void *arg), void *arg)
1105 {
1106         long          pid = kernel_thread (fn, arg, 0);
1107         unsigned long flags;
1108
1109         if (pid < 0)
1110                 return ((int)pid);
1111
1112         write_lock_irqsave(&ksocknal_data.ksnd_global_lock, flags);
1113         ksocknal_data.ksnd_nthreads++;
1114         write_unlock_irqrestore(&ksocknal_data.ksnd_global_lock, flags);
1115         return (0);
1116 }
1117
1118 void
1119 ksocknal_thread_fini (void)
1120 {
1121         unsigned long flags;
1122
1123         write_lock_irqsave(&ksocknal_data.ksnd_global_lock, flags);
1124         ksocknal_data.ksnd_nthreads--;
1125         write_unlock_irqrestore(&ksocknal_data.ksnd_global_lock, flags);
1126 }
1127
1128 void
1129 ksocknal_fmb_callback (void *arg, int error)
1130 {
1131         ksock_fmb_t       *fmb = (ksock_fmb_t *)arg;
1132         ksock_fmb_pool_t  *fmp = fmb->fmb_pool;
1133         ptl_hdr_t         *hdr = &fmb->fmb_hdr;
1134         ksock_conn_t      *conn = NULL;
1135         ksock_sched_t     *sched;
1136         unsigned long      flags;
1137         char               ipbuf[PTL_NALFMT_SIZE];
1138         char               ipbuf2[PTL_NALFMT_SIZE];
1139
1140         if (error != 0)
1141                 CERROR("Failed to route packet from "
1142                        LPX64" %s to "LPX64" %s: %d\n",
1143                        le64_to_cpu(hdr->src_nid),
1144                        portals_nid2str(SOCKNAL, le64_to_cpu(hdr->src_nid), ipbuf),
1145                        le64_to_cpu(hdr->dest_nid),
1146                        portals_nid2str(SOCKNAL, le64_to_cpu(hdr->dest_nid), ipbuf2),
1147                        error);
1148         else
1149                 CDEBUG (D_NET, "routed packet from "LPX64" to "LPX64": OK\n",
1150                         le64_to_cpu(hdr->src_nid), le64_to_cpu(hdr->dest_nid));
1151
1152         /* drop peer ref taken on init */
1153         ksocknal_put_peer (fmb->fmb_peer);
1154
1155         spin_lock_irqsave (&fmp->fmp_lock, flags);
1156
1157         list_add (&fmb->fmb_list, &fmp->fmp_idle_fmbs);
1158         fmp->fmp_nactive_fmbs--;
1159
1160         if (!list_empty (&fmp->fmp_blocked_conns)) {
1161                 conn = list_entry (fmb->fmb_pool->fmp_blocked_conns.next,
1162                                    ksock_conn_t, ksnc_rx_list);
1163                 list_del (&conn->ksnc_rx_list);
1164         }
1165
1166         spin_unlock_irqrestore (&fmp->fmp_lock, flags);
1167
1168         if (conn == NULL)
1169                 return;
1170
1171         CDEBUG (D_NET, "Scheduling conn %p\n", conn);
1172         LASSERT (conn->ksnc_rx_scheduled);
1173         LASSERT (conn->ksnc_rx_state == SOCKNAL_RX_FMB_SLEEP);
1174
1175         conn->ksnc_rx_state = SOCKNAL_RX_GET_FMB;
1176
1177         sched = conn->ksnc_scheduler;
1178
1179         spin_lock_irqsave (&sched->kss_lock, flags);
1180
1181         list_add_tail (&conn->ksnc_rx_list, &sched->kss_rx_conns);
1182         wake_up (&sched->kss_waitq);
1183
1184         spin_unlock_irqrestore (&sched->kss_lock, flags);
1185 }
1186
1187 ksock_fmb_t *
1188 ksocknal_get_idle_fmb (ksock_conn_t *conn)
1189 {
1190         int               payload_nob = conn->ksnc_rx_nob_left;
1191         unsigned long     flags;
1192         ksock_fmb_pool_t *pool;
1193         ksock_fmb_t      *fmb;
1194
1195         LASSERT (conn->ksnc_rx_state == SOCKNAL_RX_GET_FMB);
1196         LASSERT (kpr_routing(&ksocknal_data.ksnd_router));
1197
1198         if (payload_nob <= SOCKNAL_SMALL_FWD_PAGES * PAGE_SIZE)
1199                 pool = &ksocknal_data.ksnd_small_fmp;
1200         else
1201                 pool = &ksocknal_data.ksnd_large_fmp;
1202
1203         spin_lock_irqsave (&pool->fmp_lock, flags);
1204
1205         if (!list_empty (&pool->fmp_idle_fmbs)) {
1206                 fmb = list_entry(pool->fmp_idle_fmbs.next,
1207                                  ksock_fmb_t, fmb_list);
1208                 list_del (&fmb->fmb_list);
1209                 pool->fmp_nactive_fmbs++;
1210                 spin_unlock_irqrestore (&pool->fmp_lock, flags);
1211
1212                 return (fmb);
1213         }
1214
1215         /* deschedule until fmb free */
1216
1217         conn->ksnc_rx_state = SOCKNAL_RX_FMB_SLEEP;
1218
1219         list_add_tail (&conn->ksnc_rx_list,
1220                        &pool->fmp_blocked_conns);
1221
1222         spin_unlock_irqrestore (&pool->fmp_lock, flags);
1223         return (NULL);
1224 }
1225
1226 int
1227 ksocknal_init_fmb (ksock_conn_t *conn, ksock_fmb_t *fmb)
1228 {
1229         int       payload_nob = conn->ksnc_rx_nob_left;
1230         ptl_nid_t dest_nid = le64_to_cpu(conn->ksnc_hdr.dest_nid);
1231         int       niov = 0;
1232         int       nob = payload_nob;
1233
1234         LASSERT (conn->ksnc_rx_scheduled);
1235         LASSERT (conn->ksnc_rx_state == SOCKNAL_RX_GET_FMB);
1236         LASSERT (conn->ksnc_rx_nob_wanted == conn->ksnc_rx_nob_left);
1237         LASSERT (payload_nob >= 0);
1238         LASSERT (payload_nob <= fmb->fmb_pool->fmp_buff_pages * PAGE_SIZE);
1239         LASSERT (sizeof (ptl_hdr_t) < PAGE_SIZE);
1240         LASSERT (fmb->fmb_kiov[0].kiov_offset == 0);
1241
1242         /* Take a ref on the conn's peer to prevent module unload before
1243          * forwarding completes. */
1244         fmb->fmb_peer = conn->ksnc_peer;
1245         atomic_inc (&conn->ksnc_peer->ksnp_refcount);
1246
1247         /* Copy the header we just read into the forwarding buffer.  If
1248          * there's payload, start reading reading it into the buffer,
1249          * otherwise the forwarding buffer can be kicked off
1250          * immediately. */
1251         fmb->fmb_hdr = conn->ksnc_hdr;
1252
1253         while (nob > 0) {
1254                 LASSERT (niov < fmb->fmb_pool->fmp_buff_pages);
1255                 LASSERT (fmb->fmb_kiov[niov].kiov_offset == 0);
1256                 fmb->fmb_kiov[niov].kiov_len = MIN (PAGE_SIZE, nob);
1257                 nob -= PAGE_SIZE;
1258                 niov++;
1259         }
1260
1261         kpr_fwd_init(&fmb->fmb_fwd, dest_nid, &fmb->fmb_hdr,
1262                      payload_nob, niov, fmb->fmb_kiov,
1263                      ksocknal_fmb_callback, fmb);
1264
1265         if (payload_nob == 0) {         /* got complete packet already */
1266                 CDEBUG (D_NET, "%p "LPX64"->"LPX64" fwd_start (immediate)\n",
1267                         conn, le64_to_cpu(conn->ksnc_hdr.src_nid), dest_nid);
1268
1269                 kpr_fwd_start (&ksocknal_data.ksnd_router, &fmb->fmb_fwd);
1270
1271                 ksocknal_new_packet (conn, 0);  /* on to next packet */
1272                 return (1);
1273         }
1274
1275         conn->ksnc_cookie = fmb;                /* stash fmb for later */
1276         conn->ksnc_rx_state = SOCKNAL_RX_BODY_FWD; /* read in the payload */
1277
1278         /* Set up conn->ksnc_rx_kiov to read the payload into fmb's kiov-ed
1279          * buffer */
1280         LASSERT (niov <= sizeof(conn->ksnc_rx_iov_space)/sizeof(ptl_kiov_t));
1281
1282         conn->ksnc_rx_niov = 0;
1283         conn->ksnc_rx_nkiov = niov;
1284         conn->ksnc_rx_kiov = conn->ksnc_rx_iov_space.kiov;
1285         memcpy(conn->ksnc_rx_kiov, fmb->fmb_kiov, niov * sizeof(ptl_kiov_t));
1286
1287         CDEBUG (D_NET, "%p "LPX64"->"LPX64" %d reading body\n", conn,
1288                 le64_to_cpu(conn->ksnc_hdr.src_nid), dest_nid, payload_nob);
1289         return (0);
1290 }
1291
1292 void
1293 ksocknal_fwd_parse (ksock_conn_t *conn)
1294 {
1295         ksock_peer_t *peer;
1296         ptl_nid_t     dest_nid = le64_to_cpu(conn->ksnc_hdr.dest_nid);
1297         ptl_nid_t     src_nid = le64_to_cpu(conn->ksnc_hdr.src_nid);
1298         int           body_len = le32_to_cpu(conn->ksnc_hdr.payload_length);
1299         char str[PTL_NALFMT_SIZE];
1300         char str2[PTL_NALFMT_SIZE];
1301
1302         CDEBUG (D_NET, "%p "LPX64"->"LPX64" %d parsing header\n", conn,
1303                 src_nid, dest_nid, conn->ksnc_rx_nob_left);
1304
1305         LASSERT (conn->ksnc_rx_state == SOCKNAL_RX_HEADER);
1306         LASSERT (conn->ksnc_rx_scheduled);
1307
1308         if (body_len < 0) {                 /* length corrupt (overflow) */
1309                 CERROR("dropping packet from "LPX64" (%s) for "LPX64" (%s): "
1310                        "packet size %d illegal\n",
1311                        src_nid, portals_nid2str(TCPNAL, src_nid, str),
1312                        dest_nid, portals_nid2str(TCPNAL, dest_nid, str2),
1313                        body_len);
1314
1315                 ksocknal_new_packet (conn, 0);  /* on to new packet */
1316                 return;
1317         }
1318
1319         if (!kpr_routing(&ksocknal_data.ksnd_router)) {    /* not forwarding */
1320                 CERROR("dropping packet from "LPX64" (%s) for "LPX64
1321                        " (%s): not forwarding\n",
1322                        src_nid, portals_nid2str(TCPNAL, src_nid, str),
1323                        dest_nid, portals_nid2str(TCPNAL, dest_nid, str2));
1324                 /* on to new packet (skip this one's body) */
1325                 ksocknal_new_packet (conn, body_len);
1326                 return;
1327         }
1328
1329         if (body_len > PTL_MTU) {      /* too big to forward */
1330                 CERROR ("dropping packet from "LPX64" (%s) for "LPX64
1331                         "(%s): packet size %d too big\n",
1332                         src_nid, portals_nid2str(TCPNAL, src_nid, str),
1333                         dest_nid, portals_nid2str(TCPNAL, dest_nid, str2),
1334                         body_len);
1335                 /* on to new packet (skip this one's body) */
1336                 ksocknal_new_packet (conn, body_len);
1337                 return;
1338         }
1339
1340         /* should have gone direct */
1341         peer = ksocknal_get_peer (conn->ksnc_hdr.dest_nid);
1342         if (peer != NULL) {
1343                 CERROR ("dropping packet from "LPX64" (%s) for "LPX64
1344                         "(%s): target is a peer\n",
1345                         src_nid, portals_nid2str(TCPNAL, src_nid, str),
1346                         dest_nid, portals_nid2str(TCPNAL, dest_nid, str2));
1347                 ksocknal_put_peer (peer);  /* drop ref from get above */
1348
1349                 /* on to next packet (skip this one's body) */
1350                 ksocknal_new_packet (conn, body_len);
1351                 return;
1352         }
1353
1354         conn->ksnc_rx_state = SOCKNAL_RX_GET_FMB;       /* Getting FMB now */
1355         conn->ksnc_rx_nob_left = body_len;              /* stash packet size */
1356         conn->ksnc_rx_nob_wanted = body_len;            /* (no slop) */
1357 }
1358
1359 int
1360 ksocknal_new_packet (ksock_conn_t *conn, int nob_to_skip)
1361 {
1362         static char ksocknal_slop_buffer[4096];
1363
1364         int   nob;
1365         int   niov;
1366         int   skipped;
1367
1368         if (nob_to_skip == 0) {         /* right at next packet boundary now */
1369                 conn->ksnc_rx_started = 0;
1370                 mb ();                          /* racing with timeout thread */
1371
1372                 conn->ksnc_rx_state = SOCKNAL_RX_HEADER;
1373                 conn->ksnc_rx_nob_wanted = sizeof (ptl_hdr_t);
1374                 conn->ksnc_rx_nob_left = sizeof (ptl_hdr_t);
1375
1376                 conn->ksnc_rx_iov = (struct iovec *)&conn->ksnc_rx_iov_space;
1377                 conn->ksnc_rx_iov[0].iov_base = (char *)&conn->ksnc_hdr;
1378                 conn->ksnc_rx_iov[0].iov_len  = sizeof (ptl_hdr_t);
1379                 conn->ksnc_rx_niov = 1;
1380
1381                 conn->ksnc_rx_kiov = NULL;
1382                 conn->ksnc_rx_nkiov = 0;
1383                 return (1);
1384         }
1385
1386         /* Set up to skip as much a possible now.  If there's more left
1387          * (ran out of iov entries) we'll get called again */
1388
1389         conn->ksnc_rx_state = SOCKNAL_RX_SLOP;
1390         conn->ksnc_rx_nob_left = nob_to_skip;
1391         conn->ksnc_rx_iov = (struct iovec *)&conn->ksnc_rx_iov_space;
1392         skipped = 0;
1393         niov = 0;
1394
1395         do {
1396                 nob = MIN (nob_to_skip, sizeof (ksocknal_slop_buffer));
1397
1398                 conn->ksnc_rx_iov[niov].iov_base = ksocknal_slop_buffer;
1399                 conn->ksnc_rx_iov[niov].iov_len  = nob;
1400                 niov++;
1401                 skipped += nob;
1402                 nob_to_skip -=nob;
1403
1404         } while (nob_to_skip != 0 &&    /* mustn't overflow conn's rx iov */
1405                  niov < sizeof(conn->ksnc_rx_iov_space) / sizeof (struct iovec));
1406
1407         conn->ksnc_rx_niov = niov;
1408         conn->ksnc_rx_kiov = NULL;
1409         conn->ksnc_rx_nkiov = 0;
1410         conn->ksnc_rx_nob_wanted = skipped;
1411         return (0);
1412 }
1413
1414 int
1415 ksocknal_process_receive (ksock_conn_t *conn)
1416 {
1417         ksock_fmb_t  *fmb;
1418         int           rc;
1419
1420         LASSERT (atomic_read (&conn->ksnc_refcount) > 0);
1421
1422         /* doesn't need a forwarding buffer */
1423         if (conn->ksnc_rx_state != SOCKNAL_RX_GET_FMB)
1424                 goto try_read;
1425
1426  get_fmb:
1427         fmb = ksocknal_get_idle_fmb (conn);
1428         if (fmb == NULL) {
1429                 /* conn descheduled waiting for idle fmb */
1430                 return (0);
1431         }
1432
1433         if (ksocknal_init_fmb (conn, fmb)) {
1434                 /* packet forwarded */
1435                 return (0);
1436         }
1437
1438  try_read:
1439         /* NB: sched lock NOT held */
1440         LASSERT (conn->ksnc_rx_state == SOCKNAL_RX_HEADER ||
1441                  conn->ksnc_rx_state == SOCKNAL_RX_BODY ||
1442                  conn->ksnc_rx_state == SOCKNAL_RX_BODY_FWD ||
1443                  conn->ksnc_rx_state == SOCKNAL_RX_SLOP);
1444
1445         LASSERT (conn->ksnc_rx_nob_wanted > 0);
1446
1447         rc = ksocknal_receive(conn);
1448
1449         if (rc <= 0) {
1450                 LASSERT (rc != -EAGAIN);
1451
1452                 if (rc == 0)
1453                         CWARN ("[%p] EOF from "LPX64" ip %d.%d.%d.%d:%d\n",
1454                                conn, conn->ksnc_peer->ksnp_nid,
1455                                HIPQUAD(conn->ksnc_ipaddr),
1456                                conn->ksnc_port);
1457                 else if (!conn->ksnc_closing)
1458                         CERROR ("[%p] Error %d on read from "LPX64
1459                                 " ip %d.%d.%d.%d:%d\n",
1460                                 conn, rc, conn->ksnc_peer->ksnp_nid,
1461                                 HIPQUAD(conn->ksnc_ipaddr),
1462                                 conn->ksnc_port);
1463
1464                 ksocknal_close_conn_and_siblings (conn, rc);
1465                 return (rc == 0 ? -ESHUTDOWN : rc);
1466         }
1467
1468         if (conn->ksnc_rx_nob_wanted != 0) {
1469                 /* short read */
1470                 return (-EAGAIN);
1471         }
1472
1473         switch (conn->ksnc_rx_state) {
1474         case SOCKNAL_RX_HEADER:
1475                 if (conn->ksnc_hdr.type != cpu_to_le32(PTL_MSG_HELLO) &&
1476                     le64_to_cpu(conn->ksnc_hdr.dest_nid) !=
1477                     ksocknal_lib.libnal_ni.ni_pid.nid) {
1478                         /* This packet isn't for me */
1479                         ksocknal_fwd_parse (conn);
1480                         switch (conn->ksnc_rx_state) {
1481                         case SOCKNAL_RX_HEADER: /* skipped (zero payload) */
1482                                 return (0);     /* => come back later */
1483                         case SOCKNAL_RX_SLOP:   /* skipping packet's body */
1484                                 goto try_read;  /* => go read it */
1485                         case SOCKNAL_RX_GET_FMB: /* forwarding */
1486                                 goto get_fmb;   /* => go get a fwd msg buffer */
1487                         default:
1488                                 LBUG ();
1489                         }
1490                         /* Not Reached */
1491                 }
1492
1493                 /* sets wanted_len, iovs etc */
1494                 rc = lib_parse(&ksocknal_lib, &conn->ksnc_hdr, conn);
1495
1496                 if (rc != PTL_OK) {
1497                         /* I just received garbage: give up on this conn */
1498                         ksocknal_close_conn_and_siblings (conn, rc);
1499                         return (-EPROTO);
1500                 }
1501
1502                 if (conn->ksnc_rx_nob_wanted != 0) { /* need to get payload? */
1503                         conn->ksnc_rx_state = SOCKNAL_RX_BODY;
1504                         goto try_read;          /* go read the payload */
1505                 }
1506                 /* Fall through (completed packet for me) */
1507
1508         case SOCKNAL_RX_BODY:
1509                 /* payload all received */
1510                 lib_finalize(&ksocknal_lib, NULL, conn->ksnc_cookie, PTL_OK);
1511                 /* Fall through */
1512
1513         case SOCKNAL_RX_SLOP:
1514                 /* starting new packet? */
1515                 if (ksocknal_new_packet (conn, conn->ksnc_rx_nob_left))
1516                         return (0);     /* come back later */
1517                 goto try_read;          /* try to finish reading slop now */
1518
1519         case SOCKNAL_RX_BODY_FWD:
1520                 /* payload all received */
1521                 CDEBUG (D_NET, "%p "LPX64"->"LPX64" %d fwd_start (got body)\n",
1522                         conn, le64_to_cpu(conn->ksnc_hdr.src_nid),
1523                         le64_to_cpu(conn->ksnc_hdr.dest_nid),
1524                         conn->ksnc_rx_nob_left);
1525
1526                 /* forward the packet. NB ksocknal_init_fmb() put fmb into
1527                  * conn->ksnc_cookie */
1528                 fmb = (ksock_fmb_t *)conn->ksnc_cookie;
1529                 kpr_fwd_start (&ksocknal_data.ksnd_router, &fmb->fmb_fwd);
1530
1531                 /* no slop in forwarded packets */
1532                 LASSERT (conn->ksnc_rx_nob_left == 0);
1533
1534                 ksocknal_new_packet (conn, 0);  /* on to next packet */
1535                 return (0);                     /* (later) */
1536
1537         default:
1538                 break;
1539         }
1540
1541         /* Not Reached */
1542         LBUG ();
1543         return (-EINVAL);                       /* keep gcc happy */
1544 }
1545
1546 ptl_err_t
1547 ksocknal_recv (lib_nal_t *nal, void *private, lib_msg_t *msg,
1548                unsigned int niov, struct iovec *iov,
1549                size_t offset, size_t mlen, size_t rlen)
1550 {
1551         ksock_conn_t *conn = (ksock_conn_t *)private;
1552
1553         LASSERT (mlen <= rlen);
1554         LASSERT (niov <= PTL_MD_MAX_IOV);
1555
1556         conn->ksnc_cookie = msg;
1557         conn->ksnc_rx_nob_wanted = mlen;
1558         conn->ksnc_rx_nob_left   = rlen;
1559
1560         conn->ksnc_rx_nkiov = 0;
1561         conn->ksnc_rx_kiov = NULL;
1562         conn->ksnc_rx_iov = conn->ksnc_rx_iov_space.iov;
1563         conn->ksnc_rx_niov =
1564                 lib_extract_iov(PTL_MD_MAX_IOV, conn->ksnc_rx_iov,
1565                                 niov, iov, offset, mlen);
1566
1567         LASSERT (mlen ==
1568                  lib_iov_nob (conn->ksnc_rx_niov, conn->ksnc_rx_iov) +
1569                  lib_kiov_nob (conn->ksnc_rx_nkiov, conn->ksnc_rx_kiov));
1570
1571         return (PTL_OK);
1572 }
1573
1574 ptl_err_t
1575 ksocknal_recv_pages (lib_nal_t *nal, void *private, lib_msg_t *msg,
1576                      unsigned int niov, ptl_kiov_t *kiov,
1577                      size_t offset, size_t mlen, size_t rlen)
1578 {
1579         ksock_conn_t *conn = (ksock_conn_t *)private;
1580
1581         LASSERT (mlen <= rlen);
1582         LASSERT (niov <= PTL_MD_MAX_IOV);
1583
1584         conn->ksnc_cookie = msg;
1585         conn->ksnc_rx_nob_wanted = mlen;
1586         conn->ksnc_rx_nob_left   = rlen;
1587
1588         conn->ksnc_rx_niov = 0;
1589         conn->ksnc_rx_iov  = NULL;
1590         conn->ksnc_rx_kiov = conn->ksnc_rx_iov_space.kiov;
1591         conn->ksnc_rx_nkiov =
1592                 lib_extract_kiov(PTL_MD_MAX_IOV, conn->ksnc_rx_kiov,
1593                                  niov, kiov, offset, mlen);
1594
1595         LASSERT (mlen ==
1596                  lib_iov_nob (conn->ksnc_rx_niov, conn->ksnc_rx_iov) +
1597                  lib_kiov_nob (conn->ksnc_rx_nkiov, conn->ksnc_rx_kiov));
1598
1599         return (PTL_OK);
1600 }
1601
1602 static inline int
1603 ksocknal_sched_cansleep(ksock_sched_t *sched)
1604 {
1605         unsigned long flags;
1606         int           rc;
1607
1608         spin_lock_irqsave(&sched->kss_lock, flags);
1609
1610         rc = (!ksocknal_data.ksnd_shuttingdown &&
1611 #if SOCKNAL_ZC
1612               list_empty(&sched->kss_zctxdone_list) &&
1613 #endif
1614               list_empty(&sched->kss_rx_conns) &&
1615               list_empty(&sched->kss_tx_conns));
1616
1617         spin_unlock_irqrestore(&sched->kss_lock, flags);
1618         return (rc);
1619 }
1620
1621 int ksocknal_scheduler (void *arg)
1622 {
1623         ksock_sched_t     *sched = (ksock_sched_t *)arg;
1624         ksock_conn_t      *conn;
1625         ksock_tx_t        *tx;
1626         unsigned long      flags;
1627         int                rc;
1628         int                nloops = 0;
1629         int                id = sched - ksocknal_data.ksnd_schedulers;
1630         char               name[16];
1631
1632         snprintf (name, sizeof (name),"ksocknald_%02d", id);
1633         kportal_daemonize (name);
1634         kportal_blockallsigs ();
1635
1636 #if (CONFIG_SMP && CPU_AFFINITY)
1637         id = ksocknal_sched2cpu(id);
1638         if (cpu_online(id)) {
1639                 cpumask_t m;
1640                 cpu_set(id, m);
1641                 set_cpus_allowed(current, m);
1642         } else {
1643                 CERROR ("Can't set CPU affinity for %s to %d\n", name, id);
1644         }
1645 #endif /* CONFIG_SMP && CPU_AFFINITY */
1646
1647         spin_lock_irqsave (&sched->kss_lock, flags);
1648
1649         while (!ksocknal_data.ksnd_shuttingdown) {
1650                 int did_something = 0;
1651
1652                 /* Ensure I progress everything semi-fairly */
1653
1654                 if (!list_empty (&sched->kss_rx_conns)) {
1655                         conn = list_entry(sched->kss_rx_conns.next,
1656                                           ksock_conn_t, ksnc_rx_list);
1657                         list_del(&conn->ksnc_rx_list);
1658
1659                         LASSERT(conn->ksnc_rx_scheduled);
1660                         LASSERT(conn->ksnc_rx_ready);
1661
1662                         /* clear rx_ready in case receive isn't complete.
1663                          * Do it BEFORE we call process_recv, since
1664                          * data_ready can set it any time after we release
1665                          * kss_lock. */
1666                         conn->ksnc_rx_ready = 0;
1667                         spin_unlock_irqrestore(&sched->kss_lock, flags);
1668
1669                         rc = ksocknal_process_receive(conn);
1670
1671                         spin_lock_irqsave(&sched->kss_lock, flags);
1672
1673                         /* I'm the only one that can clear this flag */
1674                         LASSERT(conn->ksnc_rx_scheduled);
1675
1676                         /* Did process_receive get everything it wanted? */
1677                         if (rc == 0)
1678                                 conn->ksnc_rx_ready = 1;
1679
1680                         if (conn->ksnc_rx_state == SOCKNAL_RX_FMB_SLEEP ||
1681                             conn->ksnc_rx_state == SOCKNAL_RX_GET_FMB) {
1682                                 /* Conn blocked for a forwarding buffer.
1683                                  * It will get queued for my attention when
1684                                  * one becomes available (and it might just
1685                                  * already have been!).  Meanwhile my ref
1686                                  * on it stays put. */
1687                         } else if (conn->ksnc_rx_ready) {
1688                                 /* reschedule for rx */
1689                                 list_add_tail (&conn->ksnc_rx_list,
1690                                                &sched->kss_rx_conns);
1691                         } else {
1692                                 conn->ksnc_rx_scheduled = 0;
1693                                 /* drop my ref */
1694                                 ksocknal_put_conn(conn);
1695                         }
1696
1697                         did_something = 1;
1698                 }
1699
1700                 if (!list_empty (&sched->kss_tx_conns)) {
1701                         conn = list_entry(sched->kss_tx_conns.next,
1702                                           ksock_conn_t, ksnc_tx_list);
1703                         list_del (&conn->ksnc_tx_list);
1704
1705                         LASSERT(conn->ksnc_tx_scheduled);
1706                         LASSERT(conn->ksnc_tx_ready);
1707                         LASSERT(!list_empty(&conn->ksnc_tx_queue));
1708
1709                         tx = list_entry(conn->ksnc_tx_queue.next,
1710                                         ksock_tx_t, tx_list);
1711                         /* dequeue now so empty list => more to send */
1712                         list_del(&tx->tx_list);
1713
1714                         /* Clear tx_ready in case send isn't complete.  Do
1715                          * it BEFORE we call process_transmit, since
1716                          * write_space can set it any time after we release
1717                          * kss_lock. */
1718                         conn->ksnc_tx_ready = 0;
1719                         spin_unlock_irqrestore (&sched->kss_lock, flags);
1720
1721                         rc = ksocknal_process_transmit(conn, tx);
1722
1723                         spin_lock_irqsave (&sched->kss_lock, flags);
1724
1725                         if (rc == -ENOMEM || rc == -EAGAIN) {
1726                                 /* Incomplete send: replace tx on HEAD of tx_queue */
1727                                 list_add (&tx->tx_list, &conn->ksnc_tx_queue);
1728                         } else {
1729                                 /* Complete send; assume space for more */
1730                                 conn->ksnc_tx_ready = 1;
1731                         }
1732
1733                         if (rc == -ENOMEM) {
1734                                 /* Do nothing; after a short timeout, this
1735                                  * conn will be reposted on kss_tx_conns. */
1736                         } else if (conn->ksnc_tx_ready &&
1737                                    !list_empty (&conn->ksnc_tx_queue)) {
1738                                 /* reschedule for tx */
1739                                 list_add_tail (&conn->ksnc_tx_list,
1740                                                &sched->kss_tx_conns);
1741                         } else {
1742                                 conn->ksnc_tx_scheduled = 0;
1743                                 /* drop my ref */
1744                                 ksocknal_put_conn (conn);
1745                         }
1746
1747                         did_something = 1;
1748                 }
1749 #if SOCKNAL_ZC
1750                 if (!list_empty (&sched->kss_zctxdone_list)) {
1751                         ksock_tx_t *tx =
1752                                 list_entry(sched->kss_zctxdone_list.next,
1753                                            ksock_tx_t, tx_list);
1754                         did_something = 1;
1755
1756                         list_del (&tx->tx_list);
1757                         spin_unlock_irqrestore (&sched->kss_lock, flags);
1758
1759                         ksocknal_tx_done (tx, 1);
1760
1761                         spin_lock_irqsave (&sched->kss_lock, flags);
1762                 }
1763 #endif
1764                 if (!did_something ||           /* nothing to do */
1765                     ++nloops == SOCKNAL_RESCHED) { /* hogging CPU? */
1766                         spin_unlock_irqrestore (&sched->kss_lock, flags);
1767
1768                         nloops = 0;
1769
1770                         if (!did_something) {   /* wait for something to do */
1771                                 rc = wait_event_interruptible (sched->kss_waitq,
1772                                                                !ksocknal_sched_cansleep(sched));
1773                                 LASSERT (rc == 0);
1774                         } else
1775                                our_cond_resched();
1776
1777                         spin_lock_irqsave (&sched->kss_lock, flags);
1778                 }
1779         }
1780
1781         spin_unlock_irqrestore (&sched->kss_lock, flags);
1782         ksocknal_thread_fini ();
1783         return (0);
1784 }
1785
1786 void
1787 ksocknal_data_ready (struct sock *sk, int n)
1788 {
1789         unsigned long  flags;
1790         ksock_conn_t  *conn;
1791         ksock_sched_t *sched;
1792         ENTRY;
1793
1794         /* interleave correctly with closing sockets... */
1795         read_lock (&ksocknal_data.ksnd_global_lock);
1796
1797         conn = sk->sk_user_data;
1798         if (conn == NULL) {             /* raced with ksocknal_terminate_conn */
1799                 LASSERT (sk->sk_data_ready != &ksocknal_data_ready);
1800                 sk->sk_data_ready (sk, n);
1801         } else {
1802                 sched = conn->ksnc_scheduler;
1803
1804                 spin_lock_irqsave (&sched->kss_lock, flags);
1805
1806                 conn->ksnc_rx_ready = 1;
1807
1808                 if (!conn->ksnc_rx_scheduled) {  /* not being progressed */
1809                         list_add_tail(&conn->ksnc_rx_list,
1810                                       &sched->kss_rx_conns);
1811                         conn->ksnc_rx_scheduled = 1;
1812                         /* extra ref for scheduler */
1813                         atomic_inc (&conn->ksnc_refcount);
1814
1815                         wake_up (&sched->kss_waitq);
1816                 }
1817
1818                 spin_unlock_irqrestore (&sched->kss_lock, flags);
1819         }
1820
1821         read_unlock (&ksocknal_data.ksnd_global_lock);
1822
1823         EXIT;
1824 }
1825
1826 void
1827 ksocknal_write_space (struct sock *sk)
1828 {
1829         unsigned long  flags;
1830         ksock_conn_t  *conn;
1831         ksock_sched_t *sched;
1832
1833         /* interleave correctly with closing sockets... */
1834         read_lock (&ksocknal_data.ksnd_global_lock);
1835
1836         conn = sk->sk_user_data;
1837
1838         CDEBUG(D_NET, "sk %p wspace %d low water %d conn %p%s%s%s\n",
1839                sk, tcp_wspace(sk), SOCKNAL_TX_LOW_WATER(sk), conn,
1840                (conn == NULL) ? "" : (conn->ksnc_tx_ready ?
1841                                       " ready" : " blocked"),
1842                (conn == NULL) ? "" : (conn->ksnc_tx_scheduled ?
1843                                       " scheduled" : " idle"),
1844                (conn == NULL) ? "" : (list_empty (&conn->ksnc_tx_queue) ?
1845                                       " empty" : " queued"));
1846
1847         if (conn == NULL) {             /* raced with ksocknal_terminate_conn */
1848                 LASSERT (sk->sk_write_space != &ksocknal_write_space);
1849                 sk->sk_write_space (sk);
1850
1851                 read_unlock (&ksocknal_data.ksnd_global_lock);
1852                 return;
1853         }
1854
1855         if (tcp_wspace(sk) >= SOCKNAL_TX_LOW_WATER(sk)) { /* got enough space */
1856                 clear_bit (SOCK_NOSPACE, &sk->sk_socket->flags);
1857
1858                 sched = conn->ksnc_scheduler;
1859
1860                 spin_lock_irqsave (&sched->kss_lock, flags);
1861
1862                 conn->ksnc_tx_ready = 1;
1863
1864                 if (!conn->ksnc_tx_scheduled && // not being progressed
1865                     !list_empty(&conn->ksnc_tx_queue)){//packets to send
1866                         list_add_tail (&conn->ksnc_tx_list,
1867                                        &sched->kss_tx_conns);
1868                         conn->ksnc_tx_scheduled = 1;
1869                         /* extra ref for scheduler */
1870                         atomic_inc (&conn->ksnc_refcount);
1871
1872                         wake_up (&sched->kss_waitq);
1873                 }
1874
1875                 spin_unlock_irqrestore (&sched->kss_lock, flags);
1876         }
1877
1878         read_unlock (&ksocknal_data.ksnd_global_lock);
1879 }
1880
1881 int
1882 ksocknal_sock_write (struct socket *sock, void *buffer, int nob)
1883 {
1884         int           rc;
1885         mm_segment_t  oldmm = get_fs();
1886
1887         while (nob > 0) {
1888                 struct iovec  iov = {
1889                         .iov_base = buffer,
1890                         .iov_len  = nob
1891                 };
1892                 struct msghdr msg = {
1893                         .msg_name       = NULL,
1894                         .msg_namelen    = 0,
1895                         .msg_iov        = &iov,
1896                         .msg_iovlen     = 1,
1897                         .msg_control    = NULL,
1898                         .msg_controllen = 0,
1899                         .msg_flags      = 0
1900                 };
1901
1902                 set_fs (KERNEL_DS);
1903                 rc = sock_sendmsg (sock, &msg, iov.iov_len);
1904                 set_fs (oldmm);
1905
1906                 if (rc < 0)
1907                         return (rc);
1908
1909                 if (rc == 0) {
1910                         CERROR ("Unexpected zero rc\n");
1911                         return (-ECONNABORTED);
1912                 }
1913
1914                 buffer = ((char *)buffer) + rc;
1915                 nob -= rc;
1916         }
1917
1918         return (0);
1919 }
1920
1921 int
1922 ksocknal_sock_read (struct socket *sock, void *buffer, int nob)
1923 {
1924         int           rc;
1925         mm_segment_t  oldmm = get_fs();
1926
1927         while (nob > 0) {
1928                 struct iovec  iov = {
1929                         .iov_base = buffer,
1930                         .iov_len  = nob
1931                 };
1932                 struct msghdr msg = {
1933                         .msg_name       = NULL,
1934                         .msg_namelen    = 0,
1935                         .msg_iov        = &iov,
1936                         .msg_iovlen     = 1,
1937                         .msg_control    = NULL,
1938                         .msg_controllen = 0,
1939                         .msg_flags      = 0
1940                 };
1941
1942                 set_fs (KERNEL_DS);
1943                 rc = sock_recvmsg (sock, &msg, iov.iov_len, 0);
1944                 set_fs (oldmm);
1945
1946                 if (rc < 0)
1947                         return (rc);
1948
1949                 if (rc == 0)
1950                         return (-ECONNABORTED);
1951
1952                 buffer = ((char *)buffer) + rc;
1953                 nob -= rc;
1954         }
1955
1956         return (0);
1957 }
1958
1959 int
1960 ksocknal_send_hello (ksock_conn_t *conn, __u32 *ipaddrs, int nipaddrs)
1961 {
1962         /* CAVEAT EMPTOR: this byte flips 'ipaddrs' */
1963         struct socket      *sock = conn->ksnc_sock;
1964         ptl_hdr_t           hdr;
1965         ptl_magicversion_t *hmv = (ptl_magicversion_t *)&hdr.dest_nid;
1966         int                 i;
1967         int                 rc;
1968
1969         LASSERT (conn->ksnc_type != SOCKNAL_CONN_NONE);
1970         LASSERT (nipaddrs <= SOCKNAL_MAX_INTERFACES);
1971
1972         /* No need for getconnsock/putconnsock */
1973         LASSERT (!conn->ksnc_closing);
1974
1975         LASSERT (sizeof (*hmv) == sizeof (hdr.dest_nid));
1976         hmv->magic         = cpu_to_le32 (PORTALS_PROTO_MAGIC);
1977         hmv->version_major = cpu_to_le16 (PORTALS_PROTO_VERSION_MAJOR);
1978         hmv->version_minor = cpu_to_le16 (PORTALS_PROTO_VERSION_MINOR);
1979
1980         hdr.src_nid        = cpu_to_le64 (ksocknal_lib.libnal_ni.ni_pid.nid);
1981         hdr.type           = cpu_to_le32 (PTL_MSG_HELLO);
1982         hdr.payload_length = cpu_to_le32 (nipaddrs * sizeof(*ipaddrs));
1983
1984         hdr.msg.hello.type = cpu_to_le32 (conn->ksnc_type);
1985         hdr.msg.hello.incarnation =
1986                 cpu_to_le64 (ksocknal_data.ksnd_incarnation);
1987
1988         /* Receiver is eager */
1989         rc = ksocknal_sock_write (sock, &hdr, sizeof(hdr));
1990         if (rc != 0) {
1991                 CERROR ("Error %d sending HELLO hdr to %u.%u.%u.%u/%d\n",
1992                         rc, HIPQUAD(conn->ksnc_ipaddr), conn->ksnc_port);
1993                 return (rc);
1994         }
1995
1996         if (nipaddrs == 0)
1997                 return (0);
1998
1999         for (i = 0; i < nipaddrs; i++) {
2000                 ipaddrs[i] = __cpu_to_le32 (ipaddrs[i]);
2001         }
2002
2003         rc = ksocknal_sock_write (sock, ipaddrs, nipaddrs * sizeof(*ipaddrs));
2004         if (rc != 0)
2005                 CERROR ("Error %d sending HELLO payload (%d)"
2006                         " to %u.%u.%u.%u/%d\n", rc, nipaddrs,
2007                         HIPQUAD(conn->ksnc_ipaddr), conn->ksnc_port);
2008         return (rc);
2009 }
2010
2011 int
2012 ksocknal_invert_type(int type)
2013 {
2014         switch (type)
2015         {
2016         case SOCKNAL_CONN_ANY:
2017         case SOCKNAL_CONN_CONTROL:
2018                 return (type);
2019         case SOCKNAL_CONN_BULK_IN:
2020                 return SOCKNAL_CONN_BULK_OUT;
2021         case SOCKNAL_CONN_BULK_OUT:
2022                 return SOCKNAL_CONN_BULK_IN;
2023         default:
2024                 return (SOCKNAL_CONN_NONE);
2025         }
2026 }
2027
2028 int
2029 ksocknal_recv_hello (ksock_conn_t *conn, ptl_nid_t *nid,
2030                      __u64 *incarnation, __u32 *ipaddrs)
2031 {
2032         struct socket      *sock = conn->ksnc_sock;
2033         int                 rc;
2034         int                 nips;
2035         int                 i;
2036         int                 type;
2037         ptl_hdr_t           hdr;
2038         ptl_magicversion_t *hmv;
2039
2040         hmv = (ptl_magicversion_t *)&hdr.dest_nid;
2041         LASSERT (sizeof (*hmv) == sizeof (hdr.dest_nid));
2042
2043         rc = ksocknal_sock_read (sock, hmv, sizeof (*hmv));
2044         if (rc != 0) {
2045                 CERROR ("Error %d reading HELLO from %u.%u.%u.%u\n",
2046                         rc, HIPQUAD(conn->ksnc_ipaddr));
2047                 return (rc);
2048         }
2049
2050         if (hmv->magic != le32_to_cpu (PORTALS_PROTO_MAGIC)) {
2051                 CERROR ("Bad magic %#08x (%#08x expected) from %u.%u.%u.%u\n",
2052                         __cpu_to_le32 (hmv->magic), PORTALS_PROTO_MAGIC,
2053                         HIPQUAD(conn->ksnc_ipaddr));
2054                 return (-EPROTO);
2055         }
2056
2057         if (hmv->version_major != cpu_to_le16 (PORTALS_PROTO_VERSION_MAJOR) ||
2058             hmv->version_minor != cpu_to_le16 (PORTALS_PROTO_VERSION_MINOR)) {
2059                 CERROR ("Incompatible protocol version %d.%d (%d.%d expected)"
2060                         " from %u.%u.%u.%u\n",
2061                         le16_to_cpu (hmv->version_major),
2062                         le16_to_cpu (hmv->version_minor),
2063                         PORTALS_PROTO_VERSION_MAJOR,
2064                         PORTALS_PROTO_VERSION_MINOR,
2065                         HIPQUAD(conn->ksnc_ipaddr));
2066                 return (-EPROTO);
2067         }
2068
2069 #if (PORTALS_PROTO_VERSION_MAJOR != 1)
2070 # error "This code only understands protocol version 1.x"
2071 #endif
2072         /* version 1 sends magic/version as the dest_nid of a 'hello'
2073          * header, followed by payload full of interface IP addresses.
2074          * Read the rest of it in now... */
2075
2076         rc = ksocknal_sock_read (sock, hmv + 1, sizeof (hdr) - sizeof (*hmv));
2077         if (rc != 0) {
2078                 CERROR ("Error %d reading rest of HELLO hdr from %u.%u.%u.%u\n",
2079                         rc, HIPQUAD(conn->ksnc_ipaddr));
2080                 return (rc);
2081         }
2082
2083         /* ...and check we got what we expected */
2084         if (hdr.type != cpu_to_le32 (PTL_MSG_HELLO)) {
2085                 CERROR ("Expecting a HELLO hdr,"
2086                         " but got type %d from %u.%u.%u.%u\n",
2087                         le32_to_cpu (hdr.type),
2088                         HIPQUAD(conn->ksnc_ipaddr));
2089                 return (-EPROTO);
2090         }
2091
2092         if (le64_to_cpu(hdr.src_nid) == PTL_NID_ANY) {
2093                 CERROR("Expecting a HELLO hdr with a NID, but got PTL_NID_ANY"
2094                        "from %u.%u.%u.%u\n", HIPQUAD(conn->ksnc_ipaddr));
2095                 return (-EPROTO);
2096         }
2097
2098         if (*nid == PTL_NID_ANY) {              /* don't know peer's nid yet */
2099                 *nid = le64_to_cpu(hdr.src_nid);
2100         } else if (*nid != le64_to_cpu (hdr.src_nid)) {
2101                 CERROR ("Connected to nid "LPX64"@%u.%u.%u.%u "
2102                         "but expecting "LPX64"\n",
2103                         le64_to_cpu (hdr.src_nid),
2104                         HIPQUAD(conn->ksnc_ipaddr), *nid);
2105                 return (-EPROTO);
2106         }
2107
2108         type = __le32_to_cpu(hdr.msg.hello.type);
2109
2110         if (conn->ksnc_type == SOCKNAL_CONN_NONE) {
2111                 /* I've accepted this connection; peer determines type */
2112                 conn->ksnc_type = ksocknal_invert_type(type);
2113                 if (conn->ksnc_type == SOCKNAL_CONN_NONE) {
2114                         CERROR ("Unexpected type %d from "LPX64"@%u.%u.%u.%u\n",
2115                                 type, *nid, HIPQUAD(conn->ksnc_ipaddr));
2116                         return (-EPROTO);
2117                 }
2118         } else if (ksocknal_invert_type(type) != conn->ksnc_type) {
2119                 CERROR ("Mismatched types: me %d, "LPX64"@%u.%u.%u.%u %d\n",
2120                         conn->ksnc_type, *nid, HIPQUAD(conn->ksnc_ipaddr),
2121                         le32_to_cpu(hdr.msg.hello.type));
2122                 return (-EPROTO);
2123         }
2124
2125         *incarnation = le64_to_cpu(hdr.msg.hello.incarnation);
2126
2127         nips = __le32_to_cpu (hdr.payload_length) / sizeof (__u32);
2128
2129         if (nips > SOCKNAL_MAX_INTERFACES ||
2130             nips * sizeof(__u32) != __le32_to_cpu (hdr.payload_length)) {
2131                 CERROR("Bad payload length %d from "LPX64"@%u.%u.%u.%u\n",
2132                        __le32_to_cpu (hdr.payload_length),
2133                        *nid, HIPQUAD(conn->ksnc_ipaddr));
2134         }
2135
2136         if (nips == 0)
2137                 return (0);
2138
2139         rc = ksocknal_sock_read (sock, ipaddrs, nips * sizeof(*ipaddrs));
2140         if (rc != 0) {
2141                 CERROR ("Error %d reading IPs from "LPX64"@%u.%u.%u.%u\n",
2142                         rc, *nid, HIPQUAD(conn->ksnc_ipaddr));
2143                 return (rc);
2144         }
2145
2146         for (i = 0; i < nips; i++) {
2147                 ipaddrs[i] = __le32_to_cpu(ipaddrs[i]);
2148
2149                 if (ipaddrs[i] == 0) {
2150                         CERROR("Zero IP[%d] from "LPX64"@%u.%u.%u.%u\n",
2151                                i, *nid, HIPQUAD(conn->ksnc_ipaddr));
2152                         return (-EPROTO);
2153                 }
2154         }
2155
2156         return (nips);
2157 }
2158
2159 int
2160 ksocknal_get_conn_tunables (ksock_conn_t *conn, int *txmem, int *rxmem, int *nagle)
2161 {
2162         mm_segment_t   oldmm = get_fs ();
2163         struct socket *sock = conn->ksnc_sock;
2164         int            len;
2165         int            rc;
2166
2167         rc = ksocknal_getconnsock (conn);
2168         if (rc != 0) {
2169                 LASSERT (conn->ksnc_closing);
2170                 *txmem = *rxmem = *nagle = 0;
2171                 return (-ESHUTDOWN);
2172         }
2173
2174         set_fs (KERNEL_DS);
2175
2176         len = sizeof(*txmem);
2177         rc = sock_getsockopt(sock, SOL_SOCKET, SO_SNDBUF,
2178                              (char *)txmem, &len);
2179         if (rc == 0) {
2180                 len = sizeof(*rxmem);
2181                 rc = sock_getsockopt(sock, SOL_SOCKET, SO_RCVBUF,
2182                                      (char *)rxmem, &len);
2183         }
2184         if (rc == 0) {
2185                 len = sizeof(*nagle);
2186                 rc = sock->ops->getsockopt(sock, SOL_TCP, TCP_NODELAY,
2187                                            (char *)nagle, &len);
2188         }
2189
2190         set_fs (oldmm);
2191         ksocknal_putconnsock (conn);
2192
2193         if (rc == 0)
2194                 *nagle = !*nagle;
2195         else
2196                 *txmem = *rxmem = *nagle = 0;
2197
2198         return (rc);
2199 }
2200
2201 int
2202 ksocknal_setup_sock (struct socket *sock)
2203 {
2204         mm_segment_t    oldmm = get_fs ();
2205         int             rc;
2206         int             option;
2207         int             keep_idle;
2208         int             keep_intvl;
2209         int             keep_count;
2210         int             do_keepalive;
2211         struct linger   linger;
2212
2213         sock->sk->sk_allocation = GFP_NOFS;
2214
2215         /* Ensure this socket aborts active sends immediately when we close
2216          * it. */
2217
2218         linger.l_onoff = 0;
2219         linger.l_linger = 0;
2220
2221         set_fs (KERNEL_DS);
2222         rc = sock_setsockopt (sock, SOL_SOCKET, SO_LINGER,
2223                               (char *)&linger, sizeof (linger));
2224         set_fs (oldmm);
2225         if (rc != 0) {
2226                 CERROR ("Can't set SO_LINGER: %d\n", rc);
2227                 return (rc);
2228         }
2229
2230         option = -1;
2231         set_fs (KERNEL_DS);
2232         rc = sock->ops->setsockopt (sock, SOL_TCP, TCP_LINGER2,
2233                                     (char *)&option, sizeof (option));
2234         set_fs (oldmm);
2235         if (rc != 0) {
2236                 CERROR ("Can't set SO_LINGER2: %d\n", rc);
2237                 return (rc);
2238         }
2239
2240         if (!ksocknal_tunables.ksnd_nagle) {
2241                 option = 1;
2242
2243                 set_fs (KERNEL_DS);
2244                 rc = sock->ops->setsockopt (sock, SOL_TCP, TCP_NODELAY,
2245                                             (char *)&option, sizeof (option));
2246                 set_fs (oldmm);
2247                 if (rc != 0) {
2248                         CERROR ("Can't disable nagle: %d\n", rc);
2249                         return (rc);
2250                 }
2251         }
2252
2253         if (ksocknal_tunables.ksnd_buffer_size > 0) {
2254                 option = ksocknal_tunables.ksnd_buffer_size;
2255
2256                 set_fs (KERNEL_DS);
2257                 rc = sock_setsockopt (sock, SOL_SOCKET, SO_SNDBUF,
2258                                       (char *)&option, sizeof (option));
2259                 set_fs (oldmm);
2260                 if (rc != 0) {
2261                         CERROR ("Can't set send buffer %d: %d\n",
2262                                 option, rc);
2263                         return (rc);
2264                 }
2265
2266                 set_fs (KERNEL_DS);
2267                 rc = sock_setsockopt (sock, SOL_SOCKET, SO_RCVBUF,
2268                                       (char *)&option, sizeof (option));
2269                 set_fs (oldmm);
2270                 if (rc != 0) {
2271                         CERROR ("Can't set receive buffer %d: %d\n",
2272                                 option, rc);
2273                         return (rc);
2274                 }
2275         }
2276
2277         /* snapshot tunables */
2278         keep_idle  = ksocknal_tunables.ksnd_keepalive_idle;
2279         keep_count = ksocknal_tunables.ksnd_keepalive_count;
2280         keep_intvl = ksocknal_tunables.ksnd_keepalive_intvl;
2281
2282         do_keepalive = (keep_idle > 0 && keep_count > 0 && keep_intvl > 0);
2283
2284         option = (do_keepalive ? 1 : 0);
2285         set_fs (KERNEL_DS);
2286         rc = sock_setsockopt (sock, SOL_SOCKET, SO_KEEPALIVE,
2287                               (char *)&option, sizeof (option));
2288         set_fs (oldmm);
2289         if (rc != 0) {
2290                 CERROR ("Can't set SO_KEEPALIVE: %d\n", rc);
2291                 return (rc);
2292         }
2293
2294         if (!do_keepalive)
2295                 return (0);
2296
2297         set_fs (KERNEL_DS);
2298         rc = sock->ops->setsockopt (sock, SOL_TCP, TCP_KEEPIDLE,
2299                                     (char *)&keep_idle, sizeof (keep_idle));
2300         set_fs (oldmm);
2301         if (rc != 0) {
2302                 CERROR ("Can't set TCP_KEEPIDLE: %d\n", rc);
2303                 return (rc);
2304         }
2305
2306         set_fs (KERNEL_DS);
2307         rc = sock->ops->setsockopt (sock, SOL_TCP, TCP_KEEPINTVL,
2308                                     (char *)&keep_intvl, sizeof (keep_intvl));
2309         set_fs (oldmm);
2310         if (rc != 0) {
2311                 CERROR ("Can't set TCP_KEEPINTVL: %d\n", rc);
2312                 return (rc);
2313         }
2314
2315         set_fs (KERNEL_DS);
2316         rc = sock->ops->setsockopt (sock, SOL_TCP, TCP_KEEPCNT,
2317                                     (char *)&keep_count, sizeof (keep_count));
2318         set_fs (oldmm);
2319         if (rc != 0) {
2320                 CERROR ("Can't set TCP_KEEPCNT: %d\n", rc);
2321                 return (rc);
2322         }
2323
2324         return (0);
2325 }
2326
2327 static int
2328 ksocknal_connect_sock(struct socket **sockp, int *may_retry,
2329                       ksock_route_t *route, int local_port)
2330 {
2331         struct sockaddr_in  locaddr;
2332         struct sockaddr_in  srvaddr;
2333         struct socket      *sock;
2334         int                 rc;
2335         int                 option;
2336         mm_segment_t        oldmm = get_fs();
2337         struct timeval      tv;
2338
2339         memset(&locaddr, 0, sizeof(locaddr));
2340         locaddr.sin_family = AF_INET;
2341         locaddr.sin_port = htons(local_port);
2342         locaddr.sin_addr.s_addr =
2343                 (route->ksnr_myipaddr != 0) ? htonl(route->ksnr_myipaddr)
2344                                             : INADDR_ANY;
2345
2346         memset (&srvaddr, 0, sizeof (srvaddr));
2347         srvaddr.sin_family = AF_INET;
2348         srvaddr.sin_port = htons (route->ksnr_port);
2349         srvaddr.sin_addr.s_addr = htonl (route->ksnr_ipaddr);
2350
2351         *may_retry = 0;
2352
2353         rc = sock_create (PF_INET, SOCK_STREAM, 0, &sock);
2354         *sockp = sock;
2355         if (rc != 0) {
2356                 CERROR ("Can't create autoconnect socket: %d\n", rc);
2357                 return (rc);
2358         }
2359
2360         /* Ugh; have to map_fd for compatibility with sockets passed in
2361          * from userspace.  And we actually need the sock->file refcounting
2362          * that this gives you :) */
2363
2364         rc = sock_map_fd (sock);
2365         if (rc < 0) {
2366                 sock_release (sock);
2367                 CERROR ("sock_map_fd error %d\n", rc);
2368                 return (rc);
2369         }
2370
2371         /* NB the file descriptor (rc) now owns the ref on sock->file */
2372         LASSERT (sock->file != NULL);
2373         LASSERT (file_count(sock->file) == 1);
2374
2375         get_file(sock->file);                /* extra ref makes sock->file */
2376         sys_close(rc);                       /* survive this close */
2377
2378         /* Still got a single ref on sock->file */
2379         LASSERT (file_count(sock->file) == 1);
2380
2381         /* Set the socket timeouts, so our connection attempt completes in
2382          * finite time */
2383         tv.tv_sec = ksocknal_tunables.ksnd_io_timeout;
2384         tv.tv_usec = 0;
2385
2386         set_fs (KERNEL_DS);
2387         rc = sock_setsockopt (sock, SOL_SOCKET, SO_SNDTIMEO,
2388                               (char *)&tv, sizeof (tv));
2389         set_fs (oldmm);
2390         if (rc != 0) {
2391                 CERROR ("Can't set send timeout %d: %d\n",
2392                         ksocknal_tunables.ksnd_io_timeout, rc);
2393                 goto failed;
2394         }
2395
2396         set_fs (KERNEL_DS);
2397         rc = sock_setsockopt (sock, SOL_SOCKET, SO_RCVTIMEO,
2398                               (char *)&tv, sizeof (tv));
2399         set_fs (oldmm);
2400         if (rc != 0) {
2401                 CERROR ("Can't set receive timeout %d: %d\n",
2402                         ksocknal_tunables.ksnd_io_timeout, rc);
2403                 goto failed;
2404         }
2405
2406         set_fs (KERNEL_DS);
2407         option = 1;
2408         rc = sock_setsockopt(sock, SOL_SOCKET, SO_REUSEADDR,
2409                              (char *)&option, sizeof (option));
2410         set_fs (oldmm);
2411         if (rc != 0) {
2412                 CERROR("Can't set SO_REUSEADDR for socket: %d\n", rc);
2413                 goto failed;
2414         }
2415
2416         rc = sock->ops->bind(sock,
2417                              (struct sockaddr *)&locaddr, sizeof(locaddr));
2418         if (rc == -EADDRINUSE) {
2419                 CDEBUG(D_NET, "Port %d already in use\n", local_port);
2420                 *may_retry = 1;
2421                 goto failed;
2422         }
2423         if (rc != 0) {
2424                 CERROR("Error trying to bind to reserved port %d: %d\n",
2425                        local_port, rc);
2426                 goto failed;
2427         }
2428
2429         rc = sock->ops->connect(sock,
2430                                 (struct sockaddr *)&srvaddr, sizeof(srvaddr),
2431                                 sock->file->f_flags);
2432         if (rc == 0)
2433                 return 0;
2434
2435         /* EADDRNOTAVAIL probably means we're already connected to the same
2436          * peer/port on the same local port on a differently typed
2437          * connection.  Let our caller retry with a different local
2438          * port... */
2439         *may_retry = (rc == -EADDRNOTAVAIL);
2440
2441         CDEBUG(*may_retry ? D_NET : D_ERROR,
2442                "Error %d connecting %u.%u.%u.%u/%d -> %u.%u.%u.%u/%d\n", rc,
2443                HIPQUAD(route->ksnr_myipaddr), local_port,
2444                HIPQUAD(route->ksnr_ipaddr), route->ksnr_port);
2445
2446  failed:
2447         fput(sock->file);
2448         return rc;
2449 }
2450
2451 int
2452 ksocknal_connect_peer (ksock_route_t *route, int type)
2453 {
2454         struct socket      *sock;
2455         int                 rc;
2456         int                 port;
2457         int                 may_retry;
2458
2459         /* Iterate through reserved ports.  When typed connections are
2460          * used, we will need to bind to multiple ports, but we only know
2461          * this at connect time.  But, by that time we've already called
2462          * bind() so we need a new socket. */
2463
2464         for (port = 1023; port > 512; --port) {
2465
2466                 rc = ksocknal_connect_sock(&sock, &may_retry, route, port);
2467
2468                 if (rc == 0) {
2469                         rc = ksocknal_create_conn(route, sock, type);
2470                         fput(sock->file);
2471                         return rc;
2472                 }
2473
2474                 if (!may_retry)
2475                         return rc;
2476         }
2477
2478         CERROR("Out of ports trying to bind to a reserved port\n");
2479         return (-EADDRINUSE);
2480 }
2481
2482 void
2483 ksocknal_autoconnect (ksock_route_t *route)
2484 {
2485         LIST_HEAD        (zombies);
2486         ksock_tx_t       *tx;
2487         ksock_peer_t     *peer;
2488         unsigned long     flags;
2489         int               rc;
2490         int               type;
2491
2492         for (;;) {
2493                 for (type = 0; type < SOCKNAL_CONN_NTYPES; type++)
2494                         if ((route->ksnr_connecting & (1 << type)) != 0)
2495                                 break;
2496                 LASSERT (type < SOCKNAL_CONN_NTYPES);
2497
2498                 rc = ksocknal_connect_peer (route, type);
2499                 if (rc != 0)
2500                         break;
2501
2502                 /* successfully autoconnected: create_conn did the
2503                  * route/conn binding and scheduled any blocked packets */
2504
2505                 if (route->ksnr_connecting == 0) {
2506                         /* No more connections required */
2507                         return;
2508                 }
2509         }
2510
2511         /* Connection attempt failed */
2512
2513         write_lock_irqsave (&ksocknal_data.ksnd_global_lock, flags);
2514
2515         peer = route->ksnr_peer;
2516         route->ksnr_connecting = 0;
2517
2518         /* This is a retry rather than a new connection */
2519         LASSERT (route->ksnr_retry_interval != 0);
2520         route->ksnr_timeout = jiffies + route->ksnr_retry_interval;
2521         route->ksnr_retry_interval = MIN (route->ksnr_retry_interval * 2,
2522                                           SOCKNAL_MAX_RECONNECT_INTERVAL);
2523
2524         if (!list_empty (&peer->ksnp_tx_queue) &&
2525             ksocknal_find_connecting_route_locked (peer) == NULL) {
2526                 LASSERT (list_empty (&peer->ksnp_conns));
2527
2528                 /* None of the connections that the blocked packets are
2529                  * waiting for have been successful.  Complete them now... */
2530                 do {
2531                         tx = list_entry (peer->ksnp_tx_queue.next,
2532                                          ksock_tx_t, tx_list);
2533                         list_del (&tx->tx_list);
2534                         list_add_tail (&tx->tx_list, &zombies);
2535                 } while (!list_empty (&peer->ksnp_tx_queue));
2536         }
2537
2538 #if 0           /* irrelevent with only eager routes */
2539         if (!route->ksnr_deleted) {
2540                 /* make this route least-favourite for re-selection */
2541                 list_del(&route->ksnr_list);
2542                 list_add_tail(&route->ksnr_list, &peer->ksnp_routes);
2543         }
2544 #endif
2545         write_unlock_irqrestore (&ksocknal_data.ksnd_global_lock, flags);
2546
2547         while (!list_empty (&zombies)) {
2548                 char ipbuf[PTL_NALFMT_SIZE];
2549                 char ipbuf2[PTL_NALFMT_SIZE];
2550                 tx = list_entry (zombies.next, ksock_tx_t, tx_list);
2551
2552                 CERROR ("Deleting packet type %d len %d ("LPX64" %s->"LPX64" %s)\n",
2553                         le32_to_cpu (tx->tx_hdr->type),
2554                         le32_to_cpu (tx->tx_hdr->payload_length),
2555                         le64_to_cpu (tx->tx_hdr->src_nid),
2556                         portals_nid2str(SOCKNAL,
2557                                         le64_to_cpu(tx->tx_hdr->src_nid),
2558                                         ipbuf),
2559                         le64_to_cpu (tx->tx_hdr->dest_nid),
2560                         portals_nid2str(SOCKNAL,
2561                                         le64_to_cpu(tx->tx_hdr->src_nid),
2562                                         ipbuf2));
2563
2564                 list_del (&tx->tx_list);
2565                 /* complete now */
2566                 ksocknal_tx_done (tx, 0);
2567         }
2568 }
2569
2570 int
2571 ksocknal_autoconnectd (void *arg)
2572 {
2573         long               id = (long)arg;
2574         char               name[16];
2575         unsigned long      flags;
2576         ksock_route_t     *route;
2577         int                rc;
2578
2579         snprintf (name, sizeof (name), "ksocknal_ad%02ld", id);
2580         kportal_daemonize (name);
2581         kportal_blockallsigs ();
2582
2583         spin_lock_irqsave (&ksocknal_data.ksnd_autoconnectd_lock, flags);
2584
2585         while (!ksocknal_data.ksnd_shuttingdown) {
2586
2587                 if (!list_empty (&ksocknal_data.ksnd_autoconnectd_routes)) {
2588                         route = list_entry (ksocknal_data.ksnd_autoconnectd_routes.next,
2589                                             ksock_route_t, ksnr_connect_list);
2590
2591                         list_del (&route->ksnr_connect_list);
2592                         spin_unlock_irqrestore (&ksocknal_data.ksnd_autoconnectd_lock, flags);
2593
2594                         ksocknal_autoconnect (route);
2595                         ksocknal_put_route (route);
2596
2597                         spin_lock_irqsave(&ksocknal_data.ksnd_autoconnectd_lock,
2598                                           flags);
2599                         continue;
2600                 }
2601
2602                 spin_unlock_irqrestore(&ksocknal_data.ksnd_autoconnectd_lock,
2603                                        flags);
2604
2605                 rc = wait_event_interruptible(ksocknal_data.ksnd_autoconnectd_waitq,
2606                                               ksocknal_data.ksnd_shuttingdown ||
2607                                               !list_empty(&ksocknal_data.ksnd_autoconnectd_routes));
2608
2609                 spin_lock_irqsave(&ksocknal_data.ksnd_autoconnectd_lock, flags);
2610         }
2611
2612         spin_unlock_irqrestore (&ksocknal_data.ksnd_autoconnectd_lock, flags);
2613
2614         ksocknal_thread_fini ();
2615         return (0);
2616 }
2617
2618 ksock_conn_t *
2619 ksocknal_find_timed_out_conn (ksock_peer_t *peer)
2620 {
2621         /* We're called with a shared lock on ksnd_global_lock */
2622         ksock_conn_t      *conn;
2623         struct list_head  *ctmp;
2624
2625         list_for_each (ctmp, &peer->ksnp_conns) {
2626                 conn = list_entry (ctmp, ksock_conn_t, ksnc_list);
2627
2628                 /* Don't need the {get,put}connsock dance to deref ksnc_sock... */
2629                 LASSERT (!conn->ksnc_closing);
2630
2631                 if (conn->ksnc_sock->sk->sk_err != 0) {
2632                         /* Something (e.g. failed keepalive) set the socket error */
2633                         atomic_inc (&conn->ksnc_refcount);
2634                         CERROR ("Socket error %d: "LPX64" %p %d.%d.%d.%d\n",
2635                                 conn->ksnc_sock->sk->sk_err, peer->ksnp_nid,
2636                                 conn, HIPQUAD(conn->ksnc_ipaddr));
2637                         return (conn);
2638                 }
2639
2640                 if (conn->ksnc_rx_started &&
2641                     time_after_eq (jiffies, conn->ksnc_rx_deadline)) {
2642                         /* Timed out incomplete incoming message */
2643                         atomic_inc (&conn->ksnc_refcount);
2644                         CERROR ("Timed out RX from "LPX64" %p %d.%d.%d.%d\n",
2645                                 peer->ksnp_nid,conn,HIPQUAD(conn->ksnc_ipaddr));
2646                         return (conn);
2647                 }
2648
2649                 if ((!list_empty (&conn->ksnc_tx_queue) ||
2650                      conn->ksnc_sock->sk->sk_wmem_queued != 0) &&
2651                     time_after_eq (jiffies, conn->ksnc_tx_deadline)) {
2652                         /* Timed out messages queued for sending or
2653                          * buffered in the socket's send buffer */
2654                         atomic_inc (&conn->ksnc_refcount);
2655                         CERROR ("Timed out TX to "LPX64" %s%d %p %d.%d.%d.%d\n",
2656                                 peer->ksnp_nid,
2657                                 list_empty (&conn->ksnc_tx_queue) ? "" : "Q ",
2658                                 conn->ksnc_sock->sk->sk_wmem_queued, conn,
2659                                 HIPQUAD(conn->ksnc_ipaddr));
2660                         return (conn);
2661                 }
2662         }
2663
2664         return (NULL);
2665 }
2666
2667 void
2668 ksocknal_check_peer_timeouts (int idx)
2669 {
2670         struct list_head *peers = &ksocknal_data.ksnd_peers[idx];
2671         struct list_head *ptmp;
2672         ksock_peer_t     *peer;
2673         ksock_conn_t     *conn;
2674
2675  again:
2676         /* NB. We expect to have a look at all the peers and not find any
2677          * connections to time out, so we just use a shared lock while we
2678          * take a look... */
2679         read_lock (&ksocknal_data.ksnd_global_lock);
2680
2681         list_for_each (ptmp, peers) {
2682                 peer = list_entry (ptmp, ksock_peer_t, ksnp_list);
2683                 conn = ksocknal_find_timed_out_conn (peer);
2684
2685                 if (conn != NULL) {
2686                         read_unlock (&ksocknal_data.ksnd_global_lock);
2687
2688                         CERROR ("Timeout out conn->"LPX64" ip %d.%d.%d.%d:%d\n",
2689                                 peer->ksnp_nid,
2690                                 HIPQUAD(conn->ksnc_ipaddr),
2691                                 conn->ksnc_port);
2692                         ksocknal_close_conn_and_siblings (conn, -ETIMEDOUT);
2693
2694                         /* NB we won't find this one again, but we can't
2695                          * just proceed with the next peer, since we dropped
2696                          * ksnd_global_lock and it might be dead already! */
2697                         ksocknal_put_conn (conn);
2698                         goto again;
2699                 }
2700         }
2701
2702         read_unlock (&ksocknal_data.ksnd_global_lock);
2703 }
2704
2705 int
2706 ksocknal_reaper (void *arg)
2707 {
2708         wait_queue_t       wait;
2709         unsigned long      flags;
2710         ksock_conn_t      *conn;
2711         ksock_sched_t     *sched;
2712         struct list_head   enomem_conns;
2713         int                nenomem_conns;
2714         int                timeout;
2715         int                i;
2716         int                peer_index = 0;
2717         unsigned long      deadline = jiffies;
2718
2719         kportal_daemonize ("ksocknal_reaper");
2720         kportal_blockallsigs ();
2721
2722         INIT_LIST_HEAD(&enomem_conns);
2723         init_waitqueue_entry (&wait, current);
2724
2725         spin_lock_irqsave (&ksocknal_data.ksnd_reaper_lock, flags);
2726
2727         while (!ksocknal_data.ksnd_shuttingdown) {
2728
2729                 if (!list_empty (&ksocknal_data.ksnd_deathrow_conns)) {
2730                         conn = list_entry (ksocknal_data.ksnd_deathrow_conns.next,
2731                                            ksock_conn_t, ksnc_list);
2732                         list_del (&conn->ksnc_list);
2733
2734                         spin_unlock_irqrestore (&ksocknal_data.ksnd_reaper_lock, flags);
2735
2736                         ksocknal_terminate_conn (conn);
2737                         ksocknal_put_conn (conn);
2738
2739                         spin_lock_irqsave (&ksocknal_data.ksnd_reaper_lock, flags);
2740                         continue;
2741                 }
2742
2743                 if (!list_empty (&ksocknal_data.ksnd_zombie_conns)) {
2744                         conn = list_entry (ksocknal_data.ksnd_zombie_conns.next,
2745                                            ksock_conn_t, ksnc_list);
2746                         list_del (&conn->ksnc_list);
2747
2748                         spin_unlock_irqrestore (&ksocknal_data.ksnd_reaper_lock, flags);
2749
2750                         ksocknal_destroy_conn (conn);
2751
2752                         spin_lock_irqsave (&ksocknal_data.ksnd_reaper_lock, flags);
2753                         continue;
2754                 }
2755
2756                 if (!list_empty (&ksocknal_data.ksnd_enomem_conns)) {
2757                         list_add(&enomem_conns, &ksocknal_data.ksnd_enomem_conns);
2758                         list_del_init(&ksocknal_data.ksnd_enomem_conns);
2759                 }
2760
2761                 spin_unlock_irqrestore (&ksocknal_data.ksnd_reaper_lock, flags);
2762
2763                 /* reschedule all the connections that stalled with ENOMEM... */
2764                 nenomem_conns = 0;
2765                 while (!list_empty (&enomem_conns)) {
2766                         conn = list_entry (enomem_conns.next,
2767                                            ksock_conn_t, ksnc_tx_list);
2768                         list_del (&conn->ksnc_tx_list);
2769
2770                         sched = conn->ksnc_scheduler;
2771
2772                         spin_lock_irqsave (&sched->kss_lock, flags);
2773
2774                         LASSERT (conn->ksnc_tx_scheduled);
2775                         conn->ksnc_tx_ready = 1;
2776                         list_add_tail (&conn->ksnc_tx_list, &sched->kss_tx_conns);
2777                         wake_up (&sched->kss_waitq);
2778
2779                         spin_unlock_irqrestore (&sched->kss_lock, flags);
2780                         nenomem_conns++;
2781                 }
2782
2783                 /* careful with the jiffy wrap... */
2784                 while ((timeout = (int)(deadline - jiffies)) <= 0) {
2785                         const int n = 4;
2786                         const int p = 1;
2787                         int       chunk = ksocknal_data.ksnd_peer_hash_size;
2788
2789                         /* Time to check for timeouts on a few more peers: I do
2790                          * checks every 'p' seconds on a proportion of the peer
2791                          * table and I need to check every connection 'n' times
2792                          * within a timeout interval, to ensure I detect a
2793                          * timeout on any connection within (n+1)/n times the
2794                          * timeout interval. */
2795
2796                         if (ksocknal_tunables.ksnd_io_timeout > n * p)
2797                                 chunk = (chunk * n * p) /
2798                                         ksocknal_tunables.ksnd_io_timeout;
2799                         if (chunk == 0)
2800                                 chunk = 1;
2801
2802                         for (i = 0; i < chunk; i++) {
2803                                 ksocknal_check_peer_timeouts (peer_index);
2804                                 peer_index = (peer_index + 1) %
2805                                              ksocknal_data.ksnd_peer_hash_size;
2806                         }
2807
2808                         deadline += p * HZ;
2809                 }
2810
2811                 if (nenomem_conns != 0) {
2812                         /* Reduce my timeout if I rescheduled ENOMEM conns.
2813                          * This also prevents me getting woken immediately
2814                          * if any go back on my enomem list. */
2815                         timeout = SOCKNAL_ENOMEM_RETRY;
2816                 }
2817                 ksocknal_data.ksnd_reaper_waketime = jiffies + timeout;
2818
2819                 set_current_state (TASK_INTERRUPTIBLE);
2820                 add_wait_queue (&ksocknal_data.ksnd_reaper_waitq, &wait);
2821
2822                 if (!ksocknal_data.ksnd_shuttingdown &&
2823                     list_empty (&ksocknal_data.ksnd_deathrow_conns) &&
2824                     list_empty (&ksocknal_data.ksnd_zombie_conns))
2825                         schedule_timeout (timeout);
2826
2827                 set_current_state (TASK_RUNNING);
2828                 remove_wait_queue (&ksocknal_data.ksnd_reaper_waitq, &wait);
2829
2830                 spin_lock_irqsave (&ksocknal_data.ksnd_reaper_lock, flags);
2831         }
2832
2833         spin_unlock_irqrestore (&ksocknal_data.ksnd_reaper_lock, flags);
2834
2835         ksocknal_thread_fini ();
2836         return (0);
2837 }
2838
2839 lib_nal_t ksocknal_lib = {
2840         libnal_data:       &ksocknal_data,      /* NAL private data */
2841         libnal_send:        ksocknal_send,
2842         libnal_send_pages:  ksocknal_send_pages,
2843         libnal_recv:        ksocknal_recv,
2844         libnal_recv_pages:  ksocknal_recv_pages,
2845         libnal_dist:        ksocknal_dist
2846 };