lnet/klnds/ralnd/ralnd_cb.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) 2004, 2010, Oracle and/or its affiliates. All rights reserved.
  28  * Use is subject to license terms.
  29  *
  30  * Copyright (c) 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  * lnet/klnds/ralnd/ralnd_cb.c
  37  *
  38  * Author: Eric Barton <eric@bartonsoftware.com>
  39  */
  40
  41 #include "ralnd.h"
  42
  43 void
  44 kranal_device_callback(RAP_INT32 devid, RAP_PVOID arg)
  45 {
  46         kra_device_t *dev;
  47         int           i;
  48         unsigned long flags;
  49
  50         CDEBUG(D_NET, "callback for device %d\n", devid);
  51
  52         for (i = 0; i < kranal_data.kra_ndevs; i++) {
  53
  54                 dev = &kranal_data.kra_devices[i];
  55                 if (dev->rad_id != devid)
  56                         continue;
  57
  58                 spin_lock_irqsave(&dev->rad_lock, flags);
  59
  60                 if (!dev->rad_ready) {
  61                         dev->rad_ready = 1;
  62                         wake_up(&dev->rad_waitq);
  63                 }
  64
  65                 spin_unlock_irqrestore(&dev->rad_lock, flags);
  66                 return;
  67         }
  68
  69         CWARN("callback for unknown device %d\n", devid);
  70 }
  71
  72 void
  73 kranal_schedule_conn(kra_conn_t *conn)
  74 {
  75         kra_device_t    *dev = conn->rac_device;
  76         unsigned long    flags;
  77
  78         spin_lock_irqsave(&dev->rad_lock, flags);
  79
  80         if (!conn->rac_scheduled) {
  81                 kranal_conn_addref(conn);       /* +1 ref for scheduler */
  82                 conn->rac_scheduled = 1;
  83                 cfs_list_add_tail(&conn->rac_schedlist, &dev->rad_ready_conns);
  84                 wake_up(&dev->rad_waitq);
  85         }
  86
  87         spin_unlock_irqrestore(&dev->rad_lock, flags);
  88 }
  89
  90 kra_tx_t *
  91 kranal_get_idle_tx (void)
  92 {
  93         unsigned long  flags;
  94         kra_tx_t      *tx;
  95
  96         spin_lock_irqsave(&kranal_data.kra_tx_lock, flags);
  97
  98         if (cfs_list_empty(&kranal_data.kra_idle_txs)) {
  99                 spin_unlock_irqrestore(&kranal_data.kra_tx_lock, flags);
 100                 return NULL;
 101         }
 102
 103         tx = cfs_list_entry(kranal_data.kra_idle_txs.next, kra_tx_t, tx_list);
 104         cfs_list_del(&tx->tx_list);
 105
 106         /* Allocate a new completion cookie.  It might not be needed, but we've
 107          * got a lock right now... */
 108         tx->tx_cookie = kranal_data.kra_next_tx_cookie++;
 109
 110         spin_unlock_irqrestore(&kranal_data.kra_tx_lock, flags);
 111
 112         LASSERT (tx->tx_buftype == RANAL_BUF_NONE);
 113         LASSERT (tx->tx_msg.ram_type == RANAL_MSG_NONE);
 114         LASSERT (tx->tx_conn == NULL);
 115         LASSERT (tx->tx_lntmsg[0] == NULL);
 116         LASSERT (tx->tx_lntmsg[1] == NULL);
 117
 118         return tx;
 119 }
 120
 121 void
 122 kranal_init_msg(kra_msg_t *msg, int type)
 123 {
 124         msg->ram_magic = RANAL_MSG_MAGIC;
 125         msg->ram_version = RANAL_MSG_VERSION;
 126         msg->ram_type = type;
 127         msg->ram_srcnid = kranal_data.kra_ni->ni_nid;
 128         /* ram_connstamp gets set when FMA is sent */
 129 }
 130
 131 kra_tx_t *
 132 kranal_new_tx_msg (int type)
 133 {
 134         kra_tx_t *tx = kranal_get_idle_tx();
 135
 136         if (tx != NULL)
 137                 kranal_init_msg(&tx->tx_msg, type);
 138
 139         return tx;
 140 }
 141
 142 int
 143 kranal_setup_immediate_buffer (kra_tx_t *tx,
 144                                unsigned int niov, struct iovec *iov,
 145                                int offset, int nob)
 146
 147 {
 148         /* For now this is almost identical to kranal_setup_virt_buffer, but we
 149          * could "flatten" the payload into a single contiguous buffer ready
 150          * for sending direct over an FMA if we ever needed to. */
 151
 152         LASSERT (tx->tx_buftype == RANAL_BUF_NONE);
 153         LASSERT (nob >= 0);
 154
 155         if (nob == 0) {
 156                 tx->tx_buffer = NULL;
 157         } else {
 158                 LASSERT (niov > 0);
 159
 160                 while (offset >= iov->iov_len) {
 161                         offset -= iov->iov_len;
 162                         niov--;
 163                         iov++;
 164                         LASSERT (niov > 0);
 165                 }
 166
 167                 if (nob > iov->iov_len - offset) {
 168                         CERROR("Can't handle multiple vaddr fragments\n");
 169                         return -EMSGSIZE;
 170                 }
 171
 172                 tx->tx_buffer = (void *)(((unsigned long)iov->iov_base) + offset);
 173         }
 174
 175         tx->tx_buftype = RANAL_BUF_IMMEDIATE;
 176         tx->tx_nob = nob;
 177         return 0;
 178 }
 179
 180 int
 181 kranal_setup_virt_buffer (kra_tx_t *tx,
 182                           unsigned int niov, struct iovec *iov,
 183                           int offset, int nob)
 184
 185 {
 186         LASSERT (nob > 0);
 187         LASSERT (niov > 0);
 188         LASSERT (tx->tx_buftype == RANAL_BUF_NONE);
 189
 190         while (offset >= iov->iov_len) {
 191                 offset -= iov->iov_len;
 192                 niov--;
 193                 iov++;
 194                 LASSERT (niov > 0);
 195         }
 196
 197         if (nob > iov->iov_len - offset) {
 198                 CERROR("Can't handle multiple vaddr fragments\n");
 199                 return -EMSGSIZE;
 200         }
 201
 202         tx->tx_buftype = RANAL_BUF_VIRT_UNMAPPED;
 203         tx->tx_nob = nob;
 204         tx->tx_buffer = (void *)(((unsigned long)iov->iov_base) + offset);
 205         return 0;
 206 }
 207
 208 int
 209 kranal_setup_phys_buffer (kra_tx_t *tx, int nkiov, lnet_kiov_t *kiov,
 210                           int offset, int nob)
 211 {
 212         RAP_PHYS_REGION *phys = tx->tx_phys;
 213         int              resid;
 214
 215         CDEBUG(D_NET, "niov %d offset %d nob %d\n", nkiov, offset, nob);
 216
 217         LASSERT (nob > 0);
 218         LASSERT (nkiov > 0);
 219         LASSERT (tx->tx_buftype == RANAL_BUF_NONE);
 220
 221         while (offset >= kiov->kiov_len) {
 222                 offset -= kiov->kiov_len;
 223                 nkiov--;
 224                 kiov++;
 225                 LASSERT (nkiov > 0);
 226         }
 227
 228         tx->tx_buftype = RANAL_BUF_PHYS_UNMAPPED;
 229         tx->tx_nob = nob;
 230         tx->tx_buffer = (void *)((unsigned long)(kiov->kiov_offset + offset));
 231
 232         phys->Address = lnet_page2phys(kiov->kiov_page);
 233         phys++;
 234
 235         resid = nob - (kiov->kiov_len - offset);
 236         while (resid > 0) {
 237                 kiov++;
 238                 nkiov--;
 239                 LASSERT (nkiov > 0);
 240
 241                 if (kiov->kiov_offset != 0 ||
 242                     ((resid > PAGE_SIZE) &&
 243                      kiov->kiov_len < PAGE_SIZE)) {
 244                         /* Can't have gaps */
 245                         CERROR("Can't make payload contiguous in I/O VM:"
 246                                "page %d, offset %d, len %d \n",
 247                                (int)(phys - tx->tx_phys),
 248                                kiov->kiov_offset, kiov->kiov_len);
 249                         return -EINVAL;
 250                 }
 251
 252                 if ((phys - tx->tx_phys) == LNET_MAX_IOV) {
 253                         CERROR ("payload too big (%d)\n", (int)(phys - tx->tx_phys));
 254                         return -EMSGSIZE;
 255                 }
 256
 257                 phys->Address = lnet_page2phys(kiov->kiov_page);
 258                 phys++;
 259
 260                 resid -= PAGE_SIZE;
 261         }
 262
 263         tx->tx_phys_npages = phys - tx->tx_phys;
 264         return 0;
 265 }
 266
 267 static inline int
 268 kranal_setup_rdma_buffer (kra_tx_t *tx, unsigned int niov,
 269                           struct iovec *iov, lnet_kiov_t *kiov,
 270                           int offset, int nob)
 271 {
 272         LASSERT ((iov == NULL) != (kiov == NULL));
 273
 274         if (kiov != NULL)
 275                 return kranal_setup_phys_buffer(tx, niov, kiov, offset, nob);
 276
 277         return kranal_setup_virt_buffer(tx, niov, iov, offset, nob);
 278 }
 279
 280 int
 281 kranal_map_buffer (kra_tx_t *tx)
 282 {
 283         kra_conn_t     *conn = tx->tx_conn;
 284         kra_device_t   *dev = conn->rac_device;
 285         RAP_RETURN      rrc;
 286
 287         LASSERT (current == dev->rad_scheduler);
 288
 289         switch (tx->tx_buftype) {
 290         default:
 291                 LBUG();
 292
 293         case RANAL_BUF_NONE:
 294         case RANAL_BUF_IMMEDIATE:
 295         case RANAL_BUF_PHYS_MAPPED:
 296         case RANAL_BUF_VIRT_MAPPED:
 297                 return 0;
 298
 299         case RANAL_BUF_PHYS_UNMAPPED:
 300                 rrc = RapkRegisterPhys(dev->rad_handle,
 301                                        tx->tx_phys, tx->tx_phys_npages,
 302                                        &tx->tx_map_key);
 303                 if (rrc != RAP_SUCCESS) {
 304                         CERROR ("Can't map %d pages: dev %d "
 305                                 "phys %u pp %u, virt %u nob %lu\n",
 306                                 tx->tx_phys_npages, dev->rad_id,
 307                                 dev->rad_nphysmap, dev->rad_nppphysmap,
 308                                 dev->rad_nvirtmap, dev->rad_nobvirtmap);
 309                         return -ENOMEM; /* assume insufficient resources */
 310                 }
 311
 312                 dev->rad_nphysmap++;
 313                 dev->rad_nppphysmap += tx->tx_phys_npages;
 314
 315                 tx->tx_buftype = RANAL_BUF_PHYS_MAPPED;
 316                 return 0;
 317
 318         case RANAL_BUF_VIRT_UNMAPPED:
 319                 rrc = RapkRegisterMemory(dev->rad_handle,
 320                                          tx->tx_buffer, tx->tx_nob,
 321                                          &tx->tx_map_key);
 322                 if (rrc != RAP_SUCCESS) {
 323                         CERROR ("Can't map %d bytes: dev %d "
 324                                 "phys %u pp %u, virt %u nob %lu\n",
 325                                 tx->tx_nob, dev->rad_id,
 326                                 dev->rad_nphysmap, dev->rad_nppphysmap,
 327                                 dev->rad_nvirtmap, dev->rad_nobvirtmap);
 328                         return -ENOMEM; /* assume insufficient resources */
 329                 }
 330
 331                 dev->rad_nvirtmap++;
 332                 dev->rad_nobvirtmap += tx->tx_nob;
 333
 334                 tx->tx_buftype = RANAL_BUF_VIRT_MAPPED;
 335                 return 0;
 336         }
 337 }
 338
 339 void
 340 kranal_unmap_buffer (kra_tx_t *tx)
 341 {
 342         kra_device_t   *dev;
 343         RAP_RETURN      rrc;
 344
 345         switch (tx->tx_buftype) {
 346         default:
 347                 LBUG();
 348
 349         case RANAL_BUF_NONE:
 350         case RANAL_BUF_IMMEDIATE:
 351         case RANAL_BUF_PHYS_UNMAPPED:
 352         case RANAL_BUF_VIRT_UNMAPPED:
 353                 break;
 354
 355         case RANAL_BUF_PHYS_MAPPED:
 356                 LASSERT (tx->tx_conn != NULL);
 357                 dev = tx->tx_conn->rac_device;
 358                 LASSERT (current == dev->rad_scheduler);
 359                 rrc = RapkDeregisterMemory(dev->rad_handle, NULL,
 360                                            &tx->tx_map_key);
 361                 LASSERT (rrc == RAP_SUCCESS);
 362
 363                 dev->rad_nphysmap--;
 364                 dev->rad_nppphysmap -= tx->tx_phys_npages;
 365
 366                 tx->tx_buftype = RANAL_BUF_PHYS_UNMAPPED;
 367                 break;
 368
 369         case RANAL_BUF_VIRT_MAPPED:
 370                 LASSERT (tx->tx_conn != NULL);
 371                 dev = tx->tx_conn->rac_device;
 372                 LASSERT (current == dev->rad_scheduler);
 373                 rrc = RapkDeregisterMemory(dev->rad_handle, tx->tx_buffer,
 374                                            &tx->tx_map_key);
 375                 LASSERT (rrc == RAP_SUCCESS);
 376
 377                 dev->rad_nvirtmap--;
 378                 dev->rad_nobvirtmap -= tx->tx_nob;
 379
 380                 tx->tx_buftype = RANAL_BUF_VIRT_UNMAPPED;
 381                 break;
 382         }
 383 }
 384
 385 void
 386 kranal_tx_done (kra_tx_t *tx, int completion)
 387 {
 388         lnet_msg_t      *lnetmsg[2];
 389         unsigned long    flags;
 390         int              i;
 391
 392         LASSERT (!in_interrupt());
 393
 394         kranal_unmap_buffer(tx);
 395
 396         lnetmsg[0] = tx->tx_lntmsg[0]; tx->tx_lntmsg[0] = NULL;
 397         lnetmsg[1] = tx->tx_lntmsg[1]; tx->tx_lntmsg[1] = NULL;
 398
 399         tx->tx_buftype = RANAL_BUF_NONE;
 400         tx->tx_msg.ram_type = RANAL_MSG_NONE;
 401         tx->tx_conn = NULL;
 402
 403         spin_lock_irqsave(&kranal_data.kra_tx_lock, flags);
 404
 405         cfs_list_add_tail(&tx->tx_list, &kranal_data.kra_idle_txs);
 406
 407         spin_unlock_irqrestore(&kranal_data.kra_tx_lock, flags);
 408
 409         /* finalize AFTER freeing lnet msgs */
 410         for (i = 0; i < 2; i++) {
 411                 if (lnetmsg[i] == NULL)
 412                         continue;
 413
 414                 lnet_finalize(kranal_data.kra_ni, lnetmsg[i], completion);
 415         }
 416 }
 417
 418 kra_conn_t *
 419 kranal_find_conn_locked (kra_peer_t *peer)
 420 {
 421         cfs_list_t *tmp;
 422
 423         /* just return the first connection */
 424         cfs_list_for_each (tmp, &peer->rap_conns) {
 425                 return cfs_list_entry(tmp, kra_conn_t, rac_list);
 426         }
 427
 428         return NULL;
 429 }
 430
 431 void
 432 kranal_post_fma (kra_conn_t *conn, kra_tx_t *tx)
 433 {
 434         unsigned long    flags;
 435
 436         tx->tx_conn = conn;
 437
 438         spin_lock_irqsave(&conn->rac_lock, flags);
 439         cfs_list_add_tail(&tx->tx_list, &conn->rac_fmaq);
 440         tx->tx_qtime = jiffies;
 441         spin_unlock_irqrestore(&conn->rac_lock, flags);
 442
 443         kranal_schedule_conn(conn);
 444 }
 445
 446 void
 447 kranal_launch_tx (kra_tx_t *tx, lnet_nid_t nid)
 448 {
 449         unsigned long    flags;
 450         kra_peer_t      *peer;
 451         kra_conn_t      *conn;
 452         int              rc;
 453         int              retry;
 454         rwlock_t    *g_lock = &kranal_data.kra_global_lock;
 455
 456         /* If I get here, I've committed to send, so I complete the tx with
 457          * failure on any problems */
 458
 459         LASSERT (tx->tx_conn == NULL);      /* only set when assigned a conn */
 460
 461         for (retry = 0; ; retry = 1) {
 462
 463                 read_lock(g_lock);
 464
 465                 peer = kranal_find_peer_locked(nid);
 466                 if (peer != NULL) {
 467                         conn = kranal_find_conn_locked(peer);
 468                         if (conn != NULL) {
 469                                 kranal_post_fma(conn, tx);
 470                                 read_unlock(g_lock);
 471                                 return;
 472                         }
 473                 }
 474
 475                 /* Making connections; I'll need a write lock... */
 476                 read_unlock(g_lock);
 477                 write_lock_irqsave(g_lock, flags);
 478
 479                 peer = kranal_find_peer_locked(nid);
 480                 if (peer != NULL)
 481                         break;
 482
 483                 write_unlock_irqrestore(g_lock, flags);
 484
 485                 if (retry) {
 486                         CERROR("Can't find peer %s\n", libcfs_nid2str(nid));
 487                         kranal_tx_done(tx, -EHOSTUNREACH);
 488                         return;
 489                 }
 490
 491                 rc = kranal_add_persistent_peer(nid, LNET_NIDADDR(nid),
 492                                                 lnet_acceptor_port());
 493                 if (rc != 0) {
 494                         CERROR("Can't add peer %s: %d\n",
 495                                libcfs_nid2str(nid), rc);
 496                         kranal_tx_done(tx, rc);
 497                         return;
 498                 }
 499         }
 500
 501         conn = kranal_find_conn_locked(peer);
 502         if (conn != NULL) {
 503                 /* Connection exists; queue message on it */
 504                 kranal_post_fma(conn, tx);
 505                 write_unlock_irqrestore(g_lock, flags);
 506                 return;
 507         }
 508
 509         LASSERT (peer->rap_persistence > 0);
 510
 511         if (!peer->rap_connecting) {
 512                 LASSERT (cfs_list_empty(&peer->rap_tx_queue));
 513
 514                 if (!(peer->rap_reconnect_interval == 0 || /* first attempt */
 515                       cfs_time_aftereq(jiffies, peer->rap_reconnect_time))) {
 516                         write_unlock_irqrestore(g_lock, flags);
 517                         kranal_tx_done(tx, -EHOSTUNREACH);
 518                         return;
 519                 }
 520
 521                 peer->rap_connecting = 1;
 522                 kranal_peer_addref(peer); /* extra ref for connd */
 523
 524                 spin_lock(&kranal_data.kra_connd_lock);
 525
 526                 cfs_list_add_tail(&peer->rap_connd_list,
 527                               &kranal_data.kra_connd_peers);
 528                 wake_up(&kranal_data.kra_connd_waitq);
 529
 530                 spin_unlock(&kranal_data.kra_connd_lock);
 531         }
 532
 533         /* A connection is being established; queue the message... */
 534         cfs_list_add_tail(&tx->tx_list, &peer->rap_tx_queue);
 535
 536         write_unlock_irqrestore(g_lock, flags);
 537 }
 538
 539 void
 540 kranal_rdma(kra_tx_t *tx, int type,
 541             kra_rdma_desc_t *sink, int nob, __u64 cookie)
 542 {
 543         kra_conn_t   *conn = tx->tx_conn;
 544         RAP_RETURN    rrc;
 545         unsigned long flags;
 546
 547         LASSERT (kranal_tx_mapped(tx));
 548         LASSERT (nob <= sink->rard_nob);
 549         LASSERT (nob <= tx->tx_nob);
 550
 551         /* No actual race with scheduler sending CLOSE (I'm she!) */
 552         LASSERT (current == conn->rac_device->rad_scheduler);
 553
 554         memset(&tx->tx_rdma_desc, 0, sizeof(tx->tx_rdma_desc));
 555         tx->tx_rdma_desc.SrcPtr.AddressBits = (__u64)((unsigned long)tx->tx_buffer);
 556         tx->tx_rdma_desc.SrcKey = tx->tx_map_key;
 557         tx->tx_rdma_desc.DstPtr = sink->rard_addr;
 558         tx->tx_rdma_desc.DstKey = sink->rard_key;
 559         tx->tx_rdma_desc.Length = nob;
 560         tx->tx_rdma_desc.AppPtr = tx;
 561
 562         /* prep final completion message */
 563         kranal_init_msg(&tx->tx_msg, type);
 564         tx->tx_msg.ram_u.completion.racm_cookie = cookie;
 565
 566         if (nob == 0) { /* Immediate completion */
 567                 kranal_post_fma(conn, tx);
 568                 return;
 569         }
 570
 571         LASSERT (!conn->rac_close_sent); /* Don't lie (CLOSE == RDMA idle) */
 572
 573         rrc = RapkPostRdma(conn->rac_rihandle, &tx->tx_rdma_desc);
 574         LASSERT (rrc == RAP_SUCCESS);
 575
 576         spin_lock_irqsave(&conn->rac_lock, flags);
 577         cfs_list_add_tail(&tx->tx_list, &conn->rac_rdmaq);
 578         tx->tx_qtime = jiffies;
 579         spin_unlock_irqrestore(&conn->rac_lock, flags);
 580 }
 581
 582 int
 583 kranal_consume_rxmsg (kra_conn_t *conn, void *buffer, int nob)
 584 {
 585         __u32      nob_received = nob;
 586         RAP_RETURN rrc;
 587
 588         LASSERT (conn->rac_rxmsg != NULL);
 589         CDEBUG(D_NET, "Consuming %p\n", conn);
 590
 591         rrc = RapkFmaCopyOut(conn->rac_rihandle, buffer,
 592                              &nob_received, sizeof(kra_msg_t));
 593         LASSERT (rrc == RAP_SUCCESS);
 594
 595         conn->rac_rxmsg = NULL;
 596
 597         if (nob_received < nob) {
 598                 CWARN("Incomplete immediate msg from %s: expected %d, got %d\n",
 599                       libcfs_nid2str(conn->rac_peer->rap_nid),
 600                       nob, nob_received);
 601                 return -EPROTO;
 602         }
 603
 604         return 0;
 605 }
 606
 607 int
 608 kranal_send (lnet_ni_t *ni, void *private, lnet_msg_t *lntmsg)
 609 {
 610         lnet_hdr_t       *hdr = &lntmsg->msg_hdr;
 611         int               type = lntmsg->msg_type;
 612         lnet_process_id_t target = lntmsg->msg_target;
 613         int               target_is_router = lntmsg->msg_target_is_router;
 614         int               routing = lntmsg->msg_routing;
 615         unsigned int      niov = lntmsg->msg_niov;
 616         struct iovec     *iov = lntmsg->msg_iov;
 617         lnet_kiov_t      *kiov = lntmsg->msg_kiov;
 618         unsigned int      offset = lntmsg->msg_offset;
 619         unsigned int      nob = lntmsg->msg_len;
 620         kra_tx_t         *tx;
 621         int               rc;
 622
 623         /* NB 'private' is different depending on what we're sending.... */
 624
 625         CDEBUG(D_NET, "sending %d bytes in %d frags to %s\n",
 626                nob, niov, libcfs_id2str(target));
 627
 628         LASSERT (nob == 0 || niov > 0);
 629         LASSERT (niov <= LNET_MAX_IOV);
 630
 631         LASSERT (!in_interrupt());
 632         /* payload is either all vaddrs or all pages */
 633         LASSERT (!(kiov != NULL && iov != NULL));
 634
 635         if (routing) {
 636                 CERROR ("Can't route\n");
 637                 return -EIO;
 638         }
 639
 640         switch(type) {
 641         default:
 642                 LBUG();
 643
 644         case LNET_MSG_ACK:
 645                 LASSERT (nob == 0);
 646                 break;
 647
 648         case LNET_MSG_GET:
 649                 LASSERT (niov == 0);
 650                 LASSERT (nob == 0);
 651                 /* We have to consider the eventual sink buffer rather than any
 652                  * payload passed here (there isn't any, and strictly, looking
 653                  * inside lntmsg is a layering violation).  We send a simple
 654                  * IMMEDIATE GET if the sink buffer is mapped already and small
 655                  * enough for FMA */
 656
 657                 if (routing || target_is_router)
 658                         break;                  /* send IMMEDIATE */
 659
 660                 if ((lntmsg->msg_md->md_options & LNET_MD_KIOV) == 0 &&
 661                     lntmsg->msg_md->md_length <= RANAL_FMA_MAX_DATA &&
 662                     lntmsg->msg_md->md_length <= *kranal_tunables.kra_max_immediate)
 663                         break;                  /* send IMMEDIATE */
 664
 665                 tx = kranal_new_tx_msg(RANAL_MSG_GET_REQ);
 666                 if (tx == NULL)
 667                         return -ENOMEM;
 668
 669                 if ((lntmsg->msg_md->md_options & LNET_MD_KIOV) == 0)
 670                         rc = kranal_setup_virt_buffer(tx, lntmsg->msg_md->md_niov,
 671                                                       lntmsg->msg_md->md_iov.iov,
 672                                                       0, lntmsg->msg_md->md_length);
 673                 else
 674                         rc = kranal_setup_phys_buffer(tx, lntmsg->msg_md->md_niov,
 675                                                       lntmsg->msg_md->md_iov.kiov,
 676                                                       0, lntmsg->msg_md->md_length);
 677                 if (rc != 0) {
 678                         kranal_tx_done(tx, rc);
 679                         return -EIO;
 680                 }
 681
 682                 tx->tx_lntmsg[1] = lnet_create_reply_msg(ni, lntmsg);
 683                 if (tx->tx_lntmsg[1] == NULL) {
 684                         CERROR("Can't create reply for GET to %s\n",
 685                                libcfs_nid2str(target.nid));
 686                         kranal_tx_done(tx, rc);
 687                         return -EIO;
 688                 }
 689
 690                 tx->tx_lntmsg[0] = lntmsg;
 691                 tx->tx_msg.ram_u.get.ragm_hdr = *hdr;
 692                 /* rest of tx_msg is setup just before it is sent */
 693                 kranal_launch_tx(tx, target.nid);
 694                 return 0;
 695
 696         case LNET_MSG_REPLY:
 697         case LNET_MSG_PUT:
 698                 if (kiov == NULL &&             /* not paged */
 699                     nob <= RANAL_FMA_MAX_DATA && /* small enough */
 700                     nob <= *kranal_tunables.kra_max_immediate)
 701                         break;                  /* send IMMEDIATE */
 702
 703                 tx = kranal_new_tx_msg(RANAL_MSG_PUT_REQ);
 704                 if (tx == NULL)
 705                         return -ENOMEM;
 706
 707                 rc = kranal_setup_rdma_buffer(tx, niov, iov, kiov, offset, nob);
 708                 if (rc != 0) {
 709                         kranal_tx_done(tx, rc);
 710                         return -EIO;
 711                 }
 712
 713                 tx->tx_lntmsg[0] = lntmsg;
 714                 tx->tx_msg.ram_u.putreq.raprm_hdr = *hdr;
 715                 /* rest of tx_msg is setup just before it is sent */
 716                 kranal_launch_tx(tx, target.nid);
 717                 return 0;
 718         }
 719
 720         /* send IMMEDIATE */
 721
 722         LASSERT (kiov == NULL);
 723         LASSERT (nob <= RANAL_FMA_MAX_DATA);
 724
 725         tx = kranal_new_tx_msg(RANAL_MSG_IMMEDIATE);
 726         if (tx == NULL)
 727                 return -ENOMEM;
 728
 729         rc = kranal_setup_immediate_buffer(tx, niov, iov, offset, nob);
 730         if (rc != 0) {
 731                 kranal_tx_done(tx, rc);
 732                 return -EIO;
 733         }
 734
 735         tx->tx_msg.ram_u.immediate.raim_hdr = *hdr;
 736         tx->tx_lntmsg[0] = lntmsg;
 737         kranal_launch_tx(tx, target.nid);
 738         return 0;
 739 }
 740
 741 void
 742 kranal_reply(lnet_ni_t *ni, kra_conn_t *conn, lnet_msg_t *lntmsg)
 743 {
 744         kra_msg_t     *rxmsg = conn->rac_rxmsg;
 745         unsigned int   niov = lntmsg->msg_niov;
 746         struct iovec  *iov = lntmsg->msg_iov;
 747         lnet_kiov_t   *kiov = lntmsg->msg_kiov;
 748         unsigned int   offset = lntmsg->msg_offset;
 749         unsigned int   nob = lntmsg->msg_len;
 750         kra_tx_t      *tx;
 751         int            rc;
 752
 753         tx = kranal_get_idle_tx();
 754         if (tx == NULL)
 755                 goto failed_0;
 756
 757         rc = kranal_setup_rdma_buffer(tx, niov, iov, kiov, offset, nob);
 758         if (rc != 0)
 759                 goto failed_1;
 760
 761         tx->tx_conn = conn;
 762
 763         rc = kranal_map_buffer(tx);
 764         if (rc != 0)
 765                 goto failed_1;
 766
 767         tx->tx_lntmsg[0] = lntmsg;
 768
 769         kranal_rdma(tx, RANAL_MSG_GET_DONE,
 770                     &rxmsg->ram_u.get.ragm_desc, nob,
 771                     rxmsg->ram_u.get.ragm_cookie);
 772         return;
 773
 774  failed_1:
 775         kranal_tx_done(tx, -EIO);
 776  failed_0:
 777         lnet_finalize(ni, lntmsg, -EIO);
 778 }
 779
 780 int
 781 kranal_eager_recv (lnet_ni_t *ni, void *private, lnet_msg_t *lntmsg,
 782                    void **new_private)
 783 {
 784         kra_conn_t *conn = (kra_conn_t *)private;
 785
 786         LCONSOLE_ERROR_MSG(0x12b, "Dropping message from %s: no buffers free.\n",
 787                            libcfs_nid2str(conn->rac_peer->rap_nid));
 788
 789         return -EDEADLK;
 790 }
 791
 792 int
 793 kranal_recv (lnet_ni_t *ni, void *private, lnet_msg_t *lntmsg,
 794              int delayed, unsigned int niov,
 795              struct iovec *iov, lnet_kiov_t *kiov,
 796              unsigned int offset, unsigned int mlen, unsigned int rlen)
 797 {
 798         kra_conn_t  *conn = private;
 799         kra_msg_t   *rxmsg = conn->rac_rxmsg;
 800         kra_tx_t    *tx;
 801         void        *buffer;
 802         int          rc;
 803
 804         LASSERT (mlen <= rlen);
 805         LASSERT (!in_interrupt());
 806         /* Either all pages or all vaddrs */
 807         LASSERT (!(kiov != NULL && iov != NULL));
 808
 809         CDEBUG(D_NET, "conn %p, rxmsg %p, lntmsg %p\n", conn, rxmsg, lntmsg);
 810
 811         switch(rxmsg->ram_type) {
 812         default:
 813                 LBUG();
 814
 815         case RANAL_MSG_IMMEDIATE:
 816                 if (mlen == 0) {
 817                         buffer = NULL;
 818                 } else if (kiov != NULL) {
 819                         CERROR("Can't recv immediate into paged buffer\n");
 820                         return -EIO;
 821                 } else {
 822                         LASSERT (niov > 0);
 823                         while (offset >= iov->iov_len) {
 824                                 offset -= iov->iov_len;
 825                                 iov++;
 826                                 niov--;
 827                                 LASSERT (niov > 0);
 828                         }
 829                         if (mlen > iov->iov_len - offset) {
 830                                 CERROR("Can't handle immediate frags\n");
 831                                 return -EIO;
 832                         }
 833                         buffer = ((char *)iov->iov_base) + offset;
 834                 }
 835                 rc = kranal_consume_rxmsg(conn, buffer, mlen);
 836                 lnet_finalize(ni, lntmsg, (rc == 0) ? 0 : -EIO);
 837                 return 0;
 838
 839         case RANAL_MSG_PUT_REQ:
 840                 tx = kranal_new_tx_msg(RANAL_MSG_PUT_ACK);
 841                 if (tx == NULL) {
 842                         kranal_consume_rxmsg(conn, NULL, 0);
 843                         return -ENOMEM;
 844                 }
 845
 846                 rc = kranal_setup_rdma_buffer(tx, niov, iov, kiov, offset, mlen);
 847                 if (rc != 0) {
 848                         kranal_tx_done(tx, rc);
 849                         kranal_consume_rxmsg(conn, NULL, 0);
 850                         return -EIO;
 851                 }
 852
 853                 tx->tx_conn = conn;
 854                 rc = kranal_map_buffer(tx);
 855                 if (rc != 0) {
 856                         kranal_tx_done(tx, rc);
 857                         kranal_consume_rxmsg(conn, NULL, 0);
 858                         return -EIO;
 859                 }
 860
 861                 tx->tx_msg.ram_u.putack.rapam_src_cookie =
 862                         conn->rac_rxmsg->ram_u.putreq.raprm_cookie;
 863                 tx->tx_msg.ram_u.putack.rapam_dst_cookie = tx->tx_cookie;
 864                 tx->tx_msg.ram_u.putack.rapam_desc.rard_key = tx->tx_map_key;
 865                 tx->tx_msg.ram_u.putack.rapam_desc.rard_addr.AddressBits =
 866                         (__u64)((unsigned long)tx->tx_buffer);
 867                 tx->tx_msg.ram_u.putack.rapam_desc.rard_nob = mlen;
 868
 869                 tx->tx_lntmsg[0] = lntmsg; /* finalize this on RDMA_DONE */
 870
 871                 kranal_post_fma(conn, tx);
 872                 kranal_consume_rxmsg(conn, NULL, 0);
 873                 return 0;
 874
 875         case RANAL_MSG_GET_REQ:
 876                 if (lntmsg != NULL) {
 877                         /* Matched! */
 878                         kranal_reply(ni, conn, lntmsg);
 879                 } else {
 880                         /* No match */
 881                         tx = kranal_new_tx_msg(RANAL_MSG_GET_NAK);
 882                         if (tx != NULL) {
 883                                 tx->tx_msg.ram_u.completion.racm_cookie =
 884                                         rxmsg->ram_u.get.ragm_cookie;
 885                                 kranal_post_fma(conn, tx);
 886                         }
 887                 }
 888                 kranal_consume_rxmsg(conn, NULL, 0);
 889                 return 0;
 890         }
 891 }
 892
 893 int
 894 kranal_thread_start(int(*fn)(void *arg), void *arg, char *name)
 895 {
 896         struct task_struct *task = cfs_thread_run(fn, arg, name);
 897
 898         if (!IS_ERR(task))
 899                 cfs_atomic_inc(&kranal_data.kra_nthreads);
 900         return PTR_ERR(task);
 901 }
 902
 903 void
 904 kranal_thread_fini (void)
 905 {
 906         cfs_atomic_dec(&kranal_data.kra_nthreads);
 907 }
 908
 909 int
 910 kranal_check_conn_timeouts (kra_conn_t *conn)
 911 {
 912         kra_tx_t          *tx;
 913         cfs_list_t        *ttmp;
 914         unsigned long      flags;
 915         long               timeout;
 916         unsigned long      now = jiffies;
 917
 918         LASSERT (conn->rac_state == RANAL_CONN_ESTABLISHED ||
 919                  conn->rac_state == RANAL_CONN_CLOSING);
 920
 921         if (!conn->rac_close_sent &&
 922             cfs_time_aftereq(now, conn->rac_last_tx + conn->rac_keepalive *
 923                              HZ)) {
 924                 /* not sent in a while; schedule conn so scheduler sends a keepalive */
 925                 CDEBUG(D_NET, "Scheduling keepalive %p->%s\n",
 926                        conn, libcfs_nid2str(conn->rac_peer->rap_nid));
 927                 kranal_schedule_conn(conn);
 928         }
 929
 930         timeout = conn->rac_timeout * HZ;
 931
 932         if (!conn->rac_close_recvd &&
 933             cfs_time_aftereq(now, conn->rac_last_rx + timeout)) {
 934                 CERROR("%s received from %s within %lu seconds\n",
 935                        (conn->rac_state == RANAL_CONN_ESTABLISHED) ?
 936                        "Nothing" : "CLOSE not",
 937                        libcfs_nid2str(conn->rac_peer->rap_nid),
 938                        (now - conn->rac_last_rx)/HZ);
 939                 return -ETIMEDOUT;
 940         }
 941
 942         if (conn->rac_state != RANAL_CONN_ESTABLISHED)
 943                 return 0;
 944
 945         /* Check the conn's queues are moving.  These are "belt+braces" checks,
 946          * in case of hardware/software errors that make this conn seem
 947          * responsive even though it isn't progressing its message queues. */
 948
 949         spin_lock_irqsave(&conn->rac_lock, flags);
 950
 951         cfs_list_for_each (ttmp, &conn->rac_fmaq) {
 952                 tx = cfs_list_entry(ttmp, kra_tx_t, tx_list);
 953
 954                 if (cfs_time_aftereq(now, tx->tx_qtime + timeout)) {
 955                         spin_unlock_irqrestore(&conn->rac_lock, flags);
 956                         CERROR("tx on fmaq for %s blocked %lu seconds\n",
 957                                libcfs_nid2str(conn->rac_peer->rap_nid),
 958                                (now - tx->tx_qtime)/HZ);
 959                         return -ETIMEDOUT;
 960                 }
 961         }
 962
 963         cfs_list_for_each (ttmp, &conn->rac_rdmaq) {
 964                 tx = cfs_list_entry(ttmp, kra_tx_t, tx_list);
 965
 966                 if (cfs_time_aftereq(now, tx->tx_qtime + timeout)) {
 967                         spin_unlock_irqrestore(&conn->rac_lock, flags);
 968                         CERROR("tx on rdmaq for %s blocked %lu seconds\n",
 969                                libcfs_nid2str(conn->rac_peer->rap_nid),
 970                                (now - tx->tx_qtime)/HZ);
 971                         return -ETIMEDOUT;
 972                 }
 973         }
 974
 975         cfs_list_for_each (ttmp, &conn->rac_replyq) {
 976                 tx = cfs_list_entry(ttmp, kra_tx_t, tx_list);
 977
 978                 if (cfs_time_aftereq(now, tx->tx_qtime + timeout)) {
 979                         spin_unlock_irqrestore(&conn->rac_lock, flags);
 980                         CERROR("tx on replyq for %s blocked %lu seconds\n",
 981                                libcfs_nid2str(conn->rac_peer->rap_nid),
 982                                (now - tx->tx_qtime)/HZ);
 983                         return -ETIMEDOUT;
 984                 }
 985         }
 986
 987         spin_unlock_irqrestore(&conn->rac_lock, flags);
 988         return 0;
 989 }
 990
 991 void
 992 kranal_reaper_check (int idx, unsigned long *min_timeoutp)
 993 {
 994         cfs_list_t        *conns = &kranal_data.kra_conns[idx];
 995         cfs_list_t        *ctmp;
 996         kra_conn_t        *conn;
 997         unsigned long      flags;
 998         int                rc;
 999
1000  again:
1001         /* NB. We expect to check all the conns and not find any problems, so
1002          * we just use a shared lock while we take a look... */
1003         read_lock(&kranal_data.kra_global_lock);
1004
1005         cfs_list_for_each (ctmp, conns) {
1006                 conn = cfs_list_entry(ctmp, kra_conn_t, rac_hashlist);
1007
1008                 if (conn->rac_timeout < *min_timeoutp )
1009                         *min_timeoutp = conn->rac_timeout;
1010                 if (conn->rac_keepalive < *min_timeoutp )
1011                         *min_timeoutp = conn->rac_keepalive;
1012
1013                 rc = kranal_check_conn_timeouts(conn);
1014                 if (rc == 0)
1015                         continue;
1016
1017                 kranal_conn_addref(conn);
1018                 read_unlock(&kranal_data.kra_global_lock);
1019
1020                 CERROR("Conn to %s, cqid %d timed out\n",
1021                        libcfs_nid2str(conn->rac_peer->rap_nid),
1022                        conn->rac_cqid);
1023
1024                 write_lock_irqsave(&kranal_data.kra_global_lock, flags);
1025
1026                 switch (conn->rac_state) {
1027                 default:
1028                         LBUG();
1029
1030                 case RANAL_CONN_ESTABLISHED:
1031                         kranal_close_conn_locked(conn, -ETIMEDOUT);
1032                         break;
1033
1034                 case RANAL_CONN_CLOSING:
1035                         kranal_terminate_conn_locked(conn);
1036                         break;
1037                 }
1038
1039                 write_unlock_irqrestore(&kranal_data.kra_global_lock,
1040                                             flags);
1041
1042                 kranal_conn_decref(conn);
1043
1044                 /* start again now I've dropped the lock */
1045                 goto again;
1046         }
1047
1048         read_unlock(&kranal_data.kra_global_lock);
1049 }
1050
1051 int
1052 kranal_connd (void *arg)
1053 {
1054         long               id = (long)arg;
1055         wait_queue_t     wait;
1056         unsigned long      flags;
1057         kra_peer_t        *peer;
1058         kra_acceptsock_t  *ras;
1059         int                did_something;
1060
1061         cfs_block_allsigs();
1062
1063         init_waitqueue_entry_current(&wait);
1064
1065         spin_lock_irqsave(&kranal_data.kra_connd_lock, flags);
1066
1067         while (!kranal_data.kra_shutdown) {
1068                 did_something = 0;
1069
1070                 if (!cfs_list_empty(&kranal_data.kra_connd_acceptq)) {
1071                         ras = cfs_list_entry(kranal_data.kra_connd_acceptq.next,
1072                                              kra_acceptsock_t, ras_list);
1073                         cfs_list_del(&ras->ras_list);
1074
1075                         spin_unlock_irqrestore(&kranal_data.kra_connd_lock,
1076                                                    flags);
1077
1078                         CDEBUG(D_NET,"About to handshake someone\n");
1079
1080                         kranal_conn_handshake(ras->ras_sock, NULL);
1081                         kranal_free_acceptsock(ras);
1082
1083                         CDEBUG(D_NET,"Finished handshaking someone\n");
1084
1085                         spin_lock_irqsave(&kranal_data.kra_connd_lock,
1086                                               flags);
1087                         did_something = 1;
1088                 }
1089
1090                 if (!cfs_list_empty(&kranal_data.kra_connd_peers)) {
1091                         peer = cfs_list_entry(kranal_data.kra_connd_peers.next,
1092                                               kra_peer_t, rap_connd_list);
1093
1094                         cfs_list_del_init(&peer->rap_connd_list);
1095                         spin_unlock_irqrestore(&kranal_data.kra_connd_lock,
1096                                                    flags);
1097
1098                         kranal_connect(peer);
1099                         kranal_peer_decref(peer);
1100
1101                         spin_lock_irqsave(&kranal_data.kra_connd_lock,
1102                                               flags);
1103                         did_something = 1;
1104                 }
1105
1106                 if (did_something)
1107                         continue;
1108
1109                 set_current_state(TASK_INTERRUPTIBLE);
1110                 add_wait_queue_exclusive(&kranal_data.kra_connd_waitq, &wait);
1111
1112                 spin_unlock_irqrestore(&kranal_data.kra_connd_lock, flags);
1113
1114                 waitq_wait(&wait, TASK_INTERRUPTIBLE);
1115
1116                 set_current_state(TASK_RUNNING);
1117                 remove_wait_queue(&kranal_data.kra_connd_waitq, &wait);
1118
1119                 spin_lock_irqsave(&kranal_data.kra_connd_lock, flags);
1120         }
1121
1122         spin_unlock_irqrestore(&kranal_data.kra_connd_lock, flags);
1123
1124         kranal_thread_fini();
1125         return 0;
1126 }
1127
1128 void
1129 kranal_update_reaper_timeout(long timeout)
1130 {
1131         unsigned long   flags;
1132
1133         LASSERT (timeout > 0);
1134
1135         spin_lock_irqsave(&kranal_data.kra_reaper_lock, flags);
1136
1137         if (timeout < kranal_data.kra_new_min_timeout)
1138                 kranal_data.kra_new_min_timeout = timeout;
1139
1140         spin_unlock_irqrestore(&kranal_data.kra_reaper_lock, flags);
1141 }
1142
1143 int
1144 kranal_reaper (void *arg)
1145 {
1146         wait_queue_t     wait;
1147         unsigned long      flags;
1148         long               timeout;
1149         int                i;
1150         int                conn_entries = kranal_data.kra_conn_hash_size;
1151         int                conn_index = 0;
1152         int                base_index = conn_entries - 1;
1153         unsigned long      next_check_time = jiffies;
1154         long               next_min_timeout = MAX_SCHEDULE_TIMEOUT;
1155         long               current_min_timeout = 1;
1156
1157         cfs_block_allsigs();
1158
1159         init_waitqueue_entry_current(&wait);
1160
1161         spin_lock_irqsave(&kranal_data.kra_reaper_lock, flags);
1162
1163         while (!kranal_data.kra_shutdown) {
1164                 /* I wake up every 'p' seconds to check for timeouts on some
1165                  * more peers.  I try to check every connection 'n' times
1166                  * within the global minimum of all keepalive and timeout
1167                  * intervals, to ensure I attend to every connection within
1168                  * (n+1)/n times its timeout intervals. */
1169                 const int     p = 1;
1170                 const int     n = 3;
1171                 unsigned long min_timeout;
1172                 int           chunk;
1173
1174                 /* careful with the jiffy wrap... */
1175                 timeout = (long)(next_check_time - jiffies);
1176                 if (timeout > 0) {
1177                         set_current_state(TASK_INTERRUPTIBLE);
1178                         add_wait_queue(&kranal_data.kra_reaper_waitq, &wait);
1179
1180                         spin_unlock_irqrestore(&kranal_data.kra_reaper_lock,
1181                                                    flags);
1182
1183                         waitq_timedwait(&wait, TASK_INTERRUPTIBLE,
1184                                             timeout);
1185
1186                         spin_lock_irqsave(&kranal_data.kra_reaper_lock,
1187                                               flags);
1188
1189                         set_current_state(TASK_RUNNING);
1190                         remove_wait_queue(&kranal_data.kra_reaper_waitq, &wait);
1191                         continue;
1192                 }
1193
1194                 if (kranal_data.kra_new_min_timeout !=
1195                     MAX_SCHEDULE_TIMEOUT) {
1196                         /* new min timeout set: restart min timeout scan */
1197                         next_min_timeout = MAX_SCHEDULE_TIMEOUT;
1198                         base_index = conn_index - 1;
1199                         if (base_index < 0)
1200                                 base_index = conn_entries - 1;
1201
1202                         if (kranal_data.kra_new_min_timeout <
1203                             current_min_timeout) {
1204                                 current_min_timeout =
1205                                         kranal_data.kra_new_min_timeout;
1206                                 CDEBUG(D_NET, "Set new min timeout %ld\n",
1207                                        current_min_timeout);
1208                         }
1209
1210                         kranal_data.kra_new_min_timeout =
1211                                 MAX_SCHEDULE_TIMEOUT;
1212                 }
1213                 min_timeout = current_min_timeout;
1214
1215                 spin_unlock_irqrestore(&kranal_data.kra_reaper_lock, flags);
1216
1217                 LASSERT (min_timeout > 0);
1218
1219                 /* Compute how many table entries to check now so I get round
1220                  * the whole table fast enough given that I do this at fixed
1221                  * intervals of 'p' seconds) */
1222                 chunk = conn_entries;
1223                 if (min_timeout > n * p)
1224                         chunk = (chunk * n * p) / min_timeout;
1225                 if (chunk == 0)
1226                         chunk = 1;
1227
1228                 for (i = 0; i < chunk; i++) {
1229                         kranal_reaper_check(conn_index,
1230                                             &next_min_timeout);
1231                         conn_index = (conn_index + 1) % conn_entries;
1232                 }
1233
1234                 next_check_time += p * HZ;
1235
1236                 spin_lock_irqsave(&kranal_data.kra_reaper_lock, flags);
1237
1238                 if (((conn_index - chunk <= base_index &&
1239                       base_index < conn_index) ||
1240                      (conn_index - conn_entries - chunk <= base_index &&
1241                       base_index < conn_index - conn_entries))) {
1242
1243                         /* Scanned all conns: set current_min_timeout... */
1244                         if (current_min_timeout != next_min_timeout) {
1245                                 current_min_timeout = next_min_timeout;
1246                                 CDEBUG(D_NET, "Set new min timeout %ld\n",
1247                                        current_min_timeout);
1248                         }
1249
1250                         /* ...and restart min timeout scan */
1251                         next_min_timeout = MAX_SCHEDULE_TIMEOUT;
1252                         base_index = conn_index - 1;
1253                         if (base_index < 0)
1254                                 base_index = conn_entries - 1;
1255                 }
1256         }
1257
1258         kranal_thread_fini();
1259         return 0;
1260 }
1261
1262 void
1263 kranal_check_rdma_cq (kra_device_t *dev)
1264 {
1265         kra_conn_t          *conn;
1266         kra_tx_t            *tx;
1267         RAP_RETURN           rrc;
1268         unsigned long        flags;
1269         RAP_RDMA_DESCRIPTOR *desc;
1270         __u32                cqid;
1271         __u32                event_type;
1272
1273         for (;;) {
1274                 rrc = RapkCQDone(dev->rad_rdma_cqh, &cqid, &event_type);
1275                 if (rrc == RAP_NOT_DONE) {
1276                         CDEBUG(D_NET, "RDMA CQ %d empty\n", dev->rad_id);
1277                         return;
1278                 }
1279
1280                 LASSERT (rrc == RAP_SUCCESS);
1281                 LASSERT ((event_type & RAPK_CQ_EVENT_OVERRUN) == 0);
1282
1283                 read_lock(&kranal_data.kra_global_lock);
1284
1285                 conn = kranal_cqid2conn_locked(cqid);
1286                 if (conn == NULL) {
1287                         /* Conn was destroyed? */
1288                         CDEBUG(D_NET, "RDMA CQID lookup %d failed\n", cqid);
1289                         read_unlock(&kranal_data.kra_global_lock);
1290                         continue;
1291                 }
1292
1293                 rrc = RapkRdmaDone(conn->rac_rihandle, &desc);
1294                 LASSERT (rrc == RAP_SUCCESS);
1295
1296                 CDEBUG(D_NET, "Completed %p\n",
1297                        cfs_list_entry(conn->rac_rdmaq.next, kra_tx_t, tx_list));
1298
1299                 spin_lock_irqsave(&conn->rac_lock, flags);
1300
1301                 LASSERT (!cfs_list_empty(&conn->rac_rdmaq));
1302                 tx = cfs_list_entry(conn->rac_rdmaq.next, kra_tx_t, tx_list);
1303                 cfs_list_del(&tx->tx_list);
1304
1305                 LASSERT(desc->AppPtr == (void *)tx);
1306                 LASSERT(tx->tx_msg.ram_type == RANAL_MSG_PUT_DONE ||
1307                         tx->tx_msg.ram_type == RANAL_MSG_GET_DONE);
1308
1309                 cfs_list_add_tail(&tx->tx_list, &conn->rac_fmaq);
1310                 tx->tx_qtime = jiffies;
1311
1312                 spin_unlock_irqrestore(&conn->rac_lock, flags);
1313
1314                 /* Get conn's fmaq processed, now I've just put something
1315                  * there */
1316                 kranal_schedule_conn(conn);
1317
1318                 read_unlock(&kranal_data.kra_global_lock);
1319         }
1320 }
1321
1322 void
1323 kranal_check_fma_cq (kra_device_t *dev)
1324 {
1325         kra_conn_t         *conn;
1326         RAP_RETURN          rrc;
1327         __u32               cqid;
1328         __u32               event_type;
1329         cfs_list_t         *conns;
1330         cfs_list_t         *tmp;
1331         int                 i;
1332
1333         for (;;) {
1334                 rrc = RapkCQDone(dev->rad_fma_cqh, &cqid, &event_type);
1335                 if (rrc == RAP_NOT_DONE) {
1336                         CDEBUG(D_NET, "FMA CQ %d empty\n", dev->rad_id);
1337                         return;
1338                 }
1339
1340                 LASSERT (rrc == RAP_SUCCESS);
1341
1342                 if ((event_type & RAPK_CQ_EVENT_OVERRUN) == 0) {
1343
1344                         read_lock(&kranal_data.kra_global_lock);
1345
1346                         conn = kranal_cqid2conn_locked(cqid);
1347                         if (conn == NULL) {
1348                                 CDEBUG(D_NET, "FMA CQID lookup %d failed\n",
1349                                        cqid);
1350                         } else {
1351                                 CDEBUG(D_NET, "FMA completed: %p CQID %d\n",
1352                                        conn, cqid);
1353                                 kranal_schedule_conn(conn);
1354                         }
1355
1356                         read_unlock(&kranal_data.kra_global_lock);
1357                         continue;
1358                 }
1359
1360                 /* FMA CQ has overflowed: check ALL conns */
1361                 CWARN("FMA CQ overflow: scheduling ALL conns on device %d\n",
1362                       dev->rad_id);
1363
1364                 for (i = 0; i < kranal_data.kra_conn_hash_size; i++) {
1365
1366                         read_lock(&kranal_data.kra_global_lock);
1367
1368                         conns = &kranal_data.kra_conns[i];
1369
1370                         cfs_list_for_each (tmp, conns) {
1371                                 conn = cfs_list_entry(tmp, kra_conn_t,
1372                                                       rac_hashlist);
1373
1374                                 if (conn->rac_device == dev)
1375                                         kranal_schedule_conn(conn);
1376                         }
1377
1378                         /* don't block write lockers for too long... */
1379                         read_unlock(&kranal_data.kra_global_lock);
1380                 }
1381         }
1382 }
1383
1384 int
1385 kranal_sendmsg(kra_conn_t *conn, kra_msg_t *msg,
1386                void *immediate, int immediatenob)
1387 {
1388         int        sync = (msg->ram_type & RANAL_MSG_FENCE) != 0;
1389         RAP_RETURN rrc;
1390
1391         CDEBUG(D_NET,"%p sending msg %p %02x%s [%p for %d]\n",
1392                conn, msg, msg->ram_type, sync ? "(sync)" : "",
1393                immediate, immediatenob);
1394
1395         LASSERT (sizeof(*msg) <= RANAL_FMA_MAX_PREFIX);
1396         LASSERT ((msg->ram_type == RANAL_MSG_IMMEDIATE) ?
1397                  immediatenob <= RANAL_FMA_MAX_DATA :
1398                  immediatenob == 0);
1399
1400         msg->ram_connstamp = conn->rac_my_connstamp;
1401         msg->ram_seq = conn->rac_tx_seq;
1402
1403         if (sync)
1404                 rrc = RapkFmaSyncSend(conn->rac_rihandle,
1405                                       immediate, immediatenob,
1406                                       msg, sizeof(*msg));
1407         else
1408                 rrc = RapkFmaSend(conn->rac_rihandle,
1409                                   immediate, immediatenob,
1410                                   msg, sizeof(*msg));
1411
1412         switch (rrc) {
1413         default:
1414                 LBUG();
1415
1416         case RAP_SUCCESS:
1417                 conn->rac_last_tx = jiffies;
1418                 conn->rac_tx_seq++;
1419                 return 0;
1420
1421         case RAP_NOT_DONE:
1422                 if (cfs_time_aftereq(jiffies,
1423                                      conn->rac_last_tx + conn->rac_keepalive *
1424                                      HZ))
1425                         CWARN("EAGAIN sending %02x (idle %lu secs)\n",
1426                               msg->ram_type,
1427                               (jiffies - conn->rac_last_tx)/HZ);
1428                 return -EAGAIN;
1429         }
1430 }
1431
1432 void
1433 kranal_process_fmaq (kra_conn_t *conn)
1434 {
1435         unsigned long flags;
1436         int           more_to_do;
1437         kra_tx_t     *tx;
1438         int           rc;
1439         int           expect_reply;
1440
1441         /* NB 1. kranal_sendmsg() may fail if I'm out of credits right now.
1442          *       However I will be rescheduled by an FMA completion event
1443          *       when I eventually get some.
1444          * NB 2. Sampling rac_state here races with setting it elsewhere.
1445          *       But it doesn't matter if I try to send a "real" message just
1446          *       as I start closing because I'll get scheduled to send the
1447          *       close anyway. */
1448
1449         /* Not racing with incoming message processing! */
1450         LASSERT (current == conn->rac_device->rad_scheduler);
1451
1452         if (conn->rac_state != RANAL_CONN_ESTABLISHED) {
1453                 if (!cfs_list_empty(&conn->rac_rdmaq)) {
1454                         /* RDMAs in progress */
1455                         LASSERT (!conn->rac_close_sent);
1456
1457                         if (cfs_time_aftereq(jiffies,
1458                                              conn->rac_last_tx +
1459                                              conn->rac_keepalive * HZ)) {
1460                                 CDEBUG(D_NET, "sending NOOP (rdma in progress)\n");
1461                                 kranal_init_msg(&conn->rac_msg, RANAL_MSG_NOOP);
1462                                 kranal_sendmsg(conn, &conn->rac_msg, NULL, 0);
1463                         }
1464                         return;
1465                 }
1466
1467                 if (conn->rac_close_sent)
1468                         return;
1469
1470                 CWARN("sending CLOSE to %s\n",
1471                       libcfs_nid2str(conn->rac_peer->rap_nid));
1472                 kranal_init_msg(&conn->rac_msg, RANAL_MSG_CLOSE);
1473                 rc = kranal_sendmsg(conn, &conn->rac_msg, NULL, 0);
1474                 if (rc != 0)
1475                         return;
1476
1477                 conn->rac_close_sent = 1;
1478                 if (!conn->rac_close_recvd)
1479                         return;
1480
1481                 write_lock_irqsave(&kranal_data.kra_global_lock, flags);
1482
1483                 if (conn->rac_state == RANAL_CONN_CLOSING)
1484                         kranal_terminate_conn_locked(conn);
1485
1486                 write_unlock_irqrestore(&kranal_data.kra_global_lock,
1487                                             flags);
1488                 return;
1489         }
1490
1491         spin_lock_irqsave(&conn->rac_lock, flags);
1492
1493         if (cfs_list_empty(&conn->rac_fmaq)) {
1494
1495                 spin_unlock_irqrestore(&conn->rac_lock, flags);
1496
1497                 if (cfs_time_aftereq(jiffies,
1498                                      conn->rac_last_tx + conn->rac_keepalive *
1499                                      HZ)) {
1500                         CDEBUG(D_NET, "sending NOOP -> %s (%p idle %lu(%ld))\n",
1501                                libcfs_nid2str(conn->rac_peer->rap_nid), conn,
1502                                (jiffies - conn->rac_last_tx)/HZ,
1503                                conn->rac_keepalive);
1504                         kranal_init_msg(&conn->rac_msg, RANAL_MSG_NOOP);
1505                         kranal_sendmsg(conn, &conn->rac_msg, NULL, 0);
1506                 }
1507                 return;
1508         }
1509
1510         tx = cfs_list_entry(conn->rac_fmaq.next, kra_tx_t, tx_list);
1511         cfs_list_del(&tx->tx_list);
1512         more_to_do = !cfs_list_empty(&conn->rac_fmaq);
1513
1514         spin_unlock_irqrestore(&conn->rac_lock, flags);
1515
1516         expect_reply = 0;
1517         CDEBUG(D_NET, "sending regular msg: %p, type %02x, cookie "LPX64"\n",
1518                tx, tx->tx_msg.ram_type, tx->tx_cookie);
1519         switch (tx->tx_msg.ram_type) {
1520         default:
1521                 LBUG();
1522
1523         case RANAL_MSG_IMMEDIATE:
1524                 rc = kranal_sendmsg(conn, &tx->tx_msg,
1525                                     tx->tx_buffer, tx->tx_nob);
1526                 break;
1527
1528         case RANAL_MSG_PUT_NAK:
1529         case RANAL_MSG_PUT_DONE:
1530         case RANAL_MSG_GET_NAK:
1531         case RANAL_MSG_GET_DONE:
1532                 rc = kranal_sendmsg(conn, &tx->tx_msg, NULL, 0);
1533                 break;
1534
1535         case RANAL_MSG_PUT_REQ:
1536                 rc = kranal_map_buffer(tx);
1537                 LASSERT (rc != -EAGAIN);
1538                 if (rc != 0)
1539                         break;
1540
1541                 tx->tx_msg.ram_u.putreq.raprm_cookie = tx->tx_cookie;
1542                 rc = kranal_sendmsg(conn, &tx->tx_msg, NULL, 0);
1543                 expect_reply = 1;
1544                 break;
1545
1546         case RANAL_MSG_PUT_ACK:
1547                 rc = kranal_sendmsg(conn, &tx->tx_msg, NULL, 0);
1548                 expect_reply = 1;
1549                 break;
1550
1551         case RANAL_MSG_GET_REQ:
1552                 rc = kranal_map_buffer(tx);
1553                 LASSERT (rc != -EAGAIN);
1554                 if (rc != 0)
1555                         break;
1556
1557                 tx->tx_msg.ram_u.get.ragm_cookie = tx->tx_cookie;
1558                 tx->tx_msg.ram_u.get.ragm_desc.rard_key = tx->tx_map_key;
1559                 tx->tx_msg.ram_u.get.ragm_desc.rard_addr.AddressBits =
1560                         (__u64)((unsigned long)tx->tx_buffer);
1561                 tx->tx_msg.ram_u.get.ragm_desc.rard_nob = tx->tx_nob;
1562                 rc = kranal_sendmsg(conn, &tx->tx_msg, NULL, 0);
1563                 expect_reply = 1;
1564                 break;
1565         }
1566
1567         if (rc == -EAGAIN) {
1568                 /* I need credits to send this.  Replace tx at the head of the
1569                  * fmaq and I'll get rescheduled when credits appear */
1570                 CDEBUG(D_NET, "EAGAIN on %p\n", conn);
1571                 spin_lock_irqsave(&conn->rac_lock, flags);
1572                 cfs_list_add(&tx->tx_list, &conn->rac_fmaq);
1573                 spin_unlock_irqrestore(&conn->rac_lock, flags);
1574                 return;
1575         }
1576
1577         if (!expect_reply || rc != 0) {
1578                 kranal_tx_done(tx, rc);
1579         } else {
1580                 /* LASSERT(current) above ensures this doesn't race with reply
1581                  * processing */
1582                 spin_lock_irqsave(&conn->rac_lock, flags);
1583                 cfs_list_add_tail(&tx->tx_list, &conn->rac_replyq);
1584                 tx->tx_qtime = jiffies;
1585                 spin_unlock_irqrestore(&conn->rac_lock, flags);
1586         }
1587
1588         if (more_to_do) {
1589                 CDEBUG(D_NET, "Rescheduling %p (more to do)\n", conn);
1590                 kranal_schedule_conn(conn);
1591         }
1592 }
1593
1594 static inline void
1595 kranal_swab_rdma_desc (kra_rdma_desc_t *d)
1596 {
1597         __swab64s(&d->rard_key.Key);
1598         __swab16s(&d->rard_key.Cookie);
1599         __swab16s(&d->rard_key.MdHandle);
1600         __swab32s(&d->rard_key.Flags);
1601         __swab64s(&d->rard_addr.AddressBits);
1602         __swab32s(&d->rard_nob);
1603 }
1604
1605 kra_tx_t *
1606 kranal_match_reply(kra_conn_t *conn, int type, __u64 cookie)
1607 {
1608         cfs_list_t       *ttmp;
1609         kra_tx_t         *tx;
1610         unsigned long     flags;
1611
1612         spin_lock_irqsave(&conn->rac_lock, flags);
1613
1614         cfs_list_for_each(ttmp, &conn->rac_replyq) {
1615                 tx = cfs_list_entry(ttmp, kra_tx_t, tx_list);
1616
1617                 CDEBUG(D_NET,"Checking %p %02x/"LPX64"\n",
1618                        tx, tx->tx_msg.ram_type, tx->tx_cookie);
1619
1620                 if (tx->tx_cookie != cookie)
1621                         continue;
1622
1623                 if (tx->tx_msg.ram_type != type) {
1624                         spin_unlock_irqrestore(&conn->rac_lock, flags);
1625                         CWARN("Unexpected type %x (%x expected) "
1626                               "matched reply from %s\n",
1627                               tx->tx_msg.ram_type, type,
1628                               libcfs_nid2str(conn->rac_peer->rap_nid));
1629                         return NULL;
1630                 }
1631
1632                 cfs_list_del(&tx->tx_list);
1633                 spin_unlock_irqrestore(&conn->rac_lock, flags);
1634                 return tx;
1635         }
1636
1637         spin_unlock_irqrestore(&conn->rac_lock, flags);
1638         CWARN("Unmatched reply %02x/"LPX64" from %s\n",
1639               type, cookie, libcfs_nid2str(conn->rac_peer->rap_nid));
1640         return NULL;
1641 }
1642
1643 void
1644 kranal_check_fma_rx (kra_conn_t *conn)
1645 {
1646         unsigned long flags;
1647         __u32         seq;
1648         kra_tx_t     *tx;
1649         kra_msg_t    *msg;
1650         void         *prefix;
1651         RAP_RETURN    rrc = RapkFmaGetPrefix(conn->rac_rihandle, &prefix);
1652         kra_peer_t   *peer = conn->rac_peer;
1653         int           rc = 0;
1654         int           repost = 1;
1655
1656         if (rrc == RAP_NOT_DONE)
1657                 return;
1658
1659         CDEBUG(D_NET, "RX on %p\n", conn);
1660
1661         LASSERT (rrc == RAP_SUCCESS);
1662         conn->rac_last_rx = jiffies;
1663         seq = conn->rac_rx_seq++;
1664         msg = (kra_msg_t *)prefix;
1665
1666         /* stash message for portals callbacks they'll NULL
1667          * rac_rxmsg if they consume it */
1668         LASSERT (conn->rac_rxmsg == NULL);
1669         conn->rac_rxmsg = msg;
1670
1671         if (msg->ram_magic != RANAL_MSG_MAGIC) {
1672                 if (__swab32(msg->ram_magic) != RANAL_MSG_MAGIC) {
1673                         CERROR("Unexpected magic %08x from %s\n",
1674                                msg->ram_magic, libcfs_nid2str(peer->rap_nid));
1675                         rc = -EPROTO;
1676                         goto out;
1677                 }
1678
1679                 __swab32s(&msg->ram_magic);
1680                 __swab16s(&msg->ram_version);
1681                 __swab16s(&msg->ram_type);
1682                 __swab64s(&msg->ram_srcnid);
1683                 __swab64s(&msg->ram_connstamp);
1684                 __swab32s(&msg->ram_seq);
1685
1686                 /* NB message type checked below; NOT here... */
1687                 switch (msg->ram_type) {
1688                 case RANAL_MSG_PUT_ACK:
1689                         kranal_swab_rdma_desc(&msg->ram_u.putack.rapam_desc);
1690                         break;
1691
1692                 case RANAL_MSG_GET_REQ:
1693                         kranal_swab_rdma_desc(&msg->ram_u.get.ragm_desc);
1694                         break;
1695
1696                 default:
1697                         break;
1698                 }
1699         }
1700
1701         if (msg->ram_version != RANAL_MSG_VERSION) {
1702                 CERROR("Unexpected protocol version %d from %s\n",
1703                        msg->ram_version, libcfs_nid2str(peer->rap_nid));
1704                 rc = -EPROTO;
1705                 goto out;
1706         }
1707
1708         if (msg->ram_srcnid != peer->rap_nid) {
1709                 CERROR("Unexpected peer %s from %s\n",
1710                        libcfs_nid2str(msg->ram_srcnid),
1711                        libcfs_nid2str(peer->rap_nid));
1712                 rc = -EPROTO;
1713                 goto out;
1714         }
1715
1716         if (msg->ram_connstamp != conn->rac_peer_connstamp) {
1717                 CERROR("Unexpected connstamp "LPX64"("LPX64
1718                        " expected) from %s\n",
1719                        msg->ram_connstamp, conn->rac_peer_connstamp,
1720                        libcfs_nid2str(peer->rap_nid));
1721                 rc = -EPROTO;
1722                 goto out;
1723         }
1724
1725         if (msg->ram_seq != seq) {
1726                 CERROR("Unexpected sequence number %d(%d expected) from %s\n",
1727                        msg->ram_seq, seq, libcfs_nid2str(peer->rap_nid));
1728                 rc = -EPROTO;
1729                 goto out;
1730         }
1731
1732         if ((msg->ram_type & RANAL_MSG_FENCE) != 0) {
1733                 /* This message signals RDMA completion... */
1734                 rrc = RapkFmaSyncWait(conn->rac_rihandle);
1735                 if (rrc != RAP_SUCCESS) {
1736                         CERROR("RapkFmaSyncWait failed: %d\n", rrc);
1737                         rc = -ENETDOWN;
1738                         goto out;
1739                 }
1740         }
1741
1742         if (conn->rac_close_recvd) {
1743                 CERROR("Unexpected message %d after CLOSE from %s\n",
1744                        msg->ram_type, libcfs_nid2str(conn->rac_peer->rap_nid));
1745                 rc = -EPROTO;
1746                 goto out;
1747         }
1748
1749         if (msg->ram_type == RANAL_MSG_CLOSE) {
1750                 CWARN("RX CLOSE from %s\n", libcfs_nid2str(conn->rac_peer->rap_nid));
1751                 conn->rac_close_recvd = 1;
1752                 write_lock_irqsave(&kranal_data.kra_global_lock, flags);
1753
1754                 if (conn->rac_state == RANAL_CONN_ESTABLISHED)
1755                         kranal_close_conn_locked(conn, 0);
1756                 else if (conn->rac_state == RANAL_CONN_CLOSING &&
1757                          conn->rac_close_sent)
1758                         kranal_terminate_conn_locked(conn);
1759
1760                 write_unlock_irqrestore(&kranal_data.kra_global_lock,
1761                                             flags);
1762                 goto out;
1763         }
1764
1765         if (conn->rac_state != RANAL_CONN_ESTABLISHED)
1766                 goto out;
1767
1768         switch (msg->ram_type) {
1769         case RANAL_MSG_NOOP:
1770                 /* Nothing to do; just a keepalive */
1771                 CDEBUG(D_NET, "RX NOOP on %p\n", conn);
1772                 break;
1773
1774         case RANAL_MSG_IMMEDIATE:
1775                 CDEBUG(D_NET, "RX IMMEDIATE on %p\n", conn);
1776                 rc = lnet_parse(kranal_data.kra_ni, &msg->ram_u.immediate.raim_hdr,
1777                                 msg->ram_srcnid, conn, 0);
1778                 repost = rc < 0;
1779                 break;
1780
1781         case RANAL_MSG_PUT_REQ:
1782                 CDEBUG(D_NET, "RX PUT_REQ on %p\n", conn);
1783                 rc = lnet_parse(kranal_data.kra_ni, &msg->ram_u.putreq.raprm_hdr,
1784                                 msg->ram_srcnid, conn, 1);
1785                 repost = rc < 0;
1786                 break;
1787
1788         case RANAL_MSG_PUT_NAK:
1789                 CDEBUG(D_NET, "RX PUT_NAK on %p\n", conn);
1790                 tx = kranal_match_reply(conn, RANAL_MSG_PUT_REQ,
1791                                         msg->ram_u.completion.racm_cookie);
1792                 if (tx == NULL)
1793                         break;
1794
1795                 LASSERT (tx->tx_buftype == RANAL_BUF_PHYS_MAPPED ||
1796                          tx->tx_buftype == RANAL_BUF_VIRT_MAPPED);
1797                 kranal_tx_done(tx, -ENOENT);    /* no match */
1798                 break;
1799
1800         case RANAL_MSG_PUT_ACK:
1801                 CDEBUG(D_NET, "RX PUT_ACK on %p\n", conn);
1802                 tx = kranal_match_reply(conn, RANAL_MSG_PUT_REQ,
1803                                         msg->ram_u.putack.rapam_src_cookie);
1804                 if (tx == NULL)
1805                         break;
1806
1807                 kranal_rdma(tx, RANAL_MSG_PUT_DONE,
1808                             &msg->ram_u.putack.rapam_desc,
1809                             msg->ram_u.putack.rapam_desc.rard_nob,
1810                             msg->ram_u.putack.rapam_dst_cookie);
1811                 break;
1812
1813         case RANAL_MSG_PUT_DONE:
1814                 CDEBUG(D_NET, "RX PUT_DONE on %p\n", conn);
1815                 tx = kranal_match_reply(conn, RANAL_MSG_PUT_ACK,
1816                                         msg->ram_u.completion.racm_cookie);
1817                 if (tx == NULL)
1818                         break;
1819
1820                 LASSERT (tx->tx_buftype == RANAL_BUF_PHYS_MAPPED ||
1821                          tx->tx_buftype == RANAL_BUF_VIRT_MAPPED);
1822                 kranal_tx_done(tx, 0);
1823                 break;
1824
1825         case RANAL_MSG_GET_REQ:
1826                 CDEBUG(D_NET, "RX GET_REQ on %p\n", conn);
1827                 rc = lnet_parse(kranal_data.kra_ni, &msg->ram_u.get.ragm_hdr,
1828                                 msg->ram_srcnid, conn, 1);
1829                 repost = rc < 0;
1830                 break;
1831
1832         case RANAL_MSG_GET_NAK:
1833                 CDEBUG(D_NET, "RX GET_NAK on %p\n", conn);
1834                 tx = kranal_match_reply(conn, RANAL_MSG_GET_REQ,
1835                                         msg->ram_u.completion.racm_cookie);
1836                 if (tx == NULL)
1837                         break;
1838
1839                 LASSERT (tx->tx_buftype == RANAL_BUF_PHYS_MAPPED ||
1840                          tx->tx_buftype == RANAL_BUF_VIRT_MAPPED);
1841                 kranal_tx_done(tx, -ENOENT);    /* no match */
1842                 break;
1843
1844         case RANAL_MSG_GET_DONE:
1845                 CDEBUG(D_NET, "RX GET_DONE on %p\n", conn);
1846                 tx = kranal_match_reply(conn, RANAL_MSG_GET_REQ,
1847                                         msg->ram_u.completion.racm_cookie);
1848                 if (tx == NULL)
1849                         break;
1850
1851                 LASSERT (tx->tx_buftype == RANAL_BUF_PHYS_MAPPED ||
1852                          tx->tx_buftype == RANAL_BUF_VIRT_MAPPED);
1853 #if 0
1854                 /* completion message should send rdma length if we ever allow
1855                  * GET truncation */
1856                 lnet_set_reply_msg_len(kranal_data.kra_ni, tx->tx_lntmsg[1], ???);
1857 #endif
1858                 kranal_tx_done(tx, 0);
1859                 break;
1860         }
1861
1862  out:
1863         if (rc < 0)                             /* protocol/comms error */
1864                 kranal_close_conn (conn, rc);
1865
1866         if (repost && conn->rac_rxmsg != NULL)
1867                 kranal_consume_rxmsg(conn, NULL, 0);
1868
1869         /* check again later */
1870         kranal_schedule_conn(conn);
1871 }
1872
1873 void
1874 kranal_complete_closed_conn (kra_conn_t *conn)
1875 {
1876         kra_tx_t   *tx;
1877         int         nfma;
1878         int         nreplies;
1879
1880         LASSERT (conn->rac_state == RANAL_CONN_CLOSED);
1881         LASSERT (cfs_list_empty(&conn->rac_list));
1882         LASSERT (cfs_list_empty(&conn->rac_hashlist));
1883
1884         for (nfma = 0; !cfs_list_empty(&conn->rac_fmaq); nfma++) {
1885                 tx = cfs_list_entry(conn->rac_fmaq.next, kra_tx_t, tx_list);
1886
1887                 cfs_list_del(&tx->tx_list);
1888                 kranal_tx_done(tx, -ECONNABORTED);
1889         }
1890
1891         LASSERT (cfs_list_empty(&conn->rac_rdmaq));
1892
1893         for (nreplies = 0; !cfs_list_empty(&conn->rac_replyq); nreplies++) {
1894                 tx = cfs_list_entry(conn->rac_replyq.next, kra_tx_t, tx_list);
1895
1896                 cfs_list_del(&tx->tx_list);
1897                 kranal_tx_done(tx, -ECONNABORTED);
1898         }
1899
1900         CWARN("Closed conn %p -> %s: nmsg %d nreplies %d\n",
1901                conn, libcfs_nid2str(conn->rac_peer->rap_nid), nfma, nreplies);
1902 }
1903
1904 int kranal_process_new_conn (kra_conn_t *conn)
1905 {
1906         RAP_RETURN   rrc;
1907
1908         rrc = RapkCompleteSync(conn->rac_rihandle, 1);
1909         if (rrc == RAP_SUCCESS)
1910                 return 0;
1911
1912         LASSERT (rrc == RAP_NOT_DONE);
1913         if (!cfs_time_aftereq(jiffies, conn->rac_last_tx +
1914                               conn->rac_timeout * HZ))
1915                 return -EAGAIN;
1916
1917         /* Too late */
1918         rrc = RapkCompleteSync(conn->rac_rihandle, 0);
1919         LASSERT (rrc == RAP_SUCCESS);
1920         return -ETIMEDOUT;
1921 }
1922
1923 int
1924 kranal_scheduler (void *arg)
1925 {
1926         kra_device_t     *dev = (kra_device_t *)arg;
1927         wait_queue_t    wait;
1928         kra_conn_t       *conn;
1929         unsigned long     flags;
1930         unsigned long     deadline;
1931         unsigned long     soonest;
1932         int               nsoonest;
1933         long              timeout;
1934         cfs_list_t       *tmp;
1935         cfs_list_t       *nxt;
1936         int               rc;
1937         int               dropped_lock;
1938         int               busy_loops = 0;
1939
1940         cfs_block_allsigs();
1941
1942         dev->rad_scheduler = current;
1943         init_waitqueue_entry_current(&wait);
1944
1945         spin_lock_irqsave(&dev->rad_lock, flags);
1946
1947         while (!kranal_data.kra_shutdown) {
1948                 /* Safe: kra_shutdown only set when quiescent */
1949
1950                 if (busy_loops++ >= RANAL_RESCHED) {
1951                         spin_unlock_irqrestore(&dev->rad_lock, flags);
1952
1953                         cond_resched();
1954                         busy_loops = 0;
1955
1956                         spin_lock_irqsave(&dev->rad_lock, flags);
1957                 }
1958
1959                 dropped_lock = 0;
1960
1961                 if (dev->rad_ready) {
1962                         /* Device callback fired since I last checked it */
1963                         dev->rad_ready = 0;
1964                         spin_unlock_irqrestore(&dev->rad_lock, flags);
1965                         dropped_lock = 1;
1966
1967                         kranal_check_rdma_cq(dev);
1968                         kranal_check_fma_cq(dev);
1969
1970                         spin_lock_irqsave(&dev->rad_lock, flags);
1971                 }
1972
1973                 cfs_list_for_each_safe(tmp, nxt, &dev->rad_ready_conns) {
1974                         conn = cfs_list_entry(tmp, kra_conn_t, rac_schedlist);
1975
1976                         cfs_list_del_init(&conn->rac_schedlist);
1977                         LASSERT (conn->rac_scheduled);
1978                         conn->rac_scheduled = 0;
1979                         spin_unlock_irqrestore(&dev->rad_lock, flags);
1980                         dropped_lock = 1;
1981
1982                         kranal_check_fma_rx(conn);
1983                         kranal_process_fmaq(conn);
1984
1985                         if (conn->rac_state == RANAL_CONN_CLOSED)
1986                                 kranal_complete_closed_conn(conn);
1987
1988                         kranal_conn_decref(conn);
1989                         spin_lock_irqsave(&dev->rad_lock, flags);
1990                 }
1991
1992                 nsoonest = 0;
1993                 soonest = jiffies;
1994
1995                 cfs_list_for_each_safe(tmp, nxt, &dev->rad_new_conns) {
1996                         conn = cfs_list_entry(tmp, kra_conn_t, rac_schedlist);
1997
1998                         deadline = conn->rac_last_tx + conn->rac_keepalive;
1999                         if (cfs_time_aftereq(jiffies, deadline)) {
2000                                 /* Time to process this new conn */
2001                                 spin_unlock_irqrestore(&dev->rad_lock,
2002                                                            flags);
2003                                 dropped_lock = 1;
2004
2005                                 rc = kranal_process_new_conn(conn);
2006                                 if (rc != -EAGAIN) {
2007                                         /* All done with this conn */
2008                                         spin_lock_irqsave(&dev->rad_lock,
2009                                                               flags);
2010                                         cfs_list_del_init(&conn->rac_schedlist);
2011                                         spin_unlock_irqrestore(&dev-> \
2012                                                                    rad_lock,
2013                                                                    flags);
2014
2015                                         kranal_conn_decref(conn);
2016                                         spin_lock_irqsave(&dev->rad_lock,
2017                                                               flags);
2018                                         continue;
2019                                 }
2020
2021                                 /* retry with exponential backoff until HZ */
2022                                 if (conn->rac_keepalive == 0)
2023                                         conn->rac_keepalive = 1;
2024                                 else if (conn->rac_keepalive <= HZ)
2025                                         conn->rac_keepalive *= 2;
2026                                 else
2027                                         conn->rac_keepalive += HZ;
2028
2029                                 deadline = conn->rac_last_tx + conn->rac_keepalive;
2030                                 spin_lock_irqsave(&dev->rad_lock, flags);
2031                         }
2032
2033                         /* Does this conn need attention soonest? */
2034                         if (nsoonest++ == 0 ||
2035                             !cfs_time_aftereq(deadline, soonest))
2036                                 soonest = deadline;
2037                 }
2038
2039                 if (dropped_lock)               /* may sleep iff I didn't drop the lock */
2040                         continue;
2041
2042                 set_current_state(TASK_INTERRUPTIBLE);
2043                 add_wait_queue_exclusive(&dev->rad_waitq, &wait);
2044                 spin_unlock_irqrestore(&dev->rad_lock, flags);
2045
2046                 if (nsoonest == 0) {
2047                         busy_loops = 0;
2048                         waitq_wait(&wait, TASK_INTERRUPTIBLE);
2049                 } else {
2050                         timeout = (long)(soonest - jiffies);
2051                         if (timeout > 0) {
2052                                 busy_loops = 0;
2053                                 waitq_timedwait(&wait,
2054                                                     TASK_INTERRUPTIBLE,
2055                                                     timeout);
2056                         }
2057                 }
2058
2059                 remove_wait_queue(&dev->rad_waitq, &wait);
2060                 set_current_state(TASK_RUNNING);
2061                 spin_lock_irqsave(&dev->rad_lock, flags);
2062         }
2063
2064         spin_unlock_irqrestore(&dev->rad_lock, flags);
2065
2066         dev->rad_scheduler = NULL;
2067         kranal_thread_fini();
2068         return 0;
2069 }