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