lustre/portals/knals/qswnal/qswnal_cb.c

   1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
   2  * vim:expandtab:shiftwidth=8:tabstop=8:
   3  *
   4  * Copyright (C) 2002 Cluster File Systems, Inc.
   5  *   Author: Eric Barton <eric@bartonsoftware.com>
   6  *
   7  * Copyright (C) 2002, Lawrence Livermore National Labs (LLNL)
   8  * W. Marcus Miller - Based on ksocknal
   9  *
  10  * This file is part of Portals, http://www.sf.net/projects/sandiaportals/
  11  *
  12  * Portals is free software; you can redistribute it and/or
  13  * modify it under the terms of version 2 of the GNU General Public
  14  * License as published by the Free Software Foundation.
  15  *
  16  * Portals is distributed in the hope that it will be useful,
  17  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  18  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  19  * GNU General Public License for more details.
  20  *
  21  * You should have received a copy of the GNU General Public License
  22  * along with Portals; if not, write to the Free Software
  23  * Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
  24  *
  25  */
  26
  27 #include "qswnal.h"
  28
  29 atomic_t kqswnal_packets_launched;
  30 atomic_t kqswnal_packets_transmitted;
  31 atomic_t kqswnal_packets_received;
  32
  33
  34 /*
  35  *  LIB functions follow
  36  *
  37  */
  38 static int
  39 kqswnal_read(nal_cb_t *nal, void *private, void *dst_addr, user_ptr src_addr,
  40              size_t len)
  41 {
  42         CDEBUG (D_NET, LPX64": reading "LPSZ" bytes from %p -> %p\n",
  43                 nal->ni.nid, len, src_addr, dst_addr );
  44         memcpy( dst_addr, src_addr, len );
  45
  46         return (0);
  47 }
  48
  49 static int
  50 kqswnal_write(nal_cb_t *nal, void *private, user_ptr dst_addr, void *src_addr,
  51               size_t len)
  52 {
  53         CDEBUG (D_NET, LPX64": writing "LPSZ" bytes from %p -> %p\n",
  54                 nal->ni.nid, len, src_addr, dst_addr );
  55         memcpy( dst_addr, src_addr, len );
  56
  57         return (0);
  58 }
  59
  60 static void *
  61 kqswnal_malloc(nal_cb_t *nal, size_t len)
  62 {
  63         void *buf;
  64
  65         PORTAL_ALLOC(buf, len);
  66         return (buf);
  67 }
  68
  69 static void
  70 kqswnal_free(nal_cb_t *nal, void *buf, size_t len)
  71 {
  72         PORTAL_FREE(buf, len);
  73 }
  74
  75 static void
  76 kqswnal_printf (nal_cb_t * nal, const char *fmt, ...)
  77 {
  78         va_list ap;
  79         char msg[256];
  80
  81         va_start (ap, fmt);
  82         vsnprintf (msg, sizeof (msg), fmt, ap);        /* sprint safely */
  83         va_end (ap);
  84
  85         msg[sizeof (msg) - 1] = 0;                /* ensure terminated */
  86
  87         CDEBUG (D_NET, "%s", msg);
  88 }
  89
  90
  91 static void
  92 kqswnal_cli(nal_cb_t *nal, unsigned long *flags)
  93 {
  94         kqswnal_data_t *data= nal->nal_data;
  95
  96         spin_lock_irqsave(&data->kqn_statelock, *flags);
  97 }
  98
  99
 100 static void
 101 kqswnal_sti(nal_cb_t *nal, unsigned long *flags)
 102 {
 103         kqswnal_data_t *data= nal->nal_data;
 104
 105         spin_unlock_irqrestore(&data->kqn_statelock, *flags);
 106 }
 107
 108
 109 static int
 110 kqswnal_dist(nal_cb_t *nal, ptl_nid_t nid, unsigned long *dist)
 111 {
 112         if (nid == nal->ni.nid)
 113                 *dist = 0;                      /* it's me */
 114         else if (kqswnal_nid2elanid (nid) >= 0)
 115                 *dist = 1;                      /* it's my peer */
 116         else
 117                 *dist = 2;                      /* via router */
 118         return (0);
 119 }
 120
 121 void
 122 kqswnal_unmap_tx (kqswnal_tx_t *ktx)
 123 {
 124         if (ktx->ktx_nmappedpages == 0)
 125                 return;
 126
 127         CDEBUG (D_NET, "%p[%d] unloading pages %d for %d\n",
 128                 ktx, ktx->ktx_niov, ktx->ktx_basepage, ktx->ktx_nmappedpages);
 129
 130         LASSERT (ktx->ktx_nmappedpages <= ktx->ktx_npages);
 131         LASSERT (ktx->ktx_basepage + ktx->ktx_nmappedpages <=
 132                  kqswnal_data.kqn_eptxdmahandle->NumDvmaPages);
 133
 134         elan3_dvma_unload(kqswnal_data.kqn_epdev->DmaState,
 135                           kqswnal_data.kqn_eptxdmahandle,
 136                           ktx->ktx_basepage, ktx->ktx_nmappedpages);
 137         ktx->ktx_nmappedpages = 0;
 138 }
 139
 140 int
 141 kqswnal_map_tx_kiov (kqswnal_tx_t *ktx, int nob, int niov, ptl_kiov_t *kiov)
 142 {
 143         int       nfrags    = ktx->ktx_niov;
 144         const int maxfrags  = sizeof (ktx->ktx_iov)/sizeof (ktx->ktx_iov[0]);
 145         int       nmapped   = ktx->ktx_nmappedpages;
 146         int       maxmapped = ktx->ktx_npages;
 147         uint32_t  basepage  = ktx->ktx_basepage + nmapped;
 148         char     *ptr;
 149
 150         LASSERT (nmapped <= maxmapped);
 151         LASSERT (nfrags <= maxfrags);
 152         LASSERT (niov > 0);
 153         LASSERT (nob > 0);
 154
 155         do {
 156                 int  fraglen = kiov->kiov_len;
 157
 158                 /* nob exactly spans the iovs */
 159                 LASSERT (fraglen <= nob);
 160                 /* each frag fits in a page */
 161                 LASSERT (kiov->kiov_offset + kiov->kiov_len <= PAGE_SIZE);
 162
 163                 nmapped++;
 164                 if (nmapped > maxmapped) {
 165                         CERROR("Can't map message in %d pages (max %d)\n",
 166                                nmapped, maxmapped);
 167                         return (-EMSGSIZE);
 168                 }
 169
 170                 if (nfrags == maxfrags) {
 171                         CERROR("Message too fragmented in Elan VM (max %d frags)\n",
 172                                maxfrags);
 173                         return (-EMSGSIZE);
 174                 }
 175
 176                 /* XXX this is really crap, but we'll have to kmap until
 177                  * EKC has a page (rather than vaddr) mapping interface */
 178
 179                 ptr = ((char *)kmap (kiov->kiov_page)) + kiov->kiov_offset;
 180
 181                 CDEBUG(D_NET,
 182                        "%p[%d] loading %p for %d, page %d, %d total\n",
 183                         ktx, nfrags, ptr, fraglen, basepage, nmapped);
 184
 185                 elan3_dvma_kaddr_load (kqswnal_data.kqn_epdev->DmaState,
 186                                        kqswnal_data.kqn_eptxdmahandle,
 187                                        ptr, fraglen,
 188                                        basepage, &ktx->ktx_iov[nfrags].Base);
 189
 190                 kunmap (kiov->kiov_page);
 191
 192                 /* keep in loop for failure case */
 193                 ktx->ktx_nmappedpages = nmapped;
 194
 195                 if (nfrags > 0 &&                /* previous frag mapped */
 196                     ktx->ktx_iov[nfrags].Base == /* contiguous with this one */
 197                     (ktx->ktx_iov[nfrags-1].Base + ktx->ktx_iov[nfrags-1].Len))
 198                         /* just extend previous */
 199                         ktx->ktx_iov[nfrags - 1].Len += fraglen;
 200                 else {
 201                         ktx->ktx_iov[nfrags].Len = fraglen;
 202                         nfrags++;                /* new frag */
 203                 }
 204
 205                 basepage++;
 206                 kiov++;
 207                 niov--;
 208                 nob -= fraglen;
 209
 210                 /* iov must not run out before end of data */
 211                 LASSERT (nob == 0 || niov > 0);
 212
 213         } while (nob > 0);
 214
 215         ktx->ktx_niov = nfrags;
 216         CDEBUG (D_NET, "%p got %d frags over %d pages\n",
 217                 ktx, ktx->ktx_niov, ktx->ktx_nmappedpages);
 218
 219         return (0);
 220 }
 221
 222 int
 223 kqswnal_map_tx_iov (kqswnal_tx_t *ktx, int nob, int niov, struct iovec *iov)
 224 {
 225         int       nfrags    = ktx->ktx_niov;
 226         const int maxfrags  = sizeof (ktx->ktx_iov)/sizeof (ktx->ktx_iov[0]);
 227         int       nmapped   = ktx->ktx_nmappedpages;
 228         int       maxmapped = ktx->ktx_npages;
 229         uint32_t  basepage  = ktx->ktx_basepage + nmapped;
 230
 231         LASSERT (nmapped <= maxmapped);
 232         LASSERT (nfrags <= maxfrags);
 233         LASSERT (niov > 0);
 234         LASSERT (nob > 0);
 235
 236         do {
 237                 int  fraglen = iov->iov_len;
 238                 long npages  = kqswnal_pages_spanned (iov->iov_base, fraglen);
 239
 240                 /* nob exactly spans the iovs */
 241                 LASSERT (fraglen <= nob);
 242
 243                 nmapped += npages;
 244                 if (nmapped > maxmapped) {
 245                         CERROR("Can't map message in %d pages (max %d)\n",
 246                                nmapped, maxmapped);
 247                         return (-EMSGSIZE);
 248                 }
 249
 250                 if (nfrags == maxfrags) {
 251                         CERROR("Message too fragmented in Elan VM (max %d frags)\n",
 252                                maxfrags);
 253                         return (-EMSGSIZE);
 254                 }
 255
 256                 CDEBUG(D_NET,
 257                        "%p[%d] loading %p for %d, pages %d for %ld, %d total\n",
 258                         ktx, nfrags, iov->iov_base, fraglen, basepage, npages,
 259                         nmapped);
 260
 261                 elan3_dvma_kaddr_load (kqswnal_data.kqn_epdev->DmaState,
 262                                        kqswnal_data.kqn_eptxdmahandle,
 263                                        iov->iov_base, fraglen,
 264                                        basepage, &ktx->ktx_iov[nfrags].Base);
 265                 /* keep in loop for failure case */
 266                 ktx->ktx_nmappedpages = nmapped;
 267
 268                 if (nfrags > 0 &&                /* previous frag mapped */
 269                     ktx->ktx_iov[nfrags].Base == /* contiguous with this one */
 270                     (ktx->ktx_iov[nfrags-1].Base + ktx->ktx_iov[nfrags-1].Len))
 271                         /* just extend previous */
 272                         ktx->ktx_iov[nfrags - 1].Len += fraglen;
 273                 else {
 274                         ktx->ktx_iov[nfrags].Len = fraglen;
 275                         nfrags++;                /* new frag */
 276                 }
 277
 278                 basepage += npages;
 279                 iov++;
 280                 niov--;
 281                 nob -= fraglen;
 282
 283                 /* iov must not run out before end of data */
 284                 LASSERT (nob == 0 || niov > 0);
 285
 286         } while (nob > 0);
 287
 288         ktx->ktx_niov = nfrags;
 289         CDEBUG (D_NET, "%p got %d frags over %d pages\n",
 290                 ktx, ktx->ktx_niov, ktx->ktx_nmappedpages);
 291
 292         return (0);
 293 }
 294
 295 void
 296 kqswnal_put_idle_tx (kqswnal_tx_t *ktx)
 297 {
 298         kpr_fwd_desc_t   *fwd = NULL;
 299         struct list_head *idle = ktx->ktx_idle;
 300         unsigned long     flags;
 301
 302         kqswnal_unmap_tx (ktx);                /* release temporary mappings */
 303         ktx->ktx_state = KTX_IDLE;
 304
 305         spin_lock_irqsave (&kqswnal_data.kqn_idletxd_lock, flags);
 306
 307         list_add (&ktx->ktx_list, idle);
 308
 309         /* reserved for non-blocking tx */
 310         if (idle == &kqswnal_data.kqn_nblk_idletxds) {
 311                 spin_unlock_irqrestore (&kqswnal_data.kqn_idletxd_lock, flags);
 312                 return;
 313         }
 314
 315         /* anything blocking for a tx descriptor? */
 316         if (!list_empty(&kqswnal_data.kqn_idletxd_fwdq)) /* forwarded packet? */
 317         {
 318                 CDEBUG(D_NET,"wakeup fwd\n");
 319
 320                 fwd = list_entry (kqswnal_data.kqn_idletxd_fwdq.next,
 321                                   kpr_fwd_desc_t, kprfd_list);
 322                 list_del (&fwd->kprfd_list);
 323         }
 324
 325         if (waitqueue_active (&kqswnal_data.kqn_idletxd_waitq))  /* process? */
 326         {
 327                 /* local sender waiting for tx desc */
 328                 CDEBUG(D_NET,"wakeup process\n");
 329                 wake_up (&kqswnal_data.kqn_idletxd_waitq);
 330         }
 331
 332         spin_unlock_irqrestore (&kqswnal_data.kqn_idletxd_lock, flags);
 333
 334         if (fwd == NULL)
 335                 return;
 336
 337         /* schedule packet for forwarding again */
 338         spin_lock_irqsave (&kqswnal_data.kqn_sched_lock, flags);
 339
 340         list_add_tail (&fwd->kprfd_list, &kqswnal_data.kqn_delayedfwds);
 341         if (waitqueue_active (&kqswnal_data.kqn_sched_waitq))
 342                 wake_up (&kqswnal_data.kqn_sched_waitq);
 343
 344         spin_unlock_irqrestore (&kqswnal_data.kqn_sched_lock, flags);
 345 }
 346
 347 kqswnal_tx_t *
 348 kqswnal_get_idle_tx (kpr_fwd_desc_t *fwd, int may_block)
 349 {
 350         unsigned long  flags;
 351         kqswnal_tx_t  *ktx = NULL;
 352
 353         for (;;) {
 354                 spin_lock_irqsave (&kqswnal_data.kqn_idletxd_lock, flags);
 355
 356                 /* "normal" descriptor is free */
 357                 if (!list_empty (&kqswnal_data.kqn_idletxds)) {
 358                         ktx = list_entry (kqswnal_data.kqn_idletxds.next,
 359                                           kqswnal_tx_t, ktx_list);
 360                         list_del (&ktx->ktx_list);
 361                         break;
 362                 }
 363
 364                 /* "normal" descriptor pool is empty */
 365
 366                 if (fwd != NULL) { /* forwarded packet => queue for idle txd */
 367                         CDEBUG (D_NET, "blocked fwd [%p]\n", fwd);
 368                         list_add_tail (&fwd->kprfd_list,
 369                                        &kqswnal_data.kqn_idletxd_fwdq);
 370                         break;
 371                 }
 372
 373                 /* doing a local transmit */
 374                 if (!may_block) {
 375                         if (list_empty (&kqswnal_data.kqn_nblk_idletxds)) {
 376                                 CERROR ("intr tx desc pool exhausted\n");
 377                                 break;
 378                         }
 379
 380                         ktx = list_entry (kqswnal_data.kqn_nblk_idletxds.next,
 381                                           kqswnal_tx_t, ktx_list);
 382                         list_del (&ktx->ktx_list);
 383                         break;
 384                 }
 385
 386                 /* block for idle tx */
 387
 388                 spin_unlock_irqrestore (&kqswnal_data.kqn_idletxd_lock, flags);
 389
 390                 CDEBUG (D_NET, "blocking for tx desc\n");
 391                 wait_event (kqswnal_data.kqn_idletxd_waitq,
 392                             !list_empty (&kqswnal_data.kqn_idletxds));
 393         }
 394
 395         spin_unlock_irqrestore (&kqswnal_data.kqn_idletxd_lock, flags);
 396
 397         /* Idle descs can't have any mapped (as opposed to pre-mapped) pages */
 398         LASSERT (ktx == NULL || ktx->ktx_nmappedpages == 0);
 399         return (ktx);
 400 }
 401
 402 void
 403 kqswnal_tx_done (kqswnal_tx_t *ktx, int error)
 404 {
 405         switch (ktx->ktx_state) {
 406         case KTX_FORWARDING:       /* router asked me to forward this packet */
 407                 kpr_fwd_done (&kqswnal_data.kqn_router,
 408                               (kpr_fwd_desc_t *)ktx->ktx_args[0], error);
 409                 break;
 410
 411         case KTX_SENDING:          /* packet sourced locally */
 412                 lib_finalize (&kqswnal_lib, ktx->ktx_args[0],
 413                               (lib_msg_t *)ktx->ktx_args[1]);
 414                 break;
 415
 416         default:
 417                 LASSERT (0);
 418         }
 419
 420         kqswnal_put_idle_tx (ktx);
 421 }
 422
 423 static void
 424 kqswnal_txhandler(EP_TXD *txd, void *arg, int status)
 425 {
 426         kqswnal_tx_t      *ktx = (kqswnal_tx_t *)arg;
 427
 428         LASSERT (txd != NULL);
 429         LASSERT (ktx != NULL);
 430
 431         CDEBUG(D_NET, "txd %p, arg %p status %d\n", txd, arg, status);
 432
 433         if (status == EP_SUCCESS)
 434                 atomic_inc (&kqswnal_packets_transmitted);
 435
 436         if (status != EP_SUCCESS)
 437         {
 438                 CERROR ("kqswnal: Transmit failed with %d\n", status);
 439                 status = -EIO;
 440         }
 441
 442         kqswnal_tx_done (ktx, status);
 443 }
 444
 445 int
 446 kqswnal_launch (kqswnal_tx_t *ktx)
 447 {
 448         /* Don't block for transmit descriptor if we're in interrupt context */
 449         int   attr = in_interrupt() ? (EP_NO_SLEEP | EP_NO_ALLOC) : 0;
 450         int   dest = kqswnal_nid2elanid (ktx->ktx_nid);
 451         long  flags;
 452         int   rc;
 453
 454         LASSERT (dest >= 0);                    /* must be a peer */
 455         rc = ep_transmit_large(kqswnal_data.kqn_eptx, dest,
 456                                ktx->ktx_port, attr, kqswnal_txhandler,
 457                                ktx, ktx->ktx_iov, ktx->ktx_niov);
 458         if (rc == 0)
 459                 atomic_inc (&kqswnal_packets_launched);
 460
 461         if (rc != ENOMEM)
 462                 return (rc);
 463
 464         /* can't allocate ep txd => queue for later */
 465
 466         LASSERT (in_interrupt());      /* not called by thread (not looping) */
 467
 468         spin_lock_irqsave (&kqswnal_data.kqn_sched_lock, flags);
 469
 470         list_add_tail (&ktx->ktx_list, &kqswnal_data.kqn_delayedtxds);
 471         if (waitqueue_active (&kqswnal_data.kqn_sched_waitq))
 472                 wake_up (&kqswnal_data.kqn_sched_waitq);
 473
 474         spin_unlock_irqrestore (&kqswnal_data.kqn_sched_lock, flags);
 475
 476         return (0);
 477 }
 478
 479
 480 static char *
 481 hdr_type_string (ptl_hdr_t *hdr)
 482 {
 483         switch (hdr->type) {
 484         case PTL_MSG_ACK:
 485                 return ("ACK");
 486         case PTL_MSG_PUT:
 487                 return ("PUT");
 488         case PTL_MSG_GET:
 489                 return ("GET");
 490         case PTL_MSG_REPLY:
 491                 return ("REPLY");
 492         default:
 493                 return ("<UNKNOWN>");
 494         }
 495 }
 496
 497 static void
 498 kqswnal_cerror_hdr(ptl_hdr_t * hdr)
 499 {
 500         char *type_str = hdr_type_string (hdr);
 501
 502         CERROR("P3 Header at %p of type %s\n", hdr, type_str);
 503         CERROR("    From nid/pid "LPU64"/%u", NTOH__u64(hdr->src_nid),
 504                NTOH__u32(hdr->src_pid));
 505         CERROR("    To nid/pid "LPU64"/%u\n", NTOH__u64(hdr->dest_nid),
 506                NTOH__u32(hdr->dest_pid));
 507
 508         switch (NTOH__u32(hdr->type)) {
 509         case PTL_MSG_PUT:
 510                 CERROR("    Ptl index %d, ack md "LPX64"."LPX64", "
 511                        "match bits "LPX64"\n",
 512                        NTOH__u32 (hdr->msg.put.ptl_index),
 513                        hdr->msg.put.ack_wmd.wh_interface_cookie,
 514                        hdr->msg.put.ack_wmd.wh_object_cookie,
 515                        NTOH__u64 (hdr->msg.put.match_bits));
 516                 CERROR("    Length %d, offset %d, hdr data "LPX64"\n",
 517                        NTOH__u32(PTL_HDR_LENGTH(hdr)),
 518                        NTOH__u32(hdr->msg.put.offset),
 519                        hdr->msg.put.hdr_data);
 520                 break;
 521
 522         case PTL_MSG_GET:
 523                 CERROR("    Ptl index %d, return md "LPX64"."LPX64", "
 524                        "match bits "LPX64"\n",
 525                        NTOH__u32 (hdr->msg.get.ptl_index),
 526                        hdr->msg.get.return_wmd.wh_interface_cookie,
 527                        hdr->msg.get.return_wmd.wh_object_cookie,
 528                        hdr->msg.get.match_bits);
 529                 CERROR("    Length %d, src offset %d\n",
 530                        NTOH__u32 (hdr->msg.get.sink_length),
 531                        NTOH__u32 (hdr->msg.get.src_offset));
 532                 break;
 533
 534         case PTL_MSG_ACK:
 535                 CERROR("    dst md "LPX64"."LPX64", manipulated length %d\n",
 536                        hdr->msg.ack.dst_wmd.wh_interface_cookie,
 537                        hdr->msg.ack.dst_wmd.wh_object_cookie,
 538                        NTOH__u32 (hdr->msg.ack.mlength));
 539                 break;
 540
 541         case PTL_MSG_REPLY:
 542                 CERROR("    dst md "LPX64"."LPX64", length %d\n",
 543                        hdr->msg.reply.dst_wmd.wh_interface_cookie,
 544                        hdr->msg.reply.dst_wmd.wh_object_cookie,
 545                        NTOH__u32 (PTL_HDR_LENGTH(hdr)));
 546         }
 547
 548 }                               /* end of print_hdr() */
 549
 550 static int
 551 kqswnal_sendmsg (nal_cb_t     *nal,
 552                  void         *private,
 553                  lib_msg_t    *cookie,
 554                  ptl_hdr_t    *hdr,
 555                  int           type,
 556                  ptl_nid_t     nid,
 557                  ptl_pid_t     pid,
 558                  unsigned int  payload_niov,
 559                  struct iovec *payload_iov,
 560                  ptl_kiov_t   *payload_kiov,
 561                  size_t        payload_nob)
 562 {
 563         kqswnal_tx_t      *ktx;
 564         int                rc;
 565         ptl_nid_t          gatewaynid;
 566 #if KQSW_CHECKSUM
 567         int                i;
 568         kqsw_csum_t        csum;
 569         int                sumnob;
 570 #endif
 571
 572         /* NB, the return code from this procedure is ignored.
 573          * If we can't send, we must still complete with lib_finalize().
 574          * We'll have to wait for 3.2 to return an error event.
 575          */
 576
 577         CDEBUG(D_NET, "sending "LPSZ" bytes in %d frags to nid: "LPX64
 578                " pid %u\n", payload_nob, payload_niov, nid, pid);
 579
 580         LASSERT (payload_nob == 0 || payload_niov > 0);
 581         LASSERT (payload_niov <= PTL_MD_MAX_IOV);
 582
 583         /* It must be OK to kmap() if required */
 584         LASSERT (payload_kiov == NULL || !in_interrupt ());
 585         /* payload is either all vaddrs or all pages */
 586         LASSERT (!(payload_kiov != NULL && payload_iov != NULL));
 587
 588         if (payload_nob > KQSW_MAXPAYLOAD) {
 589                 CERROR ("request exceeds MTU size "LPSZ" (max %u).\n",
 590                         payload_nob, KQSW_MAXPAYLOAD);
 591                 lib_finalize (&kqswnal_lib, private, cookie);
 592                 return (-1);
 593         }
 594
 595         if (kqswnal_nid2elanid (nid) < 0) {     /* Can't send direct: find gateway? */
 596                 rc = kpr_lookup (&kqswnal_data.kqn_router, nid, &gatewaynid);
 597                 if (rc != 0) {
 598                         CERROR("Can't route to "LPX64": router error %d\n",
 599                                nid, rc);
 600                         lib_finalize (&kqswnal_lib, private, cookie);
 601                         return (-1);
 602                 }
 603                 if (kqswnal_nid2elanid (gatewaynid) < 0) {
 604                         CERROR("Bad gateway "LPX64" for "LPX64"\n",
 605                                gatewaynid, nid);
 606                         lib_finalize (&kqswnal_lib, private, cookie);
 607                         return (-1);
 608                 }
 609                 nid = gatewaynid;
 610         }
 611
 612         /* I may not block for a transmit descriptor if I might block the
 613          * receiver, or an interrupt handler. */
 614         ktx = kqswnal_get_idle_tx(NULL, !(type == PTL_MSG_ACK ||
 615                                           type == PTL_MSG_REPLY ||
 616                                           in_interrupt()));
 617         if (ktx == NULL) {
 618                 kqswnal_cerror_hdr (hdr);
 619                 lib_finalize (&kqswnal_lib, private, cookie);
 620                 return (-1);
 621         }
 622
 623         memcpy (ktx->ktx_buffer, hdr, sizeof (*hdr)); /* copy hdr from caller's stack */
 624
 625 #if KQSW_CHECKSUM
 626         csum = kqsw_csum (0, (char *)hdr, sizeof (*hdr));
 627         memcpy (ktx->ktx_buffer + sizeof (*hdr), &csum, sizeof (csum));
 628         for (csum = 0, i = 0, sumnob = payload_nob; sumnob > 0; i++) {
 629                 if (payload_kiov != NULL) {
 630                         ptl_kiov_t *kiov = &payload_kiov[i];
 631                         char       *addr = ((char *)kmap (kiov->kiov_page)) +
 632                                            kiov->kiov_offset;
 633
 634                         csum = kqsw_csum (csum, addr, MIN (sumnob, kiov->kiov_len));
 635                         sumnob -= kiov->kiov_len;
 636                 } else {
 637                         struct iovec *iov = &payload_iov[i];
 638
 639                         csum = kqsw_csum (csum, iov->iov_base, MIN (sumnob, kiov->iov_len));
 640                         sumnob -= iov->iov_len;
 641                 }
 642         }
 643         memcpy(ktx->ktx_buffer +sizeof(*hdr) +sizeof(csum), &csum,sizeof(csum));
 644 #endif
 645
 646         /* Set up first frag from pre-mapped buffer (it's at least the
 647          * portals header) */
 648         ktx->ktx_iov[0].Base = ktx->ktx_ebuffer;
 649         ktx->ktx_iov[0].Len = KQSW_HDR_SIZE;
 650         ktx->ktx_niov = 1;
 651
 652         if (payload_nob > 0) { /* got some payload (something more to do) */
 653                 /* make a single contiguous message? */
 654                 if (payload_nob <= KQSW_TX_MAXCONTIG) {
 655                         /* copy payload to ktx_buffer, immediately after hdr */
 656                         if (payload_kiov != NULL)
 657                                 lib_copy_kiov2buf (ktx->ktx_buffer + KQSW_HDR_SIZE,
 658                                                    payload_niov, payload_kiov, payload_nob);
 659                         else
 660                                 lib_copy_iov2buf (ktx->ktx_buffer + KQSW_HDR_SIZE,
 661                                                   payload_niov, payload_iov, payload_nob);
 662                         /* first frag includes payload */
 663                         ktx->ktx_iov[0].Len += payload_nob;
 664                 } else {
 665                         if (payload_kiov != NULL)
 666                                 rc = kqswnal_map_tx_kiov (ktx, payload_nob,
 667                                                           payload_niov, payload_kiov);
 668                         else
 669                                 rc = kqswnal_map_tx_iov (ktx, payload_nob,
 670                                                          payload_niov, payload_iov);
 671                         if (rc != 0) {
 672                                 kqswnal_put_idle_tx (ktx);
 673                                 lib_finalize (&kqswnal_lib, private, cookie);
 674                                 return (-1);
 675                         }
 676                 }
 677         }
 678
 679         ktx->ktx_port    = (payload_nob <= KQSW_SMALLPAYLOAD) ?
 680                         EP_SVC_LARGE_PORTALS_SMALL : EP_SVC_LARGE_PORTALS_LARGE;
 681         ktx->ktx_nid     = nid;
 682         ktx->ktx_state   = KTX_SENDING;   /* => lib_finalize() on completion */
 683         ktx->ktx_args[0] = private;
 684         ktx->ktx_args[1] = cookie;
 685
 686         rc = kqswnal_launch (ktx);
 687         if (rc != 0) {                    /* failed? */
 688                 CERROR ("Failed to send packet to "LPX64": %d\n", nid, rc);
 689                 lib_finalize (&kqswnal_lib, private, cookie);
 690                 return (-1);
 691         }
 692
 693         CDEBUG(D_NET, "send to "LPSZ" bytes to "LPX64"\n", payload_nob, nid);
 694         return (0);
 695 }
 696
 697 static int
 698 kqswnal_send (nal_cb_t     *nal,
 699               void         *private,
 700               lib_msg_t    *cookie,
 701               ptl_hdr_t    *hdr,
 702               int           type,
 703               ptl_nid_t     nid,
 704               ptl_pid_t     pid,
 705               unsigned int  payload_niov,
 706               struct iovec *payload_iov,
 707               size_t        payload_nob)
 708 {
 709         return (kqswnal_sendmsg (nal, private, cookie, hdr, type, nid, pid,
 710                                  payload_niov, payload_iov, NULL, payload_nob));
 711 }
 712
 713 static int
 714 kqswnal_send_pages (nal_cb_t     *nal,
 715                     void         *private,
 716                     lib_msg_t    *cookie,
 717                     ptl_hdr_t    *hdr,
 718                     int           type,
 719                     ptl_nid_t     nid,
 720                     ptl_pid_t     pid,
 721                     unsigned int  payload_niov,
 722                     ptl_kiov_t   *payload_kiov,
 723                     size_t        payload_nob)
 724 {
 725         return (kqswnal_sendmsg (nal, private, cookie, hdr, type, nid, pid,
 726                                  payload_niov, NULL, payload_kiov, payload_nob));
 727 }
 728
 729 int kqswnal_fwd_copy_contig = 0;
 730
 731 void
 732 kqswnal_fwd_packet (void *arg, kpr_fwd_desc_t *fwd)
 733 {
 734         int             rc;
 735         kqswnal_tx_t   *ktx;
 736         struct iovec   *iov = fwd->kprfd_iov;
 737         int             niov = fwd->kprfd_niov;
 738         int             nob = fwd->kprfd_nob;
 739         ptl_nid_t       nid = fwd->kprfd_gateway_nid;
 740
 741 #if KQSW_CHECKSUM
 742         CERROR ("checksums for forwarded packets not implemented\n");
 743         LBUG ();
 744 #endif
 745         /* The router wants this NAL to forward a packet */
 746         CDEBUG (D_NET, "forwarding [%p] to "LPX64", %d frags %d bytes\n",
 747                 fwd, nid, niov, nob);
 748
 749         LASSERT (niov > 0);
 750
 751         ktx = kqswnal_get_idle_tx (fwd, FALSE);
 752         if (ktx == NULL)        /* can't get txd right now */
 753                 return;         /* fwd will be scheduled when tx desc freed */
 754
 755         if (nid == kqswnal_lib.ni.nid)          /* gateway is me */
 756                 nid = fwd->kprfd_target_nid;    /* target is final dest */
 757
 758         if (kqswnal_nid2elanid (nid) < 0) {
 759                 CERROR("Can't forward [%p] to "LPX64": not a peer\n", fwd, nid);
 760                 rc = -EHOSTUNREACH;
 761                 goto failed;
 762         }
 763
 764         if (nob > KQSW_NRXMSGBYTES_LARGE) {
 765                 CERROR ("Can't forward [%p] to "LPX64
 766                         ": size %d bigger than max packet size %ld\n",
 767                         fwd, nid, nob, (long)KQSW_NRXMSGBYTES_LARGE);
 768                 rc = -EMSGSIZE;
 769                 goto failed;
 770         }
 771
 772         if ((kqswnal_fwd_copy_contig || niov > 1) &&
 773             nob <= KQSW_TX_BUFFER_SIZE)
 774         {
 775                 /* send from ktx's pre-allocated/mapped contiguous buffer? */
 776                 lib_copy_iov2buf (ktx->ktx_buffer, niov, iov, nob);
 777                 ktx->ktx_iov[0].Base = ktx->ktx_ebuffer; /* already mapped */
 778                 ktx->ktx_iov[0].Len = nob;
 779                 ktx->ktx_niov = 1;
 780         }
 781         else
 782         {
 783                 /* zero copy */
 784                 ktx->ktx_niov = 0;        /* no frags mapped yet */
 785                 rc = kqswnal_map_tx_iov (ktx, nob, niov, iov);
 786                 if (rc != 0)
 787                         goto failed;
 788         }
 789
 790         ktx->ktx_port    = (nob <= (sizeof (ptl_hdr_t) + KQSW_SMALLPAYLOAD)) ?
 791                         EP_SVC_LARGE_PORTALS_SMALL : EP_SVC_LARGE_PORTALS_LARGE;
 792         ktx->ktx_nid     = nid;
 793         ktx->ktx_state   = KTX_FORWARDING; /* kpr_put_packet() on completion */
 794         ktx->ktx_args[0] = fwd;
 795
 796         rc = kqswnal_launch (ktx);
 797         if (rc == 0)
 798                 return;
 799
 800  failed:
 801         LASSERT (rc != 0);
 802         CERROR ("Failed to forward [%p] to "LPX64": %d\n", fwd, nid, rc);
 803
 804         kqswnal_put_idle_tx (ktx);
 805         /* complete now (with failure) */
 806         kpr_fwd_done (&kqswnal_data.kqn_router, fwd, rc);
 807 }
 808
 809 void
 810 kqswnal_fwd_callback (void *arg, int error)
 811 {
 812         kqswnal_rx_t *krx = (kqswnal_rx_t *)arg;
 813
 814         /* The router has finished forwarding this packet */
 815
 816         if (error != 0)
 817         {
 818                 ptl_hdr_t *hdr = (ptl_hdr_t *)page_address (krx->krx_pages[0]);
 819
 820                 CERROR("Failed to route packet from "LPX64" to "LPX64": %d\n",
 821                        NTOH__u64(hdr->src_nid), NTOH__u64(hdr->dest_nid),error);
 822         }
 823
 824         kqswnal_requeue_rx (krx);
 825 }
 826
 827 void
 828 kqswnal_rx (kqswnal_rx_t *krx)
 829 {
 830         ptl_hdr_t      *hdr = (ptl_hdr_t *) page_address (krx->krx_pages[0]);
 831         ptl_nid_t       dest_nid = NTOH__u64 (hdr->dest_nid);
 832         int             nob;
 833         int             niov;
 834
 835         if (dest_nid == kqswnal_lib.ni.nid) { /* It's for me :) */
 836                 /* NB krx requeued when lib_parse() calls back kqswnal_recv */
 837                 lib_parse (&kqswnal_lib, hdr, krx);
 838                 return;
 839         }
 840
 841 #if KQSW_CHECKSUM
 842         CERROR ("checksums for forwarded packets not implemented\n");
 843         LBUG ();
 844 #endif
 845         if (kqswnal_nid2elanid (dest_nid) >= 0)  /* should have gone direct to peer */
 846         {
 847                 CERROR("dropping packet from "LPX64" for "LPX64
 848                        ": target is peer\n", NTOH__u64(hdr->src_nid), dest_nid);
 849                 kqswnal_requeue_rx (krx);
 850                 return;
 851         }
 852
 853         /* NB forwarding may destroy iov; rebuild every time */
 854         for (nob = krx->krx_nob, niov = 0; nob > 0; nob -= PAGE_SIZE, niov++)
 855         {
 856                 LASSERT (niov < krx->krx_npages);
 857                 krx->krx_iov[niov].iov_base= page_address(krx->krx_pages[niov]);
 858                 krx->krx_iov[niov].iov_len = MIN(PAGE_SIZE, nob);
 859         }
 860
 861         kpr_fwd_init (&krx->krx_fwd, dest_nid,
 862                       krx->krx_nob, niov, krx->krx_iov,
 863                       kqswnal_fwd_callback, krx);
 864
 865         kpr_fwd_start (&kqswnal_data.kqn_router, &krx->krx_fwd);
 866 }
 867
 868 /* Receive Interrupt Handler: posts to schedulers */
 869 void
 870 kqswnal_rxhandler(EP_RXD *rxd)
 871 {
 872         long          flags;
 873         int           nob    = ep_rxd_len (rxd);
 874         int           status = ep_rxd_status (rxd);
 875         kqswnal_rx_t *krx    = (kqswnal_rx_t *)ep_rxd_arg (rxd);
 876
 877         CDEBUG(D_NET, "kqswnal_rxhandler: rxd %p, krx %p, nob %d, status %d\n",
 878                rxd, krx, nob, status);
 879
 880         LASSERT (krx != NULL);
 881
 882         krx->krx_rxd = rxd;
 883         krx->krx_nob = nob;
 884
 885         /* must receive a whole header to be able to parse */
 886         if (status != EP_SUCCESS || nob < sizeof (ptl_hdr_t))
 887         {
 888                 /* receives complete with failure when receiver is removed */
 889                 if (kqswnal_data.kqn_shuttingdown)
 890                         return;
 891
 892                 CERROR("receive status failed with status %d nob %d\n",
 893                        ep_rxd_status(rxd), nob);
 894                 kqswnal_requeue_rx (krx);
 895                 return;
 896         }
 897
 898         atomic_inc (&kqswnal_packets_received);
 899
 900         spin_lock_irqsave (&kqswnal_data.kqn_sched_lock, flags);
 901
 902         list_add_tail (&krx->krx_list, &kqswnal_data.kqn_readyrxds);
 903         if (waitqueue_active (&kqswnal_data.kqn_sched_waitq))
 904                 wake_up (&kqswnal_data.kqn_sched_waitq);
 905
 906         spin_unlock_irqrestore (&kqswnal_data.kqn_sched_lock, flags);
 907 }
 908
 909 #if KQSW_CHECKSUM
 910 void
 911 kqswnal_csum_error (kqswnal_rx_t *krx, int ishdr)
 912 {
 913         ptl_hdr_t *hdr = (ptl_hdr_t *)page_address (krx->krx_pages[0]);
 914
 915         CERROR ("%s checksum mismatch %p: dnid "LPX64", snid "LPX64
 916                 ", dpid %d, spid %d, type %d\n",
 917                 ishdr ? "Header" : "Payload", krx,
 918                 NTOH__u64(hdr->dest_nid), NTOH__u64(hdr->src_nid)
 919                 NTOH__u32(hdr->dest_pid), NTOH__u32(hdr->src_pid),
 920                 NTOH__u32(hdr->type));
 921
 922         switch (NTOH__u32 (hdr->type))
 923         {
 924         case PTL_MSG_ACK:
 925                 CERROR("ACK: mlen %d dmd "LPX64"."LPX64" match "LPX64
 926                        " len %u\n",
 927                        NTOH__u32(hdr->msg.ack.mlength),
 928                        hdr->msg.ack.dst_wmd.handle_cookie,
 929                        hdr->msg.ack.dst_wmd.handle_idx,
 930                        NTOH__u64(hdr->msg.ack.match_bits),
 931                        NTOH__u32(hdr->msg.ack.length));
 932                 break;
 933         case PTL_MSG_PUT:
 934                 CERROR("PUT: ptl %d amd "LPX64"."LPX64" match "LPX64
 935                        " len %u off %u data "LPX64"\n",
 936                        NTOH__u32(hdr->msg.put.ptl_index),
 937                        hdr->msg.put.ack_wmd.handle_cookie,
 938                        hdr->msg.put.ack_wmd.handle_idx,
 939                        NTOH__u64(hdr->msg.put.match_bits),
 940                        NTOH__u32(hdr->msg.put.length),
 941                        NTOH__u32(hdr->msg.put.offset),
 942                        hdr->msg.put.hdr_data);
 943                 break;
 944         case PTL_MSG_GET:
 945                 CERROR ("GET: <>\n");
 946                 break;
 947         case PTL_MSG_REPLY:
 948                 CERROR ("REPLY: <>\n");
 949                 break;
 950         default:
 951                 CERROR ("TYPE?: <>\n");
 952         }
 953 }
 954 #endif
 955
 956 static int
 957 kqswnal_recvmsg (nal_cb_t     *nal,
 958                  void         *private,
 959                  lib_msg_t    *cookie,
 960                  unsigned int  niov,
 961                  struct iovec *iov,
 962                  ptl_kiov_t   *kiov,
 963                  size_t        mlen,
 964                  size_t        rlen)
 965 {
 966         kqswnal_rx_t *krx = (kqswnal_rx_t *)private;
 967         int           page;
 968         char         *page_ptr;
 969         int           page_nob;
 970         char         *iov_ptr;
 971         int           iov_nob;
 972         int           frag;
 973 #if KQSW_CHECKSUM
 974         kqsw_csum_t   senders_csum;
 975         kqsw_csum_t   payload_csum = 0;
 976         kqsw_csum_t   hdr_csum = kqsw_csum(0, page_address(krx->krx_pages[0]),
 977                                            sizeof(ptl_hdr_t));
 978         size_t        csum_len = mlen;
 979         int           csum_frags = 0;
 980         int           csum_nob = 0;
 981         static atomic_t csum_counter;
 982         int           csum_verbose = (atomic_read(&csum_counter)%1000001) == 0;
 983
 984         atomic_inc (&csum_counter);
 985
 986         memcpy (&senders_csum, ((char *)page_address (krx->krx_pages[0])) +
 987                                 sizeof (ptl_hdr_t), sizeof (kqsw_csum_t));
 988         if (senders_csum != hdr_csum)
 989                 kqswnal_csum_error (krx, 1);
 990 #endif
 991         CDEBUG(D_NET,"kqswnal_recv, mlen="LPSZ", rlen="LPSZ"\n", mlen, rlen);
 992
 993         /* What was actually received must be >= payload.
 994          * This is an LASSERT, as lib_finalize() doesn't have a completion status. */
 995         LASSERT (krx->krx_nob >= KQSW_HDR_SIZE + mlen);
 996         LASSERT (mlen <= rlen);
 997
 998         /* It must be OK to kmap() if required */
 999         LASSERT (kiov == NULL || !in_interrupt ());
1000         /* Either all pages or all vaddrs */
1001         LASSERT (!(kiov != NULL && iov != NULL));
1002
1003         if (mlen != 0)
1004         {
1005                 page     = 0;
1006                 page_ptr = ((char *) page_address(krx->krx_pages[0])) +
1007                         KQSW_HDR_SIZE;
1008                 page_nob = PAGE_SIZE - KQSW_HDR_SIZE;
1009
1010                 LASSERT (niov > 0);
1011                 if (kiov != NULL) {
1012                         iov_ptr = ((char *)kmap (kiov->kiov_page)) + kiov->kiov_offset;
1013                         iov_nob = kiov->kiov_len;
1014                 } else {
1015                         iov_ptr = iov->iov_base;
1016                         iov_nob = iov->iov_len;
1017                 }
1018
1019                 for (;;)
1020                 {
1021                         /* We expect the iov to exactly match mlen */
1022                         LASSERT (iov_nob <= mlen);
1023
1024                         frag = MIN (page_nob, iov_nob);
1025                         memcpy (iov_ptr, page_ptr, frag);
1026 #if KQSW_CHECKSUM
1027                         payload_csum = kqsw_csum (payload_csum, iov_ptr, frag);
1028                         csum_nob += frag;
1029                         csum_frags++;
1030 #endif
1031                         mlen -= frag;
1032                         if (mlen == 0)
1033                                 break;
1034
1035                         page_nob -= frag;
1036                         if (page_nob != 0)
1037                                 page_ptr += frag;
1038                         else
1039                         {
1040                                 page++;
1041                                 LASSERT (page < krx->krx_npages);
1042                                 page_ptr = page_address(krx->krx_pages[page]);
1043                                 page_nob = PAGE_SIZE;
1044                         }
1045
1046                         iov_nob -= frag;
1047                         if (iov_nob != 0)
1048                                 iov_ptr += frag;
1049                         else if (kiov != NULL) {
1050                                 kunmap (kiov->kiov_page);
1051                                 kiov++;
1052                                 niov--;
1053                                 LASSERT (niov > 0);
1054                                 iov_ptr = ((char *)kmap (kiov->kiov_page)) + kiov->kiov_offset;
1055                                 iov_nob = kiov->kiov_len;
1056                         } else {
1057                                 iov++;
1058                                 niov--;
1059                                 LASSERT (niov > 0);
1060                                 iov_ptr = iov->iov_base;
1061                                 iov_nob = iov->iov_len;
1062                         }
1063                 }
1064
1065                 if (kiov != NULL)
1066                         kunmap (kiov->kiov_page);
1067         }
1068
1069 #if KQSW_CHECKSUM
1070         memcpy (&senders_csum, ((char *)page_address (krx->krx_pages[0])) +
1071                 sizeof(ptl_hdr_t) + sizeof(kqsw_csum_t), sizeof(kqsw_csum_t));
1072
1073         if (csum_len != rlen)
1074                 CERROR("Unable to checksum data in user's buffer\n");
1075         else if (senders_csum != payload_csum)
1076                 kqswnal_csum_error (krx, 0);
1077
1078         if (csum_verbose)
1079                 CERROR("hdr csum %lx, payload_csum %lx, csum_frags %d, "
1080                        "csum_nob %d\n",
1081                         hdr_csum, payload_csum, csum_frags, csum_nob);
1082 #endif
1083         lib_finalize(nal, private, cookie);
1084
1085         kqswnal_requeue_rx (krx);
1086
1087         return (rlen);
1088 }
1089
1090 static int
1091 kqswnal_recv(nal_cb_t     *nal,
1092              void         *private,
1093              lib_msg_t    *cookie,
1094              unsigned int  niov,
1095              struct iovec *iov,
1096              size_t        mlen,
1097              size_t        rlen)
1098 {
1099         return (kqswnal_recvmsg (nal, private, cookie, niov, iov, NULL, mlen, rlen));
1100 }
1101
1102 static int
1103 kqswnal_recv_pages (nal_cb_t     *nal,
1104                     void         *private,
1105                     lib_msg_t    *cookie,
1106                     unsigned int  niov,
1107                     ptl_kiov_t   *kiov,
1108                     size_t        mlen,
1109                     size_t        rlen)
1110 {
1111         return (kqswnal_recvmsg (nal, private, cookie, niov, NULL, kiov, mlen, rlen));
1112 }
1113
1114 int
1115 kqswnal_thread_start (int (*fn)(void *arg), void *arg)
1116 {
1117         long    pid = kernel_thread (fn, arg, 0);
1118
1119         if (pid < 0)
1120                 return ((int)pid);
1121
1122         atomic_inc (&kqswnal_data.kqn_nthreads);
1123         return (0);
1124 }
1125
1126 void
1127 kqswnal_thread_fini (void)
1128 {
1129         atomic_dec (&kqswnal_data.kqn_nthreads);
1130 }
1131
1132 int
1133 kqswnal_scheduler (void *arg)
1134 {
1135         kqswnal_rx_t    *krx;
1136         kqswnal_tx_t    *ktx;
1137         kpr_fwd_desc_t  *fwd;
1138         long             flags;
1139         int              rc;
1140         int              counter = 0;
1141         int              did_something;
1142
1143         kportal_daemonize ("kqswnal_sched");
1144         kportal_blockallsigs ();
1145
1146         spin_lock_irqsave (&kqswnal_data.kqn_sched_lock, flags);
1147
1148         while (!kqswnal_data.kqn_shuttingdown)
1149         {
1150                 did_something = FALSE;
1151
1152                 if (!list_empty (&kqswnal_data.kqn_readyrxds))
1153                 {
1154                         krx = list_entry(kqswnal_data.kqn_readyrxds.next,
1155                                          kqswnal_rx_t, krx_list);
1156                         list_del (&krx->krx_list);
1157                         spin_unlock_irqrestore(&kqswnal_data.kqn_sched_lock,
1158                                                flags);
1159
1160                         kqswnal_rx (krx);
1161
1162                         did_something = TRUE;
1163                         spin_lock_irqsave(&kqswnal_data.kqn_sched_lock, flags);
1164                 }
1165
1166                 if (!list_empty (&kqswnal_data.kqn_delayedtxds))
1167                 {
1168                         ktx = list_entry(kqswnal_data.kqn_delayedtxds.next,
1169                                          kqswnal_tx_t, ktx_list);
1170                         list_del (&ktx->ktx_list);
1171                         spin_unlock_irqrestore(&kqswnal_data.kqn_sched_lock,
1172                                                flags);
1173
1174                         rc = kqswnal_launch (ktx);
1175                         if (rc != 0)          /* failed: ktx_nid down? */
1176                         {
1177                                 CERROR("Failed delayed transmit to "LPX64
1178                                        ": %d\n", ktx->ktx_nid, rc);
1179                                 kqswnal_tx_done (ktx, rc);
1180                         }
1181
1182                         did_something = TRUE;
1183                         spin_lock_irqsave (&kqswnal_data.kqn_sched_lock, flags);
1184                 }
1185
1186                 if (!list_empty (&kqswnal_data.kqn_delayedfwds))
1187                 {
1188                         fwd = list_entry (kqswnal_data.kqn_delayedfwds.next, kpr_fwd_desc_t, kprfd_list);
1189                         list_del (&fwd->kprfd_list);
1190                         spin_unlock_irqrestore (&kqswnal_data.kqn_sched_lock, flags);
1191
1192                         kqswnal_fwd_packet (NULL, fwd);
1193
1194                         did_something = TRUE;
1195                         spin_lock_irqsave (&kqswnal_data.kqn_sched_lock, flags);
1196                 }
1197
1198                     /* nothing to do or hogging CPU */
1199                 if (!did_something || counter++ == KQSW_RESCHED) {
1200                         spin_unlock_irqrestore(&kqswnal_data.kqn_sched_lock,
1201                                                flags);
1202
1203                         counter = 0;
1204
1205                         if (!did_something) {
1206                                 rc = wait_event_interruptible (kqswnal_data.kqn_sched_waitq,
1207                                                                kqswnal_data.kqn_shuttingdown ||
1208                                                                !list_empty(&kqswnal_data.kqn_readyrxds) ||
1209                                                                !list_empty(&kqswnal_data.kqn_delayedtxds) ||
1210                                                                !list_empty(&kqswnal_data.kqn_delayedfwds));
1211                                 LASSERT (rc == 0);
1212                         } else if (current->need_resched)
1213                                 schedule ();
1214
1215                         spin_lock_irqsave (&kqswnal_data.kqn_sched_lock, flags);
1216                 }
1217         }
1218
1219         spin_unlock_irqrestore (&kqswnal_data.kqn_sched_lock, flags);
1220
1221         kqswnal_thread_fini ();
1222         return (0);
1223 }
1224
1225 nal_cb_t kqswnal_lib =
1226 {
1227         nal_data:       &kqswnal_data,         /* NAL private data */
1228         cb_send:        kqswnal_send,
1229         cb_send_pages:  kqswnal_send_pages,
1230         cb_recv:        kqswnal_recv,
1231         cb_recv_pages:  kqswnal_recv_pages,
1232         cb_read:        kqswnal_read,
1233         cb_write:       kqswnal_write,
1234         cb_malloc:      kqswnal_malloc,
1235         cb_free:        kqswnal_free,
1236         cb_printf:      kqswnal_printf,
1237         cb_cli:         kqswnal_cli,
1238         cb_sti:         kqswnal_sti,
1239         cb_dist:        kqswnal_dist
1240 };