drivers/staging/lustre/lnet/klnds/socklnd/socklnd_cb.c

   1 /*
   2  * Copyright (c) 2003, 2010, Oracle and/or its affiliates. All rights reserved.
   3  *
   4  * Copyright (c) 2011, 2012, Intel Corporation.
   5  *
   6  *   Author: Zach Brown <zab@zabbo.net>
   7  *   Author: Peter J. Braam <braam@clusterfs.com>
   8  *   Author: Phil Schwan <phil@clusterfs.com>
   9  *   Author: Eric Barton <eric@bartonsoftware.com>
  10  *
  11  *   This file is part of Portals, http://www.sf.net/projects/sandiaportals/
  12  *
  13  *   Portals is free software; you can redistribute it and/or
  14  *   modify it under the terms of version 2 of the GNU General Public
  15  *   License as published by the Free Software Foundation.
  16  *
  17  *   Portals is distributed in the hope that it will be useful,
  18  *   but WITHOUT ANY WARRANTY; without even the implied warranty of
  19  *   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  20  *   GNU General Public License for more details.
  21  *
  22  *   You should have received a copy of the GNU General Public License
  23  *   along with Portals; if not, write to the Free Software
  24  *   Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
  25  */
  26
  27 #include "socklnd.h"
  28
  29 ksock_tx_t *
  30 ksocknal_alloc_tx(int type, int size)
  31 {
  32         ksock_tx_t *tx = NULL;
  33
  34         if (type == KSOCK_MSG_NOOP) {
  35                 LASSERT(size == KSOCK_NOOP_TX_SIZE);
  36
  37                 /* searching for a noop tx in free list */
  38                 spin_lock(&ksocknal_data.ksnd_tx_lock);
  39
  40                 if (!list_empty(&ksocknal_data.ksnd_idle_noop_txs)) {
  41                         tx = list_entry(ksocknal_data.ksnd_idle_noop_txs. \
  42                                             next, ksock_tx_t, tx_list);
  43                         LASSERT(tx->tx_desc_size == size);
  44                         list_del(&tx->tx_list);
  45                 }
  46
  47                 spin_unlock(&ksocknal_data.ksnd_tx_lock);
  48         }
  49
  50         if (tx == NULL)
  51                 LIBCFS_ALLOC(tx, size);
  52
  53         if (tx == NULL)
  54                 return NULL;
  55
  56         atomic_set(&tx->tx_refcount, 1);
  57         tx->tx_zc_aborted = 0;
  58         tx->tx_zc_capable = 0;
  59         tx->tx_zc_checked = 0;
  60         tx->tx_desc_size  = size;
  61
  62         atomic_inc(&ksocknal_data.ksnd_nactive_txs);
  63
  64         return tx;
  65 }
  66
  67 ksock_tx_t *
  68 ksocknal_alloc_tx_noop(__u64 cookie, int nonblk)
  69 {
  70         ksock_tx_t *tx;
  71
  72         tx = ksocknal_alloc_tx(KSOCK_MSG_NOOP, KSOCK_NOOP_TX_SIZE);
  73         if (tx == NULL) {
  74                 CERROR("Can't allocate noop tx desc\n");
  75                 return NULL;
  76         }
  77
  78         tx->tx_conn     = NULL;
  79         tx->tx_lnetmsg  = NULL;
  80         tx->tx_kiov     = NULL;
  81         tx->tx_nkiov    = 0;
  82         tx->tx_iov      = tx->tx_frags.virt.iov;
  83         tx->tx_niov     = 1;
  84         tx->tx_nonblk   = nonblk;
  85
  86         socklnd_init_msg(&tx->tx_msg, KSOCK_MSG_NOOP);
  87         tx->tx_msg.ksm_zc_cookies[1] = cookie;
  88
  89         return tx;
  90 }
  91
  92
  93 void
  94 ksocknal_free_tx (ksock_tx_t *tx)
  95 {
  96         atomic_dec(&ksocknal_data.ksnd_nactive_txs);
  97
  98         if (tx->tx_lnetmsg == NULL && tx->tx_desc_size == KSOCK_NOOP_TX_SIZE) {
  99                 /* it's a noop tx */
 100                 spin_lock(&ksocknal_data.ksnd_tx_lock);
 101
 102                 list_add(&tx->tx_list, &ksocknal_data.ksnd_idle_noop_txs);
 103
 104                 spin_unlock(&ksocknal_data.ksnd_tx_lock);
 105         } else {
 106                 LIBCFS_FREE(tx, tx->tx_desc_size);
 107         }
 108 }
 109
 110 static int
 111 ksocknal_send_iov (ksock_conn_t *conn, ksock_tx_t *tx)
 112 {
 113         struct iovec  *iov = tx->tx_iov;
 114         int    nob;
 115         int    rc;
 116
 117         LASSERT (tx->tx_niov > 0);
 118
 119         /* Never touch tx->tx_iov inside ksocknal_lib_send_iov() */
 120         rc = ksocknal_lib_send_iov(conn, tx);
 121
 122         if (rc <= 0)                        /* sent nothing? */
 123                 return rc;
 124
 125         nob = rc;
 126         LASSERT (nob <= tx->tx_resid);
 127         tx->tx_resid -= nob;
 128
 129         /* "consume" iov */
 130         do {
 131                 LASSERT (tx->tx_niov > 0);
 132
 133                 if (nob < (int) iov->iov_len) {
 134                         iov->iov_base = (void *)((char *)iov->iov_base + nob);
 135                         iov->iov_len -= nob;
 136                         return rc;
 137                 }
 138
 139                 nob -= iov->iov_len;
 140                 tx->tx_iov = ++iov;
 141                 tx->tx_niov--;
 142         } while (nob != 0);
 143
 144         return rc;
 145 }
 146
 147 static int
 148 ksocknal_send_kiov (ksock_conn_t *conn, ksock_tx_t *tx)
 149 {
 150         lnet_kiov_t    *kiov = tx->tx_kiov;
 151         int     nob;
 152         int     rc;
 153
 154         LASSERT (tx->tx_niov == 0);
 155         LASSERT (tx->tx_nkiov > 0);
 156
 157         /* Never touch tx->tx_kiov inside ksocknal_lib_send_kiov() */
 158         rc = ksocknal_lib_send_kiov(conn, tx);
 159
 160         if (rc <= 0)                        /* sent nothing? */
 161                 return rc;
 162
 163         nob = rc;
 164         LASSERT (nob <= tx->tx_resid);
 165         tx->tx_resid -= nob;
 166
 167         /* "consume" kiov */
 168         do {
 169                 LASSERT(tx->tx_nkiov > 0);
 170
 171                 if (nob < (int)kiov->kiov_len) {
 172                         kiov->kiov_offset += nob;
 173                         kiov->kiov_len -= nob;
 174                         return rc;
 175                 }
 176
 177                 nob -= (int)kiov->kiov_len;
 178                 tx->tx_kiov = ++kiov;
 179                 tx->tx_nkiov--;
 180         } while (nob != 0);
 181
 182         return rc;
 183 }
 184
 185 static int
 186 ksocknal_transmit (ksock_conn_t *conn, ksock_tx_t *tx)
 187 {
 188         int      rc;
 189         int      bufnob;
 190
 191         if (ksocknal_data.ksnd_stall_tx != 0) {
 192                 set_current_state(TASK_UNINTERRUPTIBLE);
 193                 schedule_timeout(cfs_time_seconds(ksocknal_data.ksnd_stall_tx));
 194         }
 195
 196         LASSERT (tx->tx_resid != 0);
 197
 198         rc = ksocknal_connsock_addref(conn);
 199         if (rc != 0) {
 200                 LASSERT (conn->ksnc_closing);
 201                 return -ESHUTDOWN;
 202         }
 203
 204         do {
 205                 if (ksocknal_data.ksnd_enomem_tx > 0) {
 206                         /* testing... */
 207                         ksocknal_data.ksnd_enomem_tx--;
 208                         rc = -EAGAIN;
 209                 } else if (tx->tx_niov != 0) {
 210                         rc = ksocknal_send_iov (conn, tx);
 211                 } else {
 212                         rc = ksocknal_send_kiov (conn, tx);
 213                 }
 214
 215                 bufnob = conn->ksnc_sock->sk->sk_wmem_queued;
 216                 if (rc > 0)                  /* sent something? */
 217                         conn->ksnc_tx_bufnob += rc; /* account it */
 218
 219                 if (bufnob < conn->ksnc_tx_bufnob) {
 220                         /* allocated send buffer bytes < computed; infer
 221                          * something got ACKed */
 222                         conn->ksnc_tx_deadline =
 223                                 cfs_time_shift(*ksocknal_tunables.ksnd_timeout);
 224                         conn->ksnc_peer->ksnp_last_alive = cfs_time_current();
 225                         conn->ksnc_tx_bufnob = bufnob;
 226                         mb();
 227                 }
 228
 229                 if (rc <= 0) { /* Didn't write anything? */
 230
 231                         if (rc == 0) /* some stacks return 0 instead of -EAGAIN */
 232                                 rc = -EAGAIN;
 233
 234                         /* Check if EAGAIN is due to memory pressure */
 235                         if (rc == -EAGAIN && ksocknal_lib_memory_pressure(conn))
 236                                 rc = -ENOMEM;
 237
 238                         break;
 239                 }
 240
 241                 /* socket's wmem_queued now includes 'rc' bytes */
 242                 atomic_sub (rc, &conn->ksnc_tx_nob);
 243                 rc = 0;
 244
 245         } while (tx->tx_resid != 0);
 246
 247         ksocknal_connsock_decref(conn);
 248         return rc;
 249 }
 250
 251 static int
 252 ksocknal_recv_iov (ksock_conn_t *conn)
 253 {
 254         struct iovec *iov = conn->ksnc_rx_iov;
 255         int     nob;
 256         int     rc;
 257
 258         LASSERT (conn->ksnc_rx_niov > 0);
 259
 260         /* Never touch conn->ksnc_rx_iov or change connection
 261          * status inside ksocknal_lib_recv_iov */
 262         rc = ksocknal_lib_recv_iov(conn);
 263
 264         if (rc <= 0)
 265                 return rc;
 266
 267         /* received something... */
 268         nob = rc;
 269
 270         conn->ksnc_peer->ksnp_last_alive = cfs_time_current();
 271         conn->ksnc_rx_deadline =
 272                 cfs_time_shift(*ksocknal_tunables.ksnd_timeout);
 273         mb();                  /* order with setting rx_started */
 274         conn->ksnc_rx_started = 1;
 275
 276         conn->ksnc_rx_nob_wanted -= nob;
 277         conn->ksnc_rx_nob_left -= nob;
 278
 279         do {
 280                 LASSERT (conn->ksnc_rx_niov > 0);
 281
 282                 if (nob < (int)iov->iov_len) {
 283                         iov->iov_len -= nob;
 284                         iov->iov_base += nob;
 285                         return -EAGAIN;
 286                 }
 287
 288                 nob -= iov->iov_len;
 289                 conn->ksnc_rx_iov = ++iov;
 290                 conn->ksnc_rx_niov--;
 291         } while (nob != 0);
 292
 293         return rc;
 294 }
 295
 296 static int
 297 ksocknal_recv_kiov (ksock_conn_t *conn)
 298 {
 299         lnet_kiov_t   *kiov = conn->ksnc_rx_kiov;
 300         int     nob;
 301         int     rc;
 302         LASSERT (conn->ksnc_rx_nkiov > 0);
 303
 304         /* Never touch conn->ksnc_rx_kiov or change connection
 305          * status inside ksocknal_lib_recv_iov */
 306         rc = ksocknal_lib_recv_kiov(conn);
 307
 308         if (rc <= 0)
 309                 return rc;
 310
 311         /* received something... */
 312         nob = rc;
 313
 314         conn->ksnc_peer->ksnp_last_alive = cfs_time_current();
 315         conn->ksnc_rx_deadline =
 316                 cfs_time_shift(*ksocknal_tunables.ksnd_timeout);
 317         mb();                  /* order with setting rx_started */
 318         conn->ksnc_rx_started = 1;
 319
 320         conn->ksnc_rx_nob_wanted -= nob;
 321         conn->ksnc_rx_nob_left -= nob;
 322
 323         do {
 324                 LASSERT (conn->ksnc_rx_nkiov > 0);
 325
 326                 if (nob < (int) kiov->kiov_len) {
 327                         kiov->kiov_offset += nob;
 328                         kiov->kiov_len -= nob;
 329                         return -EAGAIN;
 330                 }
 331
 332                 nob -= kiov->kiov_len;
 333                 conn->ksnc_rx_kiov = ++kiov;
 334                 conn->ksnc_rx_nkiov--;
 335         } while (nob != 0);
 336
 337         return 1;
 338 }
 339
 340 static int
 341 ksocknal_receive (ksock_conn_t *conn)
 342 {
 343         /* Return 1 on success, 0 on EOF, < 0 on error.
 344          * Caller checks ksnc_rx_nob_wanted to determine
 345          * progress/completion. */
 346         int     rc;
 347
 348         if (ksocknal_data.ksnd_stall_rx != 0) {
 349                 set_current_state(TASK_UNINTERRUPTIBLE);
 350                 schedule_timeout(cfs_time_seconds(ksocknal_data.ksnd_stall_rx));
 351         }
 352
 353         rc = ksocknal_connsock_addref(conn);
 354         if (rc != 0) {
 355                 LASSERT (conn->ksnc_closing);
 356                 return -ESHUTDOWN;
 357         }
 358
 359         for (;;) {
 360                 if (conn->ksnc_rx_niov != 0)
 361                         rc = ksocknal_recv_iov (conn);
 362                 else
 363                         rc = ksocknal_recv_kiov (conn);
 364
 365                 if (rc <= 0) {
 366                         /* error/EOF or partial receive */
 367                         if (rc == -EAGAIN) {
 368                                 rc = 1;
 369                         } else if (rc == 0 && conn->ksnc_rx_started) {
 370                                 /* EOF in the middle of a message */
 371                                 rc = -EPROTO;
 372                         }
 373                         break;
 374                 }
 375
 376                 /* Completed a fragment */
 377
 378                 if (conn->ksnc_rx_nob_wanted == 0) {
 379                         rc = 1;
 380                         break;
 381                 }
 382         }
 383
 384         ksocknal_connsock_decref(conn);
 385         return rc;
 386 }
 387
 388 void
 389 ksocknal_tx_done (lnet_ni_t *ni, ksock_tx_t *tx)
 390 {
 391         lnet_msg_t  *lnetmsg = tx->tx_lnetmsg;
 392         int       rc = (tx->tx_resid == 0 && !tx->tx_zc_aborted) ? 0 : -EIO;
 393
 394         LASSERT(ni != NULL || tx->tx_conn != NULL);
 395
 396         if (tx->tx_conn != NULL)
 397                 ksocknal_conn_decref(tx->tx_conn);
 398
 399         if (ni == NULL && tx->tx_conn != NULL)
 400                 ni = tx->tx_conn->ksnc_peer->ksnp_ni;
 401
 402         ksocknal_free_tx (tx);
 403         if (lnetmsg != NULL) /* KSOCK_MSG_NOOP go without lnetmsg */
 404                 lnet_finalize (ni, lnetmsg, rc);
 405 }
 406
 407 void
 408 ksocknal_txlist_done (lnet_ni_t *ni, struct list_head *txlist, int error)
 409 {
 410         ksock_tx_t *tx;
 411
 412         while (!list_empty (txlist)) {
 413                 tx = list_entry (txlist->next, ksock_tx_t, tx_list);
 414
 415                 if (error && tx->tx_lnetmsg != NULL) {
 416                         CNETERR("Deleting packet type %d len %d %s->%s\n",
 417                                 le32_to_cpu (tx->tx_lnetmsg->msg_hdr.type),
 418                                 le32_to_cpu (tx->tx_lnetmsg->msg_hdr.payload_length),
 419                                 libcfs_nid2str(le64_to_cpu(tx->tx_lnetmsg->msg_hdr.src_nid)),
 420                                 libcfs_nid2str(le64_to_cpu(tx->tx_lnetmsg->msg_hdr.dest_nid)));
 421                 } else if (error) {
 422                         CNETERR("Deleting noop packet\n");
 423                 }
 424
 425                 list_del (&tx->tx_list);
 426
 427                 LASSERT (atomic_read(&tx->tx_refcount) == 1);
 428                 ksocknal_tx_done (ni, tx);
 429         }
 430 }
 431
 432 static void
 433 ksocknal_check_zc_req(ksock_tx_t *tx)
 434 {
 435         ksock_conn_t   *conn = tx->tx_conn;
 436         ksock_peer_t   *peer = conn->ksnc_peer;
 437
 438         /* Set tx_msg.ksm_zc_cookies[0] to a unique non-zero cookie and add tx
 439          * to ksnp_zc_req_list if some fragment of this message should be sent
 440          * zero-copy.  Our peer will send an ACK containing this cookie when
 441          * she has received this message to tell us we can signal completion.
 442          * tx_msg.ksm_zc_cookies[0] remains non-zero while tx is on
 443          * ksnp_zc_req_list. */
 444         LASSERT (tx->tx_msg.ksm_type != KSOCK_MSG_NOOP);
 445         LASSERT (tx->tx_zc_capable);
 446
 447         tx->tx_zc_checked = 1;
 448
 449         if (conn->ksnc_proto == &ksocknal_protocol_v1x ||
 450             !conn->ksnc_zc_capable)
 451                 return;
 452
 453         /* assign cookie and queue tx to pending list, it will be released when
 454          * a matching ack is received. See ksocknal_handle_zcack() */
 455
 456         ksocknal_tx_addref(tx);
 457
 458         spin_lock(&peer->ksnp_lock);
 459
 460         /* ZC_REQ is going to be pinned to the peer */
 461         tx->tx_deadline =
 462                 cfs_time_shift(*ksocknal_tunables.ksnd_timeout);
 463
 464         LASSERT (tx->tx_msg.ksm_zc_cookies[0] == 0);
 465
 466         tx->tx_msg.ksm_zc_cookies[0] = peer->ksnp_zc_next_cookie++;
 467
 468         if (peer->ksnp_zc_next_cookie == 0)
 469                 peer->ksnp_zc_next_cookie = SOCKNAL_KEEPALIVE_PING + 1;
 470
 471         list_add_tail(&tx->tx_zc_list, &peer->ksnp_zc_req_list);
 472
 473         spin_unlock(&peer->ksnp_lock);
 474 }
 475
 476 static void
 477 ksocknal_uncheck_zc_req(ksock_tx_t *tx)
 478 {
 479         ksock_peer_t   *peer = tx->tx_conn->ksnc_peer;
 480
 481         LASSERT(tx->tx_msg.ksm_type != KSOCK_MSG_NOOP);
 482         LASSERT(tx->tx_zc_capable);
 483
 484         tx->tx_zc_checked = 0;
 485
 486         spin_lock(&peer->ksnp_lock);
 487
 488         if (tx->tx_msg.ksm_zc_cookies[0] == 0) {
 489                 /* Not waiting for an ACK */
 490                 spin_unlock(&peer->ksnp_lock);
 491                 return;
 492         }
 493
 494         tx->tx_msg.ksm_zc_cookies[0] = 0;
 495         list_del(&tx->tx_zc_list);
 496
 497         spin_unlock(&peer->ksnp_lock);
 498
 499         ksocknal_tx_decref(tx);
 500 }
 501
 502 static int
 503 ksocknal_process_transmit (ksock_conn_t *conn, ksock_tx_t *tx)
 504 {
 505         int         rc;
 506
 507         if (tx->tx_zc_capable && !tx->tx_zc_checked)
 508                 ksocknal_check_zc_req(tx);
 509
 510         rc = ksocknal_transmit (conn, tx);
 511
 512         CDEBUG (D_NET, "send(%d) %d\n", tx->tx_resid, rc);
 513
 514         if (tx->tx_resid == 0) {
 515                 /* Sent everything OK */
 516                 LASSERT (rc == 0);
 517
 518                 return 0;
 519         }
 520
 521         if (rc == -EAGAIN)
 522                 return rc;
 523
 524         if (rc == -ENOMEM) {
 525                 static int counter;
 526
 527                 counter++;   /* exponential backoff warnings */
 528                 if ((counter & (-counter)) == counter)
 529                         CWARN("%u ENOMEM tx %p (%u allocated)\n",
 530                               counter, conn, atomic_read(&libcfs_kmemory));
 531
 532                 /* Queue on ksnd_enomem_conns for retry after a timeout */
 533                 spin_lock_bh(&ksocknal_data.ksnd_reaper_lock);
 534
 535                 /* enomem list takes over scheduler's ref... */
 536                 LASSERT (conn->ksnc_tx_scheduled);
 537                 list_add_tail(&conn->ksnc_tx_list,
 538                                   &ksocknal_data.ksnd_enomem_conns);
 539                 if (!cfs_time_aftereq(cfs_time_add(cfs_time_current(),
 540                                                    SOCKNAL_ENOMEM_RETRY),
 541                                    ksocknal_data.ksnd_reaper_waketime))
 542                         wake_up (&ksocknal_data.ksnd_reaper_waitq);
 543
 544                 spin_unlock_bh(&ksocknal_data.ksnd_reaper_lock);
 545                 return rc;
 546         }
 547
 548         /* Actual error */
 549         LASSERT (rc < 0);
 550
 551         if (!conn->ksnc_closing) {
 552                 switch (rc) {
 553                 case -ECONNRESET:
 554                         LCONSOLE_WARN("Host %pI4h reset our connection "
 555                                       "while we were sending data; it may have "
 556                                       "rebooted.\n",
 557                                       &conn->ksnc_ipaddr);
 558                         break;
 559                 default:
 560                         LCONSOLE_WARN("There was an unexpected network error "
 561                                       "while writing to %pI4h: %d.\n",
 562                                       &conn->ksnc_ipaddr, rc);
 563                         break;
 564                 }
 565                 CDEBUG(D_NET, "[%p] Error %d on write to %s"
 566                        " ip %pI4h:%d\n", conn, rc,
 567                        libcfs_id2str(conn->ksnc_peer->ksnp_id),
 568                        &conn->ksnc_ipaddr,
 569                        conn->ksnc_port);
 570         }
 571
 572         if (tx->tx_zc_checked)
 573                 ksocknal_uncheck_zc_req(tx);
 574
 575         /* it's not an error if conn is being closed */
 576         ksocknal_close_conn_and_siblings (conn,
 577                                           (conn->ksnc_closing) ? 0 : rc);
 578
 579         return rc;
 580 }
 581
 582 static void
 583 ksocknal_launch_connection_locked (ksock_route_t *route)
 584 {
 585
 586         /* called holding write lock on ksnd_global_lock */
 587
 588         LASSERT (!route->ksnr_scheduled);
 589         LASSERT (!route->ksnr_connecting);
 590         LASSERT ((ksocknal_route_mask() & ~route->ksnr_connected) != 0);
 591
 592         route->ksnr_scheduled = 1;            /* scheduling conn for connd */
 593         ksocknal_route_addref(route);      /* extra ref for connd */
 594
 595         spin_lock_bh(&ksocknal_data.ksnd_connd_lock);
 596
 597         list_add_tail(&route->ksnr_connd_list,
 598                           &ksocknal_data.ksnd_connd_routes);
 599         wake_up(&ksocknal_data.ksnd_connd_waitq);
 600
 601         spin_unlock_bh(&ksocknal_data.ksnd_connd_lock);
 602 }
 603
 604 void
 605 ksocknal_launch_all_connections_locked (ksock_peer_t *peer)
 606 {
 607         ksock_route_t *route;
 608
 609         /* called holding write lock on ksnd_global_lock */
 610         for (;;) {
 611                 /* launch any/all connections that need it */
 612                 route = ksocknal_find_connectable_route_locked(peer);
 613                 if (route == NULL)
 614                         return;
 615
 616                 ksocknal_launch_connection_locked(route);
 617         }
 618 }
 619
 620 ksock_conn_t *
 621 ksocknal_find_conn_locked(ksock_peer_t *peer, ksock_tx_t *tx, int nonblk)
 622 {
 623         struct list_head       *tmp;
 624         ksock_conn_t     *conn;
 625         ksock_conn_t     *typed = NULL;
 626         ksock_conn_t     *fallback = NULL;
 627         int            tnob     = 0;
 628         int            fnob     = 0;
 629
 630         list_for_each (tmp, &peer->ksnp_conns) {
 631                 ksock_conn_t *c  = list_entry(tmp, ksock_conn_t, ksnc_list);
 632                 int        nob = atomic_read(&c->ksnc_tx_nob) +
 633                                     c->ksnc_sock->sk->sk_wmem_queued;
 634                 int        rc;
 635
 636                 LASSERT (!c->ksnc_closing);
 637                 LASSERT (c->ksnc_proto != NULL &&
 638                          c->ksnc_proto->pro_match_tx != NULL);
 639
 640                 rc = c->ksnc_proto->pro_match_tx(c, tx, nonblk);
 641
 642                 switch (rc) {
 643                 default:
 644                         LBUG();
 645                 case SOCKNAL_MATCH_NO: /* protocol rejected the tx */
 646                         continue;
 647
 648                 case SOCKNAL_MATCH_YES: /* typed connection */
 649                         if (typed == NULL || tnob > nob ||
 650                             (tnob == nob && *ksocknal_tunables.ksnd_round_robin &&
 651                              cfs_time_after(typed->ksnc_tx_last_post, c->ksnc_tx_last_post))) {
 652                                 typed = c;
 653                                 tnob  = nob;
 654                         }
 655                         break;
 656
 657                 case SOCKNAL_MATCH_MAY: /* fallback connection */
 658                         if (fallback == NULL || fnob > nob ||
 659                             (fnob == nob && *ksocknal_tunables.ksnd_round_robin &&
 660                              cfs_time_after(fallback->ksnc_tx_last_post, c->ksnc_tx_last_post))) {
 661                                 fallback = c;
 662                                 fnob     = nob;
 663                         }
 664                         break;
 665                 }
 666         }
 667
 668         /* prefer the typed selection */
 669         conn = (typed != NULL) ? typed : fallback;
 670
 671         if (conn != NULL)
 672                 conn->ksnc_tx_last_post = cfs_time_current();
 673
 674         return conn;
 675 }
 676
 677 void
 678 ksocknal_tx_prep(ksock_conn_t *conn, ksock_tx_t *tx)
 679 {
 680         conn->ksnc_proto->pro_pack(tx);
 681
 682         atomic_add (tx->tx_nob, &conn->ksnc_tx_nob);
 683         ksocknal_conn_addref(conn); /* +1 ref for tx */
 684         tx->tx_conn = conn;
 685 }
 686
 687 void
 688 ksocknal_queue_tx_locked (ksock_tx_t *tx, ksock_conn_t *conn)
 689 {
 690         ksock_sched_t *sched = conn->ksnc_scheduler;
 691         ksock_msg_t   *msg = &tx->tx_msg;
 692         ksock_tx_t    *ztx = NULL;
 693         int         bufnob = 0;
 694
 695         /* called holding global lock (read or irq-write) and caller may
 696          * not have dropped this lock between finding conn and calling me,
 697          * so we don't need the {get,put}connsock dance to deref
 698          * ksnc_sock... */
 699         LASSERT(!conn->ksnc_closing);
 700
 701         CDEBUG(D_NET, "Sending to %s ip %pI4h:%d\n",
 702                 libcfs_id2str(conn->ksnc_peer->ksnp_id),
 703                 &conn->ksnc_ipaddr,
 704                 conn->ksnc_port);
 705
 706         ksocknal_tx_prep(conn, tx);
 707
 708         /* Ensure the frags we've been given EXACTLY match the number of
 709          * bytes we want to send.  Many TCP/IP stacks disregard any total
 710          * size parameters passed to them and just look at the frags.
 711          *
 712          * We always expect at least 1 mapped fragment containing the
 713          * complete ksocknal message header. */
 714         LASSERT (lnet_iov_nob (tx->tx_niov, tx->tx_iov) +
 715                  lnet_kiov_nob(tx->tx_nkiov, tx->tx_kiov) ==
 716                  (unsigned int)tx->tx_nob);
 717         LASSERT (tx->tx_niov >= 1);
 718         LASSERT (tx->tx_resid == tx->tx_nob);
 719
 720         CDEBUG (D_NET, "Packet %p type %d, nob %d niov %d nkiov %d\n",
 721                 tx, (tx->tx_lnetmsg != NULL) ? tx->tx_lnetmsg->msg_hdr.type:
 722                                                KSOCK_MSG_NOOP,
 723                 tx->tx_nob, tx->tx_niov, tx->tx_nkiov);
 724
 725         /*
 726          * FIXME: SOCK_WMEM_QUEUED and SOCK_ERROR could block in __DARWIN8__
 727          * but they're used inside spinlocks a lot.
 728          */
 729         bufnob = conn->ksnc_sock->sk->sk_wmem_queued;
 730         spin_lock_bh(&sched->kss_lock);
 731
 732         if (list_empty(&conn->ksnc_tx_queue) && bufnob == 0) {
 733                 /* First packet starts the timeout */
 734                 conn->ksnc_tx_deadline =
 735                         cfs_time_shift(*ksocknal_tunables.ksnd_timeout);
 736                 if (conn->ksnc_tx_bufnob > 0) /* something got ACKed */
 737                         conn->ksnc_peer->ksnp_last_alive = cfs_time_current();
 738                 conn->ksnc_tx_bufnob = 0;
 739                 mb(); /* order with adding to tx_queue */
 740         }
 741
 742         if (msg->ksm_type == KSOCK_MSG_NOOP) {
 743                 /* The packet is noop ZC ACK, try to piggyback the ack_cookie
 744                  * on a normal packet so I don't need to send it */
 745                 LASSERT (msg->ksm_zc_cookies[1] != 0);
 746                 LASSERT (conn->ksnc_proto->pro_queue_tx_zcack != NULL);
 747
 748                 if (conn->ksnc_proto->pro_queue_tx_zcack(conn, tx, 0))
 749                         ztx = tx; /* ZC ACK piggybacked on ztx release tx later */
 750
 751         } else {
 752                 /* It's a normal packet - can it piggback a noop zc-ack that
 753                  * has been queued already? */
 754                 LASSERT (msg->ksm_zc_cookies[1] == 0);
 755                 LASSERT (conn->ksnc_proto->pro_queue_tx_msg != NULL);
 756
 757                 ztx = conn->ksnc_proto->pro_queue_tx_msg(conn, tx);
 758                 /* ztx will be released later */
 759         }
 760
 761         if (ztx != NULL) {
 762                 atomic_sub (ztx->tx_nob, &conn->ksnc_tx_nob);
 763                 list_add_tail(&ztx->tx_list, &sched->kss_zombie_noop_txs);
 764         }
 765
 766         if (conn->ksnc_tx_ready &&      /* able to send */
 767             !conn->ksnc_tx_scheduled) { /* not scheduled to send */
 768                 /* +1 ref for scheduler */
 769                 ksocknal_conn_addref(conn);
 770                 list_add_tail (&conn->ksnc_tx_list,
 771                                    &sched->kss_tx_conns);
 772                 conn->ksnc_tx_scheduled = 1;
 773                 wake_up (&sched->kss_waitq);
 774         }
 775
 776         spin_unlock_bh(&sched->kss_lock);
 777 }
 778
 779
 780 ksock_route_t *
 781 ksocknal_find_connectable_route_locked (ksock_peer_t *peer)
 782 {
 783         unsigned long     now = cfs_time_current();
 784         struct list_head    *tmp;
 785         ksock_route_t *route;
 786
 787         list_for_each (tmp, &peer->ksnp_routes) {
 788                 route = list_entry (tmp, ksock_route_t, ksnr_list);
 789
 790                 LASSERT (!route->ksnr_connecting || route->ksnr_scheduled);
 791
 792                 if (route->ksnr_scheduled)      /* connections being established */
 793                         continue;
 794
 795                 /* all route types connected ? */
 796                 if ((ksocknal_route_mask() & ~route->ksnr_connected) == 0)
 797                         continue;
 798
 799                 if (!(route->ksnr_retry_interval == 0 || /* first attempt */
 800                       cfs_time_aftereq(now, route->ksnr_timeout))) {
 801                         CDEBUG(D_NET,
 802                                "Too soon to retry route %pI4h "
 803                                "(cnted %d, interval %ld, %ld secs later)\n",
 804                                &route->ksnr_ipaddr,
 805                                route->ksnr_connected,
 806                                route->ksnr_retry_interval,
 807                                cfs_duration_sec(route->ksnr_timeout - now));
 808                         continue;
 809                 }
 810
 811                 return route;
 812         }
 813
 814         return NULL;
 815 }
 816
 817 ksock_route_t *
 818 ksocknal_find_connecting_route_locked (ksock_peer_t *peer)
 819 {
 820         struct list_head        *tmp;
 821         ksock_route_t     *route;
 822
 823         list_for_each (tmp, &peer->ksnp_routes) {
 824                 route = list_entry (tmp, ksock_route_t, ksnr_list);
 825
 826                 LASSERT (!route->ksnr_connecting || route->ksnr_scheduled);
 827
 828                 if (route->ksnr_scheduled)
 829                         return route;
 830         }
 831
 832         return NULL;
 833 }
 834
 835 int
 836 ksocknal_launch_packet (lnet_ni_t *ni, ksock_tx_t *tx, lnet_process_id_t id)
 837 {
 838         ksock_peer_t     *peer;
 839         ksock_conn_t     *conn;
 840         rwlock_t     *g_lock;
 841         int            retry;
 842         int            rc;
 843
 844         LASSERT (tx->tx_conn == NULL);
 845
 846         g_lock = &ksocknal_data.ksnd_global_lock;
 847
 848         for (retry = 0;; retry = 1) {
 849                 read_lock(g_lock);
 850                 peer = ksocknal_find_peer_locked(ni, id);
 851                 if (peer != NULL) {
 852                         if (ksocknal_find_connectable_route_locked(peer) == NULL) {
 853                                 conn = ksocknal_find_conn_locked(peer, tx, tx->tx_nonblk);
 854                                 if (conn != NULL) {
 855                                         /* I've got no routes that need to be
 856                                          * connecting and I do have an actual
 857                                          * connection... */
 858                                         ksocknal_queue_tx_locked (tx, conn);
 859                                         read_unlock(g_lock);
 860                                         return 0;
 861                                 }
 862                         }
 863                 }
 864
 865                 /* I'll need a write lock... */
 866                 read_unlock(g_lock);
 867
 868                 write_lock_bh(g_lock);
 869
 870                 peer = ksocknal_find_peer_locked(ni, id);
 871                 if (peer != NULL)
 872                         break;
 873
 874                 write_unlock_bh(g_lock);
 875
 876                 if ((id.pid & LNET_PID_USERFLAG) != 0) {
 877                         CERROR("Refusing to create a connection to "
 878                                "userspace process %s\n", libcfs_id2str(id));
 879                         return -EHOSTUNREACH;
 880                 }
 881
 882                 if (retry) {
 883                         CERROR("Can't find peer %s\n", libcfs_id2str(id));
 884                         return -EHOSTUNREACH;
 885                 }
 886
 887                 rc = ksocknal_add_peer(ni, id,
 888                                        LNET_NIDADDR(id.nid),
 889                                        lnet_acceptor_port());
 890                 if (rc != 0) {
 891                         CERROR("Can't add peer %s: %d\n",
 892                                libcfs_id2str(id), rc);
 893                         return rc;
 894                 }
 895         }
 896
 897         ksocknal_launch_all_connections_locked(peer);
 898
 899         conn = ksocknal_find_conn_locked(peer, tx, tx->tx_nonblk);
 900         if (conn != NULL) {
 901                 /* Connection exists; queue message on it */
 902                 ksocknal_queue_tx_locked (tx, conn);
 903                 write_unlock_bh(g_lock);
 904                 return 0;
 905         }
 906
 907         if (peer->ksnp_accepting > 0 ||
 908             ksocknal_find_connecting_route_locked (peer) != NULL) {
 909                 /* the message is going to be pinned to the peer */
 910                 tx->tx_deadline =
 911                         cfs_time_shift(*ksocknal_tunables.ksnd_timeout);
 912
 913                 /* Queue the message until a connection is established */
 914                 list_add_tail (&tx->tx_list, &peer->ksnp_tx_queue);
 915                 write_unlock_bh(g_lock);
 916                 return 0;
 917         }
 918
 919         write_unlock_bh(g_lock);
 920
 921         /* NB Routes may be ignored if connections to them failed recently */
 922         CNETERR("No usable routes to %s\n", libcfs_id2str(id));
 923         return -EHOSTUNREACH;
 924 }
 925
 926 int
 927 ksocknal_send(lnet_ni_t *ni, void *private, lnet_msg_t *lntmsg)
 928 {
 929         int            mpflag = 1;
 930         int            type = lntmsg->msg_type;
 931         lnet_process_id_t target = lntmsg->msg_target;
 932         unsigned int      payload_niov = lntmsg->msg_niov;
 933         struct iovec     *payload_iov = lntmsg->msg_iov;
 934         lnet_kiov_t      *payload_kiov = lntmsg->msg_kiov;
 935         unsigned int      payload_offset = lntmsg->msg_offset;
 936         unsigned int      payload_nob = lntmsg->msg_len;
 937         ksock_tx_t       *tx;
 938         int            desc_size;
 939         int            rc;
 940
 941         /* NB 'private' is different depending on what we're sending.
 942          * Just ignore it... */
 943
 944         CDEBUG(D_NET, "sending %u bytes in %d frags to %s\n",
 945                payload_nob, payload_niov, libcfs_id2str(target));
 946
 947         LASSERT (payload_nob == 0 || payload_niov > 0);
 948         LASSERT (payload_niov <= LNET_MAX_IOV);
 949         /* payload is either all vaddrs or all pages */
 950         LASSERT (!(payload_kiov != NULL && payload_iov != NULL));
 951         LASSERT (!in_interrupt ());
 952
 953         if (payload_iov != NULL)
 954                 desc_size = offsetof(ksock_tx_t,
 955                                      tx_frags.virt.iov[1 + payload_niov]);
 956         else
 957                 desc_size = offsetof(ksock_tx_t,
 958                                      tx_frags.paged.kiov[payload_niov]);
 959
 960         if (lntmsg->msg_vmflush)
 961                 mpflag = cfs_memory_pressure_get_and_set();
 962         tx = ksocknal_alloc_tx(KSOCK_MSG_LNET, desc_size);
 963         if (tx == NULL) {
 964                 CERROR("Can't allocate tx desc type %d size %d\n",
 965                        type, desc_size);
 966                 if (lntmsg->msg_vmflush)
 967                         cfs_memory_pressure_restore(mpflag);
 968                 return -ENOMEM;
 969         }
 970
 971         tx->tx_conn = NULL;                  /* set when assigned a conn */
 972         tx->tx_lnetmsg = lntmsg;
 973
 974         if (payload_iov != NULL) {
 975                 tx->tx_kiov = NULL;
 976                 tx->tx_nkiov = 0;
 977                 tx->tx_iov = tx->tx_frags.virt.iov;
 978                 tx->tx_niov = 1 +
 979                               lnet_extract_iov(payload_niov, &tx->tx_iov[1],
 980                                                payload_niov, payload_iov,
 981                                                payload_offset, payload_nob);
 982         } else {
 983                 tx->tx_niov = 1;
 984                 tx->tx_iov = &tx->tx_frags.paged.iov;
 985                 tx->tx_kiov = tx->tx_frags.paged.kiov;
 986                 tx->tx_nkiov = lnet_extract_kiov(payload_niov, tx->tx_kiov,
 987                                                  payload_niov, payload_kiov,
 988                                                  payload_offset, payload_nob);
 989
 990                 if (payload_nob >= *ksocknal_tunables.ksnd_zc_min_payload)
 991                         tx->tx_zc_capable = 1;
 992         }
 993
 994         socklnd_init_msg(&tx->tx_msg, KSOCK_MSG_LNET);
 995
 996         /* The first fragment will be set later in pro_pack */
 997         rc = ksocknal_launch_packet(ni, tx, target);
 998         if (!mpflag)
 999                 cfs_memory_pressure_restore(mpflag);
1000
1001         if (rc == 0)
1002                 return 0;
1003
1004         ksocknal_free_tx(tx);
1005         return -EIO;
1006 }
1007
1008 int
1009 ksocknal_thread_start(int (*fn)(void *arg), void *arg, char *name)
1010 {
1011         struct task_struct *task = kthread_run(fn, arg, "%s", name);
1012
1013         if (IS_ERR(task))
1014                 return PTR_ERR(task);
1015
1016         write_lock_bh(&ksocknal_data.ksnd_global_lock);
1017         ksocknal_data.ksnd_nthreads++;
1018         write_unlock_bh(&ksocknal_data.ksnd_global_lock);
1019         return 0;
1020 }
1021
1022 void
1023 ksocknal_thread_fini (void)
1024 {
1025         write_lock_bh(&ksocknal_data.ksnd_global_lock);
1026         ksocknal_data.ksnd_nthreads--;
1027         write_unlock_bh(&ksocknal_data.ksnd_global_lock);
1028 }
1029
1030 int
1031 ksocknal_new_packet (ksock_conn_t *conn, int nob_to_skip)
1032 {
1033         static char ksocknal_slop_buffer[4096];
1034
1035         int         nob;
1036         unsigned int   niov;
1037         int         skipped;
1038
1039         LASSERT(conn->ksnc_proto != NULL);
1040
1041         if ((*ksocknal_tunables.ksnd_eager_ack & conn->ksnc_type) != 0) {
1042                 /* Remind the socket to ack eagerly... */
1043                 ksocknal_lib_eager_ack(conn);
1044         }
1045
1046         if (nob_to_skip == 0) {  /* right at next packet boundary now */
1047                 conn->ksnc_rx_started = 0;
1048                 mb();                  /* racing with timeout thread */
1049
1050                 switch (conn->ksnc_proto->pro_version) {
1051                 case  KSOCK_PROTO_V2:
1052                 case  KSOCK_PROTO_V3:
1053                         conn->ksnc_rx_state = SOCKNAL_RX_KSM_HEADER;
1054                         conn->ksnc_rx_iov = (struct iovec *)&conn->ksnc_rx_iov_space;
1055                         conn->ksnc_rx_iov[0].iov_base = (char *)&conn->ksnc_msg;
1056
1057                         conn->ksnc_rx_nob_wanted = offsetof(ksock_msg_t, ksm_u);
1058                         conn->ksnc_rx_nob_left = offsetof(ksock_msg_t, ksm_u);
1059                         conn->ksnc_rx_iov[0].iov_len  = offsetof(ksock_msg_t, ksm_u);
1060                         break;
1061
1062                 case KSOCK_PROTO_V1:
1063                         /* Receiving bare lnet_hdr_t */
1064                         conn->ksnc_rx_state = SOCKNAL_RX_LNET_HEADER;
1065                         conn->ksnc_rx_nob_wanted = sizeof(lnet_hdr_t);
1066                         conn->ksnc_rx_nob_left = sizeof(lnet_hdr_t);
1067
1068                         conn->ksnc_rx_iov = (struct iovec *)&conn->ksnc_rx_iov_space;
1069                         conn->ksnc_rx_iov[0].iov_base = (char *)&conn->ksnc_msg.ksm_u.lnetmsg;
1070                         conn->ksnc_rx_iov[0].iov_len  = sizeof (lnet_hdr_t);
1071                         break;
1072
1073                 default:
1074                         LBUG ();
1075                 }
1076                 conn->ksnc_rx_niov = 1;
1077
1078                 conn->ksnc_rx_kiov = NULL;
1079                 conn->ksnc_rx_nkiov = 0;
1080                 conn->ksnc_rx_csum = ~0;
1081                 return 1;
1082         }
1083
1084         /* Set up to skip as much as possible now.  If there's more left
1085          * (ran out of iov entries) we'll get called again */
1086
1087         conn->ksnc_rx_state = SOCKNAL_RX_SLOP;
1088         conn->ksnc_rx_nob_left = nob_to_skip;
1089         conn->ksnc_rx_iov = (struct iovec *)&conn->ksnc_rx_iov_space;
1090         skipped = 0;
1091         niov = 0;
1092
1093         do {
1094                 nob = MIN (nob_to_skip, sizeof (ksocknal_slop_buffer));
1095
1096                 conn->ksnc_rx_iov[niov].iov_base = ksocknal_slop_buffer;
1097                 conn->ksnc_rx_iov[niov].iov_len  = nob;
1098                 niov++;
1099                 skipped += nob;
1100                 nob_to_skip -=nob;
1101
1102         } while (nob_to_skip != 0 &&    /* mustn't overflow conn's rx iov */
1103                  niov < sizeof(conn->ksnc_rx_iov_space) / sizeof (struct iovec));
1104
1105         conn->ksnc_rx_niov = niov;
1106         conn->ksnc_rx_kiov = NULL;
1107         conn->ksnc_rx_nkiov = 0;
1108         conn->ksnc_rx_nob_wanted = skipped;
1109         return 0;
1110 }
1111
1112 static int
1113 ksocknal_process_receive (ksock_conn_t *conn)
1114 {
1115         lnet_hdr_t      *lhdr;
1116         lnet_process_id_t *id;
1117         int             rc;
1118
1119         LASSERT (atomic_read(&conn->ksnc_conn_refcount) > 0);
1120
1121         /* NB: sched lock NOT held */
1122         /* SOCKNAL_RX_LNET_HEADER is here for backward compatibility */
1123         LASSERT (conn->ksnc_rx_state == SOCKNAL_RX_KSM_HEADER ||
1124                  conn->ksnc_rx_state == SOCKNAL_RX_LNET_PAYLOAD ||
1125                  conn->ksnc_rx_state == SOCKNAL_RX_LNET_HEADER ||
1126                  conn->ksnc_rx_state == SOCKNAL_RX_SLOP);
1127  again:
1128         if (conn->ksnc_rx_nob_wanted != 0) {
1129                 rc = ksocknal_receive(conn);
1130
1131                 if (rc <= 0) {
1132                         LASSERT (rc != -EAGAIN);
1133
1134                         if (rc == 0)
1135                                 CDEBUG(D_NET, "[%p] EOF from %s"
1136                                         " ip %pI4h:%d\n", conn,
1137                                         libcfs_id2str(conn->ksnc_peer->ksnp_id),
1138                                         &conn->ksnc_ipaddr,
1139                                         conn->ksnc_port);
1140                         else if (!conn->ksnc_closing)
1141                                 CERROR("[%p] Error %d on read from %s"
1142                                         " ip %pI4h:%d\n",
1143                                         conn, rc,
1144                                         libcfs_id2str(conn->ksnc_peer->ksnp_id),
1145                                         &conn->ksnc_ipaddr,
1146                                         conn->ksnc_port);
1147
1148                         /* it's not an error if conn is being closed */
1149                         ksocknal_close_conn_and_siblings (conn,
1150                                                           (conn->ksnc_closing) ? 0 : rc);
1151                         return (rc == 0 ? -ESHUTDOWN : rc);
1152                 }
1153
1154                 if (conn->ksnc_rx_nob_wanted != 0) {
1155                         /* short read */
1156                         return -EAGAIN;
1157                 }
1158         }
1159         switch (conn->ksnc_rx_state) {
1160         case SOCKNAL_RX_KSM_HEADER:
1161                 if (conn->ksnc_flip) {
1162                         __swab32s(&conn->ksnc_msg.ksm_type);
1163                         __swab32s(&conn->ksnc_msg.ksm_csum);
1164                         __swab64s(&conn->ksnc_msg.ksm_zc_cookies[0]);
1165                         __swab64s(&conn->ksnc_msg.ksm_zc_cookies[1]);
1166                 }
1167
1168                 if (conn->ksnc_msg.ksm_type != KSOCK_MSG_NOOP &&
1169                     conn->ksnc_msg.ksm_type != KSOCK_MSG_LNET) {
1170                         CERROR("%s: Unknown message type: %x\n",
1171                                libcfs_id2str(conn->ksnc_peer->ksnp_id),
1172                                conn->ksnc_msg.ksm_type);
1173                         ksocknal_new_packet(conn, 0);
1174                         ksocknal_close_conn_and_siblings(conn, -EPROTO);
1175                         return -EPROTO;
1176                 }
1177
1178                 if (conn->ksnc_msg.ksm_type == KSOCK_MSG_NOOP &&
1179                     conn->ksnc_msg.ksm_csum != 0 &&     /* has checksum */
1180                     conn->ksnc_msg.ksm_csum != conn->ksnc_rx_csum) {
1181                         /* NOOP Checksum error */
1182                         CERROR("%s: Checksum error, wire:0x%08X data:0x%08X\n",
1183                                libcfs_id2str(conn->ksnc_peer->ksnp_id),
1184                                conn->ksnc_msg.ksm_csum, conn->ksnc_rx_csum);
1185                         ksocknal_new_packet(conn, 0);
1186                         ksocknal_close_conn_and_siblings(conn, -EPROTO);
1187                         return -EIO;
1188                 }
1189
1190                 if (conn->ksnc_msg.ksm_zc_cookies[1] != 0) {
1191                         __u64 cookie = 0;
1192
1193                         LASSERT (conn->ksnc_proto != &ksocknal_protocol_v1x);
1194
1195                         if (conn->ksnc_msg.ksm_type == KSOCK_MSG_NOOP)
1196                                 cookie = conn->ksnc_msg.ksm_zc_cookies[0];
1197
1198                         rc = conn->ksnc_proto->pro_handle_zcack(conn, cookie,
1199                                                conn->ksnc_msg.ksm_zc_cookies[1]);
1200
1201                         if (rc != 0) {
1202                                 CERROR("%s: Unknown ZC-ACK cookie: %llu, %llu\n",
1203                                        libcfs_id2str(conn->ksnc_peer->ksnp_id),
1204                                        cookie, conn->ksnc_msg.ksm_zc_cookies[1]);
1205                                 ksocknal_new_packet(conn, 0);
1206                                 ksocknal_close_conn_and_siblings(conn, -EPROTO);
1207                                 return rc;
1208                         }
1209                 }
1210
1211                 if (conn->ksnc_msg.ksm_type == KSOCK_MSG_NOOP) {
1212                         ksocknal_new_packet (conn, 0);
1213                         return 0;       /* NOOP is done and just return */
1214                 }
1215
1216                 conn->ksnc_rx_state = SOCKNAL_RX_LNET_HEADER;
1217                 conn->ksnc_rx_nob_wanted = sizeof(ksock_lnet_msg_t);
1218                 conn->ksnc_rx_nob_left = sizeof(ksock_lnet_msg_t);
1219
1220                 conn->ksnc_rx_iov = (struct iovec *)&conn->ksnc_rx_iov_space;
1221                 conn->ksnc_rx_iov[0].iov_base = (char *)&conn->ksnc_msg.ksm_u.lnetmsg;
1222                 conn->ksnc_rx_iov[0].iov_len  = sizeof(ksock_lnet_msg_t);
1223
1224                 conn->ksnc_rx_niov = 1;
1225                 conn->ksnc_rx_kiov = NULL;
1226                 conn->ksnc_rx_nkiov = 0;
1227
1228                 goto again;     /* read lnet header now */
1229
1230         case SOCKNAL_RX_LNET_HEADER:
1231                 /* unpack message header */
1232                 conn->ksnc_proto->pro_unpack(&conn->ksnc_msg);
1233
1234                 if ((conn->ksnc_peer->ksnp_id.pid & LNET_PID_USERFLAG) != 0) {
1235                         /* Userspace peer */
1236                         lhdr = &conn->ksnc_msg.ksm_u.lnetmsg.ksnm_hdr;
1237                         id   = &conn->ksnc_peer->ksnp_id;
1238
1239                         /* Substitute process ID assigned at connection time */
1240                         lhdr->src_pid = cpu_to_le32(id->pid);
1241                         lhdr->src_nid = cpu_to_le64(id->nid);
1242                 }
1243
1244                 conn->ksnc_rx_state = SOCKNAL_RX_PARSE;
1245                 ksocknal_conn_addref(conn);     /* ++ref while parsing */
1246
1247                 rc = lnet_parse(conn->ksnc_peer->ksnp_ni,
1248                                 &conn->ksnc_msg.ksm_u.lnetmsg.ksnm_hdr,
1249                                 conn->ksnc_peer->ksnp_id.nid, conn, 0);
1250                 if (rc < 0) {
1251                         /* I just received garbage: give up on this conn */
1252                         ksocknal_new_packet(conn, 0);
1253                         ksocknal_close_conn_and_siblings (conn, rc);
1254                         ksocknal_conn_decref(conn);
1255                         return -EPROTO;
1256                 }
1257
1258                 /* I'm racing with ksocknal_recv() */
1259                 LASSERT (conn->ksnc_rx_state == SOCKNAL_RX_PARSE ||
1260                          conn->ksnc_rx_state == SOCKNAL_RX_LNET_PAYLOAD);
1261
1262                 if (conn->ksnc_rx_state != SOCKNAL_RX_LNET_PAYLOAD)
1263                         return 0;
1264
1265                 /* ksocknal_recv() got called */
1266                 goto again;
1267
1268         case SOCKNAL_RX_LNET_PAYLOAD:
1269                 /* payload all received */
1270                 rc = 0;
1271
1272                 if (conn->ksnc_rx_nob_left == 0 &&   /* not truncating */
1273                     conn->ksnc_msg.ksm_csum != 0 &&  /* has checksum */
1274                     conn->ksnc_msg.ksm_csum != conn->ksnc_rx_csum) {
1275                         CERROR("%s: Checksum error, wire:0x%08X data:0x%08X\n",
1276                                libcfs_id2str(conn->ksnc_peer->ksnp_id),
1277                                conn->ksnc_msg.ksm_csum, conn->ksnc_rx_csum);
1278                         rc = -EIO;
1279                 }
1280
1281                 if (rc == 0 && conn->ksnc_msg.ksm_zc_cookies[0] != 0) {
1282                         LASSERT(conn->ksnc_proto != &ksocknal_protocol_v1x);
1283
1284                         lhdr = &conn->ksnc_msg.ksm_u.lnetmsg.ksnm_hdr;
1285                         id   = &conn->ksnc_peer->ksnp_id;
1286
1287                         rc = conn->ksnc_proto->pro_handle_zcreq(conn,
1288                                         conn->ksnc_msg.ksm_zc_cookies[0],
1289                                         *ksocknal_tunables.ksnd_nonblk_zcack ||
1290                                         le64_to_cpu(lhdr->src_nid) != id->nid);
1291                 }
1292
1293                 lnet_finalize(conn->ksnc_peer->ksnp_ni, conn->ksnc_cookie, rc);
1294
1295                 if (rc != 0) {
1296                         ksocknal_new_packet(conn, 0);
1297                         ksocknal_close_conn_and_siblings (conn, rc);
1298                         return -EPROTO;
1299                 }
1300                 /* Fall through */
1301
1302         case SOCKNAL_RX_SLOP:
1303                 /* starting new packet? */
1304                 if (ksocknal_new_packet (conn, conn->ksnc_rx_nob_left))
1305                         return 0;       /* come back later */
1306                 goto again;          /* try to finish reading slop now */
1307
1308         default:
1309                 break;
1310         }
1311
1312         /* Not Reached */
1313         LBUG ();
1314         return -EINVAL;                /* keep gcc happy */
1315 }
1316
1317 int
1318 ksocknal_recv (lnet_ni_t *ni, void *private, lnet_msg_t *msg, int delayed,
1319                unsigned int niov, struct iovec *iov, lnet_kiov_t *kiov,
1320                unsigned int offset, unsigned int mlen, unsigned int rlen)
1321 {
1322         ksock_conn_t  *conn = (ksock_conn_t *)private;
1323         ksock_sched_t *sched = conn->ksnc_scheduler;
1324
1325         LASSERT (mlen <= rlen);
1326         LASSERT (niov <= LNET_MAX_IOV);
1327
1328         conn->ksnc_cookie = msg;
1329         conn->ksnc_rx_nob_wanted = mlen;
1330         conn->ksnc_rx_nob_left   = rlen;
1331
1332         if (mlen == 0 || iov != NULL) {
1333                 conn->ksnc_rx_nkiov = 0;
1334                 conn->ksnc_rx_kiov = NULL;
1335                 conn->ksnc_rx_iov = conn->ksnc_rx_iov_space.iov;
1336                 conn->ksnc_rx_niov =
1337                         lnet_extract_iov(LNET_MAX_IOV, conn->ksnc_rx_iov,
1338                                          niov, iov, offset, mlen);
1339         } else {
1340                 conn->ksnc_rx_niov = 0;
1341                 conn->ksnc_rx_iov  = NULL;
1342                 conn->ksnc_rx_kiov = conn->ksnc_rx_iov_space.kiov;
1343                 conn->ksnc_rx_nkiov =
1344                         lnet_extract_kiov(LNET_MAX_IOV, conn->ksnc_rx_kiov,
1345                                           niov, kiov, offset, mlen);
1346         }
1347
1348         LASSERT (mlen ==
1349                  lnet_iov_nob (conn->ksnc_rx_niov, conn->ksnc_rx_iov) +
1350                  lnet_kiov_nob (conn->ksnc_rx_nkiov, conn->ksnc_rx_kiov));
1351
1352         LASSERT (conn->ksnc_rx_scheduled);
1353
1354         spin_lock_bh(&sched->kss_lock);
1355
1356         switch (conn->ksnc_rx_state) {
1357         case SOCKNAL_RX_PARSE_WAIT:
1358                 list_add_tail(&conn->ksnc_rx_list, &sched->kss_rx_conns);
1359                 wake_up (&sched->kss_waitq);
1360                 LASSERT (conn->ksnc_rx_ready);
1361                 break;
1362
1363         case SOCKNAL_RX_PARSE:
1364                 /* scheduler hasn't noticed I'm parsing yet */
1365                 break;
1366         }
1367
1368         conn->ksnc_rx_state = SOCKNAL_RX_LNET_PAYLOAD;
1369
1370         spin_unlock_bh(&sched->kss_lock);
1371         ksocknal_conn_decref(conn);
1372         return 0;
1373 }
1374
1375 static inline int
1376 ksocknal_sched_cansleep(ksock_sched_t *sched)
1377 {
1378         int        rc;
1379
1380         spin_lock_bh(&sched->kss_lock);
1381
1382         rc = (!ksocknal_data.ksnd_shuttingdown &&
1383               list_empty(&sched->kss_rx_conns) &&
1384               list_empty(&sched->kss_tx_conns));
1385
1386         spin_unlock_bh(&sched->kss_lock);
1387         return rc;
1388 }
1389
1390 int ksocknal_scheduler(void *arg)
1391 {
1392         struct ksock_sched_info *info;
1393         ksock_sched_t           *sched;
1394         ksock_conn_t            *conn;
1395         ksock_tx_t              *tx;
1396         int                     rc;
1397         int                     nloops = 0;
1398         long                    id = (long)arg;
1399
1400         info = ksocknal_data.ksnd_sched_info[KSOCK_THREAD_CPT(id)];
1401         sched = &info->ksi_scheds[KSOCK_THREAD_SID(id)];
1402
1403         cfs_block_allsigs();
1404
1405         rc = cfs_cpt_bind(lnet_cpt_table(), info->ksi_cpt);
1406         if (rc != 0) {
1407                 CERROR("Can't set CPT affinity to %d: %d\n",
1408                        info->ksi_cpt, rc);
1409         }
1410
1411         spin_lock_bh(&sched->kss_lock);
1412
1413         while (!ksocknal_data.ksnd_shuttingdown) {
1414                 int did_something = 0;
1415
1416                 /* Ensure I progress everything semi-fairly */
1417
1418                 if (!list_empty (&sched->kss_rx_conns)) {
1419                         conn = list_entry(sched->kss_rx_conns.next,
1420                                               ksock_conn_t, ksnc_rx_list);
1421                         list_del(&conn->ksnc_rx_list);
1422
1423                         LASSERT(conn->ksnc_rx_scheduled);
1424                         LASSERT(conn->ksnc_rx_ready);
1425
1426                         /* clear rx_ready in case receive isn't complete.
1427                          * Do it BEFORE we call process_recv, since
1428                          * data_ready can set it any time after we release
1429                          * kss_lock. */
1430                         conn->ksnc_rx_ready = 0;
1431                         spin_unlock_bh(&sched->kss_lock);
1432
1433                         rc = ksocknal_process_receive(conn);
1434
1435                         spin_lock_bh(&sched->kss_lock);
1436
1437                         /* I'm the only one that can clear this flag */
1438                         LASSERT(conn->ksnc_rx_scheduled);
1439
1440                         /* Did process_receive get everything it wanted? */
1441                         if (rc == 0)
1442                                 conn->ksnc_rx_ready = 1;
1443
1444                         if (conn->ksnc_rx_state == SOCKNAL_RX_PARSE) {
1445                                 /* Conn blocked waiting for ksocknal_recv()
1446                                  * I change its state (under lock) to signal
1447                                  * it can be rescheduled */
1448                                 conn->ksnc_rx_state = SOCKNAL_RX_PARSE_WAIT;
1449                         } else if (conn->ksnc_rx_ready) {
1450                                 /* reschedule for rx */
1451                                 list_add_tail (&conn->ksnc_rx_list,
1452                                                    &sched->kss_rx_conns);
1453                         } else {
1454                                 conn->ksnc_rx_scheduled = 0;
1455                                 /* drop my ref */
1456                                 ksocknal_conn_decref(conn);
1457                         }
1458
1459                         did_something = 1;
1460                 }
1461
1462                 if (!list_empty (&sched->kss_tx_conns)) {
1463                         LIST_HEAD    (zlist);
1464
1465                         if (!list_empty(&sched->kss_zombie_noop_txs)) {
1466                                 list_add(&zlist,
1467                                              &sched->kss_zombie_noop_txs);
1468                                 list_del_init(&sched->kss_zombie_noop_txs);
1469                         }
1470
1471                         conn = list_entry(sched->kss_tx_conns.next,
1472                                               ksock_conn_t, ksnc_tx_list);
1473                         list_del (&conn->ksnc_tx_list);
1474
1475                         LASSERT(conn->ksnc_tx_scheduled);
1476                         LASSERT(conn->ksnc_tx_ready);
1477                         LASSERT(!list_empty(&conn->ksnc_tx_queue));
1478
1479                         tx = list_entry(conn->ksnc_tx_queue.next,
1480                                             ksock_tx_t, tx_list);
1481
1482                         if (conn->ksnc_tx_carrier == tx)
1483                                 ksocknal_next_tx_carrier(conn);
1484
1485                         /* dequeue now so empty list => more to send */
1486                         list_del(&tx->tx_list);
1487
1488                         /* Clear tx_ready in case send isn't complete.  Do
1489                          * it BEFORE we call process_transmit, since
1490                          * write_space can set it any time after we release
1491                          * kss_lock. */
1492                         conn->ksnc_tx_ready = 0;
1493                         spin_unlock_bh(&sched->kss_lock);
1494
1495                         if (!list_empty(&zlist)) {
1496                                 /* free zombie noop txs, it's fast because
1497                                  * noop txs are just put in freelist */
1498                                 ksocknal_txlist_done(NULL, &zlist, 0);
1499                         }
1500
1501                         rc = ksocknal_process_transmit(conn, tx);
1502
1503                         if (rc == -ENOMEM || rc == -EAGAIN) {
1504                                 /* Incomplete send: replace tx on HEAD of tx_queue */
1505                                 spin_lock_bh(&sched->kss_lock);
1506                                 list_add(&tx->tx_list,
1507                                              &conn->ksnc_tx_queue);
1508                         } else {
1509                                 /* Complete send; tx -ref */
1510                                 ksocknal_tx_decref(tx);
1511
1512                                 spin_lock_bh(&sched->kss_lock);
1513                                 /* assume space for more */
1514                                 conn->ksnc_tx_ready = 1;
1515                         }
1516
1517                         if (rc == -ENOMEM) {
1518                                 /* Do nothing; after a short timeout, this
1519                                  * conn will be reposted on kss_tx_conns. */
1520                         } else if (conn->ksnc_tx_ready &&
1521                                    !list_empty (&conn->ksnc_tx_queue)) {
1522                                 /* reschedule for tx */
1523                                 list_add_tail (&conn->ksnc_tx_list,
1524                                                    &sched->kss_tx_conns);
1525                         } else {
1526                                 conn->ksnc_tx_scheduled = 0;
1527                                 /* drop my ref */
1528                                 ksocknal_conn_decref(conn);
1529                         }
1530
1531                         did_something = 1;
1532                 }
1533                 if (!did_something ||      /* nothing to do */
1534                     ++nloops == SOCKNAL_RESCHED) { /* hogging CPU? */
1535                         spin_unlock_bh(&sched->kss_lock);
1536
1537                         nloops = 0;
1538
1539                         if (!did_something) {   /* wait for something to do */
1540                                 rc = wait_event_interruptible_exclusive(
1541                                         sched->kss_waitq,
1542                                         !ksocknal_sched_cansleep(sched));
1543                                 LASSERT (rc == 0);
1544                         } else {
1545                                 cond_resched();
1546                         }
1547
1548                         spin_lock_bh(&sched->kss_lock);
1549                 }
1550         }
1551
1552         spin_unlock_bh(&sched->kss_lock);
1553         ksocknal_thread_fini();
1554         return 0;
1555 }
1556
1557 /*
1558  * Add connection to kss_rx_conns of scheduler
1559  * and wakeup the scheduler.
1560  */
1561 void ksocknal_read_callback (ksock_conn_t *conn)
1562 {
1563         ksock_sched_t *sched;
1564
1565         sched = conn->ksnc_scheduler;
1566
1567         spin_lock_bh(&sched->kss_lock);
1568
1569         conn->ksnc_rx_ready = 1;
1570
1571         if (!conn->ksnc_rx_scheduled) {  /* not being progressed */
1572                 list_add_tail(&conn->ksnc_rx_list,
1573                                   &sched->kss_rx_conns);
1574                 conn->ksnc_rx_scheduled = 1;
1575                 /* extra ref for scheduler */
1576                 ksocknal_conn_addref(conn);
1577
1578                 wake_up (&sched->kss_waitq);
1579         }
1580         spin_unlock_bh(&sched->kss_lock);
1581 }
1582
1583 /*
1584  * Add connection to kss_tx_conns of scheduler
1585  * and wakeup the scheduler.
1586  */
1587 void ksocknal_write_callback (ksock_conn_t *conn)
1588 {
1589         ksock_sched_t *sched;
1590
1591         sched = conn->ksnc_scheduler;
1592
1593         spin_lock_bh(&sched->kss_lock);
1594
1595         conn->ksnc_tx_ready = 1;
1596
1597         if (!conn->ksnc_tx_scheduled && /* not being progressed */
1598             !list_empty(&conn->ksnc_tx_queue)) { /* packets to send */
1599                 list_add_tail (&conn->ksnc_tx_list,
1600                                    &sched->kss_tx_conns);
1601                 conn->ksnc_tx_scheduled = 1;
1602                 /* extra ref for scheduler */
1603                 ksocknal_conn_addref(conn);
1604
1605                 wake_up (&sched->kss_waitq);
1606         }
1607
1608         spin_unlock_bh(&sched->kss_lock);
1609 }
1610
1611 static ksock_proto_t *
1612 ksocknal_parse_proto_version (ksock_hello_msg_t *hello)
1613 {
1614         __u32   version = 0;
1615
1616         if (hello->kshm_magic == LNET_PROTO_MAGIC)
1617                 version = hello->kshm_version;
1618         else if (hello->kshm_magic == __swab32(LNET_PROTO_MAGIC))
1619                 version = __swab32(hello->kshm_version);
1620
1621         if (version != 0) {
1622 #if SOCKNAL_VERSION_DEBUG
1623                 if (*ksocknal_tunables.ksnd_protocol == 1)
1624                         return NULL;
1625
1626                 if (*ksocknal_tunables.ksnd_protocol == 2 &&
1627                     version == KSOCK_PROTO_V3)
1628                         return NULL;
1629 #endif
1630                 if (version == KSOCK_PROTO_V2)
1631                         return &ksocknal_protocol_v2x;
1632
1633                 if (version == KSOCK_PROTO_V3)
1634                         return &ksocknal_protocol_v3x;
1635
1636                 return NULL;
1637         }
1638
1639         if (hello->kshm_magic == le32_to_cpu(LNET_PROTO_TCP_MAGIC)) {
1640                 lnet_magicversion_t *hmv = (lnet_magicversion_t *)hello;
1641
1642                 CLASSERT (sizeof (lnet_magicversion_t) ==
1643                           offsetof (ksock_hello_msg_t, kshm_src_nid));
1644
1645                 if (hmv->version_major == cpu_to_le16 (KSOCK_PROTO_V1_MAJOR) &&
1646                     hmv->version_minor == cpu_to_le16 (KSOCK_PROTO_V1_MINOR))
1647                         return &ksocknal_protocol_v1x;
1648         }
1649
1650         return NULL;
1651 }
1652
1653 int
1654 ksocknal_send_hello (lnet_ni_t *ni, ksock_conn_t *conn,
1655                      lnet_nid_t peer_nid, ksock_hello_msg_t *hello)
1656 {
1657         /* CAVEAT EMPTOR: this byte flips 'ipaddrs' */
1658         ksock_net_t      *net = (ksock_net_t *)ni->ni_data;
1659
1660         LASSERT (hello->kshm_nips <= LNET_MAX_INTERFACES);
1661
1662         /* rely on caller to hold a ref on socket so it wouldn't disappear */
1663         LASSERT (conn->ksnc_proto != NULL);
1664
1665         hello->kshm_src_nid      = ni->ni_nid;
1666         hello->kshm_dst_nid      = peer_nid;
1667         hello->kshm_src_pid      = the_lnet.ln_pid;
1668
1669         hello->kshm_src_incarnation = net->ksnn_incarnation;
1670         hello->kshm_ctype          = conn->ksnc_type;
1671
1672         return conn->ksnc_proto->pro_send_hello(conn, hello);
1673 }
1674
1675 static int
1676 ksocknal_invert_type(int type)
1677 {
1678         switch (type) {
1679         case SOCKLND_CONN_ANY:
1680         case SOCKLND_CONN_CONTROL:
1681                 return type;
1682         case SOCKLND_CONN_BULK_IN:
1683                 return SOCKLND_CONN_BULK_OUT;
1684         case SOCKLND_CONN_BULK_OUT:
1685                 return SOCKLND_CONN_BULK_IN;
1686         default:
1687                 return SOCKLND_CONN_NONE;
1688         }
1689 }
1690
1691 int
1692 ksocknal_recv_hello (lnet_ni_t *ni, ksock_conn_t *conn,
1693                      ksock_hello_msg_t *hello, lnet_process_id_t *peerid,
1694                      __u64 *incarnation)
1695 {
1696         /* Return < 0   fatal error
1697          *      0         success
1698          *      EALREADY   lost connection race
1699          *      EPROTO     protocol version mismatch
1700          */
1701         struct socket   *sock = conn->ksnc_sock;
1702         int               active = (conn->ksnc_proto != NULL);
1703         int               timeout;
1704         int               proto_match;
1705         int               rc;
1706         ksock_proto_t       *proto;
1707         lnet_process_id_t    recv_id;
1708
1709         /* socket type set on active connections - not set on passive */
1710         LASSERT (!active == !(conn->ksnc_type != SOCKLND_CONN_NONE));
1711
1712         timeout = active ? *ksocknal_tunables.ksnd_timeout :
1713                             lnet_acceptor_timeout();
1714
1715         rc = libcfs_sock_read(sock, &hello->kshm_magic, sizeof (hello->kshm_magic), timeout);
1716         if (rc != 0) {
1717                 CERROR("Error %d reading HELLO from %pI4h\n",
1718                         rc, &conn->ksnc_ipaddr);
1719                 LASSERT (rc < 0);
1720                 return rc;
1721         }
1722
1723         if (hello->kshm_magic != LNET_PROTO_MAGIC &&
1724             hello->kshm_magic != __swab32(LNET_PROTO_MAGIC) &&
1725             hello->kshm_magic != le32_to_cpu (LNET_PROTO_TCP_MAGIC)) {
1726                 /* Unexpected magic! */
1727                 CERROR("Bad magic(1) %#08x (%#08x expected) from "
1728                         "%pI4h\n", __cpu_to_le32 (hello->kshm_magic),
1729                         LNET_PROTO_TCP_MAGIC,
1730                         &conn->ksnc_ipaddr);
1731                 return -EPROTO;
1732         }
1733
1734         rc = libcfs_sock_read(sock, &hello->kshm_version,
1735                               sizeof(hello->kshm_version), timeout);
1736         if (rc != 0) {
1737                 CERROR("Error %d reading HELLO from %pI4h\n",
1738                         rc, &conn->ksnc_ipaddr);
1739                 LASSERT (rc < 0);
1740                 return rc;
1741         }
1742
1743         proto = ksocknal_parse_proto_version(hello);
1744         if (proto == NULL) {
1745                 if (!active) {
1746                         /* unknown protocol from peer, tell peer my protocol */
1747                         conn->ksnc_proto = &ksocknal_protocol_v3x;
1748 #if SOCKNAL_VERSION_DEBUG
1749                         if (*ksocknal_tunables.ksnd_protocol == 2)
1750                                 conn->ksnc_proto = &ksocknal_protocol_v2x;
1751                         else if (*ksocknal_tunables.ksnd_protocol == 1)
1752                                 conn->ksnc_proto = &ksocknal_protocol_v1x;
1753 #endif
1754                         hello->kshm_nips = 0;
1755                         ksocknal_send_hello(ni, conn, ni->ni_nid, hello);
1756                 }
1757
1758                 CERROR("Unknown protocol version (%d.x expected)"
1759                         " from %pI4h\n",
1760                         conn->ksnc_proto->pro_version,
1761                         &conn->ksnc_ipaddr);
1762
1763                 return -EPROTO;
1764         }
1765
1766         proto_match = (conn->ksnc_proto == proto);
1767         conn->ksnc_proto = proto;
1768
1769         /* receive the rest of hello message anyway */
1770         rc = conn->ksnc_proto->pro_recv_hello(conn, hello, timeout);
1771         if (rc != 0) {
1772                 CERROR("Error %d reading or checking hello from from %pI4h\n",
1773                        rc, &conn->ksnc_ipaddr);
1774                 LASSERT (rc < 0);
1775                 return rc;
1776         }
1777
1778         *incarnation = hello->kshm_src_incarnation;
1779
1780         if (hello->kshm_src_nid == LNET_NID_ANY) {
1781                 CERROR("Expecting a HELLO hdr with a NID, but got LNET_NID_ANY"
1782                        "from %pI4h\n", &conn->ksnc_ipaddr);
1783                 return -EPROTO;
1784         }
1785
1786         if (!active &&
1787             conn->ksnc_port > LNET_ACCEPTOR_MAX_RESERVED_PORT) {
1788                 /* Userspace NAL assigns peer process ID from socket */
1789                 recv_id.pid = conn->ksnc_port | LNET_PID_USERFLAG;
1790                 recv_id.nid = LNET_MKNID(LNET_NIDNET(ni->ni_nid), conn->ksnc_ipaddr);
1791         } else {
1792                 recv_id.nid = hello->kshm_src_nid;
1793                 recv_id.pid = hello->kshm_src_pid;
1794         }
1795
1796         if (!active) {
1797                 *peerid = recv_id;
1798
1799                 /* peer determines type */
1800                 conn->ksnc_type = ksocknal_invert_type(hello->kshm_ctype);
1801                 if (conn->ksnc_type == SOCKLND_CONN_NONE) {
1802                         CERROR("Unexpected type %d from %s ip %pI4h\n",
1803                                 hello->kshm_ctype, libcfs_id2str(*peerid),
1804                                 &conn->ksnc_ipaddr);
1805                         return -EPROTO;
1806                 }
1807
1808                 return 0;
1809         }
1810
1811         if (peerid->pid != recv_id.pid ||
1812             peerid->nid != recv_id.nid) {
1813                 LCONSOLE_ERROR_MSG(0x130, "Connected successfully to %s on host"
1814                                    " %pI4h, but they claimed they were "
1815                                    "%s; please check your Lustre "
1816                                    "configuration.\n",
1817                                    libcfs_id2str(*peerid),
1818                                    &conn->ksnc_ipaddr,
1819                                    libcfs_id2str(recv_id));
1820                 return -EPROTO;
1821         }
1822
1823         if (hello->kshm_ctype == SOCKLND_CONN_NONE) {
1824                 /* Possible protocol mismatch or I lost the connection race */
1825                 return proto_match ? EALREADY : EPROTO;
1826         }
1827
1828         if (ksocknal_invert_type(hello->kshm_ctype) != conn->ksnc_type) {
1829                 CERROR("Mismatched types: me %d, %s ip %pI4h %d\n",
1830                         conn->ksnc_type, libcfs_id2str(*peerid),
1831                         &conn->ksnc_ipaddr,
1832                         hello->kshm_ctype);
1833                 return -EPROTO;
1834         }
1835
1836         return 0;
1837 }
1838
1839 static int
1840 ksocknal_connect (ksock_route_t *route)
1841 {
1842         LIST_HEAD    (zombies);
1843         ksock_peer_t     *peer = route->ksnr_peer;
1844         int            type;
1845         int            wanted;
1846         struct socket     *sock;
1847         unsigned long   deadline;
1848         int            retry_later = 0;
1849         int            rc = 0;
1850
1851         deadline = cfs_time_add(cfs_time_current(),
1852                                 cfs_time_seconds(*ksocknal_tunables.ksnd_timeout));
1853
1854         write_lock_bh(&ksocknal_data.ksnd_global_lock);
1855
1856         LASSERT (route->ksnr_scheduled);
1857         LASSERT (!route->ksnr_connecting);
1858
1859         route->ksnr_connecting = 1;
1860
1861         for (;;) {
1862                 wanted = ksocknal_route_mask() & ~route->ksnr_connected;
1863
1864                 /* stop connecting if peer/route got closed under me, or
1865                  * route got connected while queued */
1866                 if (peer->ksnp_closing || route->ksnr_deleted ||
1867                     wanted == 0) {
1868                         retry_later = 0;
1869                         break;
1870                 }
1871
1872                 /* reschedule if peer is connecting to me */
1873                 if (peer->ksnp_accepting > 0) {
1874                         CDEBUG(D_NET,
1875                                "peer %s(%d) already connecting to me, retry later.\n",
1876                                libcfs_nid2str(peer->ksnp_id.nid), peer->ksnp_accepting);
1877                         retry_later = 1;
1878                 }
1879
1880                 if (retry_later) /* needs reschedule */
1881                         break;
1882
1883                 if ((wanted & (1 << SOCKLND_CONN_ANY)) != 0) {
1884                         type = SOCKLND_CONN_ANY;
1885                 } else if ((wanted & (1 << SOCKLND_CONN_CONTROL)) != 0) {
1886                         type = SOCKLND_CONN_CONTROL;
1887                 } else if ((wanted & (1 << SOCKLND_CONN_BULK_IN)) != 0) {
1888                         type = SOCKLND_CONN_BULK_IN;
1889                 } else {
1890                         LASSERT ((wanted & (1 << SOCKLND_CONN_BULK_OUT)) != 0);
1891                         type = SOCKLND_CONN_BULK_OUT;
1892                 }
1893
1894                 write_unlock_bh(&ksocknal_data.ksnd_global_lock);
1895
1896                 if (cfs_time_aftereq(cfs_time_current(), deadline)) {
1897                         rc = -ETIMEDOUT;
1898                         lnet_connect_console_error(rc, peer->ksnp_id.nid,
1899                                                    route->ksnr_ipaddr,
1900                                                    route->ksnr_port);
1901                         goto failed;
1902                 }
1903
1904                 rc = lnet_connect(&sock, peer->ksnp_id.nid,
1905                                   route->ksnr_myipaddr,
1906                                   route->ksnr_ipaddr, route->ksnr_port);
1907                 if (rc != 0)
1908                         goto failed;
1909
1910                 rc = ksocknal_create_conn(peer->ksnp_ni, route, sock, type);
1911                 if (rc < 0) {
1912                         lnet_connect_console_error(rc, peer->ksnp_id.nid,
1913                                                    route->ksnr_ipaddr,
1914                                                    route->ksnr_port);
1915                         goto failed;
1916                 }
1917
1918                 /* A +ve RC means I have to retry because I lost the connection
1919                  * race or I have to renegotiate protocol version */
1920                 retry_later = (rc != 0);
1921                 if (retry_later)
1922                         CDEBUG(D_NET, "peer %s: conn race, retry later.\n",
1923                                libcfs_nid2str(peer->ksnp_id.nid));
1924
1925                 write_lock_bh(&ksocknal_data.ksnd_global_lock);
1926         }
1927
1928         route->ksnr_scheduled = 0;
1929         route->ksnr_connecting = 0;
1930
1931         if (retry_later) {
1932                 /* re-queue for attention; this frees me up to handle
1933                  * the peer's incoming connection request */
1934
1935                 if (rc == EALREADY ||
1936                     (rc == 0 && peer->ksnp_accepting > 0)) {
1937                         /* We want to introduce a delay before next
1938                          * attempt to connect if we lost conn race,
1939                          * but the race is resolved quickly usually,
1940                          * so min_reconnectms should be good heuristic */
1941                         route->ksnr_retry_interval =
1942                                 cfs_time_seconds(*ksocknal_tunables.ksnd_min_reconnectms)/1000;
1943                         route->ksnr_timeout = cfs_time_add(cfs_time_current(),
1944                                                            route->ksnr_retry_interval);
1945                 }
1946
1947                 ksocknal_launch_connection_locked(route);
1948         }
1949
1950         write_unlock_bh(&ksocknal_data.ksnd_global_lock);
1951         return retry_later;
1952
1953  failed:
1954         write_lock_bh(&ksocknal_data.ksnd_global_lock);
1955
1956         route->ksnr_scheduled = 0;
1957         route->ksnr_connecting = 0;
1958
1959         /* This is a retry rather than a new connection */
1960         route->ksnr_retry_interval *= 2;
1961         route->ksnr_retry_interval =
1962                 MAX(route->ksnr_retry_interval,
1963                     cfs_time_seconds(*ksocknal_tunables.ksnd_min_reconnectms)/1000);
1964         route->ksnr_retry_interval =
1965                 MIN(route->ksnr_retry_interval,
1966                     cfs_time_seconds(*ksocknal_tunables.ksnd_max_reconnectms)/1000);
1967
1968         LASSERT (route->ksnr_retry_interval != 0);
1969         route->ksnr_timeout = cfs_time_add(cfs_time_current(),
1970                                            route->ksnr_retry_interval);
1971
1972         if (!list_empty(&peer->ksnp_tx_queue) &&
1973             peer->ksnp_accepting == 0 &&
1974             ksocknal_find_connecting_route_locked(peer) == NULL) {
1975                 ksock_conn_t *conn;
1976
1977                 /* ksnp_tx_queue is queued on a conn on successful
1978                  * connection for V1.x and V2.x */
1979                 if (!list_empty (&peer->ksnp_conns)) {
1980                         conn = list_entry(peer->ksnp_conns.next,
1981                                               ksock_conn_t, ksnc_list);
1982                         LASSERT (conn->ksnc_proto == &ksocknal_protocol_v3x);
1983                 }
1984
1985                 /* take all the blocked packets while I've got the lock and
1986                  * complete below... */
1987                 list_splice_init(&peer->ksnp_tx_queue, &zombies);
1988         }
1989
1990 #if 0      /* irrelevant with only eager routes */
1991         if (!route->ksnr_deleted) {
1992                 /* make this route least-favourite for re-selection */
1993                 list_del(&route->ksnr_list);
1994                 list_add_tail(&route->ksnr_list, &peer->ksnp_routes);
1995         }
1996 #endif
1997         write_unlock_bh(&ksocknal_data.ksnd_global_lock);
1998
1999         ksocknal_peer_failed(peer);
2000         ksocknal_txlist_done(peer->ksnp_ni, &zombies, 1);
2001         return 0;
2002 }
2003
2004 /*
2005  * check whether we need to create more connds.
2006  * It will try to create new thread if it's necessary, @timeout can
2007  * be updated if failed to create, so caller wouldn't keep try while
2008  * running out of resource.
2009  */
2010 static int
2011 ksocknal_connd_check_start(long sec, long *timeout)
2012 {
2013         char name[16];
2014         int rc;
2015         int total = ksocknal_data.ksnd_connd_starting +
2016                     ksocknal_data.ksnd_connd_running;
2017
2018         if (unlikely(ksocknal_data.ksnd_init < SOCKNAL_INIT_ALL)) {
2019                 /* still in initializing */
2020                 return 0;
2021         }
2022
2023         if (total >= *ksocknal_tunables.ksnd_nconnds_max ||
2024             total > ksocknal_data.ksnd_connd_connecting + SOCKNAL_CONND_RESV) {
2025                 /* can't create more connd, or still have enough
2026                  * threads to handle more connecting */
2027                 return 0;
2028         }
2029
2030         if (list_empty(&ksocknal_data.ksnd_connd_routes)) {
2031                 /* no pending connecting request */
2032                 return 0;
2033         }
2034
2035         if (sec - ksocknal_data.ksnd_connd_failed_stamp <= 1) {
2036                 /* may run out of resource, retry later */
2037                 *timeout = cfs_time_seconds(1);
2038                 return 0;
2039         }
2040
2041         if (ksocknal_data.ksnd_connd_starting > 0) {
2042                 /* serialize starting to avoid flood */
2043                 return 0;
2044         }
2045
2046         ksocknal_data.ksnd_connd_starting_stamp = sec;
2047         ksocknal_data.ksnd_connd_starting++;
2048         spin_unlock_bh(&ksocknal_data.ksnd_connd_lock);
2049
2050         /* NB: total is the next id */
2051         snprintf(name, sizeof(name), "socknal_cd%02d", total);
2052         rc = ksocknal_thread_start(ksocknal_connd, NULL, name);
2053
2054         spin_lock_bh(&ksocknal_data.ksnd_connd_lock);
2055         if (rc == 0)
2056                 return 1;
2057
2058         /* we tried ... */
2059         LASSERT(ksocknal_data.ksnd_connd_starting > 0);
2060         ksocknal_data.ksnd_connd_starting--;
2061         ksocknal_data.ksnd_connd_failed_stamp = get_seconds();
2062
2063         return 1;
2064 }
2065
2066 /*
2067  * check whether current thread can exit, it will return 1 if there are too
2068  * many threads and no creating in past 120 seconds.
2069  * Also, this function may update @timeout to make caller come back
2070  * again to recheck these conditions.
2071  */
2072 static int
2073 ksocknal_connd_check_stop(long sec, long *timeout)
2074 {
2075         int val;
2076
2077         if (unlikely(ksocknal_data.ksnd_init < SOCKNAL_INIT_ALL)) {
2078                 /* still in initializing */
2079                 return 0;
2080         }
2081
2082         if (ksocknal_data.ksnd_connd_starting > 0) {
2083                 /* in progress of starting new thread */
2084                 return 0;
2085         }
2086
2087         if (ksocknal_data.ksnd_connd_running <=
2088             *ksocknal_tunables.ksnd_nconnds) { /* can't shrink */
2089                 return 0;
2090         }
2091
2092         /* created thread in past 120 seconds? */
2093         val = (int)(ksocknal_data.ksnd_connd_starting_stamp +
2094                     SOCKNAL_CONND_TIMEOUT - sec);
2095
2096         *timeout = (val > 0) ? cfs_time_seconds(val) :
2097                                cfs_time_seconds(SOCKNAL_CONND_TIMEOUT);
2098         if (val > 0)
2099                 return 0;
2100
2101         /* no creating in past 120 seconds */
2102
2103         return ksocknal_data.ksnd_connd_running >
2104                ksocknal_data.ksnd_connd_connecting + SOCKNAL_CONND_RESV;
2105 }
2106
2107 /* Go through connd_routes queue looking for a route that we can process
2108  * right now, @timeout_p can be updated if we need to come back later */
2109 static ksock_route_t *
2110 ksocknal_connd_get_route_locked(signed long *timeout_p)
2111 {
2112         ksock_route_t *route;
2113         unsigned long     now;
2114
2115         now = cfs_time_current();
2116
2117         /* connd_routes can contain both pending and ordinary routes */
2118         list_for_each_entry (route, &ksocknal_data.ksnd_connd_routes,
2119                                  ksnr_connd_list) {
2120
2121                 if (route->ksnr_retry_interval == 0 ||
2122                     cfs_time_aftereq(now, route->ksnr_timeout))
2123                         return route;
2124
2125                 if (*timeout_p == MAX_SCHEDULE_TIMEOUT ||
2126                     (int)*timeout_p > (int)(route->ksnr_timeout - now))
2127                         *timeout_p = (int)(route->ksnr_timeout - now);
2128         }
2129
2130         return NULL;
2131 }
2132
2133 int
2134 ksocknal_connd (void *arg)
2135 {
2136         spinlock_t    *connd_lock = &ksocknal_data.ksnd_connd_lock;
2137         ksock_connreq_t   *cr;
2138         wait_queue_t     wait;
2139         int             nloops = 0;
2140         int             cons_retry = 0;
2141
2142         cfs_block_allsigs ();
2143
2144         init_waitqueue_entry(&wait, current);
2145
2146         spin_lock_bh(connd_lock);
2147
2148         LASSERT(ksocknal_data.ksnd_connd_starting > 0);
2149         ksocknal_data.ksnd_connd_starting--;
2150         ksocknal_data.ksnd_connd_running++;
2151
2152         while (!ksocknal_data.ksnd_shuttingdown) {
2153                 ksock_route_t *route = NULL;
2154                 long sec = get_seconds();
2155                 long timeout = MAX_SCHEDULE_TIMEOUT;
2156                 int  dropped_lock = 0;
2157
2158                 if (ksocknal_connd_check_stop(sec, &timeout)) {
2159                         /* wakeup another one to check stop */
2160                         wake_up(&ksocknal_data.ksnd_connd_waitq);
2161                         break;
2162                 }
2163
2164                 if (ksocknal_connd_check_start(sec, &timeout)) {
2165                         /* created new thread */
2166                         dropped_lock = 1;
2167                 }
2168
2169                 if (!list_empty(&ksocknal_data.ksnd_connd_connreqs)) {
2170                         /* Connection accepted by the listener */
2171                         cr = list_entry(ksocknal_data.ksnd_connd_connreqs. \
2172                                             next, ksock_connreq_t, ksncr_list);
2173
2174                         list_del(&cr->ksncr_list);
2175                         spin_unlock_bh(connd_lock);
2176                         dropped_lock = 1;
2177
2178                         ksocknal_create_conn(cr->ksncr_ni, NULL,
2179                                              cr->ksncr_sock, SOCKLND_CONN_NONE);
2180                         lnet_ni_decref(cr->ksncr_ni);
2181                         LIBCFS_FREE(cr, sizeof(*cr));
2182
2183                         spin_lock_bh(connd_lock);
2184                 }
2185
2186                 /* Only handle an outgoing connection request if there
2187                  * is a thread left to handle incoming connections and
2188                  * create new connd */
2189                 if (ksocknal_data.ksnd_connd_connecting + SOCKNAL_CONND_RESV <
2190                     ksocknal_data.ksnd_connd_running) {
2191                         route = ksocknal_connd_get_route_locked(&timeout);
2192                 }
2193                 if (route != NULL) {
2194                         list_del (&route->ksnr_connd_list);
2195                         ksocknal_data.ksnd_connd_connecting++;
2196                         spin_unlock_bh(connd_lock);
2197                         dropped_lock = 1;
2198
2199                         if (ksocknal_connect(route)) {
2200                                 /* consecutive retry */
2201                                 if (cons_retry++ > SOCKNAL_INSANITY_RECONN) {
2202                                         CWARN("massive consecutive "
2203                                               "re-connecting to %pI4h\n",
2204                                               &route->ksnr_ipaddr);
2205                                         cons_retry = 0;
2206                                 }
2207                         } else {
2208                                 cons_retry = 0;
2209                         }
2210
2211                         ksocknal_route_decref(route);
2212
2213                         spin_lock_bh(connd_lock);
2214                         ksocknal_data.ksnd_connd_connecting--;
2215                 }
2216
2217                 if (dropped_lock) {
2218                         if (++nloops < SOCKNAL_RESCHED)
2219                                 continue;
2220                         spin_unlock_bh(connd_lock);
2221                         nloops = 0;
2222                         cond_resched();
2223                         spin_lock_bh(connd_lock);
2224                         continue;
2225                 }
2226
2227                 /* Nothing to do for 'timeout'  */
2228                 set_current_state(TASK_INTERRUPTIBLE);
2229                 add_wait_queue_exclusive(&ksocknal_data.ksnd_connd_waitq, &wait);
2230                 spin_unlock_bh(connd_lock);
2231
2232                 nloops = 0;
2233                 schedule_timeout(timeout);
2234
2235                 remove_wait_queue(&ksocknal_data.ksnd_connd_waitq, &wait);
2236                 spin_lock_bh(connd_lock);
2237         }
2238         ksocknal_data.ksnd_connd_running--;
2239         spin_unlock_bh(connd_lock);
2240
2241         ksocknal_thread_fini();
2242         return 0;
2243 }
2244
2245 static ksock_conn_t *
2246 ksocknal_find_timed_out_conn (ksock_peer_t *peer)
2247 {
2248         /* We're called with a shared lock on ksnd_global_lock */
2249         ksock_conn_t      *conn;
2250         struct list_head        *ctmp;
2251
2252         list_for_each (ctmp, &peer->ksnp_conns) {
2253                 int     error;
2254                 conn = list_entry (ctmp, ksock_conn_t, ksnc_list);
2255
2256                 /* Don't need the {get,put}connsock dance to deref ksnc_sock */
2257                 LASSERT (!conn->ksnc_closing);
2258
2259                 /* SOCK_ERROR will reset error code of socket in
2260                  * some platform (like Darwin8.x) */
2261                 error = conn->ksnc_sock->sk->sk_err;
2262                 if (error != 0) {
2263                         ksocknal_conn_addref(conn);
2264
2265                         switch (error) {
2266                         case ECONNRESET:
2267                                 CNETERR("A connection with %s "
2268                                         "(%pI4h:%d) was reset; "
2269                                         "it may have rebooted.\n",
2270                                         libcfs_id2str(peer->ksnp_id),
2271                                         &conn->ksnc_ipaddr,
2272                                         conn->ksnc_port);
2273                                 break;
2274                         case ETIMEDOUT:
2275                                 CNETERR("A connection with %s "
2276                                         "(%pI4h:%d) timed out; the "
2277                                         "network or node may be down.\n",
2278                                         libcfs_id2str(peer->ksnp_id),
2279                                         &conn->ksnc_ipaddr,
2280                                         conn->ksnc_port);
2281                                 break;
2282                         default:
2283                                 CNETERR("An unexpected network error %d "
2284                                         "occurred with %s "
2285                                         "(%pI4h:%d\n", error,
2286                                         libcfs_id2str(peer->ksnp_id),
2287                                         &conn->ksnc_ipaddr,
2288                                         conn->ksnc_port);
2289                                 break;
2290                         }
2291
2292                         return conn;
2293                 }
2294
2295                 if (conn->ksnc_rx_started &&
2296                     cfs_time_aftereq(cfs_time_current(),
2297                                      conn->ksnc_rx_deadline)) {
2298                         /* Timed out incomplete incoming message */
2299                         ksocknal_conn_addref(conn);
2300                         CNETERR("Timeout receiving from %s (%pI4h:%d), "
2301                                 "state %d wanted %d left %d\n",
2302                                 libcfs_id2str(peer->ksnp_id),
2303                                 &conn->ksnc_ipaddr,
2304                                 conn->ksnc_port,
2305                                 conn->ksnc_rx_state,
2306                                 conn->ksnc_rx_nob_wanted,
2307                                 conn->ksnc_rx_nob_left);
2308                         return conn;
2309                 }
2310
2311                 if ((!list_empty(&conn->ksnc_tx_queue) ||
2312                      conn->ksnc_sock->sk->sk_wmem_queued != 0) &&
2313                     cfs_time_aftereq(cfs_time_current(),
2314                                      conn->ksnc_tx_deadline)) {
2315                         /* Timed out messages queued for sending or
2316                          * buffered in the socket's send buffer */
2317                         ksocknal_conn_addref(conn);
2318                         CNETERR("Timeout sending data to %s (%pI4h:%d) "
2319                                 "the network or that node may be down.\n",
2320                                 libcfs_id2str(peer->ksnp_id),
2321                                 &conn->ksnc_ipaddr,
2322                                 conn->ksnc_port);
2323                         return conn;
2324                 }
2325         }
2326
2327         return NULL;
2328 }
2329
2330 static inline void
2331 ksocknal_flush_stale_txs(ksock_peer_t *peer)
2332 {
2333         ksock_tx_t      *tx;
2334         LIST_HEAD      (stale_txs);
2335
2336         write_lock_bh(&ksocknal_data.ksnd_global_lock);
2337
2338         while (!list_empty (&peer->ksnp_tx_queue)) {
2339                 tx = list_entry (peer->ksnp_tx_queue.next,
2340                                      ksock_tx_t, tx_list);
2341
2342                 if (!cfs_time_aftereq(cfs_time_current(),
2343                                       tx->tx_deadline))
2344                         break;
2345
2346                 list_del (&tx->tx_list);
2347                 list_add_tail (&tx->tx_list, &stale_txs);
2348         }
2349
2350         write_unlock_bh(&ksocknal_data.ksnd_global_lock);
2351
2352         ksocknal_txlist_done(peer->ksnp_ni, &stale_txs, 1);
2353 }
2354
2355 static int
2356 ksocknal_send_keepalive_locked(ksock_peer_t *peer)
2357 {
2358         ksock_sched_t  *sched;
2359         ksock_conn_t   *conn;
2360         ksock_tx_t     *tx;
2361
2362         if (list_empty(&peer->ksnp_conns)) /* last_alive will be updated by create_conn */
2363                 return 0;
2364
2365         if (peer->ksnp_proto != &ksocknal_protocol_v3x)
2366                 return 0;
2367
2368         if (*ksocknal_tunables.ksnd_keepalive <= 0 ||
2369             time_before(cfs_time_current(),
2370                         cfs_time_add(peer->ksnp_last_alive,
2371                                      cfs_time_seconds(*ksocknal_tunables.ksnd_keepalive))))
2372                 return 0;
2373
2374         if (time_before(cfs_time_current(), peer->ksnp_send_keepalive))
2375                 return 0;
2376
2377         /* retry 10 secs later, so we wouldn't put pressure
2378          * on this peer if we failed to send keepalive this time */
2379         peer->ksnp_send_keepalive = cfs_time_shift(10);
2380
2381         conn = ksocknal_find_conn_locked(peer, NULL, 1);
2382         if (conn != NULL) {
2383                 sched = conn->ksnc_scheduler;
2384
2385                 spin_lock_bh(&sched->kss_lock);
2386                 if (!list_empty(&conn->ksnc_tx_queue)) {
2387                         spin_unlock_bh(&sched->kss_lock);
2388                         /* there is an queued ACK, don't need keepalive */
2389                         return 0;
2390                 }
2391
2392                 spin_unlock_bh(&sched->kss_lock);
2393         }
2394
2395         read_unlock(&ksocknal_data.ksnd_global_lock);
2396
2397         /* cookie = 1 is reserved for keepalive PING */
2398         tx = ksocknal_alloc_tx_noop(1, 1);
2399         if (tx == NULL) {
2400                 read_lock(&ksocknal_data.ksnd_global_lock);
2401                 return -ENOMEM;
2402         }
2403
2404         if (ksocknal_launch_packet(peer->ksnp_ni, tx, peer->ksnp_id) == 0) {
2405                 read_lock(&ksocknal_data.ksnd_global_lock);
2406                 return 1;
2407         }
2408
2409         ksocknal_free_tx(tx);
2410         read_lock(&ksocknal_data.ksnd_global_lock);
2411
2412         return -EIO;
2413 }
2414
2415
2416 static void
2417 ksocknal_check_peer_timeouts (int idx)
2418 {
2419         struct list_head       *peers = &ksocknal_data.ksnd_peers[idx];
2420         ksock_peer_t     *peer;
2421         ksock_conn_t     *conn;
2422         ksock_tx_t       *tx;
2423
2424  again:
2425         /* NB. We expect to have a look at all the peers and not find any
2426          * connections to time out, so we just use a shared lock while we
2427          * take a look... */
2428         read_lock(&ksocknal_data.ksnd_global_lock);
2429
2430         list_for_each_entry(peer, peers, ksnp_list) {
2431                 unsigned long  deadline = 0;
2432                 int      resid = 0;
2433                 int      n     = 0;
2434
2435                 if (ksocknal_send_keepalive_locked(peer) != 0) {
2436                         read_unlock(&ksocknal_data.ksnd_global_lock);
2437                         goto again;
2438                 }
2439
2440                 conn = ksocknal_find_timed_out_conn (peer);
2441
2442                 if (conn != NULL) {
2443                         read_unlock(&ksocknal_data.ksnd_global_lock);
2444
2445                         ksocknal_close_conn_and_siblings (conn, -ETIMEDOUT);
2446
2447                         /* NB we won't find this one again, but we can't
2448                          * just proceed with the next peer, since we dropped
2449                          * ksnd_global_lock and it might be dead already! */
2450                         ksocknal_conn_decref(conn);
2451                         goto again;
2452                 }
2453
2454                 /* we can't process stale txs right here because we're
2455                  * holding only shared lock */
2456                 if (!list_empty (&peer->ksnp_tx_queue)) {
2457                         ksock_tx_t *tx =
2458                                 list_entry (peer->ksnp_tx_queue.next,
2459                                                 ksock_tx_t, tx_list);
2460
2461                         if (cfs_time_aftereq(cfs_time_current(),
2462                                              tx->tx_deadline)) {
2463
2464                                 ksocknal_peer_addref(peer);
2465                                 read_unlock(&ksocknal_data.ksnd_global_lock);
2466
2467                                 ksocknal_flush_stale_txs(peer);
2468
2469                                 ksocknal_peer_decref(peer);
2470                                 goto again;
2471                         }
2472                 }
2473
2474                 if (list_empty(&peer->ksnp_zc_req_list))
2475                         continue;
2476
2477                 spin_lock(&peer->ksnp_lock);
2478                 list_for_each_entry(tx, &peer->ksnp_zc_req_list, tx_zc_list) {
2479                         if (!cfs_time_aftereq(cfs_time_current(),
2480                                               tx->tx_deadline))
2481                                 break;
2482                         /* ignore the TX if connection is being closed */
2483                         if (tx->tx_conn->ksnc_closing)
2484                                 continue;
2485                         n++;
2486                 }
2487
2488                 if (n == 0) {
2489                         spin_unlock(&peer->ksnp_lock);
2490                         continue;
2491                 }
2492
2493                 tx = list_entry(peer->ksnp_zc_req_list.next,
2494                                     ksock_tx_t, tx_zc_list);
2495                 deadline = tx->tx_deadline;
2496                 resid    = tx->tx_resid;
2497                 conn     = tx->tx_conn;
2498                 ksocknal_conn_addref(conn);
2499
2500                 spin_unlock(&peer->ksnp_lock);
2501                 read_unlock(&ksocknal_data.ksnd_global_lock);
2502
2503                 CERROR("Total %d stale ZC_REQs for peer %s detected; the "
2504                        "oldest(%p) timed out %ld secs ago, "
2505                        "resid: %d, wmem: %d\n",
2506                        n, libcfs_nid2str(peer->ksnp_id.nid), tx,
2507                        cfs_duration_sec(cfs_time_current() - deadline),
2508                        resid, conn->ksnc_sock->sk->sk_wmem_queued);
2509
2510                 ksocknal_close_conn_and_siblings (conn, -ETIMEDOUT);
2511                 ksocknal_conn_decref(conn);
2512                 goto again;
2513         }
2514
2515         read_unlock(&ksocknal_data.ksnd_global_lock);
2516 }
2517
2518 int
2519 ksocknal_reaper (void *arg)
2520 {
2521         wait_queue_t     wait;
2522         ksock_conn_t      *conn;
2523         ksock_sched_t     *sched;
2524         struct list_head         enomem_conns;
2525         int             nenomem_conns;
2526         long     timeout;
2527         int             i;
2528         int             peer_index = 0;
2529         unsigned long    deadline = cfs_time_current();
2530
2531         cfs_block_allsigs ();
2532
2533         INIT_LIST_HEAD(&enomem_conns);
2534         init_waitqueue_entry(&wait, current);
2535
2536         spin_lock_bh(&ksocknal_data.ksnd_reaper_lock);
2537
2538         while (!ksocknal_data.ksnd_shuttingdown) {
2539
2540                 if (!list_empty (&ksocknal_data.ksnd_deathrow_conns)) {
2541                         conn = list_entry (ksocknal_data. \
2542                                                ksnd_deathrow_conns.next,
2543                                                ksock_conn_t, ksnc_list);
2544                         list_del (&conn->ksnc_list);
2545
2546                         spin_unlock_bh(&ksocknal_data.ksnd_reaper_lock);
2547
2548                         ksocknal_terminate_conn(conn);
2549                         ksocknal_conn_decref(conn);
2550
2551                         spin_lock_bh(&ksocknal_data.ksnd_reaper_lock);
2552                         continue;
2553                 }
2554
2555                 if (!list_empty (&ksocknal_data.ksnd_zombie_conns)) {
2556                         conn = list_entry (ksocknal_data.ksnd_zombie_conns.\
2557                                                next, ksock_conn_t, ksnc_list);
2558                         list_del (&conn->ksnc_list);
2559
2560                         spin_unlock_bh(&ksocknal_data.ksnd_reaper_lock);
2561
2562                         ksocknal_destroy_conn(conn);
2563
2564                         spin_lock_bh(&ksocknal_data.ksnd_reaper_lock);
2565                         continue;
2566                 }
2567
2568                 if (!list_empty (&ksocknal_data.ksnd_enomem_conns)) {
2569                         list_add(&enomem_conns,
2570                                      &ksocknal_data.ksnd_enomem_conns);
2571                         list_del_init(&ksocknal_data.ksnd_enomem_conns);
2572                 }
2573
2574                 spin_unlock_bh(&ksocknal_data.ksnd_reaper_lock);
2575
2576                 /* reschedule all the connections that stalled with ENOMEM... */
2577                 nenomem_conns = 0;
2578                 while (!list_empty (&enomem_conns)) {
2579                         conn = list_entry (enomem_conns.next,
2580                                                ksock_conn_t, ksnc_tx_list);
2581                         list_del (&conn->ksnc_tx_list);
2582
2583                         sched = conn->ksnc_scheduler;
2584
2585                         spin_lock_bh(&sched->kss_lock);
2586
2587                         LASSERT(conn->ksnc_tx_scheduled);
2588                         conn->ksnc_tx_ready = 1;
2589                         list_add_tail(&conn->ksnc_tx_list,
2590                                           &sched->kss_tx_conns);
2591                         wake_up(&sched->kss_waitq);
2592
2593                         spin_unlock_bh(&sched->kss_lock);
2594                         nenomem_conns++;
2595                 }
2596
2597                 /* careful with the jiffy wrap... */
2598                 while ((timeout = cfs_time_sub(deadline,
2599                                                cfs_time_current())) <= 0) {
2600                         const int n = 4;
2601                         const int p = 1;
2602                         int       chunk = ksocknal_data.ksnd_peer_hash_size;
2603
2604                         /* Time to check for timeouts on a few more peers: I do
2605                          * checks every 'p' seconds on a proportion of the peer
2606                          * table and I need to check every connection 'n' times
2607                          * within a timeout interval, to ensure I detect a
2608                          * timeout on any connection within (n+1)/n times the
2609                          * timeout interval. */
2610
2611                         if (*ksocknal_tunables.ksnd_timeout > n * p)
2612                                 chunk = (chunk * n * p) /
2613                                         *ksocknal_tunables.ksnd_timeout;
2614                         if (chunk == 0)
2615                                 chunk = 1;
2616
2617                         for (i = 0; i < chunk; i++) {
2618                                 ksocknal_check_peer_timeouts (peer_index);
2619                                 peer_index = (peer_index + 1) %
2620                                              ksocknal_data.ksnd_peer_hash_size;
2621                         }
2622
2623                         deadline = cfs_time_add(deadline, cfs_time_seconds(p));
2624                 }
2625
2626                 if (nenomem_conns != 0) {
2627                         /* Reduce my timeout if I rescheduled ENOMEM conns.
2628                          * This also prevents me getting woken immediately
2629                          * if any go back on my enomem list. */
2630                         timeout = SOCKNAL_ENOMEM_RETRY;
2631                 }
2632                 ksocknal_data.ksnd_reaper_waketime =
2633                         cfs_time_add(cfs_time_current(), timeout);
2634
2635                 set_current_state (TASK_INTERRUPTIBLE);
2636                 add_wait_queue (&ksocknal_data.ksnd_reaper_waitq, &wait);
2637
2638                 if (!ksocknal_data.ksnd_shuttingdown &&
2639                     list_empty (&ksocknal_data.ksnd_deathrow_conns) &&
2640                     list_empty (&ksocknal_data.ksnd_zombie_conns))
2641                         schedule_timeout(timeout);
2642
2643                 set_current_state (TASK_RUNNING);
2644                 remove_wait_queue (&ksocknal_data.ksnd_reaper_waitq, &wait);
2645
2646                 spin_lock_bh(&ksocknal_data.ksnd_reaper_lock);
2647         }
2648
2649         spin_unlock_bh(&ksocknal_data.ksnd_reaper_lock);
2650
2651         ksocknal_thread_fini();
2652         return 0;
2653 }