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