lnet/klnds/socklnd/socklnd_lib.c

   1 /*
   2  * GPL HEADER START
   3  *
   4  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
   5  *
   6  * This program is free software; you can redistribute it and/or modify
   7  * it under the terms of the GNU General Public License version 2 only,
   8  * as published by the Free Software Foundation.
   9  *
  10  * This program is distributed in the hope that it will be useful, but
  11  * WITHOUT ANY WARRANTY; without even the implied warranty of
  12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  13  * General Public License version 2 for more details (a copy is included
  14  * in the LICENSE file that accompanied this code).
  15  *
  16  * You should have received a copy of the GNU General Public License
  17  * version 2 along with this program; If not, see
  18  * http://www.sun.com/software/products/lustre/docs/GPLv2.pdf
  19  *
  20  * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
  21  * CA 95054 USA or visit www.sun.com if you need additional information or
  22  * have any questions.
  23  *
  24  * GPL HEADER END
  25  */
  26 /*
  27  * Copyright (c) 2005, 2010, Oracle and/or its affiliates. All rights reserved.
  28  * Use is subject to license terms.
  29  *
  30  * Copyright (c) 2011, 2012, Intel Corporation.
  31  */
  32 /*
  33  * This file is part of Lustre, http://www.lustre.org/
  34  * Lustre is a trademark of Sun Microsystems, Inc.
  35  */
  36
  37 #include "socklnd.h"
  38
  39 # if defined(CONFIG_SYSCTL) && !CFS_SYSFS_MODULE_PARM
  40
  41 static struct ctl_table ksocknal_ctl_table[] = {
  42         {
  43                 INIT_CTL_NAME
  44                 .procname       = "timeout",
  45                 .data           = &ksocknal_tunables.ksnd_timeout,
  46                 .maxlen         = sizeof (int),
  47                 .mode           = 0644,
  48                 .proc_handler   = &proc_dointvec,
  49                 INIT_STRATEGY
  50         },
  51         {
  52                 INIT_CTL_NAME
  53                 .procname       = "credits",
  54                 .data           = &ksocknal_tunables.ksnd_credits,
  55                 .maxlen         = sizeof (int),
  56                 .mode           = 0444,
  57                 .proc_handler   = &proc_dointvec,
  58                 INIT_STRATEGY
  59         },
  60         {
  61                 INIT_CTL_NAME
  62                 .procname       = "peer_credits",
  63                 .data           = &ksocknal_tunables.ksnd_peertxcredits,
  64                 .maxlen         = sizeof (int),
  65                 .mode           = 0444,
  66                 .proc_handler   = &proc_dointvec,
  67                 INIT_STRATEGY
  68         },
  69         {
  70                 INIT_CTL_NAME
  71                 .procname       = "peer_buffer_credits",
  72                 .data           = &ksocknal_tunables.ksnd_peerrtrcredits,
  73                 .maxlen         = sizeof (int),
  74                 .mode           = 0444,
  75                 .proc_handler   = &proc_dointvec,
  76                 INIT_STRATEGY
  77         },
  78         {
  79                 INIT_CTL_NAME
  80                 .procname       = "peer_timeout",
  81                 .data           = &ksocknal_tunables.ksnd_peertimeout,
  82                 .maxlen         = sizeof (int),
  83                 .mode           = 0444,
  84                 .proc_handler   = &proc_dointvec
  85                 INIT_STRATEGY
  86         },
  87         {
  88                 INIT_CTL_NAME
  89                 .procname       = "nconnds",
  90                 .data           = &ksocknal_tunables.ksnd_nconnds,
  91                 .maxlen         = sizeof (int),
  92                 .mode           = 0444,
  93                 .proc_handler   = &proc_dointvec,
  94                 INIT_STRATEGY
  95         },
  96         {
  97                 INIT_CTL_NAME
  98                 .procname       = "min_reconnectms",
  99                 .data           = &ksocknal_tunables.ksnd_min_reconnectms,
 100                 .maxlen         = sizeof (int),
 101                 .mode           = 0444,
 102                 .proc_handler   = &proc_dointvec,
 103                 INIT_STRATEGY
 104         },
 105         {
 106                 INIT_CTL_NAME
 107                 .procname       = "max_reconnectms",
 108                 .data           = &ksocknal_tunables.ksnd_max_reconnectms,
 109                 .maxlen         = sizeof (int),
 110                 .mode           = 0444,
 111                 .proc_handler   = &proc_dointvec,
 112                 INIT_STRATEGY
 113         },
 114         {
 115                 INIT_CTL_NAME
 116                 .procname       = "eager_ack",
 117                 .data           = &ksocknal_tunables.ksnd_eager_ack,
 118                 .maxlen         = sizeof (int),
 119                 .mode           = 0644,
 120                 .proc_handler   = &proc_dointvec,
 121                 INIT_STRATEGY
 122         },
 123         {
 124                 INIT_CTL_NAME
 125                 .procname       = "zero_copy",
 126                 .data           = &ksocknal_tunables.ksnd_zc_min_payload,
 127                 .maxlen         = sizeof (int),
 128                 .mode           = 0644,
 129                 .proc_handler   = &proc_dointvec,
 130                 INIT_STRATEGY
 131         },
 132         {
 133                 INIT_CTL_NAME
 134                 .procname       = "zero_copy_recv",
 135                 .data           = &ksocknal_tunables.ksnd_zc_recv,
 136                 .maxlen         = sizeof (int),
 137                 .mode           = 0644,
 138                 .proc_handler   = &proc_dointvec,
 139                 INIT_STRATEGY
 140         },
 141         {
 142                 INIT_CTL_NAME
 143                 .procname       = "zero_copy_recv",
 144                 .data           = &ksocknal_tunables.ksnd_zc_recv_min_nfrags,
 145                 .maxlen         = sizeof (int),
 146                 .mode           = 0644,
 147                 .proc_handler   = &proc_dointvec,
 148                 INIT_STRATEGY
 149         },
 150         {
 151                 INIT_CTL_NAME
 152                 .procname       = "typed",
 153                 .data           = &ksocknal_tunables.ksnd_typed_conns,
 154                 .maxlen         = sizeof (int),
 155                 .mode           = 0444,
 156                 .proc_handler   = &proc_dointvec,
 157                 INIT_STRATEGY
 158         },
 159         {
 160                 INIT_CTL_NAME
 161                 .procname       = "min_bulk",
 162                 .data           = &ksocknal_tunables.ksnd_min_bulk,
 163                 .maxlen         = sizeof (int),
 164                 .mode           = 0644,
 165                 .proc_handler   = &proc_dointvec,
 166                 INIT_STRATEGY
 167         },
 168         {
 169                 INIT_CTL_NAME
 170                 .procname       = "rx_buffer_size",
 171                 .data           = &ksocknal_tunables.ksnd_rx_buffer_size,
 172                 .maxlen         = sizeof(int),
 173                 .mode           = 0644,
 174                 .proc_handler   = &proc_dointvec,
 175                 INIT_STRATEGY
 176         },
 177         {
 178                 INIT_CTL_NAME
 179                 .procname       = "tx_buffer_size",
 180                 .data           = &ksocknal_tunables.ksnd_tx_buffer_size,
 181                 .maxlen         = sizeof(int),
 182                 .mode           = 0644,
 183                 .proc_handler   = &proc_dointvec,
 184                 INIT_STRATEGY
 185         },
 186         {
 187                 INIT_CTL_NAME
 188                 .procname       = "nagle",
 189                 .data           = &ksocknal_tunables.ksnd_nagle,
 190                 .maxlen         = sizeof(int),
 191                 .mode           = 0644,
 192                 .proc_handler   = &proc_dointvec,
 193                 INIT_STRATEGY
 194         },
 195 #ifdef CPU_AFFINITY
 196         {
 197                 INIT_CTL_NAME
 198                 .procname       = "irq_affinity",
 199                 .data           = &ksocknal_tunables.ksnd_irq_affinity,
 200                 .maxlen         = sizeof(int),
 201                 .mode           = 0644,
 202                 .proc_handler   = &proc_dointvec,
 203                 INIT_STRATEGY
 204         },
 205 #endif
 206         {
 207                 INIT_CTL_NAME
 208                 .procname       = "round_robin",
 209                 .data           = &ksocknal_tunables.ksnd_round_robin,
 210                 .maxlen         = sizeof(int),
 211                 .mode           = 0644,
 212                 .proc_handler   = &proc_dointvec,
 213                 INIT_STRATEGY
 214         },
 215         {
 216                 INIT_CTL_NAME
 217                 .procname       = "keepalive",
 218                 .data           = &ksocknal_tunables.ksnd_keepalive,
 219                 .maxlen         = sizeof(int),
 220                 .mode           = 0644,
 221                 .proc_handler   = &proc_dointvec,
 222                 INIT_STRATEGY
 223         },
 224         {
 225                 INIT_CTL_NAME
 226                 .procname       = "keepalive_idle",
 227                 .data           = &ksocknal_tunables.ksnd_keepalive_idle,
 228                 .maxlen         = sizeof(int),
 229                 .mode           = 0644,
 230                 .proc_handler   = &proc_dointvec,
 231                 INIT_STRATEGY
 232         },
 233         {
 234                 INIT_CTL_NAME
 235                 .procname       = "keepalive_count",
 236                 .data           = &ksocknal_tunables.ksnd_keepalive_count,
 237                 .maxlen         = sizeof(int),
 238                 .mode           = 0644,
 239                 .proc_handler   = &proc_dointvec,
 240                 INIT_STRATEGY
 241         },
 242         {
 243                 INIT_CTL_NAME
 244                 .procname       = "keepalive_intvl",
 245                 .data           = &ksocknal_tunables.ksnd_keepalive_intvl,
 246                 .maxlen         = sizeof(int),
 247                 .mode           = 0644,
 248                 .proc_handler   = &proc_dointvec,
 249                 INIT_STRATEGY
 250         },
 251 #ifdef SOCKNAL_BACKOFF
 252         {
 253                 INIT_CTL_NAME
 254                 .procname       = "backoff_init",
 255                 .data           = &ksocknal_tunables.ksnd_backoff_init,
 256                 .maxlen         = sizeof(int),
 257                 .mode           = 0644,
 258                 .proc_handler   = &proc_dointvec,
 259                 INIT_STRATEGY
 260         },
 261         {
 262                 INIT_CTL_NAME
 263                 .procname       = "backoff_max",
 264                 .data           = &ksocknal_tunables.ksnd_backoff_max,
 265                 .maxlen         = sizeof(int),
 266                 .mode           = 0644,
 267                 .proc_handler   = &proc_dointvec,
 268                 INIT_STRATEGY
 269         },
 270 #endif
 271 #if SOCKNAL_VERSION_DEBUG
 272         {
 273                 INIT_CTL_NAME
 274                 .procname       = "protocol",
 275                 .data           = &ksocknal_tunables.ksnd_protocol,
 276                 .maxlen         = sizeof(int),
 277                 .mode           = 0644,
 278                 .proc_handler   = &proc_dointvec,
 279                 INIT_STRATEGY
 280         },
 281 #endif
 282         { 0 }
 283 };
 284
 285 struct ctl_table ksocknal_top_ctl_table[] = {
 286         {
 287                 INIT_CTL_NAME
 288                 .procname       = "socknal",
 289                 .data           = NULL,
 290                 .maxlen         = 0,
 291                 .mode           = 0555,
 292                 .child          = ksocknal_ctl_table
 293         },
 294         { 0 }
 295 };
 296
 297 int
 298 ksocknal_lib_tunables_init ()
 299 {
 300         if (!*ksocknal_tunables.ksnd_typed_conns) {
 301                 int rc = -EINVAL;
 302 #if SOCKNAL_VERSION_DEBUG
 303                 if (*ksocknal_tunables.ksnd_protocol < 3)
 304                         rc = 0;
 305 #endif
 306                 if (rc != 0) {
 307                         CERROR("Protocol V3.x MUST have typed connections\n");
 308                         return rc;
 309                 }
 310         }
 311
 312         if (*ksocknal_tunables.ksnd_zc_recv_min_nfrags < 2)
 313                 *ksocknal_tunables.ksnd_zc_recv_min_nfrags = 2;
 314         if (*ksocknal_tunables.ksnd_zc_recv_min_nfrags > LNET_MAX_IOV)
 315                 *ksocknal_tunables.ksnd_zc_recv_min_nfrags = LNET_MAX_IOV;
 316
 317         ksocknal_tunables.ksnd_sysctl =
 318                 register_sysctl_table(ksocknal_top_ctl_table);
 319
 320         if (ksocknal_tunables.ksnd_sysctl == NULL)
 321                 CWARN("Can't setup /proc tunables\n");
 322
 323         return 0;
 324 }
 325
 326 void
 327 ksocknal_lib_tunables_fini ()
 328 {
 329         if (ksocknal_tunables.ksnd_sysctl != NULL)
 330                 unregister_sysctl_table(ksocknal_tunables.ksnd_sysctl);
 331 }
 332 #else
 333 int
 334 ksocknal_lib_tunables_init ()
 335 {
 336         return 0;
 337 }
 338
 339 void
 340 ksocknal_lib_tunables_fini ()
 341 {
 342 }
 343 #endif /* # if CONFIG_SYSCTL && !CFS_SYSFS_MODULE_PARM */
 344
 345 int
 346 ksocknal_lib_get_conn_addrs (ksock_conn_t *conn)
 347 {
 348         int rc = lnet_sock_getaddr(conn->ksnc_sock, true,
 349                                      &conn->ksnc_ipaddr,
 350                                      &conn->ksnc_port);
 351
 352         /* Didn't need the {get,put}connsock dance to deref ksnc_sock... */
 353         LASSERT (!conn->ksnc_closing);
 354
 355         if (rc != 0) {
 356                 CERROR ("Error %d getting sock peer IP\n", rc);
 357                 return rc;
 358         }
 359
 360         rc = lnet_sock_getaddr(conn->ksnc_sock, false,
 361                                  &conn->ksnc_myipaddr, NULL);
 362         if (rc != 0) {
 363                 CERROR ("Error %d getting sock local IP\n", rc);
 364                 return rc;
 365         }
 366
 367         return 0;
 368 }
 369
 370 int
 371 ksocknal_lib_zc_capable(ksock_conn_t *conn)
 372 {
 373         int  caps = conn->ksnc_sock->sk->sk_route_caps;
 374
 375         if (conn->ksnc_proto == &ksocknal_protocol_v1x)
 376                 return 0;
 377
 378         /* ZC if the socket supports scatter/gather and doesn't need software
 379          * checksums */
 380         return ((caps & NETIF_F_SG) != 0 && (caps & NETIF_F_ALL_CSUM) != 0);
 381 }
 382
 383 int
 384 ksocknal_lib_send_iov(ksock_conn_t *conn, ksock_tx_t *tx)
 385 {
 386         struct socket  *sock = conn->ksnc_sock;
 387         int             nob;
 388         int             rc;
 389
 390         if (*ksocknal_tunables.ksnd_enable_csum        && /* checksum enabled */
 391             conn->ksnc_proto == &ksocknal_protocol_v2x && /* V2.x connection  */
 392             tx->tx_nob == tx->tx_resid                 && /* frist sending    */
 393             tx->tx_msg.ksm_csum == 0)                     /* not checksummed  */
 394                 ksocknal_lib_csum_tx(tx);
 395
 396         /* NB we can't trust socket ops to either consume our iovs
 397          * or leave them alone. */
 398
 399         {
 400 #if SOCKNAL_SINGLE_FRAG_TX
 401                 struct iovec    scratch;
 402                 struct iovec   *scratchiov = &scratch;
 403                 unsigned int    niov = 1;
 404 #else
 405                 struct iovec   *scratchiov = conn->ksnc_scheduler->kss_scratch_iov;
 406                 unsigned int    niov = tx->tx_niov;
 407 #endif
 408                 struct msghdr msg = { .msg_flags = MSG_DONTWAIT };
 409                 int  i;
 410
 411                 for (nob = i = 0; i < niov; i++) {
 412                         scratchiov[i] = tx->tx_iov[i];
 413                         nob += scratchiov[i].iov_len;
 414                 }
 415
 416                 if (!list_empty(&conn->ksnc_tx_queue) ||
 417                     nob < tx->tx_resid)
 418                         msg.msg_flags |= MSG_MORE;
 419
 420                 rc = kernel_sendmsg(sock, &msg, (struct kvec *)scratchiov,
 421                                     niov, nob);
 422         }
 423         return rc;
 424 }
 425
 426 int
 427 ksocknal_lib_send_kiov(ksock_conn_t *conn, ksock_tx_t *tx)
 428 {
 429         struct socket *sock = conn->ksnc_sock;
 430         lnet_kiov_t   *kiov = tx->tx_kiov;
 431         int            rc;
 432         int            nob;
 433
 434         /* Not NOOP message */
 435         LASSERT (tx->tx_lnetmsg != NULL);
 436
 437         /* NB we can't trust socket ops to either consume our iovs
 438          * or leave them alone. */
 439         if (tx->tx_msg.ksm_zc_cookies[0] != 0) {
 440                 /* Zero copy is enabled */
 441                 struct sock   *sk = sock->sk;
 442                 struct page   *page = kiov->kiov_page;
 443                 int            offset = kiov->kiov_offset;
 444                 int            fragsize = kiov->kiov_len;
 445                 int            msgflg = MSG_DONTWAIT;
 446
 447                 CDEBUG(D_NET, "page %p + offset %x for %d\n",
 448                                page, offset, kiov->kiov_len);
 449
 450                 if (!list_empty(&conn->ksnc_tx_queue) ||
 451                     fragsize < tx->tx_resid)
 452                         msgflg |= MSG_MORE;
 453
 454                 if (sk->sk_prot->sendpage != NULL) {
 455                         rc = sk->sk_prot->sendpage(sk, page,
 456                                                    offset, fragsize, msgflg);
 457                 } else {
 458                         rc = cfs_tcp_sendpage(sk, page, offset, fragsize,
 459                                               msgflg);
 460                 }
 461         } else {
 462 #if SOCKNAL_SINGLE_FRAG_TX || !SOCKNAL_RISK_KMAP_DEADLOCK
 463                 struct iovec    scratch;
 464                 struct iovec   *scratchiov = &scratch;
 465                 unsigned int    niov = 1;
 466 #else
 467 #ifdef CONFIG_HIGHMEM
 468 #warning "XXX risk of kmap deadlock on multiple frags..."
 469 #endif
 470                 struct iovec *scratchiov = conn->ksnc_scheduler->kss_scratch_iov;
 471                 unsigned int  niov = tx->tx_nkiov;
 472 #endif
 473                 struct msghdr msg = { .msg_flags = MSG_DONTWAIT };
 474                 int           i;
 475
 476                 for (nob = i = 0; i < niov; i++) {
 477                         scratchiov[i].iov_base = kmap(kiov[i].kiov_page) +
 478                                                  kiov[i].kiov_offset;
 479                         nob += scratchiov[i].iov_len = kiov[i].kiov_len;
 480                 }
 481
 482                 if (!list_empty(&conn->ksnc_tx_queue) ||
 483                     nob < tx->tx_resid)
 484                         msg.msg_flags |= MSG_MORE;
 485
 486                 rc = kernel_sendmsg(sock, &msg, (struct kvec *)scratchiov, niov, nob);
 487
 488                 for (i = 0; i < niov; i++)
 489                         kunmap(kiov[i].kiov_page);
 490         }
 491         return rc;
 492 }
 493
 494 void
 495 ksocknal_lib_eager_ack (ksock_conn_t *conn)
 496 {
 497         int            opt = 1;
 498         mm_segment_t   oldmm = get_fs();
 499         struct socket *sock = conn->ksnc_sock;
 500
 501         /* Remind the socket to ACK eagerly.  If I don't, the socket might
 502          * think I'm about to send something it could piggy-back the ACK
 503          * on, introducing delay in completing zero-copy sends in my
 504          * peer. */
 505
 506         set_fs(KERNEL_DS);
 507         sock->ops->setsockopt (sock, SOL_TCP, TCP_QUICKACK,
 508                                (char *)&opt, sizeof (opt));
 509         set_fs(oldmm);
 510 }
 511
 512 int
 513 ksocknal_lib_recv_iov (ksock_conn_t *conn)
 514 {
 515 #if SOCKNAL_SINGLE_FRAG_RX
 516         struct iovec  scratch;
 517         struct iovec *scratchiov = &scratch;
 518         unsigned int  niov = 1;
 519 #else
 520         struct iovec *scratchiov = conn->ksnc_scheduler->kss_scratch_iov;
 521         unsigned int  niov = conn->ksnc_rx_niov;
 522 #endif
 523         struct iovec *iov = conn->ksnc_rx_iov;
 524         struct msghdr msg = {
 525                 .msg_flags      = 0
 526         };
 527         int          nob;
 528         int          i;
 529         int          rc;
 530         int          fragnob;
 531         int          sum;
 532         __u32        saved_csum;
 533
 534         /* NB we can't trust socket ops to either consume our iovs
 535          * or leave them alone. */
 536         LASSERT (niov > 0);
 537
 538         for (nob = i = 0; i < niov; i++) {
 539                 scratchiov[i] = iov[i];
 540                 nob += scratchiov[i].iov_len;
 541         }
 542         LASSERT (nob <= conn->ksnc_rx_nob_wanted);
 543
 544         rc = kernel_recvmsg(conn->ksnc_sock, &msg,
 545                 (struct kvec *)scratchiov, niov, nob, MSG_DONTWAIT);
 546
 547         saved_csum = 0;
 548         if (conn->ksnc_proto == &ksocknal_protocol_v2x) {
 549                 saved_csum = conn->ksnc_msg.ksm_csum;
 550                 conn->ksnc_msg.ksm_csum = 0;
 551         }
 552
 553         if (saved_csum != 0) {
 554                 /* accumulate checksum */
 555                 for (i = 0, sum = rc; sum > 0; i++, sum -= fragnob) {
 556                         LASSERT (i < niov);
 557
 558                         fragnob = iov[i].iov_len;
 559                         if (fragnob > sum)
 560                                 fragnob = sum;
 561
 562                         conn->ksnc_rx_csum = ksocknal_csum(conn->ksnc_rx_csum,
 563                                                            iov[i].iov_base, fragnob);
 564                 }
 565                 conn->ksnc_msg.ksm_csum = saved_csum;
 566         }
 567
 568         return rc;
 569 }
 570
 571 static void
 572 ksocknal_lib_kiov_vunmap(void *addr)
 573 {
 574         if (addr == NULL)
 575                 return;
 576
 577         vunmap(addr);
 578 }
 579
 580 static void *
 581 ksocknal_lib_kiov_vmap(lnet_kiov_t *kiov, int niov,
 582                        struct iovec *iov, struct page **pages)
 583 {
 584         void             *addr;
 585         int               nob;
 586         int               i;
 587
 588         if (!*ksocknal_tunables.ksnd_zc_recv || pages == NULL)
 589                 return NULL;
 590
 591         LASSERT (niov <= LNET_MAX_IOV);
 592
 593         if (niov < 2 ||
 594             niov < *ksocknal_tunables.ksnd_zc_recv_min_nfrags)
 595                 return NULL;
 596
 597         for (nob = i = 0; i < niov; i++) {
 598                 if ((kiov[i].kiov_offset != 0 && i > 0) ||
 599                     (kiov[i].kiov_offset + kiov[i].kiov_len !=
 600                      PAGE_CACHE_SIZE && i < niov - 1))
 601                         return NULL;
 602
 603                 pages[i] = kiov[i].kiov_page;
 604                 nob += kiov[i].kiov_len;
 605         }
 606
 607         addr = vmap(pages, niov, VM_MAP, PAGE_KERNEL);
 608         if (addr == NULL)
 609                 return NULL;
 610
 611         iov->iov_base = addr + kiov[0].kiov_offset;
 612         iov->iov_len = nob;
 613
 614         return addr;
 615 }
 616
 617 int
 618 ksocknal_lib_recv_kiov (ksock_conn_t *conn)
 619 {
 620 #if SOCKNAL_SINGLE_FRAG_RX || !SOCKNAL_RISK_KMAP_DEADLOCK
 621         struct iovec   scratch;
 622         struct iovec  *scratchiov = &scratch;
 623         struct page  **pages      = NULL;
 624         unsigned int   niov       = 1;
 625 #else
 626 #ifdef CONFIG_HIGHMEM
 627 #warning "XXX risk of kmap deadlock on multiple frags..."
 628 #endif
 629         struct iovec  *scratchiov = conn->ksnc_scheduler->kss_scratch_iov;
 630         struct page  **pages      = conn->ksnc_scheduler->kss_rx_scratch_pgs;
 631         unsigned int   niov       = conn->ksnc_rx_nkiov;
 632 #endif
 633         lnet_kiov_t   *kiov = conn->ksnc_rx_kiov;
 634         struct msghdr msg = {
 635                 .msg_flags      = 0
 636         };
 637         int          nob;
 638         int          i;
 639         int          rc;
 640         void        *base;
 641         void        *addr;
 642         int          sum;
 643         int          fragnob;
 644         int n;
 645
 646         /* NB we can't trust socket ops to either consume our iovs
 647          * or leave them alone. */
 648         if ((addr = ksocknal_lib_kiov_vmap(kiov, niov, scratchiov, pages)) != NULL) {
 649                 nob = scratchiov[0].iov_len;
 650                 n = 1;
 651
 652         } else {
 653                 for (nob = i = 0; i < niov; i++) {
 654                         nob += scratchiov[i].iov_len = kiov[i].kiov_len;
 655                         scratchiov[i].iov_base = kmap(kiov[i].kiov_page) +
 656                                                  kiov[i].kiov_offset;
 657                 }
 658                 n = niov;
 659         }
 660
 661         LASSERT (nob <= conn->ksnc_rx_nob_wanted);
 662
 663         rc = kernel_recvmsg(conn->ksnc_sock, &msg,
 664                         (struct kvec *)scratchiov, n, nob, MSG_DONTWAIT);
 665
 666         if (conn->ksnc_msg.ksm_csum != 0) {
 667                 for (i = 0, sum = rc; sum > 0; i++, sum -= fragnob) {
 668                         LASSERT (i < niov);
 669
 670                         /* Dang! have to kmap again because I have nowhere to stash the
 671                          * mapped address.  But by doing it while the page is still
 672                          * mapped, the kernel just bumps the map count and returns me
 673                          * the address it stashed. */
 674                         base = kmap(kiov[i].kiov_page) + kiov[i].kiov_offset;
 675                         fragnob = kiov[i].kiov_len;
 676                         if (fragnob > sum)
 677                                 fragnob = sum;
 678
 679                         conn->ksnc_rx_csum = ksocknal_csum(conn->ksnc_rx_csum,
 680                                                            base, fragnob);
 681
 682                         kunmap(kiov[i].kiov_page);
 683                 }
 684         }
 685
 686         if (addr != NULL) {
 687                 ksocknal_lib_kiov_vunmap(addr);
 688         } else {
 689                 for (i = 0; i < niov; i++)
 690                         kunmap(kiov[i].kiov_page);
 691         }
 692
 693         return (rc);
 694 }
 695
 696 void
 697 ksocknal_lib_csum_tx(ksock_tx_t *tx)
 698 {
 699         int          i;
 700         __u32        csum;
 701         void        *base;
 702
 703         LASSERT(tx->tx_iov[0].iov_base == (void *)&tx->tx_msg);
 704         LASSERT(tx->tx_conn != NULL);
 705         LASSERT(tx->tx_conn->ksnc_proto == &ksocknal_protocol_v2x);
 706
 707         tx->tx_msg.ksm_csum = 0;
 708
 709         csum = ksocknal_csum(~0, (void *)tx->tx_iov[0].iov_base,
 710                              tx->tx_iov[0].iov_len);
 711
 712         if (tx->tx_kiov != NULL) {
 713                 for (i = 0; i < tx->tx_nkiov; i++) {
 714                         base = kmap(tx->tx_kiov[i].kiov_page) +
 715                                tx->tx_kiov[i].kiov_offset;
 716
 717                         csum = ksocknal_csum(csum, base, tx->tx_kiov[i].kiov_len);
 718
 719                         kunmap(tx->tx_kiov[i].kiov_page);
 720                 }
 721         } else {
 722                 for (i = 1; i < tx->tx_niov; i++)
 723                         csum = ksocknal_csum(csum, tx->tx_iov[i].iov_base,
 724                                              tx->tx_iov[i].iov_len);
 725         }
 726
 727         if (*ksocknal_tunables.ksnd_inject_csum_error) {
 728                 csum++;
 729                 *ksocknal_tunables.ksnd_inject_csum_error = 0;
 730         }
 731
 732         tx->tx_msg.ksm_csum = csum;
 733 }
 734
 735 int
 736 ksocknal_lib_get_conn_tunables (ksock_conn_t *conn, int *txmem, int *rxmem, int *nagle)
 737 {
 738         mm_segment_t   oldmm = get_fs ();
 739         struct socket *sock = conn->ksnc_sock;
 740         int            len;
 741         int            rc;
 742
 743         rc = ksocknal_connsock_addref(conn);
 744         if (rc != 0) {
 745                 LASSERT (conn->ksnc_closing);
 746                 *txmem = *rxmem = *nagle = 0;
 747                 return (-ESHUTDOWN);
 748         }
 749
 750         rc = lnet_sock_getbuf(sock, txmem, rxmem);
 751         if (rc == 0) {
 752                 len = sizeof(*nagle);
 753                 set_fs(KERNEL_DS);
 754                 rc = sock->ops->getsockopt(sock, SOL_TCP, TCP_NODELAY,
 755                                            (char *)nagle, &len);
 756                 set_fs(oldmm);
 757         }
 758
 759         ksocknal_connsock_decref(conn);
 760
 761         if (rc == 0)
 762                 *nagle = !*nagle;
 763         else
 764                 *txmem = *rxmem = *nagle = 0;
 765
 766         return (rc);
 767 }
 768
 769 int
 770 ksocknal_lib_setup_sock (struct socket *sock)
 771 {
 772         mm_segment_t    oldmm = get_fs ();
 773         int             rc;
 774         int             option;
 775         int             keep_idle;
 776         int             keep_intvl;
 777         int             keep_count;
 778         int             do_keepalive;
 779         struct linger   linger;
 780
 781         sock->sk->sk_allocation = GFP_NOFS;
 782
 783         /* Ensure this socket aborts active sends immediately when we close
 784          * it. */
 785
 786         linger.l_onoff = 0;
 787         linger.l_linger = 0;
 788
 789         set_fs (KERNEL_DS);
 790         rc = sock_setsockopt (sock, SOL_SOCKET, SO_LINGER,
 791                               (char *)&linger, sizeof (linger));
 792         set_fs (oldmm);
 793         if (rc != 0) {
 794                 CERROR ("Can't set SO_LINGER: %d\n", rc);
 795                 return (rc);
 796         }
 797
 798         option = -1;
 799         set_fs (KERNEL_DS);
 800         rc = sock->ops->setsockopt (sock, SOL_TCP, TCP_LINGER2,
 801                                     (char *)&option, sizeof (option));
 802         set_fs (oldmm);
 803         if (rc != 0) {
 804                 CERROR ("Can't set SO_LINGER2: %d\n", rc);
 805                 return (rc);
 806         }
 807
 808         if (!*ksocknal_tunables.ksnd_nagle) {
 809                 option = 1;
 810
 811                 set_fs (KERNEL_DS);
 812                 rc = sock->ops->setsockopt (sock, SOL_TCP, TCP_NODELAY,
 813                                             (char *)&option, sizeof (option));
 814                 set_fs (oldmm);
 815                 if (rc != 0) {
 816                         CERROR ("Can't disable nagle: %d\n", rc);
 817                         return (rc);
 818                 }
 819         }
 820
 821         rc = lnet_sock_setbuf(sock,
 822                               *ksocknal_tunables.ksnd_tx_buffer_size,
 823                               *ksocknal_tunables.ksnd_rx_buffer_size);
 824         if (rc != 0) {
 825                 CERROR ("Can't set buffer tx %d, rx %d buffers: %d\n",
 826                         *ksocknal_tunables.ksnd_tx_buffer_size,
 827                         *ksocknal_tunables.ksnd_rx_buffer_size, rc);
 828                 return (rc);
 829         }
 830
 831 /* TCP_BACKOFF_* sockopt tunables unsupported in stock kernels */
 832 #ifdef SOCKNAL_BACKOFF
 833         if (*ksocknal_tunables.ksnd_backoff_init > 0) {
 834                 option = *ksocknal_tunables.ksnd_backoff_init;
 835 #ifdef SOCKNAL_BACKOFF_MS
 836                 option *= 1000;
 837 #endif
 838
 839                 set_fs (KERNEL_DS);
 840                 rc = sock->ops->setsockopt (sock, SOL_TCP, TCP_BACKOFF_INIT,
 841                                             (char *)&option, sizeof (option));
 842                 set_fs (oldmm);
 843                 if (rc != 0) {
 844                         CERROR ("Can't set initial tcp backoff %d: %d\n",
 845                                 option, rc);
 846                         return (rc);
 847                 }
 848         }
 849
 850         if (*ksocknal_tunables.ksnd_backoff_max > 0) {
 851                 option = *ksocknal_tunables.ksnd_backoff_max;
 852 #ifdef SOCKNAL_BACKOFF_MS
 853                 option *= 1000;
 854 #endif
 855
 856                 set_fs (KERNEL_DS);
 857                 rc = sock->ops->setsockopt (sock, SOL_TCP, TCP_BACKOFF_MAX,
 858                                             (char *)&option, sizeof (option));
 859                 set_fs (oldmm);
 860                 if (rc != 0) {
 861                         CERROR ("Can't set maximum tcp backoff %d: %d\n",
 862                                 option, rc);
 863                         return (rc);
 864                 }
 865         }
 866 #endif
 867
 868         /* snapshot tunables */
 869         keep_idle  = *ksocknal_tunables.ksnd_keepalive_idle;
 870         keep_count = *ksocknal_tunables.ksnd_keepalive_count;
 871         keep_intvl = *ksocknal_tunables.ksnd_keepalive_intvl;
 872
 873         do_keepalive = (keep_idle > 0 && keep_count > 0 && keep_intvl > 0);
 874
 875         option = (do_keepalive ? 1 : 0);
 876         set_fs (KERNEL_DS);
 877         rc = sock_setsockopt (sock, SOL_SOCKET, SO_KEEPALIVE,
 878                               (char *)&option, sizeof (option));
 879         set_fs (oldmm);
 880         if (rc != 0) {
 881                 CERROR ("Can't set SO_KEEPALIVE: %d\n", rc);
 882                 return (rc);
 883         }
 884
 885         if (!do_keepalive)
 886                 return (0);
 887
 888         set_fs (KERNEL_DS);
 889         rc = sock->ops->setsockopt (sock, SOL_TCP, TCP_KEEPIDLE,
 890                                     (char *)&keep_idle, sizeof (keep_idle));
 891         set_fs (oldmm);
 892         if (rc != 0) {
 893                 CERROR ("Can't set TCP_KEEPIDLE: %d\n", rc);
 894                 return (rc);
 895         }
 896
 897         set_fs (KERNEL_DS);
 898         rc = sock->ops->setsockopt (sock, SOL_TCP, TCP_KEEPINTVL,
 899                                     (char *)&keep_intvl, sizeof (keep_intvl));
 900         set_fs (oldmm);
 901         if (rc != 0) {
 902                 CERROR ("Can't set TCP_KEEPINTVL: %d\n", rc);
 903                 return (rc);
 904         }
 905
 906         set_fs (KERNEL_DS);
 907         rc = sock->ops->setsockopt (sock, SOL_TCP, TCP_KEEPCNT,
 908                                     (char *)&keep_count, sizeof (keep_count));
 909         set_fs (oldmm);
 910         if (rc != 0) {
 911                 CERROR ("Can't set TCP_KEEPCNT: %d\n", rc);
 912                 return (rc);
 913         }
 914
 915         return (0);
 916 }
 917
 918 void
 919 ksocknal_lib_push_conn (ksock_conn_t *conn)
 920 {
 921         struct sock    *sk;
 922         struct tcp_sock *tp;
 923         int             nonagle;
 924         int             val = 1;
 925         int             rc;
 926         mm_segment_t    oldmm;
 927
 928         rc = ksocknal_connsock_addref(conn);
 929         if (rc != 0)                            /* being shut down */
 930                 return;
 931
 932         sk = conn->ksnc_sock->sk;
 933         tp = tcp_sk(sk);
 934
 935         lock_sock (sk);
 936         nonagle = tp->nonagle;
 937         tp->nonagle = 1;
 938         release_sock (sk);
 939
 940         oldmm = get_fs ();
 941         set_fs (KERNEL_DS);
 942
 943         rc = sk->sk_prot->setsockopt (sk, SOL_TCP, TCP_NODELAY,
 944                                       (char *)&val, sizeof (val));
 945         LASSERT (rc == 0);
 946
 947         set_fs (oldmm);
 948
 949         lock_sock (sk);
 950         tp->nonagle = nonagle;
 951         release_sock (sk);
 952
 953         ksocknal_connsock_decref(conn);
 954 }
 955
 956 extern void ksocknal_read_callback (ksock_conn_t *conn);
 957 extern void ksocknal_write_callback (ksock_conn_t *conn);
 958 /*
 959  * socket call back in Linux
 960  */
 961 static void
 962 #ifdef HAVE_SK_DATA_READY_ONE_ARG
 963 ksocknal_data_ready(struct sock *sk)
 964 #else
 965 ksocknal_data_ready(struct sock *sk, int n)
 966 #endif
 967 {
 968         ksock_conn_t  *conn;
 969         ENTRY;
 970
 971         /* interleave correctly with closing sockets... */
 972         LASSERT(!in_irq());
 973         read_lock(&ksocknal_data.ksnd_global_lock);
 974
 975         conn = sk->sk_user_data;
 976         if (conn == NULL) {     /* raced with ksocknal_terminate_conn */
 977                 LASSERT(sk->sk_data_ready != &ksocknal_data_ready);
 978 #ifdef HAVE_SK_DATA_READY_ONE_ARG
 979                 sk->sk_data_ready(sk);
 980 #else
 981                 sk->sk_data_ready(sk, n);
 982 #endif
 983         } else
 984                 ksocknal_read_callback(conn);
 985
 986         read_unlock(&ksocknal_data.ksnd_global_lock);
 987
 988         EXIT;
 989 }
 990
 991 static void
 992 ksocknal_write_space (struct sock *sk)
 993 {
 994         ksock_conn_t  *conn;
 995         int            wspace;
 996         int            min_wpace;
 997
 998         /* interleave correctly with closing sockets... */
 999         LASSERT(!in_irq());
1000         read_lock(&ksocknal_data.ksnd_global_lock);
1001
1002         conn = sk->sk_user_data;
1003         wspace = sk_stream_wspace(sk);
1004         min_wpace = sk_stream_min_wspace(sk);
1005
1006         CDEBUG(D_NET, "sk %p wspace %d low water %d conn %p%s%s%s\n",
1007                sk, wspace, min_wpace, conn,
1008                (conn == NULL) ? "" : (conn->ksnc_tx_ready ?
1009                                       " ready" : " blocked"),
1010                (conn == NULL) ? "" : (conn->ksnc_tx_scheduled ?
1011                                       " scheduled" : " idle"),
1012                (conn == NULL) ? "" : (list_empty(&conn->ksnc_tx_queue) ?
1013                                       " empty" : " queued"));
1014
1015         if (conn == NULL) {             /* raced with ksocknal_terminate_conn */
1016                 LASSERT (sk->sk_write_space != &ksocknal_write_space);
1017                 sk->sk_write_space (sk);
1018
1019                 read_unlock(&ksocknal_data.ksnd_global_lock);
1020                 return;
1021         }
1022
1023         if (wspace >= min_wpace) {              /* got enough space */
1024                 ksocknal_write_callback(conn);
1025
1026                 /* Clear SOCK_NOSPACE _after_ ksocknal_write_callback so the
1027                  * ENOMEM check in ksocknal_transmit is race-free (think about
1028                  * it). */
1029
1030                 clear_bit (SOCK_NOSPACE, &sk->sk_socket->flags);
1031         }
1032
1033         read_unlock(&ksocknal_data.ksnd_global_lock);
1034 }
1035
1036 void
1037 ksocknal_lib_save_callback(struct socket *sock, ksock_conn_t *conn)
1038 {
1039         conn->ksnc_saved_data_ready = sock->sk->sk_data_ready;
1040         conn->ksnc_saved_write_space = sock->sk->sk_write_space;
1041 }
1042
1043 void
1044 ksocknal_lib_set_callback(struct socket *sock,  ksock_conn_t *conn)
1045 {
1046         sock->sk->sk_user_data = conn;
1047         sock->sk->sk_data_ready = ksocknal_data_ready;
1048         sock->sk->sk_write_space = ksocknal_write_space;
1049         return;
1050 }
1051
1052 void
1053 ksocknal_lib_reset_callback(struct socket *sock, ksock_conn_t *conn)
1054 {
1055         /* Remove conn's network callbacks.
1056          * NB I _have_ to restore the callback, rather than storing a noop,
1057          * since the socket could survive past this module being unloaded!! */
1058         sock->sk->sk_data_ready = conn->ksnc_saved_data_ready;
1059         sock->sk->sk_write_space = conn->ksnc_saved_write_space;
1060
1061         /* A callback could be in progress already; they hold a read lock
1062          * on ksnd_global_lock (to serialise with me) and NOOP if
1063          * sk_user_data is NULL. */
1064         sock->sk->sk_user_data = NULL;
1065
1066         return ;
1067 }
1068
1069 int
1070 ksocknal_lib_memory_pressure(ksock_conn_t *conn)
1071 {
1072         int            rc = 0;
1073         ksock_sched_t *sched;
1074
1075         sched = conn->ksnc_scheduler;
1076         spin_lock_bh(&sched->kss_lock);
1077
1078         if (!test_bit(SOCK_NOSPACE, &conn->ksnc_sock->flags) &&
1079             !conn->ksnc_tx_ready) {
1080                 /* SOCK_NOSPACE is set when the socket fills
1081                  * and cleared in the write_space callback
1082                  * (which also sets ksnc_tx_ready).  If
1083                  * SOCK_NOSPACE and ksnc_tx_ready are BOTH
1084                  * zero, I didn't fill the socket and
1085                  * write_space won't reschedule me, so I
1086                  * return -ENOMEM to get my caller to retry
1087                  * after a timeout */
1088                 rc = -ENOMEM;
1089         }
1090
1091         spin_unlock_bh(&sched->kss_lock);
1092
1093         return rc;
1094 }