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 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 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 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 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 = (void *)((char *)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 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 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 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 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 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: "LPU64", "LPU64"\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                                 cfs_wait_event_interruptible_exclusive(
1541                                         sched->kss_waitq,
1542                                         !ksocknal_sched_cansleep(sched), rc);
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 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 int
1676 ksocknal_invert_type(int type)
1677 {
1678         switch (type)
1679         {
1680         case SOCKLND_CONN_ANY:
1681         case SOCKLND_CONN_CONTROL:
1682                 return type;
1683         case SOCKLND_CONN_BULK_IN:
1684                 return SOCKLND_CONN_BULK_OUT;
1685         case SOCKLND_CONN_BULK_OUT:
1686                 return SOCKLND_CONN_BULK_IN;
1687         default:
1688                 return SOCKLND_CONN_NONE;
1689         }
1690 }
1691
1692 int
1693 ksocknal_recv_hello (lnet_ni_t *ni, ksock_conn_t *conn,
1694                      ksock_hello_msg_t *hello, lnet_process_id_t *peerid,
1695                      __u64 *incarnation)
1696 {
1697         /* Return < 0   fatal error
1698          *      0         success
1699          *      EALREADY   lost connection race
1700          *      EPROTO     protocol version mismatch
1701          */
1702         struct socket   *sock = conn->ksnc_sock;
1703         int               active = (conn->ksnc_proto != NULL);
1704         int               timeout;
1705         int               proto_match;
1706         int               rc;
1707         ksock_proto_t       *proto;
1708         lnet_process_id_t    recv_id;
1709
1710         /* socket type set on active connections - not set on passive */
1711         LASSERT (!active == !(conn->ksnc_type != SOCKLND_CONN_NONE));
1712
1713         timeout = active ? *ksocknal_tunables.ksnd_timeout :
1714                             lnet_acceptor_timeout();
1715
1716         rc = libcfs_sock_read(sock, &hello->kshm_magic, sizeof (hello->kshm_magic), timeout);
1717         if (rc != 0) {
1718                 CERROR("Error %d reading HELLO from %pI4h\n",
1719                         rc, &conn->ksnc_ipaddr);
1720                 LASSERT (rc < 0);
1721                 return rc;
1722         }
1723
1724         if (hello->kshm_magic != LNET_PROTO_MAGIC &&
1725             hello->kshm_magic != __swab32(LNET_PROTO_MAGIC) &&
1726             hello->kshm_magic != le32_to_cpu (LNET_PROTO_TCP_MAGIC)) {
1727                 /* Unexpected magic! */
1728                 CERROR("Bad magic(1) %#08x (%#08x expected) from "
1729                         "%pI4h\n", __cpu_to_le32 (hello->kshm_magic),
1730                         LNET_PROTO_TCP_MAGIC,
1731                         &conn->ksnc_ipaddr);
1732                 return -EPROTO;
1733         }
1734
1735         rc = libcfs_sock_read(sock, &hello->kshm_version,
1736                               sizeof(hello->kshm_version), timeout);
1737         if (rc != 0) {
1738                 CERROR("Error %d reading HELLO from %pI4h\n",
1739                         rc, &conn->ksnc_ipaddr);
1740                 LASSERT (rc < 0);
1741                 return rc;
1742         }
1743
1744         proto = ksocknal_parse_proto_version(hello);
1745         if (proto == NULL) {
1746                 if (!active) {
1747                         /* unknown protocol from peer, tell peer my protocol */
1748                         conn->ksnc_proto = &ksocknal_protocol_v3x;
1749 #if SOCKNAL_VERSION_DEBUG
1750                         if (*ksocknal_tunables.ksnd_protocol == 2)
1751                                 conn->ksnc_proto = &ksocknal_protocol_v2x;
1752                         else if (*ksocknal_tunables.ksnd_protocol == 1)
1753                                 conn->ksnc_proto = &ksocknal_protocol_v1x;
1754 #endif
1755                         hello->kshm_nips = 0;
1756                         ksocknal_send_hello(ni, conn, ni->ni_nid, hello);
1757                 }
1758
1759                 CERROR("Unknown protocol version (%d.x expected)"
1760                         " from %pI4h\n",
1761                         conn->ksnc_proto->pro_version,
1762                         &conn->ksnc_ipaddr);
1763
1764                 return -EPROTO;
1765         }
1766
1767         proto_match = (conn->ksnc_proto == proto);
1768         conn->ksnc_proto = proto;
1769
1770         /* receive the rest of hello message anyway */
1771         rc = conn->ksnc_proto->pro_recv_hello(conn, hello, timeout);
1772         if (rc != 0) {
1773                 CERROR("Error %d reading or checking hello from from %pI4h\n",
1774                        rc, &conn->ksnc_ipaddr);
1775                 LASSERT (rc < 0);
1776                 return rc;
1777         }
1778
1779         *incarnation = hello->kshm_src_incarnation;
1780
1781         if (hello->kshm_src_nid == LNET_NID_ANY) {
1782                 CERROR("Expecting a HELLO hdr with a NID, but got LNET_NID_ANY"
1783                        "from %pI4h\n", &conn->ksnc_ipaddr);
1784                 return -EPROTO;
1785         }
1786
1787         if (!active &&
1788             conn->ksnc_port > LNET_ACCEPTOR_MAX_RESERVED_PORT) {
1789                 /* Userspace NAL assigns peer process ID from socket */
1790                 recv_id.pid = conn->ksnc_port | LNET_PID_USERFLAG;
1791                 recv_id.nid = LNET_MKNID(LNET_NIDNET(ni->ni_nid), conn->ksnc_ipaddr);
1792         } else {
1793                 recv_id.nid = hello->kshm_src_nid;
1794                 recv_id.pid = hello->kshm_src_pid;
1795         }
1796
1797         if (!active) {
1798                 *peerid = recv_id;
1799
1800                 /* peer determines type */
1801                 conn->ksnc_type = ksocknal_invert_type(hello->kshm_ctype);
1802                 if (conn->ksnc_type == SOCKLND_CONN_NONE) {
1803                         CERROR("Unexpected type %d from %s ip %pI4h\n",
1804                                 hello->kshm_ctype, libcfs_id2str(*peerid),
1805                                 &conn->ksnc_ipaddr);
1806                         return -EPROTO;
1807                 }
1808
1809                 return 0;
1810         }
1811
1812         if (peerid->pid != recv_id.pid ||
1813             peerid->nid != recv_id.nid) {
1814                 LCONSOLE_ERROR_MSG(0x130, "Connected successfully to %s on host"
1815                                    " %pI4h, but they claimed they were "
1816                                    "%s; please check your Lustre "
1817                                    "configuration.\n",
1818                                    libcfs_id2str(*peerid),
1819                                    &conn->ksnc_ipaddr,
1820                                    libcfs_id2str(recv_id));
1821                 return -EPROTO;
1822         }
1823
1824         if (hello->kshm_ctype == SOCKLND_CONN_NONE) {
1825                 /* Possible protocol mismatch or I lost the connection race */
1826                 return proto_match ? EALREADY : EPROTO;
1827         }
1828
1829         if (ksocknal_invert_type(hello->kshm_ctype) != conn->ksnc_type) {
1830                 CERROR("Mismatched types: me %d, %s ip %pI4h %d\n",
1831                         conn->ksnc_type, libcfs_id2str(*peerid),
1832                         &conn->ksnc_ipaddr,
1833                         hello->kshm_ctype);
1834                 return -EPROTO;
1835         }
1836
1837         return 0;
1838 }
1839
1840 int
1841 ksocknal_connect (ksock_route_t *route)
1842 {
1843         LIST_HEAD    (zombies);
1844         ksock_peer_t     *peer = route->ksnr_peer;
1845         int            type;
1846         int            wanted;
1847         struct socket     *sock;
1848         unsigned long   deadline;
1849         int            retry_later = 0;
1850         int            rc = 0;
1851
1852         deadline = cfs_time_add(cfs_time_current(),
1853                                 cfs_time_seconds(*ksocknal_tunables.ksnd_timeout));
1854
1855         write_lock_bh(&ksocknal_data.ksnd_global_lock);
1856
1857         LASSERT (route->ksnr_scheduled);
1858         LASSERT (!route->ksnr_connecting);
1859
1860         route->ksnr_connecting = 1;
1861
1862         for (;;) {
1863                 wanted = ksocknal_route_mask() & ~route->ksnr_connected;
1864
1865                 /* stop connecting if peer/route got closed under me, or
1866                  * route got connected while queued */
1867                 if (peer->ksnp_closing || route->ksnr_deleted ||
1868                     wanted == 0) {
1869                         retry_later = 0;
1870                         break;
1871                 }
1872
1873                 /* reschedule if peer is connecting to me */
1874                 if (peer->ksnp_accepting > 0) {
1875                         CDEBUG(D_NET,
1876                                "peer %s(%d) already connecting to me, retry later.\n",
1877                                libcfs_nid2str(peer->ksnp_id.nid), peer->ksnp_accepting);
1878                         retry_later = 1;
1879                 }
1880
1881                 if (retry_later) /* needs reschedule */
1882                         break;
1883
1884                 if ((wanted & (1 << SOCKLND_CONN_ANY)) != 0) {
1885                         type = SOCKLND_CONN_ANY;
1886                 } else if ((wanted & (1 << SOCKLND_CONN_CONTROL)) != 0) {
1887                         type = SOCKLND_CONN_CONTROL;
1888                 } else if ((wanted & (1 << SOCKLND_CONN_BULK_IN)) != 0) {
1889                         type = SOCKLND_CONN_BULK_IN;
1890                 } else {
1891                         LASSERT ((wanted & (1 << SOCKLND_CONN_BULK_OUT)) != 0);
1892                         type = SOCKLND_CONN_BULK_OUT;
1893                 }
1894
1895                 write_unlock_bh(&ksocknal_data.ksnd_global_lock);
1896
1897                 if (cfs_time_aftereq(cfs_time_current(), deadline)) {
1898                         rc = -ETIMEDOUT;
1899                         lnet_connect_console_error(rc, peer->ksnp_id.nid,
1900                                                    route->ksnr_ipaddr,
1901                                                    route->ksnr_port);
1902                         goto failed;
1903                 }
1904
1905                 rc = lnet_connect(&sock, peer->ksnp_id.nid,
1906                                   route->ksnr_myipaddr,
1907                                   route->ksnr_ipaddr, route->ksnr_port);
1908                 if (rc != 0)
1909                         goto failed;
1910
1911                 rc = ksocknal_create_conn(peer->ksnp_ni, route, sock, type);
1912                 if (rc < 0) {
1913                         lnet_connect_console_error(rc, peer->ksnp_id.nid,
1914                                                    route->ksnr_ipaddr,
1915                                                    route->ksnr_port);
1916                         goto failed;
1917                 }
1918
1919                 /* A +ve RC means I have to retry because I lost the connection
1920                  * race or I have to renegotiate protocol version */
1921                 retry_later = (rc != 0);
1922                 if (retry_later)
1923                         CDEBUG(D_NET, "peer %s: conn race, retry later.\n",
1924                                libcfs_nid2str(peer->ksnp_id.nid));
1925
1926                 write_lock_bh(&ksocknal_data.ksnd_global_lock);
1927         }
1928
1929         route->ksnr_scheduled = 0;
1930         route->ksnr_connecting = 0;
1931
1932         if (retry_later) {
1933                 /* re-queue for attention; this frees me up to handle
1934                  * the peer's incoming connection request */
1935
1936                 if (rc == EALREADY ||
1937                     (rc == 0 && peer->ksnp_accepting > 0)) {
1938                         /* We want to introduce a delay before next
1939                          * attempt to connect if we lost conn race,
1940                          * but the race is resolved quickly usually,
1941                          * so min_reconnectms should be good heuristic */
1942                         route->ksnr_retry_interval =
1943                                 cfs_time_seconds(*ksocknal_tunables.ksnd_min_reconnectms)/1000;
1944                         route->ksnr_timeout = cfs_time_add(cfs_time_current(),
1945                                                            route->ksnr_retry_interval);
1946                 }
1947
1948                 ksocknal_launch_connection_locked(route);
1949         }
1950
1951         write_unlock_bh(&ksocknal_data.ksnd_global_lock);
1952         return retry_later;
1953
1954  failed:
1955         write_lock_bh(&ksocknal_data.ksnd_global_lock);
1956
1957         route->ksnr_scheduled = 0;
1958         route->ksnr_connecting = 0;
1959
1960         /* This is a retry rather than a new connection */
1961         route->ksnr_retry_interval *= 2;
1962         route->ksnr_retry_interval =
1963                 MAX(route->ksnr_retry_interval,
1964                     cfs_time_seconds(*ksocknal_tunables.ksnd_min_reconnectms)/1000);
1965         route->ksnr_retry_interval =
1966                 MIN(route->ksnr_retry_interval,
1967                     cfs_time_seconds(*ksocknal_tunables.ksnd_max_reconnectms)/1000);
1968
1969         LASSERT (route->ksnr_retry_interval != 0);
1970         route->ksnr_timeout = cfs_time_add(cfs_time_current(),
1971                                            route->ksnr_retry_interval);
1972
1973         if (!list_empty(&peer->ksnp_tx_queue) &&
1974             peer->ksnp_accepting == 0 &&
1975             ksocknal_find_connecting_route_locked(peer) == NULL) {
1976                 ksock_conn_t *conn;
1977
1978                 /* ksnp_tx_queue is queued on a conn on successful
1979                  * connection for V1.x and V2.x */
1980                 if (!list_empty (&peer->ksnp_conns)) {
1981                         conn = list_entry(peer->ksnp_conns.next,
1982                                               ksock_conn_t, ksnc_list);
1983                         LASSERT (conn->ksnc_proto == &ksocknal_protocol_v3x);
1984                 }
1985
1986                 /* take all the blocked packets while I've got the lock and
1987                  * complete below... */
1988                 list_splice_init(&peer->ksnp_tx_queue, &zombies);
1989         }
1990
1991 #if 0      /* irrelevant with only eager routes */
1992         if (!route->ksnr_deleted) {
1993                 /* make this route least-favourite for re-selection */
1994                 list_del(&route->ksnr_list);
1995                 list_add_tail(&route->ksnr_list, &peer->ksnp_routes);
1996         }
1997 #endif
1998         write_unlock_bh(&ksocknal_data.ksnd_global_lock);
1999
2000         ksocknal_peer_failed(peer);
2001         ksocknal_txlist_done(peer->ksnp_ni, &zombies, 1);
2002         return 0;
2003 }
2004
2005 /*
2006  * check whether we need to create more connds.
2007  * It will try to create new thread if it's necessary, @timeout can
2008  * be updated if failed to create, so caller wouldn't keep try while
2009  * running out of resource.
2010  */
2011 static int
2012 ksocknal_connd_check_start(long sec, long *timeout)
2013 {
2014         char name[16];
2015         int rc;
2016         int total = ksocknal_data.ksnd_connd_starting +
2017                     ksocknal_data.ksnd_connd_running;
2018
2019         if (unlikely(ksocknal_data.ksnd_init < SOCKNAL_INIT_ALL)) {
2020                 /* still in initializing */
2021                 return 0;
2022         }
2023
2024         if (total >= *ksocknal_tunables.ksnd_nconnds_max ||
2025             total > ksocknal_data.ksnd_connd_connecting + SOCKNAL_CONND_RESV) {
2026                 /* can't create more connd, or still have enough
2027                  * threads to handle more connecting */
2028                 return 0;
2029         }
2030
2031         if (list_empty(&ksocknal_data.ksnd_connd_routes)) {
2032                 /* no pending connecting request */
2033                 return 0;
2034         }
2035
2036         if (sec - ksocknal_data.ksnd_connd_failed_stamp <= 1) {
2037                 /* may run out of resource, retry later */
2038                 *timeout = cfs_time_seconds(1);
2039                 return 0;
2040         }
2041
2042         if (ksocknal_data.ksnd_connd_starting > 0) {
2043                 /* serialize starting to avoid flood */
2044                 return 0;
2045         }
2046
2047         ksocknal_data.ksnd_connd_starting_stamp = sec;
2048         ksocknal_data.ksnd_connd_starting++;
2049         spin_unlock_bh(&ksocknal_data.ksnd_connd_lock);
2050
2051         /* NB: total is the next id */
2052         snprintf(name, sizeof(name), "socknal_cd%02d", total);
2053         rc = ksocknal_thread_start(ksocknal_connd, NULL, name);
2054
2055         spin_lock_bh(&ksocknal_data.ksnd_connd_lock);
2056         if (rc == 0)
2057                 return 1;
2058
2059         /* we tried ... */
2060         LASSERT(ksocknal_data.ksnd_connd_starting > 0);
2061         ksocknal_data.ksnd_connd_starting--;
2062         ksocknal_data.ksnd_connd_failed_stamp = get_seconds();
2063
2064         return 1;
2065 }
2066
2067 /*
2068  * check whether current thread can exit, it will return 1 if there are too
2069  * many threads and no creating in past 120 seconds.
2070  * Also, this function may update @timeout to make caller come back
2071  * again to recheck these conditions.
2072  */
2073 static int
2074 ksocknal_connd_check_stop(long sec, long *timeout)
2075 {
2076         int val;
2077
2078         if (unlikely(ksocknal_data.ksnd_init < SOCKNAL_INIT_ALL)) {
2079                 /* still in initializing */
2080                 return 0;
2081         }
2082
2083         if (ksocknal_data.ksnd_connd_starting > 0) {
2084                 /* in progress of starting new thread */
2085                 return 0;
2086         }
2087
2088         if (ksocknal_data.ksnd_connd_running <=
2089             *ksocknal_tunables.ksnd_nconnds) { /* can't shrink */
2090                 return 0;
2091         }
2092
2093         /* created thread in past 120 seconds? */
2094         val = (int)(ksocknal_data.ksnd_connd_starting_stamp +
2095                     SOCKNAL_CONND_TIMEOUT - sec);
2096
2097         *timeout = (val > 0) ? cfs_time_seconds(val) :
2098                                cfs_time_seconds(SOCKNAL_CONND_TIMEOUT);
2099         if (val > 0)
2100                 return 0;
2101
2102         /* no creating in past 120 seconds */
2103
2104         return ksocknal_data.ksnd_connd_running >
2105                ksocknal_data.ksnd_connd_connecting + SOCKNAL_CONND_RESV;
2106 }
2107
2108 /* Go through connd_routes queue looking for a route that we can process
2109  * right now, @timeout_p can be updated if we need to come back later */
2110 static ksock_route_t *
2111 ksocknal_connd_get_route_locked(signed long *timeout_p)
2112 {
2113         ksock_route_t *route;
2114         unsigned long     now;
2115
2116         now = cfs_time_current();
2117
2118         /* connd_routes can contain both pending and ordinary routes */
2119         list_for_each_entry (route, &ksocknal_data.ksnd_connd_routes,
2120                                  ksnr_connd_list) {
2121
2122                 if (route->ksnr_retry_interval == 0 ||
2123                     cfs_time_aftereq(now, route->ksnr_timeout))
2124                         return route;
2125
2126                 if (*timeout_p == MAX_SCHEDULE_TIMEOUT ||
2127                     (int)*timeout_p > (int)(route->ksnr_timeout - now))
2128                         *timeout_p = (int)(route->ksnr_timeout - now);
2129         }
2130
2131         return NULL;
2132 }
2133
2134 int
2135 ksocknal_connd (void *arg)
2136 {
2137         spinlock_t    *connd_lock = &ksocknal_data.ksnd_connd_lock;
2138         ksock_connreq_t   *cr;
2139         wait_queue_t     wait;
2140         int             nloops = 0;
2141         int             cons_retry = 0;
2142
2143         cfs_block_allsigs ();
2144
2145         init_waitqueue_entry(&wait, current);
2146
2147         spin_lock_bh(connd_lock);
2148
2149         LASSERT(ksocknal_data.ksnd_connd_starting > 0);
2150         ksocknal_data.ksnd_connd_starting--;
2151         ksocknal_data.ksnd_connd_running++;
2152
2153         while (!ksocknal_data.ksnd_shuttingdown) {
2154                 ksock_route_t *route = NULL;
2155                 long sec = get_seconds();
2156                 long timeout = MAX_SCHEDULE_TIMEOUT;
2157                 int  dropped_lock = 0;
2158
2159                 if (ksocknal_connd_check_stop(sec, &timeout)) {
2160                         /* wakeup another one to check stop */
2161                         wake_up(&ksocknal_data.ksnd_connd_waitq);
2162                         break;
2163                 }
2164
2165                 if (ksocknal_connd_check_start(sec, &timeout)) {
2166                         /* created new thread */
2167                         dropped_lock = 1;
2168                 }
2169
2170                 if (!list_empty(&ksocknal_data.ksnd_connd_connreqs)) {
2171                         /* Connection accepted by the listener */
2172                         cr = list_entry(ksocknal_data.ksnd_connd_connreqs. \
2173                                             next, ksock_connreq_t, ksncr_list);
2174
2175                         list_del(&cr->ksncr_list);
2176                         spin_unlock_bh(connd_lock);
2177                         dropped_lock = 1;
2178
2179                         ksocknal_create_conn(cr->ksncr_ni, NULL,
2180                                              cr->ksncr_sock, SOCKLND_CONN_NONE);
2181                         lnet_ni_decref(cr->ksncr_ni);
2182                         LIBCFS_FREE(cr, sizeof(*cr));
2183
2184                         spin_lock_bh(connd_lock);
2185                 }
2186
2187                 /* Only handle an outgoing connection request if there
2188                  * is a thread left to handle incoming connections and
2189                  * create new connd */
2190                 if (ksocknal_data.ksnd_connd_connecting + SOCKNAL_CONND_RESV <
2191                     ksocknal_data.ksnd_connd_running) {
2192                         route = ksocknal_connd_get_route_locked(&timeout);
2193                 }
2194                 if (route != NULL) {
2195                         list_del (&route->ksnr_connd_list);
2196                         ksocknal_data.ksnd_connd_connecting++;
2197                         spin_unlock_bh(connd_lock);
2198                         dropped_lock = 1;
2199
2200                         if (ksocknal_connect(route)) {
2201                                 /* consecutive retry */
2202                                 if (cons_retry++ > SOCKNAL_INSANITY_RECONN) {
2203                                         CWARN("massive consecutive "
2204                                               "re-connecting to %pI4h\n",
2205                                               &route->ksnr_ipaddr);
2206                                         cons_retry = 0;
2207                                 }
2208                         } else {
2209                                 cons_retry = 0;
2210                         }
2211
2212                         ksocknal_route_decref(route);
2213
2214                         spin_lock_bh(connd_lock);
2215                         ksocknal_data.ksnd_connd_connecting--;
2216                 }
2217
2218                 if (dropped_lock) {
2219                         if (++nloops < SOCKNAL_RESCHED)
2220                                 continue;
2221                         spin_unlock_bh(connd_lock);
2222                         nloops = 0;
2223                         cond_resched();
2224                         spin_lock_bh(connd_lock);
2225                         continue;
2226                 }
2227
2228                 /* Nothing to do for 'timeout'  */
2229                 set_current_state(TASK_INTERRUPTIBLE);
2230                 add_wait_queue_exclusive(&ksocknal_data.ksnd_connd_waitq, &wait);
2231                 spin_unlock_bh(connd_lock);
2232
2233                 nloops = 0;
2234                 schedule_timeout(timeout);
2235
2236                 set_current_state(TASK_RUNNING);
2237                 remove_wait_queue(&ksocknal_data.ksnd_connd_waitq, &wait);
2238                 spin_lock_bh(connd_lock);
2239         }
2240         ksocknal_data.ksnd_connd_running--;
2241         spin_unlock_bh(connd_lock);
2242
2243         ksocknal_thread_fini();
2244         return 0;
2245 }
2246
2247 ksock_conn_t *
2248 ksocknal_find_timed_out_conn (ksock_peer_t *peer)
2249 {
2250         /* We're called with a shared lock on ksnd_global_lock */
2251         ksock_conn_t      *conn;
2252         struct list_head        *ctmp;
2253
2254         list_for_each (ctmp, &peer->ksnp_conns) {
2255                 int     error;
2256                 conn = list_entry (ctmp, ksock_conn_t, ksnc_list);
2257
2258                 /* Don't need the {get,put}connsock dance to deref ksnc_sock */
2259                 LASSERT (!conn->ksnc_closing);
2260
2261                 /* SOCK_ERROR will reset error code of socket in
2262                  * some platform (like Darwin8.x) */
2263                 error = conn->ksnc_sock->sk->sk_err;
2264                 if (error != 0) {
2265                         ksocknal_conn_addref(conn);
2266
2267                         switch (error) {
2268                         case ECONNRESET:
2269                                 CNETERR("A connection with %s "
2270                                         "(%pI4h:%d) was reset; "
2271                                         "it may have rebooted.\n",
2272                                         libcfs_id2str(peer->ksnp_id),
2273                                         &conn->ksnc_ipaddr,
2274                                         conn->ksnc_port);
2275                                 break;
2276                         case ETIMEDOUT:
2277                                 CNETERR("A connection with %s "
2278                                         "(%pI4h:%d) timed out; the "
2279                                         "network or node may be down.\n",
2280                                         libcfs_id2str(peer->ksnp_id),
2281                                         &conn->ksnc_ipaddr,
2282                                         conn->ksnc_port);
2283                                 break;
2284                         default:
2285                                 CNETERR("An unexpected network error %d "
2286                                         "occurred with %s "
2287                                         "(%pI4h:%d\n", error,
2288                                         libcfs_id2str(peer->ksnp_id),
2289                                         &conn->ksnc_ipaddr,
2290                                         conn->ksnc_port);
2291                                 break;
2292                         }
2293
2294                         return conn;
2295                 }
2296
2297                 if (conn->ksnc_rx_started &&
2298                     cfs_time_aftereq(cfs_time_current(),
2299                                      conn->ksnc_rx_deadline)) {
2300                         /* Timed out incomplete incoming message */
2301                         ksocknal_conn_addref(conn);
2302                         CNETERR("Timeout receiving from %s (%pI4h:%d), "
2303                                 "state %d wanted %d left %d\n",
2304                                 libcfs_id2str(peer->ksnp_id),
2305                                 &conn->ksnc_ipaddr,
2306                                 conn->ksnc_port,
2307                                 conn->ksnc_rx_state,
2308                                 conn->ksnc_rx_nob_wanted,
2309                                 conn->ksnc_rx_nob_left);
2310                         return conn;
2311                 }
2312
2313                 if ((!list_empty(&conn->ksnc_tx_queue) ||
2314                      conn->ksnc_sock->sk->sk_wmem_queued != 0) &&
2315                     cfs_time_aftereq(cfs_time_current(),
2316                                      conn->ksnc_tx_deadline)) {
2317                         /* Timed out messages queued for sending or
2318                          * buffered in the socket's send buffer */
2319                         ksocknal_conn_addref(conn);
2320                         CNETERR("Timeout sending data to %s (%pI4h:%d) "
2321                                 "the network or that node may be down.\n",
2322                                 libcfs_id2str(peer->ksnp_id),
2323                                 &conn->ksnc_ipaddr,
2324                                 conn->ksnc_port);
2325                         return conn;
2326                 }
2327         }
2328
2329         return NULL;
2330 }
2331
2332 static inline void
2333 ksocknal_flush_stale_txs(ksock_peer_t *peer)
2334 {
2335         ksock_tx_t      *tx;
2336         LIST_HEAD      (stale_txs);
2337
2338         write_lock_bh(&ksocknal_data.ksnd_global_lock);
2339
2340         while (!list_empty (&peer->ksnp_tx_queue)) {
2341                 tx = list_entry (peer->ksnp_tx_queue.next,
2342                                      ksock_tx_t, tx_list);
2343
2344                 if (!cfs_time_aftereq(cfs_time_current(),
2345                                       tx->tx_deadline))
2346                         break;
2347
2348                 list_del (&tx->tx_list);
2349                 list_add_tail (&tx->tx_list, &stale_txs);
2350         }
2351
2352         write_unlock_bh(&ksocknal_data.ksnd_global_lock);
2353
2354         ksocknal_txlist_done(peer->ksnp_ni, &stale_txs, 1);
2355 }
2356
2357 int
2358 ksocknal_send_keepalive_locked(ksock_peer_t *peer)
2359 {
2360         ksock_sched_t  *sched;
2361         ksock_conn_t   *conn;
2362         ksock_tx_t     *tx;
2363
2364         if (list_empty(&peer->ksnp_conns)) /* last_alive will be updated by create_conn */
2365                 return 0;
2366
2367         if (peer->ksnp_proto != &ksocknal_protocol_v3x)
2368                 return 0;
2369
2370         if (*ksocknal_tunables.ksnd_keepalive <= 0 ||
2371             cfs_time_before(cfs_time_current(),
2372                             cfs_time_add(peer->ksnp_last_alive,
2373                                          cfs_time_seconds(*ksocknal_tunables.ksnd_keepalive))))
2374                 return 0;
2375
2376         if (cfs_time_before(cfs_time_current(),
2377                             peer->ksnp_send_keepalive))
2378                 return 0;
2379
2380         /* retry 10 secs later, so we wouldn't put pressure
2381          * on this peer if we failed to send keepalive this time */
2382         peer->ksnp_send_keepalive = cfs_time_shift(10);
2383
2384         conn = ksocknal_find_conn_locked(peer, NULL, 1);
2385         if (conn != NULL) {
2386                 sched = conn->ksnc_scheduler;
2387
2388                 spin_lock_bh(&sched->kss_lock);
2389                 if (!list_empty(&conn->ksnc_tx_queue)) {
2390                         spin_unlock_bh(&sched->kss_lock);
2391                         /* there is an queued ACK, don't need keepalive */
2392                         return 0;
2393                 }
2394
2395                 spin_unlock_bh(&sched->kss_lock);
2396         }
2397
2398         read_unlock(&ksocknal_data.ksnd_global_lock);
2399
2400         /* cookie = 1 is reserved for keepalive PING */
2401         tx = ksocknal_alloc_tx_noop(1, 1);
2402         if (tx == NULL) {
2403                 read_lock(&ksocknal_data.ksnd_global_lock);
2404                 return -ENOMEM;
2405         }
2406
2407         if (ksocknal_launch_packet(peer->ksnp_ni, tx, peer->ksnp_id) == 0) {
2408                 read_lock(&ksocknal_data.ksnd_global_lock);
2409                 return 1;
2410         }
2411
2412         ksocknal_free_tx(tx);
2413         read_lock(&ksocknal_data.ksnd_global_lock);
2414
2415         return -EIO;
2416 }
2417
2418
2419 void
2420 ksocknal_check_peer_timeouts (int idx)
2421 {
2422         struct list_head       *peers = &ksocknal_data.ksnd_peers[idx];
2423         ksock_peer_t     *peer;
2424         ksock_conn_t     *conn;
2425         ksock_tx_t       *tx;
2426
2427  again:
2428         /* NB. We expect to have a look at all the peers and not find any
2429          * connections to time out, so we just use a shared lock while we
2430          * take a look... */
2431         read_lock(&ksocknal_data.ksnd_global_lock);
2432
2433         list_for_each_entry(peer, peers, ksnp_list) {
2434                 unsigned long  deadline = 0;
2435                 int      resid = 0;
2436                 int      n     = 0;
2437
2438                 if (ksocknal_send_keepalive_locked(peer) != 0) {
2439                         read_unlock(&ksocknal_data.ksnd_global_lock);
2440                         goto again;
2441                 }
2442
2443                 conn = ksocknal_find_timed_out_conn (peer);
2444
2445                 if (conn != NULL) {
2446                         read_unlock(&ksocknal_data.ksnd_global_lock);
2447
2448                         ksocknal_close_conn_and_siblings (conn, -ETIMEDOUT);
2449
2450                         /* NB we won't find this one again, but we can't
2451                          * just proceed with the next peer, since we dropped
2452                          * ksnd_global_lock and it might be dead already! */
2453                         ksocknal_conn_decref(conn);
2454                         goto again;
2455                 }
2456
2457                 /* we can't process stale txs right here because we're
2458                  * holding only shared lock */
2459                 if (!list_empty (&peer->ksnp_tx_queue)) {
2460                         ksock_tx_t *tx =
2461                                 list_entry (peer->ksnp_tx_queue.next,
2462                                                 ksock_tx_t, tx_list);
2463
2464                         if (cfs_time_aftereq(cfs_time_current(),
2465                                              tx->tx_deadline)) {
2466
2467                                 ksocknal_peer_addref(peer);
2468                                 read_unlock(&ksocknal_data.ksnd_global_lock);
2469
2470                                 ksocknal_flush_stale_txs(peer);
2471
2472                                 ksocknal_peer_decref(peer);
2473                                 goto again;
2474                         }
2475                 }
2476
2477                 if (list_empty(&peer->ksnp_zc_req_list))
2478                         continue;
2479
2480                 spin_lock(&peer->ksnp_lock);
2481                 list_for_each_entry(tx, &peer->ksnp_zc_req_list, tx_zc_list) {
2482                         if (!cfs_time_aftereq(cfs_time_current(),
2483                                               tx->tx_deadline))
2484                                 break;
2485                         /* ignore the TX if connection is being closed */
2486                         if (tx->tx_conn->ksnc_closing)
2487                                 continue;
2488                         n++;
2489                 }
2490
2491                 if (n == 0) {
2492                         spin_unlock(&peer->ksnp_lock);
2493                         continue;
2494                 }
2495
2496                 tx = list_entry(peer->ksnp_zc_req_list.next,
2497                                     ksock_tx_t, tx_zc_list);
2498                 deadline = tx->tx_deadline;
2499                 resid    = tx->tx_resid;
2500                 conn     = tx->tx_conn;
2501                 ksocknal_conn_addref(conn);
2502
2503                 spin_unlock(&peer->ksnp_lock);
2504                 read_unlock(&ksocknal_data.ksnd_global_lock);
2505
2506                 CERROR("Total %d stale ZC_REQs for peer %s detected; the "
2507                        "oldest(%p) timed out %ld secs ago, "
2508                        "resid: %d, wmem: %d\n",
2509                        n, libcfs_nid2str(peer->ksnp_id.nid), tx,
2510                        cfs_duration_sec(cfs_time_current() - deadline),
2511                        resid, conn->ksnc_sock->sk->sk_wmem_queued);
2512
2513                 ksocknal_close_conn_and_siblings (conn, -ETIMEDOUT);
2514                 ksocknal_conn_decref(conn);
2515                 goto again;
2516         }
2517
2518         read_unlock(&ksocknal_data.ksnd_global_lock);
2519 }
2520
2521 int
2522 ksocknal_reaper (void *arg)
2523 {
2524         wait_queue_t     wait;
2525         ksock_conn_t      *conn;
2526         ksock_sched_t     *sched;
2527         struct list_head         enomem_conns;
2528         int             nenomem_conns;
2529         long     timeout;
2530         int             i;
2531         int             peer_index = 0;
2532         unsigned long    deadline = cfs_time_current();
2533
2534         cfs_block_allsigs ();
2535
2536         INIT_LIST_HEAD(&enomem_conns);
2537         init_waitqueue_entry(&wait, current);
2538
2539         spin_lock_bh(&ksocknal_data.ksnd_reaper_lock);
2540
2541         while (!ksocknal_data.ksnd_shuttingdown) {
2542
2543                 if (!list_empty (&ksocknal_data.ksnd_deathrow_conns)) {
2544                         conn = list_entry (ksocknal_data. \
2545                                                ksnd_deathrow_conns.next,
2546                                                ksock_conn_t, ksnc_list);
2547                         list_del (&conn->ksnc_list);
2548
2549                         spin_unlock_bh(&ksocknal_data.ksnd_reaper_lock);
2550
2551                         ksocknal_terminate_conn(conn);
2552                         ksocknal_conn_decref(conn);
2553
2554                         spin_lock_bh(&ksocknal_data.ksnd_reaper_lock);
2555                         continue;
2556                 }
2557
2558                 if (!list_empty (&ksocknal_data.ksnd_zombie_conns)) {
2559                         conn = list_entry (ksocknal_data.ksnd_zombie_conns.\
2560                                                next, ksock_conn_t, ksnc_list);
2561                         list_del (&conn->ksnc_list);
2562
2563                         spin_unlock_bh(&ksocknal_data.ksnd_reaper_lock);
2564
2565                         ksocknal_destroy_conn(conn);
2566
2567                         spin_lock_bh(&ksocknal_data.ksnd_reaper_lock);
2568                         continue;
2569                 }
2570
2571                 if (!list_empty (&ksocknal_data.ksnd_enomem_conns)) {
2572                         list_add(&enomem_conns,
2573                                      &ksocknal_data.ksnd_enomem_conns);
2574                         list_del_init(&ksocknal_data.ksnd_enomem_conns);
2575                 }
2576
2577                 spin_unlock_bh(&ksocknal_data.ksnd_reaper_lock);
2578
2579                 /* reschedule all the connections that stalled with ENOMEM... */
2580                 nenomem_conns = 0;
2581                 while (!list_empty (&enomem_conns)) {
2582                         conn = list_entry (enomem_conns.next,
2583                                                ksock_conn_t, ksnc_tx_list);
2584                         list_del (&conn->ksnc_tx_list);
2585
2586                         sched = conn->ksnc_scheduler;
2587
2588                         spin_lock_bh(&sched->kss_lock);
2589
2590                         LASSERT(conn->ksnc_tx_scheduled);
2591                         conn->ksnc_tx_ready = 1;
2592                         list_add_tail(&conn->ksnc_tx_list,
2593                                           &sched->kss_tx_conns);
2594                         wake_up(&sched->kss_waitq);
2595
2596                         spin_unlock_bh(&sched->kss_lock);
2597                         nenomem_conns++;
2598                 }
2599
2600                 /* careful with the jiffy wrap... */
2601                 while ((timeout = cfs_time_sub(deadline,
2602                                                cfs_time_current())) <= 0) {
2603                         const int n = 4;
2604                         const int p = 1;
2605                         int       chunk = ksocknal_data.ksnd_peer_hash_size;
2606
2607                         /* Time to check for timeouts on a few more peers: I do
2608                          * checks every 'p' seconds on a proportion of the peer
2609                          * table and I need to check every connection 'n' times
2610                          * within a timeout interval, to ensure I detect a
2611                          * timeout on any connection within (n+1)/n times the
2612                          * timeout interval. */
2613
2614                         if (*ksocknal_tunables.ksnd_timeout > n * p)
2615                                 chunk = (chunk * n * p) /
2616                                         *ksocknal_tunables.ksnd_timeout;
2617                         if (chunk == 0)
2618                                 chunk = 1;
2619
2620                         for (i = 0; i < chunk; i++) {
2621                                 ksocknal_check_peer_timeouts (peer_index);
2622                                 peer_index = (peer_index + 1) %
2623                                              ksocknal_data.ksnd_peer_hash_size;
2624                         }
2625
2626                         deadline = cfs_time_add(deadline, cfs_time_seconds(p));
2627                 }
2628
2629                 if (nenomem_conns != 0) {
2630                         /* Reduce my timeout if I rescheduled ENOMEM conns.
2631                          * This also prevents me getting woken immediately
2632                          * if any go back on my enomem list. */
2633                         timeout = SOCKNAL_ENOMEM_RETRY;
2634                 }
2635                 ksocknal_data.ksnd_reaper_waketime =
2636                         cfs_time_add(cfs_time_current(), timeout);
2637
2638                 set_current_state (TASK_INTERRUPTIBLE);
2639                 add_wait_queue (&ksocknal_data.ksnd_reaper_waitq, &wait);
2640
2641                 if (!ksocknal_data.ksnd_shuttingdown &&
2642                     list_empty (&ksocknal_data.ksnd_deathrow_conns) &&
2643                     list_empty (&ksocknal_data.ksnd_zombie_conns))
2644                         schedule_timeout(timeout);
2645
2646                 set_current_state (TASK_RUNNING);
2647                 remove_wait_queue (&ksocknal_data.ksnd_reaper_waitq, &wait);
2648
2649                 spin_lock_bh(&ksocknal_data.ksnd_reaper_lock);
2650         }
2651
2652         spin_unlock_bh(&ksocknal_data.ksnd_reaper_lock);
2653
2654         ksocknal_thread_fini();
2655         return 0;
2656 }