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