lnet/klnds/socklnd/socklnd_lib-winnt.c

   1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
   2  * vim:expandtab:shiftwidth=8:tabstop=8:
   3  *
   4  * Copyright (C) 2006 Cluster File Systems, Inc, All rights reserved.
   5  * Author: Matt Wu
   6  *
   7  * This file is part of Lustre, http://www.lustre.org.
   8  *
   9  * This Lustre Software is proprietary - please refer to the license
  10  * agreement you received with your software.
  11  *
  12  * windows socknal library
  13  *
  14  */
  15
  16 #include "socklnd.h"
  17
  18 # if CONFIG_SYSCTL && !CFS_SYSFS_MODULE_PARM
  19 static ctl_table ksocknal_ctl_table[18];
  20
  21 ctl_table ksocknal_top_ctl_table[] = {
  22         {200, "socknal", NULL, 0, 0555, ksocknal_ctl_table},
  23         { 0 }
  24 };
  25
  26 int
  27 ksocknal_lib_tunables_init ()
  28 {
  29             int    i = 0;
  30             int    j = 1;
  31
  32         ksocknal_ctl_table[i++] = (ctl_table)
  33                 {j++, "timeout", ksocknal_tunables.ksnd_timeout,
  34                  sizeof (int), 0644, NULL, &proc_dointvec};
  35         ksocknal_ctl_table[i++] = (ctl_table)
  36                 {j++, "credits", ksocknal_tunables.ksnd_credits,
  37                  sizeof (int), 0444, NULL, &proc_dointvec};
  38         ksocknal_ctl_table[i++] = (ctl_table)
  39                 {j++, "peer_credits", ksocknal_tunables.ksnd_peercredits,
  40                  sizeof (int), 0444, NULL, &proc_dointvec};
  41         ksocknal_ctl_table[i++] = (ctl_table)
  42                 {j++, "nconnds", ksocknal_tunables.ksnd_nconnds,
  43                  sizeof (int), 0444, NULL, &proc_dointvec};
  44         ksocknal_ctl_table[i++] = (ctl_table)
  45                 {j++, "min_reconnectms", ksocknal_tunables.ksnd_min_reconnectms,
  46                  sizeof (int), 0444, NULL, &proc_dointvec};
  47         ksocknal_ctl_table[i++] = (ctl_table)
  48                 {j++, "max_reconnectms", ksocknal_tunables.ksnd_max_reconnectms,
  49                  sizeof (int), 0444, NULL, &proc_dointvec};
  50         ksocknal_ctl_table[i++] = (ctl_table)
  51                 {j++, "eager_ack", ksocknal_tunables.ksnd_eager_ack,
  52                  sizeof (int), 0644, NULL, &proc_dointvec};
  53 #if SOCKNAL_ZC
  54         ksocknal_ctl_table[i++] = (ctl_table)
  55                 {j++, "zero_copy", ksocknal_tunables.ksnd_zc_min_frag,
  56                  sizeof (int), 0644, NULL, &proc_dointvec};
  57 #endif
  58         ksocknal_ctl_table[i++] = (ctl_table)
  59                 {j++, "typed", ksocknal_tunables.ksnd_typed_conns,
  60                  sizeof (int), 0444, NULL, &proc_dointvec};
  61         ksocknal_ctl_table[i++] = (ctl_table)
  62                 {j++, "min_bulk", ksocknal_tunables.ksnd_min_bulk,
  63                  sizeof (int), 0644, NULL, &proc_dointvec};
  64         ksocknal_ctl_table[i++] = (ctl_table)
  65                 {j++, "buffer_size", ksocknal_tunables.ksnd_buffer_size,
  66                  sizeof(int), 0644, NULL, &proc_dointvec};
  67         ksocknal_ctl_table[i++] = (ctl_table)
  68                 {j++, "nagle", ksocknal_tunables.ksnd_nagle,
  69                  sizeof(int), 0644, NULL, &proc_dointvec};
  70 #if CPU_AFFINITY
  71         ksocknal_ctl_table[i++] = (ctl_table)
  72                 {j++, "irq_affinity", ksocknal_tunables.ksnd_irq_affinity,
  73                  sizeof(int), 0644, NULL, &proc_dointvec};
  74 #endif
  75         ksocknal_ctl_table[i++] = (ctl_table)
  76                 {j++, "keepalive_idle", ksocknal_tunables.ksnd_keepalive_idle,
  77                  sizeof(int), 0644, NULL, &proc_dointvec};
  78         ksocknal_ctl_table[i++] = (ctl_table)
  79                 {j++, "keepalive_count", ksocknal_tunables.ksnd_keepalive_count,
  80                  sizeof(int), 0644, NULL, &proc_dointvec};
  81         ksocknal_ctl_table[i++] = (ctl_table)
  82                 {j++, "keepalive_intvl", ksocknal_tunables.ksnd_keepalive_intvl,
  83                  sizeof(int), 0644, NULL, &proc_dointvec};
  84
  85         LASSERT (j == i+1);
  86         LASSERT (i < sizeof(ksocknal_ctl_table)/sizeof(ksocknal_ctl_table[0]));
  87
  88         ksocknal_tunables.ksnd_sysctl =
  89                 register_sysctl_table(ksocknal_top_ctl_table, 0);
  90
  91         if (ksocknal_tunables.ksnd_sysctl == NULL)
  92                 CWARN("Can't setup /proc tunables\n");
  93
  94         return 0;
  95 }
  96
  97 void
  98 ksocknal_lib_tunables_fini ()
  99 {
 100         if (ksocknal_tunables.ksnd_sysctl != NULL)
 101                 unregister_sysctl_table(ksocknal_tunables.ksnd_sysctl);
 102 }
 103 #else
 104 int
 105 ksocknal_lib_tunables_init ()
 106 {
 107         return 0;
 108 }
 109
 110 void
 111 ksocknal_lib_tunables_fini ()
 112 {
 113 }
 114 #endif
 115
 116 void
 117 ksocknal_lib_bind_irq (unsigned int irq)
 118 {
 119 }
 120
 121 int
 122 ksocknal_lib_get_conn_addrs (ksock_conn_t *conn)
 123 {
 124         int rc = libcfs_sock_getaddr(conn->ksnc_sock, 1,
 125                                      &conn->ksnc_ipaddr, &conn->ksnc_port);
 126
 127         /* Didn't need the {get,put}connsock dance to deref ksnc_sock... */
 128         LASSERT (!conn->ksnc_closing);
 129
 130         if (rc != 0) {
 131                 CERROR ("Error %d getting sock peer IP\n", rc);
 132                 return rc;
 133         }
 134
 135         rc = libcfs_sock_getaddr(conn->ksnc_sock, 0,
 136                                  &conn->ksnc_myipaddr, NULL);
 137         if (rc != 0) {
 138                 CERROR ("Error %d getting sock local IP\n", rc);
 139                 return rc;
 140         }
 141
 142         return 0;
 143 }
 144
 145 unsigned int
 146 ksocknal_lib_sock_irq (struct socket *sock)
 147 {
 148     return 0;
 149 }
 150
 151 #if (SOCKNAL_ZC && SOCKNAL_VADDR_ZC)
 152 static struct page *
 153 ksocknal_kvaddr_to_page (unsigned long vaddr)
 154 {
 155         struct page *page;
 156
 157         if (vaddr >= VMALLOC_START &&
 158             vaddr < VMALLOC_END)
 159                 page = vmalloc_to_page ((void *)vaddr);
 160 #if CONFIG_HIGHMEM
 161         else if (vaddr >= PKMAP_BASE &&
 162                  vaddr < (PKMAP_BASE + LAST_PKMAP * PAGE_SIZE))
 163                 page = vmalloc_to_page ((void *)vaddr);
 164                 /* in 2.4 ^ just walks the page tables */
 165 #endif
 166         else
 167                 page = virt_to_page (vaddr);
 168
 169         if (page == NULL ||
 170             !VALID_PAGE (page))
 171                 return (NULL);
 172
 173         return (page);
 174 }
 175 #endif
 176
 177 /*
 178  * ks_lock_iovs
 179  *   Lock the i/o vector buffers into MDL structure
 180  *
 181  * Arguments:
 182  *   iov:  the array of i/o vectors
 183  *   niov: number of i/o vectors to be locked
 184  *   len:  the real length of the iov vectors
 185  *
 186  * Return Value:
 187  *   ksock_mdl_t *: the Mdl of the locked buffers or
 188  *         NULL pointer in failure case
 189  *
 190  * Notes:
 191  *   N/A
 192  */
 193
 194 ksock_mdl_t *
 195 ks_lock_iovs(
 196     IN struct iovec  *iov,
 197     IN int            niov,
 198     IN int            recving,
 199     IN int *          len )
 200 {
 201     int             rc = 0;
 202
 203     int             i = 0;
 204     int             total = 0;
 205     ksock_mdl_t *   mdl = NULL;
 206     ksock_mdl_t *   tail = NULL;
 207
 208     LASSERT(iov != NULL);
 209     LASSERT(niov > 0);
 210     LASSERT(len != NULL);
 211
 212     for (i=0; i < niov; i++) {
 213
 214         ksock_mdl_t * Iovec = NULL;
 215
 216         rc = ks_lock_buffer(
 217                 iov[i].iov_base,
 218                 FALSE,
 219                 iov[i].iov_len,
 220                 recving ? IoWriteAccess : IoReadAccess,
 221                 &Iovec );
 222
 223         if (rc < 0) {
 224             break;
 225         }
 226
 227         if (tail) {
 228             tail->Next = Iovec;
 229         } else {
 230             mdl = Iovec;
 231         }
 232
 233         tail = Iovec;
 234
 235         total +=iov[i].iov_len;
 236     }
 237
 238     if (rc >= 0) {
 239         *len = total;
 240     } else {
 241         if (mdl) {
 242             ks_release_mdl(mdl, FALSE);
 243             mdl = NULL;
 244         }
 245     }
 246
 247     return mdl;
 248 }
 249
 250 /*
 251  * ks_lock_kiovs
 252  *   Lock the kiov pages into MDL structure
 253  *
 254  * Arguments:
 255  *   kiov:  the array of kiov pages
 256  *   niov:  number of kiov to be locked
 257  *   len:   the real length of the kiov arrary
 258  *
 259  * Return Value:
 260  *   PMDL: the Mdl of the locked buffers or NULL
 261  *         pointer in failure case
 262  *
 263  * Notes:
 264  *   N/A
 265  */
 266 ksock_mdl_t *
 267 ks_lock_kiovs(
 268     IN lnet_kiov_t *  kiov,
 269     IN int            nkiov,
 270     IN int            recving,
 271     IN int *          len )
 272 {
 273     int             rc = 0;
 274     int             i = 0;
 275     int             total = 0;
 276     ksock_mdl_t *   mdl = NULL;
 277     ksock_mdl_t *   tail = NULL;
 278
 279     LASSERT(kiov != NULL);
 280     LASSERT(nkiov > 0);
 281     LASSERT(len != NULL);
 282
 283     for (i=0; i < nkiov; i++) {
 284
 285         ksock_mdl_t *        Iovec = NULL;
 286
 287
 288         //
 289         //  Lock the kiov page into Iovec ¡
 290         //
 291
 292         rc = ks_lock_buffer(
 293                 (PUCHAR)kiov[i].kiov_page->addr +
 294                      kiov[i].kiov_offset,
 295                 FALSE,
 296                 kiov[i].kiov_len,
 297                 recving ? IoWriteAccess : IoReadAccess,
 298                 &Iovec
 299             );
 300
 301         if (rc < 0) {
 302             break;
 303         }
 304
 305         //
 306         // Attach the Iovec to the mdl chain
 307         //
 308
 309         if (tail) {
 310             tail->Next = Iovec;
 311         } else {
 312             mdl = Iovec;
 313         }
 314
 315         tail = Iovec;
 316
 317         total += kiov[i].kiov_len;
 318
 319     }
 320
 321     if (rc >= 0) {
 322         *len = total;
 323     } else {
 324         if (mdl) {
 325             ks_release_mdl(mdl, FALSE);
 326             mdl = NULL;
 327         }
 328     }
 329
 330     return mdl;
 331 }
 332
 333
 334 int
 335 ksocknal_lib_send_iov (ksock_conn_t *conn, ksock_tx_t *tx)
 336 {
 337         struct socket *sock = conn->ksnc_sock;
 338 #if (SOCKNAL_ZC && SOCKNAL_VADDR_ZC)
 339         unsigned long  vaddr = (unsigned long)iov->iov_base
 340         int            offset = vaddr & (PAGE_SIZE - 1);
 341         int            zcsize = MIN (iov->iov_len, PAGE_SIZE - offset);
 342         struct page   *page;
 343 #endif
 344         int            nob;
 345         int            rc;
 346         ksock_mdl_t *  mdl;
 347
 348         /* NB we can't trust socket ops to either consume our iovs
 349          * or leave them alone. */
 350
 351 #if (SOCKNAL_ZC && SOCKNAL_VADDR_ZC)
 352         if (zcsize >= ksocknal_data.ksnd_zc_min_frag &&
 353             (sock->sk->sk_route_caps & NETIF_F_SG) &&
 354             (sock->sk->sk_route_caps & (NETIF_F_IP_CSUM | NETIF_F_NO_CSUM | NETIF_F_HW_CSUM)) &&
 355             (page = ksocknal_kvaddr_to_page (vaddr)) != NULL) {
 356                 int msgflg = MSG_DONTWAIT;
 357
 358                 CDEBUG(D_NET, "vaddr %p, page %p->%p + offset %x for %d\n",
 359                        (void *)vaddr, page, page_address(page), offset, zcsize);
 360
 361                 if (!list_empty (&conn->ksnc_tx_queue) ||
 362                     zcsize < tx->tx_resid)
 363                         msgflg |= MSG_MORE;
 364
 365                 rc = tcp_sendpage_zccd(sock, page, offset, zcsize, msgflg, &tx->tx_zccd);
 366         } else
 367 #endif
 368         {
 369                 /* lock the whole tx iovs into a single mdl chain */
 370                 mdl = ks_lock_iovs(tx->tx_iov, tx->tx_niov, FALSE, &nob);
 371
 372                 if (mdl) {
 373                         /* send the total mdl chain */
 374                         rc = ks_send_mdl( conn->ksnc_sock, tx, mdl, nob,
 375                                     (!list_empty (&conn->ksnc_tx_queue) || nob < tx->tx_resid) ?
 376                                     (MSG_DONTWAIT | MSG_MORE) : MSG_DONTWAIT);
 377                 } else {
 378                         rc = -ENOMEM;
 379                 }
 380         }
 381
 382             return rc;
 383 }
 384
 385 int
 386 ksocknal_lib_send_kiov (ksock_conn_t *conn, ksock_tx_t *tx)
 387 {
 388         struct socket *sock = conn->ksnc_sock;
 389         lnet_kiov_t    *kiov = tx->tx_kiov;
 390         int            rc;
 391         int            nob;
 392         ksock_mdl_t *  mdl;
 393
 394         /* NB we can't trust socket ops to either consume our iovs
 395          * or leave them alone. */
 396
 397 #if SOCKNAL_ZC
 398         if (kiov->kiov_len >= *ksocknal_tunables.ksnd_zc_min_frag &&
 399             (sock->sk->sk_route_caps & NETIF_F_SG) &&
 400             (sock->sk->sk_route_caps & (NETIF_F_IP_CSUM | NETIF_F_NO_CSUM | NETIF_F_HW_CSUM))) {
 401                 struct page   *page = kiov->kiov_page;
 402                 int            offset = kiov->kiov_offset;
 403                 int            fragsize = kiov->kiov_len;
 404                 int            msgflg = MSG_DONTWAIT;
 405
 406                 CDEBUG(D_NET, "page %p + offset %x for %d\n",
 407                                page, offset, kiov->kiov_len);
 408
 409                 if (!list_empty(&conn->ksnc_tx_queue) ||
 410                     fragsize < tx->tx_resid)
 411                         msgflg |= MSG_MORE;
 412
 413                 rc = tcp_sendpage_zccd(sock, page, offset, fragsize, msgflg,
 414                                        &tx->tx_zccd);
 415         } else
 416 #endif
 417         {
 418                 /* lock the whole tx kiovs into a single mdl chain */
 419                 mdl = ks_lock_kiovs(tx->tx_kiov, tx->tx_nkiov, FALSE, &nob);
 420
 421                 if (mdl) {
 422                         /* send the total mdl chain */
 423                         rc = ks_send_mdl(
 424                                     conn->ksnc_sock, tx, mdl, nob,
 425                                     (!list_empty(&conn->ksnc_tx_queue) || nob < tx->tx_resid) ?
 426                                     (MSG_DONTWAIT | MSG_MORE) : MSG_DONTWAIT);
 427                 } else {
 428                         rc = -ENOMEM;
 429                 }
 430         }
 431
 432             return rc;
 433 }
 434
 435
 436 int
 437 ksocknal_lib_recv_iov (ksock_conn_t *conn)
 438 {
 439         struct iovec *iov = conn->ksnc_rx_iov;
 440         int           rc;
 441         int           size;
 442         ksock_mdl_t * mdl;
 443
 444         /* lock the whole tx iovs into a single mdl chain */
 445         mdl = ks_lock_iovs(iov, conn->ksnc_rx_niov, TRUE, &size);
 446
 447         if (!mdl) {
 448             return (-ENOMEM);
 449         }
 450
 451         LASSERT (size <= conn->ksnc_rx_nob_wanted);
 452
 453         /* try to request data for the whole mdl chain */
 454         rc = ks_recv_mdl (conn->ksnc_sock, mdl, size, MSG_DONTWAIT);
 455
 456         return rc;
 457 }
 458
 459 int
 460 ksocknal_lib_recv_kiov (ksock_conn_t *conn)
 461 {
 462         lnet_kiov_t  *kiov = conn->ksnc_rx_kiov;
 463         int           size;
 464         int           rc;
 465         ksock_mdl_t * mdl;
 466
 467         /* NB we can't trust socket ops to either consume our iovs
 468          * or leave them alone, so we only receive 1 frag at a time. */
 469         LASSERT (conn->ksnc_rx_nkiov > 0);
 470
 471         /* lock the whole tx kiovs into a single mdl chain */
 472         mdl = ks_lock_kiovs(kiov, conn->ksnc_rx_nkiov, TRUE, &size);
 473
 474         if (!mdl) {
 475             rc = -ENOMEM;
 476             return (rc);
 477         }
 478
 479         LASSERT (size <= conn->ksnc_rx_nob_wanted);
 480
 481         /* try to request data for the whole mdl chain */
 482         rc = ks_recv_mdl (conn->ksnc_sock, mdl, size, MSG_DONTWAIT);
 483
 484         return rc;
 485 }
 486
 487 void
 488 ksocknal_lib_eager_ack (ksock_conn_t *conn)
 489 {
 490         __u32   option = 1;
 491         int     rc = 0;
 492
 493         rc = ks_set_tcp_option(
 494                 conn->ksnc_sock, TCP_SOCKET_NODELAY,
 495                 &option, sizeof(option) );
 496         if (rc != 0) {
 497                 CERROR("Can't disable nagle: %d\n", rc);
 498         }
 499 }
 500
 501 int
 502 ksocknal_lib_get_conn_tunables (ksock_conn_t *conn, int *txmem, int *rxmem, int *nagle)
 503 {
 504         ksock_tconn_t * tconn = conn->ksnc_sock;
 505         int             len;
 506         int             rc;
 507
 508         ks_get_tconn (tconn);
 509
 510         *txmem = *rxmem = 0;
 511
 512         len = sizeof(*nagle);
 513
 514         rc = ks_get_tcp_option(
 515                     tconn, TCP_SOCKET_NODELAY,
 516                     (__u32 *)nagle, &len);
 517
 518         ks_put_tconn (tconn);
 519
 520         printk("ksocknal_get_conn_tunables: nodelay = %d rc = %d\n", *nagle, rc);
 521
 522         if (rc == 0)
 523                 *nagle = !*nagle;
 524         else
 525                 *txmem = *rxmem = *nagle = 0;
 526
 527         return (rc);
 528 }
 529
 530 int
 531 ksocknal_lib_buffersize (int current_sz, int tunable_sz)
 532 {
 533             /* ensure >= SOCKNAL_MIN_BUFFER */
 534             if (current_sz < SOCKNAL_MIN_BUFFER)
 535                         return MAX(SOCKNAL_MIN_BUFFER, tunable_sz);
 536
 537             if (tunable_sz > SOCKNAL_MIN_BUFFER)
 538                         return tunable_sz;
 539
 540             /* leave alone */
 541             return 0;
 542 }
 543
 544 int
 545 ksocknal_lib_setup_sock (struct socket *sock)
 546 {
 547         int             rc;
 548
 549         int             keep_idle;
 550         int             keep_count;
 551         int             keep_intvl;
 552         int             keep_alive;
 553
 554         __u32           option;
 555
 556         /* set the window size */
 557
 558 #if 0
 559         tconn->kstc_snd_wnd = ksocknal_tunables.ksnd_buffer_size;
 560         tconn->kstc_rcv_wnd = ksocknal_tunables.ksnd_buffer_size;
 561 #endif
 562
 563         /* disable nagle */
 564         if (!ksocknal_tunables.ksnd_nagle) {
 565                 option = 1;
 566
 567                 rc = ks_set_tcp_option(
 568                             sock, TCP_SOCKET_NODELAY,
 569                             &option, sizeof (option));
 570                 if (rc != 0) {
 571                         printk ("Can't disable nagle: %d\n", rc);
 572                         return (rc);
 573                 }
 574         }
 575
 576         /* snapshot tunables */
 577         keep_idle  = *ksocknal_tunables.ksnd_keepalive_idle;
 578         keep_count = *ksocknal_tunables.ksnd_keepalive_count;
 579         keep_intvl = *ksocknal_tunables.ksnd_keepalive_intvl;
 580
 581         keep_alive = (keep_idle > 0 && keep_count > 0 && keep_intvl > 0);
 582
 583         option = (__u32)(keep_alive ? 1 : 0);
 584
 585         rc = ks_set_tcp_option(
 586                     sock, TCP_SOCKET_KEEPALIVE,
 587                     &option, sizeof (option));
 588         if (rc != 0) {
 589                 CERROR ("Can't disable nagle: %d\n", rc);
 590                 return (rc);
 591         }
 592
 593         return (0);
 594 }
 595
 596 void
 597 ksocknal_lib_push_conn (ksock_conn_t *conn)
 598 {
 599         ksock_tconn_t * tconn;
 600         __u32           nagle;
 601         __u32           val = 1;
 602         int             rc;
 603
 604         tconn = conn->ksnc_sock;
 605
 606         ks_get_tconn(tconn);
 607
 608         spin_lock(&tconn->kstc_lock);
 609         if (tconn->kstc_type == kstt_sender) {
 610             nagle = tconn->sender.kstc_info.nagle;
 611             tconn->sender.kstc_info.nagle = 0;
 612         } else {
 613             LASSERT(tconn->kstc_type == kstt_child);
 614             nagle = tconn->child.kstc_info.nagle;
 615             tconn->child.kstc_info.nagle = 0;
 616         }
 617
 618         spin_unlock(&tconn->kstc_lock);
 619
 620         val = 1;
 621         rc = ks_set_tcp_option(
 622                     tconn,
 623                     TCP_SOCKET_NODELAY,
 624                     &(val),
 625                     sizeof(__u32)
 626                     );
 627
 628         LASSERT (rc == 0);
 629         spin_lock(&tconn->kstc_lock);
 630
 631         if (tconn->kstc_type == kstt_sender) {
 632             tconn->sender.kstc_info.nagle = nagle;
 633         } else {
 634             LASSERT(tconn->kstc_type == kstt_child);
 635             tconn->child.kstc_info.nagle = nagle;
 636         }
 637         spin_unlock(&tconn->kstc_lock);
 638
 639         ks_put_tconn(tconn);
 640 }
 641
 642 /* @mode: 0: receiving mode / 1: sending mode */
 643 void
 644 ksocknal_sched_conn (ksock_conn_t *conn, int mode, ksock_tx_t *tx)
 645 {
 646         int             flags;
 647         ksock_sched_t * sched;
 648         ENTRY;
 649
 650         /* interleave correctly with closing sockets... */
 651         read_lock (&ksocknal_data.ksnd_global_lock);
 652
 653         sched = conn->ksnc_scheduler;
 654
 655         spin_lock_irqsave (&sched->kss_lock, flags);
 656
 657         if (mode) { /* transmission can continue ... */
 658
 659                 conn->ksnc_tx_ready = 1;
 660
 661                 if (tx) {
 662                     /* Incomplete send: place tx on HEAD of tx_queue */
 663                     list_add (&tx->tx_list, &conn->ksnc_tx_queue);
 664                 }
 665
 666                 if ( !conn->ksnc_tx_scheduled &&
 667                      !list_empty(&conn->ksnc_tx_queue)) {  //packets to send
 668                         list_add_tail (&conn->ksnc_tx_list,
 669                                        &sched->kss_tx_conns);
 670                         conn->ksnc_tx_scheduled = 1;
 671                         /* extra ref for scheduler */
 672                         atomic_inc (&conn->ksnc_conn_refcount);
 673
 674                         cfs_waitq_signal (&sched->kss_waitq);
 675                 }
 676         } else {    /* receiving can continue ... */
 677
 678                 conn->ksnc_rx_ready = 1;
 679
 680                 if ( !conn->ksnc_rx_scheduled) {  /* not being progressed */
 681                         list_add_tail(&conn->ksnc_rx_list,
 682                                       &sched->kss_rx_conns);
 683                         conn->ksnc_rx_scheduled = 1;
 684                         /* extra ref for scheduler */
 685                         atomic_inc (&conn->ksnc_conn_refcount);
 686
 687                         cfs_waitq_signal (&sched->kss_waitq);
 688                 }
 689         }
 690
 691         spin_unlock_irqrestore (&sched->kss_lock, flags);
 692         read_unlock (&ksocknal_data.ksnd_global_lock);
 693
 694         EXIT;
 695 }
 696
 697 void ksocknal_schedule_callback(struct socket*sock, int mode, void * tx, ulong_ptr bytes)
 698 {
 699     ksock_conn_t * conn = (ksock_conn_t *) sock->kstc_conn;
 700
 701     if (mode) {
 702         ksocknal_sched_conn(conn, mode, tx);
 703     } else {
 704         if ( CAN_BE_SCHED(bytes, (ulong_ptr)conn->ksnc_rx_nob_wanted )) {
 705             ksocknal_sched_conn(conn, mode, tx);
 706         }
 707     }
 708 }
 709
 710 extern void
 711 ksocknal_tx_launched (ksock_tx_t *tx);
 712
 713 void
 714 ksocknal_fini_sending(ksock_tcpx_fini_t *tcpx)
 715 {
 716     ksocknal_tx_launched(tcpx->tx);
 717     cfs_free(tcpx);
 718 }
 719
 720 void *
 721 ksocknal_update_tx(
 722     struct socket*  tconn,
 723     void *          txp,
 724     ulong_ptr       rc
 725     )
 726 {
 727     ksock_tx_t *    tx = (ksock_tx_t *)txp;
 728
 729     /*
 730      *  the transmission was done, we need update the tx
 731      */
 732
 733     LASSERT(tx->tx_resid >= (int)rc);
 734     tx->tx_resid -= (int)rc;
 735
 736     /*
 737      *  just partial of tx is sent out, we need update
 738      *  the fields of tx and schedule later transmission.
 739      */
 740
 741     if (tx->tx_resid) {
 742
 743         if (tx->tx_niov > 0) {
 744
 745             /* if there's iov, we need process iov first */
 746             while (rc > 0 ) {
 747                 if (rc < tx->tx_iov->iov_len) {
 748                     /* didn't send whole iov entry... */
 749                     tx->tx_iov->iov_base =
 750                         (char *)(tx->tx_iov->iov_base) + rc;
 751                     tx->tx_iov->iov_len -= rc;
 752                     rc = 0;
 753                  } else {
 754                     /* the whole of iov was sent out */
 755                     rc -= tx->tx_iov->iov_len;
 756                     tx->tx_iov++;
 757                     tx->tx_niov--;
 758                 }
 759             }
 760
 761         } else {
 762
 763             /* now we need process the kiov queues ... */
 764
 765             while (rc > 0 ) {
 766
 767                 if (rc < tx->tx_kiov->kiov_len) {
 768                     /* didn't send whole kiov entry... */
 769                     tx->tx_kiov->kiov_offset += rc;
 770                     tx->tx_kiov->kiov_len -= rc;
 771                     rc = 0;
 772                 } else {
 773                     /* whole kiov was sent out */
 774                     rc -= tx->tx_kiov->kiov_len;
 775                     tx->tx_kiov++;
 776                     tx->tx_nkiov--;
 777                 }
 778             }
 779         }
 780
 781     } else {
 782
 783         ksock_tcpx_fini_t * tcpx =
 784                 cfs_alloc(sizeof(ksock_tcpx_fini_t), CFS_ALLOC_ZERO);
 785
 786         ASSERT(tx->tx_resid == 0);
 787
 788         if (!tcpx) {
 789
 790             ksocknal_tx_launched (tx);
 791
 792         } else {
 793
 794             tcpx->tx = tx;
 795             ExInitializeWorkItem(
 796                     &(tcpx->item),
 797                     ksocknal_fini_sending,
 798                     tcpx
 799             );
 800             ExQueueWorkItem(
 801                     &(tcpx->item),
 802                     CriticalWorkQueue
 803                     );
 804         }
 805
 806         tx = NULL;
 807     }
 808
 809     return (void *)tx;
 810 }
 811
 812 void
 813 ksocknal_lib_save_callback(struct socket *sock, ksock_conn_t *conn)
 814 {
 815 }
 816
 817 void
 818 ksocknal_lib_set_callback(struct socket *sock,  ksock_conn_t *conn)
 819 {
 820         sock->kstc_conn      = conn;
 821         sock->kstc_sched_cb  = ksocknal_schedule_callback;
 822         sock->kstc_update_tx = ksocknal_update_tx;
 823 }
 824
 825 void
 826 ksocknal_lib_reset_callback(struct socket *sock, ksock_conn_t *conn)
 827 {
 828         sock->kstc_conn      = NULL;
 829         sock->kstc_sched_cb  = NULL;
 830         sock->kstc_update_tx = NULL;
 831 }
 832