lnet/klnds/ralnd/ralnd_cb.c

   1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
   2  * vim:expandtab:shiftwidth=8:tabstop=8:
   3  *
   4  * GPL HEADER START
   5  *
   6  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
   7  *
   8  * This program is free software; you can redistribute it and/or modify
   9  * it under the terms of the GNU General Public License version 2 only,
  10  * as published by the Free Software Foundation.
  11  *
  12  * This program is distributed in the hope that it will be useful, but
  13  * WITHOUT ANY WARRANTY; without even the implied warranty of
  14  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  15  * General Public License version 2 for more details (a copy is included
  16  * in the LICENSE file that accompanied this code).
  17  *
  18  * You should have received a copy of the GNU General Public License
  19  * version 2 along with this program; If not, see [sun.com URL with a
  20  * copy of GPLv2].
  21  *
  22  * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
  23  * CA 95054 USA or visit www.sun.com if you need additional information or
  24  * have any questions.
  25  *
  26  * GPL HEADER END
  27  */
  28 /*
  29  * Copyright  2008 Sun Microsystems, Inc. All rights reserved
  30  * Use is subject to license terms.
  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                 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 (list_empty(&kranal_data.kra_idle_txs)) {
  99                 spin_unlock_irqrestore(&kranal_data.kra_tx_lock, flags);
 100                 return NULL;
 101         }
 102
 103         tx = list_entry(kranal_data.kra_idle_txs.next, kra_tx_t, tx_list);
 104         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         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         struct list_head *tmp;
 422
 423         /* just return the first connection */
 424         list_for_each (tmp, &peer->rap_conns) {
 425                 return 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         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 (list_empty(&peer->rap_tx_queue));
 513
 514                 if (!(peer->rap_reconnect_interval == 0 || /* first attempt */
 515                       time_after_eq(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                 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         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         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)
 895 {
 896         long    pid = kernel_thread(fn, arg, 0);
 897
 898         if (pid < 0)
 899                 return(int)pid;
 900
 901         atomic_inc(&kranal_data.kra_nthreads);
 902         return 0;
 903 }
 904
 905 void
 906 kranal_thread_fini (void)
 907 {
 908         atomic_dec(&kranal_data.kra_nthreads);
 909 }
 910
 911 int
 912 kranal_check_conn_timeouts (kra_conn_t *conn)
 913 {
 914         kra_tx_t          *tx;
 915         struct list_head  *ttmp;
 916         unsigned long      flags;
 917         long               timeout;
 918         unsigned long      now = jiffies;
 919
 920         LASSERT (conn->rac_state == RANAL_CONN_ESTABLISHED ||
 921                  conn->rac_state == RANAL_CONN_CLOSING);
 922
 923         if (!conn->rac_close_sent &&
 924             time_after_eq(now, conn->rac_last_tx + conn->rac_keepalive * HZ)) {
 925                 /* not sent in a while; schedule conn so scheduler sends a keepalive */
 926                 CDEBUG(D_NET, "Scheduling keepalive %p->%s\n",
 927                        conn, libcfs_nid2str(conn->rac_peer->rap_nid));
 928                 kranal_schedule_conn(conn);
 929         }
 930
 931         timeout = conn->rac_timeout * HZ;
 932
 933         if (!conn->rac_close_recvd &&
 934             time_after_eq(now, conn->rac_last_rx + timeout)) {
 935                 CERROR("%s received from %s within %lu seconds\n",
 936                        (conn->rac_state == RANAL_CONN_ESTABLISHED) ?
 937                        "Nothing" : "CLOSE not",
 938                        libcfs_nid2str(conn->rac_peer->rap_nid),
 939                        (now - conn->rac_last_rx)/HZ);
 940                 return -ETIMEDOUT;
 941         }
 942
 943         if (conn->rac_state != RANAL_CONN_ESTABLISHED)
 944                 return 0;
 945
 946         /* Check the conn's queues are moving.  These are "belt+braces" checks,
 947          * in case of hardware/software errors that make this conn seem
 948          * responsive even though it isn't progressing its message queues. */
 949
 950         spin_lock_irqsave(&conn->rac_lock, flags);
 951
 952         list_for_each (ttmp, &conn->rac_fmaq) {
 953                 tx = list_entry(ttmp, kra_tx_t, tx_list);
 954
 955                 if (time_after_eq(now, tx->tx_qtime + timeout)) {
 956                         spin_unlock_irqrestore(&conn->rac_lock, flags);
 957                         CERROR("tx on fmaq for %s blocked %lu seconds\n",
 958                                libcfs_nid2str(conn->rac_peer->rap_nid),
 959                                (now - tx->tx_qtime)/HZ);
 960                         return -ETIMEDOUT;
 961                 }
 962         }
 963
 964         list_for_each (ttmp, &conn->rac_rdmaq) {
 965                 tx = list_entry(ttmp, kra_tx_t, tx_list);
 966
 967                 if (time_after_eq(now, tx->tx_qtime + timeout)) {
 968                         spin_unlock_irqrestore(&conn->rac_lock, flags);
 969                         CERROR("tx on rdmaq for %s blocked %lu seconds\n",
 970                                libcfs_nid2str(conn->rac_peer->rap_nid),
 971                                (now - tx->tx_qtime)/HZ);
 972                         return -ETIMEDOUT;
 973                 }
 974         }
 975
 976         list_for_each (ttmp, &conn->rac_replyq) {
 977                 tx = list_entry(ttmp, kra_tx_t, tx_list);
 978
 979                 if (time_after_eq(now, tx->tx_qtime + timeout)) {
 980                         spin_unlock_irqrestore(&conn->rac_lock, flags);
 981                         CERROR("tx on replyq for %s blocked %lu seconds\n",
 982                                libcfs_nid2str(conn->rac_peer->rap_nid),
 983                                (now - tx->tx_qtime)/HZ);
 984                         return -ETIMEDOUT;
 985                 }
 986         }
 987
 988         spin_unlock_irqrestore(&conn->rac_lock, flags);
 989         return 0;
 990 }
 991
 992 void
 993 kranal_reaper_check (int idx, unsigned long *min_timeoutp)
 994 {
 995         struct list_head  *conns = &kranal_data.kra_conns[idx];
 996         struct list_head  *ctmp;
 997         kra_conn_t        *conn;
 998         unsigned long      flags;
 999         int                rc;
1000
1001  again:
1002         /* NB. We expect to check all the conns and not find any problems, so
1003          * we just use a shared lock while we take a look... */
1004         read_lock(&kranal_data.kra_global_lock);
1005
1006         list_for_each (ctmp, conns) {
1007                 conn = list_entry(ctmp, kra_conn_t, rac_hashlist);
1008
1009                 if (conn->rac_timeout < *min_timeoutp )
1010                         *min_timeoutp = conn->rac_timeout;
1011                 if (conn->rac_keepalive < *min_timeoutp )
1012                         *min_timeoutp = conn->rac_keepalive;
1013
1014                 rc = kranal_check_conn_timeouts(conn);
1015                 if (rc == 0)
1016                         continue;
1017
1018                 kranal_conn_addref(conn);
1019                 read_unlock(&kranal_data.kra_global_lock);
1020
1021                 CERROR("Conn to %s, cqid %d timed out\n",
1022                        libcfs_nid2str(conn->rac_peer->rap_nid),
1023                        conn->rac_cqid);
1024
1025                 write_lock_irqsave(&kranal_data.kra_global_lock, flags);
1026
1027                 switch (conn->rac_state) {
1028                 default:
1029                         LBUG();
1030
1031                 case RANAL_CONN_ESTABLISHED:
1032                         kranal_close_conn_locked(conn, -ETIMEDOUT);
1033                         break;
1034
1035                 case RANAL_CONN_CLOSING:
1036                         kranal_terminate_conn_locked(conn);
1037                         break;
1038                 }
1039
1040                 write_unlock_irqrestore(&kranal_data.kra_global_lock, 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         char               name[16];
1056         wait_queue_t       wait;
1057         unsigned long      flags;
1058         kra_peer_t        *peer;
1059         kra_acceptsock_t  *ras;
1060         int                did_something;
1061
1062         snprintf(name, sizeof(name), "kranal_connd_%02ld", id);
1063         cfs_daemonize(name);
1064         cfs_block_allsigs();
1065
1066         init_waitqueue_entry(&wait, current);
1067
1068         spin_lock_irqsave(&kranal_data.kra_connd_lock, flags);
1069
1070         while (!kranal_data.kra_shutdown) {
1071                 did_something = 0;
1072
1073                 if (!list_empty(&kranal_data.kra_connd_acceptq)) {
1074                         ras = list_entry(kranal_data.kra_connd_acceptq.next,
1075                                          kra_acceptsock_t, ras_list);
1076                         list_del(&ras->ras_list);
1077
1078                         spin_unlock_irqrestore(&kranal_data.kra_connd_lock, flags);
1079
1080                         CDEBUG(D_NET,"About to handshake someone\n");
1081
1082                         kranal_conn_handshake(ras->ras_sock, NULL);
1083                         kranal_free_acceptsock(ras);
1084
1085                         CDEBUG(D_NET,"Finished handshaking someone\n");
1086
1087                         spin_lock_irqsave(&kranal_data.kra_connd_lock, flags);
1088                         did_something = 1;
1089                 }
1090
1091                 if (!list_empty(&kranal_data.kra_connd_peers)) {
1092                         peer = list_entry(kranal_data.kra_connd_peers.next,
1093                                           kra_peer_t, rap_connd_list);
1094
1095                         list_del_init(&peer->rap_connd_list);
1096                         spin_unlock_irqrestore(&kranal_data.kra_connd_lock, flags);
1097
1098                         kranal_connect(peer);
1099                         kranal_peer_decref(peer);
1100
1101                         spin_lock_irqsave(&kranal_data.kra_connd_lock, flags);
1102                         did_something = 1;
1103                 }
1104
1105                 if (did_something)
1106                         continue;
1107
1108                 set_current_state(TASK_INTERRUPTIBLE);
1109                 add_wait_queue_exclusive(&kranal_data.kra_connd_waitq, &wait);
1110
1111                 spin_unlock_irqrestore(&kranal_data.kra_connd_lock, flags);
1112
1113                 schedule ();
1114
1115                 set_current_state(TASK_RUNNING);
1116                 remove_wait_queue(&kranal_data.kra_connd_waitq, &wait);
1117
1118                 spin_lock_irqsave(&kranal_data.kra_connd_lock, flags);
1119         }
1120
1121         spin_unlock_irqrestore(&kranal_data.kra_connd_lock, flags);
1122
1123         kranal_thread_fini();
1124         return 0;
1125 }
1126
1127 void
1128 kranal_update_reaper_timeout(long timeout)
1129 {
1130         unsigned long   flags;
1131
1132         LASSERT (timeout > 0);
1133
1134         spin_lock_irqsave(&kranal_data.kra_reaper_lock, flags);
1135
1136         if (timeout < kranal_data.kra_new_min_timeout)
1137                 kranal_data.kra_new_min_timeout = timeout;
1138
1139         spin_unlock_irqrestore(&kranal_data.kra_reaper_lock, flags);
1140 }
1141
1142 int
1143 kranal_reaper (void *arg)
1144 {
1145         wait_queue_t       wait;
1146         unsigned long      flags;
1147         long               timeout;
1148         int                i;
1149         int                conn_entries = kranal_data.kra_conn_hash_size;
1150         int                conn_index = 0;
1151         int                base_index = conn_entries - 1;
1152         unsigned long      next_check_time = jiffies;
1153         long               next_min_timeout = MAX_SCHEDULE_TIMEOUT;
1154         long               current_min_timeout = 1;
1155
1156         cfs_daemonize("kranal_reaper");
1157         cfs_block_allsigs();
1158
1159         init_waitqueue_entry(&wait, current);
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, flags);
1181
1182                         schedule_timeout(timeout);
1183
1184                         spin_lock_irqsave(&kranal_data.kra_reaper_lock, flags);
1185
1186                         set_current_state(TASK_RUNNING);
1187                         remove_wait_queue(&kranal_data.kra_reaper_waitq, &wait);
1188                         continue;
1189                 }
1190
1191                 if (kranal_data.kra_new_min_timeout != MAX_SCHEDULE_TIMEOUT) {
1192                         /* new min timeout set: restart min timeout scan */
1193                         next_min_timeout = MAX_SCHEDULE_TIMEOUT;
1194                         base_index = conn_index - 1;
1195                         if (base_index < 0)
1196                                 base_index = conn_entries - 1;
1197
1198                         if (kranal_data.kra_new_min_timeout < current_min_timeout) {
1199                                 current_min_timeout = kranal_data.kra_new_min_timeout;
1200                                 CDEBUG(D_NET, "Set new min timeout %ld\n",
1201                                        current_min_timeout);
1202                         }
1203
1204                         kranal_data.kra_new_min_timeout = MAX_SCHEDULE_TIMEOUT;
1205                 }
1206                 min_timeout = current_min_timeout;
1207
1208                 spin_unlock_irqrestore(&kranal_data.kra_reaper_lock, flags);
1209
1210                 LASSERT (min_timeout > 0);
1211
1212                 /* Compute how many table entries to check now so I get round
1213                  * the whole table fast enough given that I do this at fixed
1214                  * intervals of 'p' seconds) */
1215                 chunk = conn_entries;
1216                 if (min_timeout > n * p)
1217                         chunk = (chunk * n * p) / min_timeout;
1218                 if (chunk == 0)
1219                         chunk = 1;
1220
1221                 for (i = 0; i < chunk; i++) {
1222                         kranal_reaper_check(conn_index,
1223                                             &next_min_timeout);
1224                         conn_index = (conn_index + 1) % conn_entries;
1225                 }
1226
1227                 next_check_time += p * HZ;
1228
1229                 spin_lock_irqsave(&kranal_data.kra_reaper_lock, flags);
1230
1231                 if (((conn_index - chunk <= base_index &&
1232                       base_index < conn_index) ||
1233                      (conn_index - conn_entries - chunk <= base_index &&
1234                       base_index < conn_index - conn_entries))) {
1235
1236                         /* Scanned all conns: set current_min_timeout... */
1237                         if (current_min_timeout != next_min_timeout) {
1238                                 current_min_timeout = next_min_timeout;
1239                                 CDEBUG(D_NET, "Set new min timeout %ld\n",
1240                                        current_min_timeout);
1241                         }
1242
1243                         /* ...and restart min timeout scan */
1244                         next_min_timeout = MAX_SCHEDULE_TIMEOUT;
1245                         base_index = conn_index - 1;
1246                         if (base_index < 0)
1247                                 base_index = conn_entries - 1;
1248                 }
1249         }
1250
1251         kranal_thread_fini();
1252         return 0;
1253 }
1254
1255 void
1256 kranal_check_rdma_cq (kra_device_t *dev)
1257 {
1258         kra_conn_t          *conn;
1259         kra_tx_t            *tx;
1260         RAP_RETURN           rrc;
1261         unsigned long        flags;
1262         RAP_RDMA_DESCRIPTOR *desc;
1263         __u32                cqid;
1264         __u32                event_type;
1265
1266         for (;;) {
1267                 rrc = RapkCQDone(dev->rad_rdma_cqh, &cqid, &event_type);
1268                 if (rrc == RAP_NOT_DONE) {
1269                         CDEBUG(D_NET, "RDMA CQ %d empty\n", dev->rad_id);
1270                         return;
1271                 }
1272
1273                 LASSERT (rrc == RAP_SUCCESS);
1274                 LASSERT ((event_type & RAPK_CQ_EVENT_OVERRUN) == 0);
1275
1276                 read_lock(&kranal_data.kra_global_lock);
1277
1278                 conn = kranal_cqid2conn_locked(cqid);
1279                 if (conn == NULL) {
1280                         /* Conn was destroyed? */
1281                         CDEBUG(D_NET, "RDMA CQID lookup %d failed\n", cqid);
1282                         read_unlock(&kranal_data.kra_global_lock);
1283                         continue;
1284                 }
1285
1286                 rrc = RapkRdmaDone(conn->rac_rihandle, &desc);
1287                 LASSERT (rrc == RAP_SUCCESS);
1288
1289                 CDEBUG(D_NET, "Completed %p\n",
1290                        list_entry(conn->rac_rdmaq.next, kra_tx_t, tx_list));
1291
1292                 spin_lock_irqsave(&conn->rac_lock, flags);
1293
1294                 LASSERT (!list_empty(&conn->rac_rdmaq));
1295                 tx = list_entry(conn->rac_rdmaq.next, kra_tx_t, tx_list);
1296                 list_del(&tx->tx_list);
1297
1298                 LASSERT(desc->AppPtr == (void *)tx);
1299                 LASSERT(tx->tx_msg.ram_type == RANAL_MSG_PUT_DONE ||
1300                         tx->tx_msg.ram_type == RANAL_MSG_GET_DONE);
1301
1302                 list_add_tail(&tx->tx_list, &conn->rac_fmaq);
1303                 tx->tx_qtime = jiffies;
1304
1305                 spin_unlock_irqrestore(&conn->rac_lock, flags);
1306
1307                 /* Get conn's fmaq processed, now I've just put something
1308                  * there */
1309                 kranal_schedule_conn(conn);
1310
1311                 read_unlock(&kranal_data.kra_global_lock);
1312         }
1313 }
1314
1315 void
1316 kranal_check_fma_cq (kra_device_t *dev)
1317 {
1318         kra_conn_t         *conn;
1319         RAP_RETURN          rrc;
1320         __u32               cqid;
1321         __u32               event_type;
1322         struct list_head   *conns;
1323         struct list_head   *tmp;
1324         int                 i;
1325
1326         for (;;) {
1327                 rrc = RapkCQDone(dev->rad_fma_cqh, &cqid, &event_type);
1328                 if (rrc == RAP_NOT_DONE) {
1329                         CDEBUG(D_NET, "FMA CQ %d empty\n", dev->rad_id);
1330                         return;
1331                 }
1332
1333                 LASSERT (rrc == RAP_SUCCESS);
1334
1335                 if ((event_type & RAPK_CQ_EVENT_OVERRUN) == 0) {
1336
1337                         read_lock(&kranal_data.kra_global_lock);
1338
1339                         conn = kranal_cqid2conn_locked(cqid);
1340                         if (conn == NULL) {
1341                                 CDEBUG(D_NET, "FMA CQID lookup %d failed\n",
1342                                        cqid);
1343                         } else {
1344                                 CDEBUG(D_NET, "FMA completed: %p CQID %d\n",
1345                                        conn, cqid);
1346                                 kranal_schedule_conn(conn);
1347                         }
1348
1349                         read_unlock(&kranal_data.kra_global_lock);
1350                         continue;
1351                 }
1352
1353                 /* FMA CQ has overflowed: check ALL conns */
1354                 CWARN("FMA CQ overflow: scheduling ALL conns on device %d\n",
1355                       dev->rad_id);
1356
1357                 for (i = 0; i < kranal_data.kra_conn_hash_size; i++) {
1358
1359                         read_lock(&kranal_data.kra_global_lock);
1360
1361                         conns = &kranal_data.kra_conns[i];
1362
1363                         list_for_each (tmp, conns) {
1364                                 conn = list_entry(tmp, kra_conn_t,
1365                                                   rac_hashlist);
1366
1367                                 if (conn->rac_device == dev)
1368                                         kranal_schedule_conn(conn);
1369                         }
1370
1371                         /* don't block write lockers for too long... */
1372                         read_unlock(&kranal_data.kra_global_lock);
1373                 }
1374         }
1375 }
1376
1377 int
1378 kranal_sendmsg(kra_conn_t *conn, kra_msg_t *msg,
1379                void *immediate, int immediatenob)
1380 {
1381         int        sync = (msg->ram_type & RANAL_MSG_FENCE) != 0;
1382         RAP_RETURN rrc;
1383
1384         CDEBUG(D_NET,"%p sending msg %p %02x%s [%p for %d]\n",
1385                conn, msg, msg->ram_type, sync ? "(sync)" : "",
1386                immediate, immediatenob);
1387
1388         LASSERT (sizeof(*msg) <= RANAL_FMA_MAX_PREFIX);
1389         LASSERT ((msg->ram_type == RANAL_MSG_IMMEDIATE) ?
1390                  immediatenob <= RANAL_FMA_MAX_DATA :
1391                  immediatenob == 0);
1392
1393         msg->ram_connstamp = conn->rac_my_connstamp;
1394         msg->ram_seq = conn->rac_tx_seq;
1395
1396         if (sync)
1397                 rrc = RapkFmaSyncSend(conn->rac_rihandle,
1398                                       immediate, immediatenob,
1399                                       msg, sizeof(*msg));
1400         else
1401                 rrc = RapkFmaSend(conn->rac_rihandle,
1402                                   immediate, immediatenob,
1403                                   msg, sizeof(*msg));
1404
1405         switch (rrc) {
1406         default:
1407                 LBUG();
1408
1409         case RAP_SUCCESS:
1410                 conn->rac_last_tx = jiffies;
1411                 conn->rac_tx_seq++;
1412                 return 0;
1413
1414         case RAP_NOT_DONE:
1415                 if (time_after_eq(jiffies,
1416                                   conn->rac_last_tx + conn->rac_keepalive*HZ))
1417                         CWARN("EAGAIN sending %02x (idle %lu secs)\n",
1418                                msg->ram_type, (jiffies - conn->rac_last_tx)/HZ);
1419                 return -EAGAIN;
1420         }
1421 }
1422
1423 void
1424 kranal_process_fmaq (kra_conn_t *conn)
1425 {
1426         unsigned long flags;
1427         int           more_to_do;
1428         kra_tx_t     *tx;
1429         int           rc;
1430         int           expect_reply;
1431
1432         /* NB 1. kranal_sendmsg() may fail if I'm out of credits right now.
1433          *       However I will be rescheduled by an FMA completion event
1434          *       when I eventually get some.
1435          * NB 2. Sampling rac_state here races with setting it elsewhere.
1436          *       But it doesn't matter if I try to send a "real" message just
1437          *       as I start closing because I'll get scheduled to send the
1438          *       close anyway. */
1439
1440         /* Not racing with incoming message processing! */
1441         LASSERT (current == conn->rac_device->rad_scheduler);
1442
1443         if (conn->rac_state != RANAL_CONN_ESTABLISHED) {
1444                 if (!list_empty(&conn->rac_rdmaq)) {
1445                         /* RDMAs in progress */
1446                         LASSERT (!conn->rac_close_sent);
1447
1448                         if (time_after_eq(jiffies,
1449                                           conn->rac_last_tx +
1450                                           conn->rac_keepalive * HZ)) {
1451                                 CDEBUG(D_NET, "sending NOOP (rdma in progress)\n");
1452                                 kranal_init_msg(&conn->rac_msg, RANAL_MSG_NOOP);
1453                                 kranal_sendmsg(conn, &conn->rac_msg, NULL, 0);
1454                         }
1455                         return;
1456                 }
1457
1458                 if (conn->rac_close_sent)
1459                         return;
1460
1461                 CWARN("sending CLOSE to %s\n",
1462                       libcfs_nid2str(conn->rac_peer->rap_nid));
1463                 kranal_init_msg(&conn->rac_msg, RANAL_MSG_CLOSE);
1464                 rc = kranal_sendmsg(conn, &conn->rac_msg, NULL, 0);
1465                 if (rc != 0)
1466                         return;
1467
1468                 conn->rac_close_sent = 1;
1469                 if (!conn->rac_close_recvd)
1470                         return;
1471
1472                 write_lock_irqsave(&kranal_data.kra_global_lock, flags);
1473
1474                 if (conn->rac_state == RANAL_CONN_CLOSING)
1475                         kranal_terminate_conn_locked(conn);
1476
1477                 write_unlock_irqrestore(&kranal_data.kra_global_lock, flags);
1478                 return;
1479         }
1480
1481         spin_lock_irqsave(&conn->rac_lock, flags);
1482
1483         if (list_empty(&conn->rac_fmaq)) {
1484
1485                 spin_unlock_irqrestore(&conn->rac_lock, flags);
1486
1487                 if (time_after_eq(jiffies,
1488                                   conn->rac_last_tx + conn->rac_keepalive * HZ)) {
1489                         CDEBUG(D_NET, "sending NOOP -> %s (%p idle %lu(%ld))\n",
1490                                libcfs_nid2str(conn->rac_peer->rap_nid), conn,
1491                                (jiffies - conn->rac_last_tx)/HZ, conn->rac_keepalive);
1492                         kranal_init_msg(&conn->rac_msg, RANAL_MSG_NOOP);
1493                         kranal_sendmsg(conn, &conn->rac_msg, NULL, 0);
1494                 }
1495                 return;
1496         }
1497
1498         tx = list_entry(conn->rac_fmaq.next, kra_tx_t, tx_list);
1499         list_del(&tx->tx_list);
1500         more_to_do = !list_empty(&conn->rac_fmaq);
1501
1502         spin_unlock_irqrestore(&conn->rac_lock, flags);
1503
1504         expect_reply = 0;
1505         CDEBUG(D_NET, "sending regular msg: %p, type %02x, cookie "LPX64"\n",
1506                tx, tx->tx_msg.ram_type, tx->tx_cookie);
1507         switch (tx->tx_msg.ram_type) {
1508         default:
1509                 LBUG();
1510
1511         case RANAL_MSG_IMMEDIATE:
1512                 rc = kranal_sendmsg(conn, &tx->tx_msg,
1513                                     tx->tx_buffer, tx->tx_nob);
1514                 break;
1515
1516         case RANAL_MSG_PUT_NAK:
1517         case RANAL_MSG_PUT_DONE:
1518         case RANAL_MSG_GET_NAK:
1519         case RANAL_MSG_GET_DONE:
1520                 rc = kranal_sendmsg(conn, &tx->tx_msg, NULL, 0);
1521                 break;
1522
1523         case RANAL_MSG_PUT_REQ:
1524                 rc = kranal_map_buffer(tx);
1525                 LASSERT (rc != -EAGAIN);
1526                 if (rc != 0)
1527                         break;
1528
1529                 tx->tx_msg.ram_u.putreq.raprm_cookie = tx->tx_cookie;
1530                 rc = kranal_sendmsg(conn, &tx->tx_msg, NULL, 0);
1531                 expect_reply = 1;
1532                 break;
1533
1534         case RANAL_MSG_PUT_ACK:
1535                 rc = kranal_sendmsg(conn, &tx->tx_msg, NULL, 0);
1536                 expect_reply = 1;
1537                 break;
1538
1539         case RANAL_MSG_GET_REQ:
1540                 rc = kranal_map_buffer(tx);
1541                 LASSERT (rc != -EAGAIN);
1542                 if (rc != 0)
1543                         break;
1544
1545                 tx->tx_msg.ram_u.get.ragm_cookie = tx->tx_cookie;
1546                 tx->tx_msg.ram_u.get.ragm_desc.rard_key = tx->tx_map_key;
1547                 tx->tx_msg.ram_u.get.ragm_desc.rard_addr.AddressBits =
1548                         (__u64)((unsigned long)tx->tx_buffer);
1549                 tx->tx_msg.ram_u.get.ragm_desc.rard_nob = tx->tx_nob;
1550                 rc = kranal_sendmsg(conn, &tx->tx_msg, NULL, 0);
1551                 expect_reply = 1;
1552                 break;
1553         }
1554
1555         if (rc == -EAGAIN) {
1556                 /* I need credits to send this.  Replace tx at the head of the
1557                  * fmaq and I'll get rescheduled when credits appear */
1558                 CDEBUG(D_NET, "EAGAIN on %p\n", conn);
1559                 spin_lock_irqsave(&conn->rac_lock, flags);
1560                 list_add(&tx->tx_list, &conn->rac_fmaq);
1561                 spin_unlock_irqrestore(&conn->rac_lock, flags);
1562                 return;
1563         }
1564
1565         if (!expect_reply || rc != 0) {
1566                 kranal_tx_done(tx, rc);
1567         } else {
1568                 /* LASSERT(current) above ensures this doesn't race with reply
1569                  * processing */
1570                 spin_lock_irqsave(&conn->rac_lock, flags);
1571                 list_add_tail(&tx->tx_list, &conn->rac_replyq);
1572                 tx->tx_qtime = jiffies;
1573                 spin_unlock_irqrestore(&conn->rac_lock, flags);
1574         }
1575
1576         if (more_to_do) {
1577                 CDEBUG(D_NET, "Rescheduling %p (more to do)\n", conn);
1578                 kranal_schedule_conn(conn);
1579         }
1580 }
1581
1582 static inline void
1583 kranal_swab_rdma_desc (kra_rdma_desc_t *d)
1584 {
1585         __swab64s(&d->rard_key.Key);
1586         __swab16s(&d->rard_key.Cookie);
1587         __swab16s(&d->rard_key.MdHandle);
1588         __swab32s(&d->rard_key.Flags);
1589         __swab64s(&d->rard_addr.AddressBits);
1590         __swab32s(&d->rard_nob);
1591 }
1592
1593 kra_tx_t *
1594 kranal_match_reply(kra_conn_t *conn, int type, __u64 cookie)
1595 {
1596         struct list_head *ttmp;
1597         kra_tx_t         *tx;
1598         unsigned long     flags;
1599
1600         spin_lock_irqsave(&conn->rac_lock, flags);
1601
1602         list_for_each(ttmp, &conn->rac_replyq) {
1603                 tx = list_entry(ttmp, kra_tx_t, tx_list);
1604
1605                 CDEBUG(D_NET,"Checking %p %02x/"LPX64"\n",
1606                        tx, tx->tx_msg.ram_type, tx->tx_cookie);
1607
1608                 if (tx->tx_cookie != cookie)
1609                         continue;
1610
1611                 if (tx->tx_msg.ram_type != type) {
1612                         spin_unlock_irqrestore(&conn->rac_lock, flags);
1613                         CWARN("Unexpected type %x (%x expected) "
1614                               "matched reply from %s\n",
1615                               tx->tx_msg.ram_type, type,
1616                               libcfs_nid2str(conn->rac_peer->rap_nid));
1617                         return NULL;
1618                 }
1619
1620                 list_del(&tx->tx_list);
1621                 spin_unlock_irqrestore(&conn->rac_lock, flags);
1622                 return tx;
1623         }
1624
1625         spin_unlock_irqrestore(&conn->rac_lock, flags);
1626         CWARN("Unmatched reply %02x/"LPX64" from %s\n",
1627               type, cookie, libcfs_nid2str(conn->rac_peer->rap_nid));
1628         return NULL;
1629 }
1630
1631 void
1632 kranal_check_fma_rx (kra_conn_t *conn)
1633 {
1634         unsigned long flags;
1635         __u32         seq;
1636         kra_tx_t     *tx;
1637         kra_msg_t    *msg;
1638         void         *prefix;
1639         RAP_RETURN    rrc = RapkFmaGetPrefix(conn->rac_rihandle, &prefix);
1640         kra_peer_t   *peer = conn->rac_peer;
1641         int           rc = 0;
1642         int           repost = 1;
1643
1644         if (rrc == RAP_NOT_DONE)
1645                 return;
1646
1647         CDEBUG(D_NET, "RX on %p\n", conn);
1648
1649         LASSERT (rrc == RAP_SUCCESS);
1650         conn->rac_last_rx = jiffies;
1651         seq = conn->rac_rx_seq++;
1652         msg = (kra_msg_t *)prefix;
1653
1654         /* stash message for portals callbacks they'll NULL
1655          * rac_rxmsg if they consume it */
1656         LASSERT (conn->rac_rxmsg == NULL);
1657         conn->rac_rxmsg = msg;
1658
1659         if (msg->ram_magic != RANAL_MSG_MAGIC) {
1660                 if (__swab32(msg->ram_magic) != RANAL_MSG_MAGIC) {
1661                         CERROR("Unexpected magic %08x from %s\n",
1662                                msg->ram_magic, libcfs_nid2str(peer->rap_nid));
1663                         rc = -EPROTO;
1664                         goto out;
1665                 }
1666
1667                 __swab32s(&msg->ram_magic);
1668                 __swab16s(&msg->ram_version);
1669                 __swab16s(&msg->ram_type);
1670                 __swab64s(&msg->ram_srcnid);
1671                 __swab64s(&msg->ram_connstamp);
1672                 __swab32s(&msg->ram_seq);
1673
1674                 /* NB message type checked below; NOT here... */
1675                 switch (msg->ram_type) {
1676                 case RANAL_MSG_PUT_ACK:
1677                         kranal_swab_rdma_desc(&msg->ram_u.putack.rapam_desc);
1678                         break;
1679
1680                 case RANAL_MSG_GET_REQ:
1681                         kranal_swab_rdma_desc(&msg->ram_u.get.ragm_desc);
1682                         break;
1683
1684                 default:
1685                         break;
1686                 }
1687         }
1688
1689         if (msg->ram_version != RANAL_MSG_VERSION) {
1690                 CERROR("Unexpected protocol version %d from %s\n",
1691                        msg->ram_version, libcfs_nid2str(peer->rap_nid));
1692                 rc = -EPROTO;
1693                 goto out;
1694         }
1695
1696         if (msg->ram_srcnid != peer->rap_nid) {
1697                 CERROR("Unexpected peer %s from %s\n",
1698                        libcfs_nid2str(msg->ram_srcnid),
1699                        libcfs_nid2str(peer->rap_nid));
1700                 rc = -EPROTO;
1701                 goto out;
1702         }
1703
1704         if (msg->ram_connstamp != conn->rac_peer_connstamp) {
1705                 CERROR("Unexpected connstamp "LPX64"("LPX64
1706                        " expected) from %s\n",
1707                        msg->ram_connstamp, conn->rac_peer_connstamp,
1708                        libcfs_nid2str(peer->rap_nid));
1709                 rc = -EPROTO;
1710                 goto out;
1711         }
1712
1713         if (msg->ram_seq != seq) {
1714                 CERROR("Unexpected sequence number %d(%d expected) from %s\n",
1715                        msg->ram_seq, seq, libcfs_nid2str(peer->rap_nid));
1716                 rc = -EPROTO;
1717                 goto out;
1718         }
1719
1720         if ((msg->ram_type & RANAL_MSG_FENCE) != 0) {
1721                 /* This message signals RDMA completion... */
1722                 rrc = RapkFmaSyncWait(conn->rac_rihandle);
1723                 if (rrc != RAP_SUCCESS) {
1724                         CERROR("RapkFmaSyncWait failed: %d\n", rrc);
1725                         rc = -ENETDOWN;
1726                         goto out;
1727                 }
1728         }
1729
1730         if (conn->rac_close_recvd) {
1731                 CERROR("Unexpected message %d after CLOSE from %s\n",
1732                        msg->ram_type, libcfs_nid2str(conn->rac_peer->rap_nid));
1733                 rc = -EPROTO;
1734                 goto out;
1735         }
1736
1737         if (msg->ram_type == RANAL_MSG_CLOSE) {
1738                 CWARN("RX CLOSE from %s\n", libcfs_nid2str(conn->rac_peer->rap_nid));
1739                 conn->rac_close_recvd = 1;
1740                 write_lock_irqsave(&kranal_data.kra_global_lock, flags);
1741
1742                 if (conn->rac_state == RANAL_CONN_ESTABLISHED)
1743                         kranal_close_conn_locked(conn, 0);
1744                 else if (conn->rac_state == RANAL_CONN_CLOSING &&
1745                          conn->rac_close_sent)
1746                         kranal_terminate_conn_locked(conn);
1747
1748                 write_unlock_irqrestore(&kranal_data.kra_global_lock, flags);
1749                 goto out;
1750         }
1751
1752         if (conn->rac_state != RANAL_CONN_ESTABLISHED)
1753                 goto out;
1754
1755         switch (msg->ram_type) {
1756         case RANAL_MSG_NOOP:
1757                 /* Nothing to do; just a keepalive */
1758                 CDEBUG(D_NET, "RX NOOP on %p\n", conn);
1759                 break;
1760
1761         case RANAL_MSG_IMMEDIATE:
1762                 CDEBUG(D_NET, "RX IMMEDIATE on %p\n", conn);
1763                 rc = lnet_parse(kranal_data.kra_ni, &msg->ram_u.immediate.raim_hdr,
1764                                 msg->ram_srcnid, conn, 0);
1765                 repost = rc < 0;
1766                 break;
1767
1768         case RANAL_MSG_PUT_REQ:
1769                 CDEBUG(D_NET, "RX PUT_REQ on %p\n", conn);
1770                 rc = lnet_parse(kranal_data.kra_ni, &msg->ram_u.putreq.raprm_hdr,
1771                                 msg->ram_srcnid, conn, 1);
1772                 repost = rc < 0;
1773                 break;
1774
1775         case RANAL_MSG_PUT_NAK:
1776                 CDEBUG(D_NET, "RX PUT_NAK on %p\n", conn);
1777                 tx = kranal_match_reply(conn, RANAL_MSG_PUT_REQ,
1778                                         msg->ram_u.completion.racm_cookie);
1779                 if (tx == NULL)
1780                         break;
1781
1782                 LASSERT (tx->tx_buftype == RANAL_BUF_PHYS_MAPPED ||
1783                          tx->tx_buftype == RANAL_BUF_VIRT_MAPPED);
1784                 kranal_tx_done(tx, -ENOENT);    /* no match */
1785                 break;
1786
1787         case RANAL_MSG_PUT_ACK:
1788                 CDEBUG(D_NET, "RX PUT_ACK on %p\n", conn);
1789                 tx = kranal_match_reply(conn, RANAL_MSG_PUT_REQ,
1790                                         msg->ram_u.putack.rapam_src_cookie);
1791                 if (tx == NULL)
1792                         break;
1793
1794                 kranal_rdma(tx, RANAL_MSG_PUT_DONE,
1795                             &msg->ram_u.putack.rapam_desc,
1796                             msg->ram_u.putack.rapam_desc.rard_nob,
1797                             msg->ram_u.putack.rapam_dst_cookie);
1798                 break;
1799
1800         case RANAL_MSG_PUT_DONE:
1801                 CDEBUG(D_NET, "RX PUT_DONE on %p\n", conn);
1802                 tx = kranal_match_reply(conn, RANAL_MSG_PUT_ACK,
1803                                         msg->ram_u.completion.racm_cookie);
1804                 if (tx == NULL)
1805                         break;
1806
1807                 LASSERT (tx->tx_buftype == RANAL_BUF_PHYS_MAPPED ||
1808                          tx->tx_buftype == RANAL_BUF_VIRT_MAPPED);
1809                 kranal_tx_done(tx, 0);
1810                 break;
1811
1812         case RANAL_MSG_GET_REQ:
1813                 CDEBUG(D_NET, "RX GET_REQ on %p\n", conn);
1814                 rc = lnet_parse(kranal_data.kra_ni, &msg->ram_u.get.ragm_hdr,
1815                                 msg->ram_srcnid, conn, 1);
1816                 repost = rc < 0;
1817                 break;
1818
1819         case RANAL_MSG_GET_NAK:
1820                 CDEBUG(D_NET, "RX GET_NAK on %p\n", conn);
1821                 tx = kranal_match_reply(conn, RANAL_MSG_GET_REQ,
1822                                         msg->ram_u.completion.racm_cookie);
1823                 if (tx == NULL)
1824                         break;
1825
1826                 LASSERT (tx->tx_buftype == RANAL_BUF_PHYS_MAPPED ||
1827                          tx->tx_buftype == RANAL_BUF_VIRT_MAPPED);
1828                 kranal_tx_done(tx, -ENOENT);    /* no match */
1829                 break;
1830
1831         case RANAL_MSG_GET_DONE:
1832                 CDEBUG(D_NET, "RX GET_DONE on %p\n", conn);
1833                 tx = kranal_match_reply(conn, RANAL_MSG_GET_REQ,
1834                                         msg->ram_u.completion.racm_cookie);
1835                 if (tx == NULL)
1836                         break;
1837
1838                 LASSERT (tx->tx_buftype == RANAL_BUF_PHYS_MAPPED ||
1839                          tx->tx_buftype == RANAL_BUF_VIRT_MAPPED);
1840 #if 0
1841                 /* completion message should send rdma length if we ever allow
1842                  * GET truncation */
1843                 lnet_set_reply_msg_len(kranal_data.kra_ni, tx->tx_lntmsg[1], ???);
1844 #endif
1845                 kranal_tx_done(tx, 0);
1846                 break;
1847         }
1848
1849  out:
1850         if (rc < 0)                             /* protocol/comms error */
1851                 kranal_close_conn (conn, rc);
1852
1853         if (repost && conn->rac_rxmsg != NULL)
1854                 kranal_consume_rxmsg(conn, NULL, 0);
1855
1856         /* check again later */
1857         kranal_schedule_conn(conn);
1858 }
1859
1860 void
1861 kranal_complete_closed_conn (kra_conn_t *conn)
1862 {
1863         kra_tx_t   *tx;
1864         int         nfma;
1865         int         nreplies;
1866
1867         LASSERT (conn->rac_state == RANAL_CONN_CLOSED);
1868         LASSERT (list_empty(&conn->rac_list));
1869         LASSERT (list_empty(&conn->rac_hashlist));
1870
1871         for (nfma = 0; !list_empty(&conn->rac_fmaq); nfma++) {
1872                 tx = list_entry(conn->rac_fmaq.next, kra_tx_t, tx_list);
1873
1874                 list_del(&tx->tx_list);
1875                 kranal_tx_done(tx, -ECONNABORTED);
1876         }
1877
1878         LASSERT (list_empty(&conn->rac_rdmaq));
1879
1880         for (nreplies = 0; !list_empty(&conn->rac_replyq); nreplies++) {
1881                 tx = list_entry(conn->rac_replyq.next, kra_tx_t, tx_list);
1882
1883                 list_del(&tx->tx_list);
1884                 kranal_tx_done(tx, -ECONNABORTED);
1885         }
1886
1887         CWARN("Closed conn %p -> %s: nmsg %d nreplies %d\n",
1888                conn, libcfs_nid2str(conn->rac_peer->rap_nid), nfma, nreplies);
1889 }
1890
1891 int
1892 kranal_process_new_conn (kra_conn_t *conn)
1893 {
1894         RAP_RETURN   rrc;
1895
1896         rrc = RapkCompleteSync(conn->rac_rihandle, 1);
1897         if (rrc == RAP_SUCCESS)
1898                 return 0;
1899
1900         LASSERT (rrc == RAP_NOT_DONE);
1901         if (!time_after_eq(jiffies, conn->rac_last_tx +
1902                            conn->rac_timeout * HZ))
1903                 return -EAGAIN;
1904
1905         /* Too late */
1906         rrc = RapkCompleteSync(conn->rac_rihandle, 0);
1907         LASSERT (rrc == RAP_SUCCESS);
1908         return -ETIMEDOUT;
1909 }
1910
1911 int
1912 kranal_scheduler (void *arg)
1913 {
1914         kra_device_t     *dev = (kra_device_t *)arg;
1915         wait_queue_t      wait;
1916         char              name[16];
1917         kra_conn_t       *conn;
1918         unsigned long     flags;
1919         unsigned long     deadline;
1920         unsigned long     soonest;
1921         int               nsoonest;
1922         long              timeout;
1923         struct list_head *tmp;
1924         struct list_head *nxt;
1925         int               rc;
1926         int               dropped_lock;
1927         int               busy_loops = 0;
1928
1929         snprintf(name, sizeof(name), "kranal_sd_%02d", dev->rad_idx);
1930         cfs_daemonize(name);
1931         cfs_block_allsigs();
1932
1933         dev->rad_scheduler = current;
1934         init_waitqueue_entry(&wait, current);
1935
1936         spin_lock_irqsave(&dev->rad_lock, flags);
1937
1938         while (!kranal_data.kra_shutdown) {
1939                 /* Safe: kra_shutdown only set when quiescent */
1940
1941                 if (busy_loops++ >= RANAL_RESCHED) {
1942                         spin_unlock_irqrestore(&dev->rad_lock, flags);
1943
1944                         our_cond_resched();
1945                         busy_loops = 0;
1946
1947                         spin_lock_irqsave(&dev->rad_lock, flags);
1948                 }
1949
1950                 dropped_lock = 0;
1951
1952                 if (dev->rad_ready) {
1953                         /* Device callback fired since I last checked it */
1954                         dev->rad_ready = 0;
1955                         spin_unlock_irqrestore(&dev->rad_lock, flags);
1956                         dropped_lock = 1;
1957
1958                         kranal_check_rdma_cq(dev);
1959                         kranal_check_fma_cq(dev);
1960
1961                         spin_lock_irqsave(&dev->rad_lock, flags);
1962                 }
1963
1964                 list_for_each_safe(tmp, nxt, &dev->rad_ready_conns) {
1965                         conn = list_entry(tmp, kra_conn_t, rac_schedlist);
1966
1967                         list_del_init(&conn->rac_schedlist);
1968                         LASSERT (conn->rac_scheduled);
1969                         conn->rac_scheduled = 0;
1970                         spin_unlock_irqrestore(&dev->rad_lock, flags);
1971                         dropped_lock = 1;
1972
1973                         kranal_check_fma_rx(conn);
1974                         kranal_process_fmaq(conn);
1975
1976                         if (conn->rac_state == RANAL_CONN_CLOSED)
1977                                 kranal_complete_closed_conn(conn);
1978
1979                         kranal_conn_decref(conn);
1980                         spin_lock_irqsave(&dev->rad_lock, flags);
1981                 }
1982
1983                 nsoonest = 0;
1984                 soonest = jiffies;
1985
1986                 list_for_each_safe(tmp, nxt, &dev->rad_new_conns) {
1987                         conn = list_entry(tmp, kra_conn_t, rac_schedlist);
1988
1989                         deadline = conn->rac_last_tx + conn->rac_keepalive;
1990                         if (time_after_eq(jiffies, deadline)) {
1991                                 /* Time to process this new conn */
1992                                 spin_unlock_irqrestore(&dev->rad_lock, flags);
1993                                 dropped_lock = 1;
1994
1995                                 rc = kranal_process_new_conn(conn);
1996                                 if (rc != -EAGAIN) {
1997                                         /* All done with this conn */
1998                                         spin_lock_irqsave(&dev->rad_lock, flags);
1999                                         list_del_init(&conn->rac_schedlist);
2000                                         spin_unlock_irqrestore(&dev->rad_lock, flags);
2001
2002                                         kranal_conn_decref(conn);
2003                                         spin_lock_irqsave(&dev->rad_lock, flags);
2004                                         continue;
2005                                 }
2006
2007                                 /* retry with exponential backoff until HZ */
2008                                 if (conn->rac_keepalive == 0)
2009                                         conn->rac_keepalive = 1;
2010                                 else if (conn->rac_keepalive <= HZ)
2011                                         conn->rac_keepalive *= 2;
2012                                 else
2013                                         conn->rac_keepalive += HZ;
2014
2015                                 deadline = conn->rac_last_tx + conn->rac_keepalive;
2016                                 spin_lock_irqsave(&dev->rad_lock, flags);
2017                         }
2018
2019                         /* Does this conn need attention soonest? */
2020                         if (nsoonest++ == 0 ||
2021                             !time_after_eq(deadline, soonest))
2022                                 soonest = deadline;
2023                 }
2024
2025                 if (dropped_lock)               /* may sleep iff I didn't drop the lock */
2026                         continue;
2027
2028                 set_current_state(TASK_INTERRUPTIBLE);
2029                 add_wait_queue_exclusive(&dev->rad_waitq, &wait);
2030                 spin_unlock_irqrestore(&dev->rad_lock, flags);
2031
2032                 if (nsoonest == 0) {
2033                         busy_loops = 0;
2034                         schedule();
2035                 } else {
2036                         timeout = (long)(soonest - jiffies);
2037                         if (timeout > 0) {
2038                                 busy_loops = 0;
2039                                 schedule_timeout(timeout);
2040                         }
2041                 }
2042
2043                 remove_wait_queue(&dev->rad_waitq, &wait);
2044                 set_current_state(TASK_RUNNING);
2045                 spin_lock_irqsave(&dev->rad_lock, flags);
2046         }
2047
2048         spin_unlock_irqrestore(&dev->rad_lock, flags);
2049
2050         dev->rad_scheduler = NULL;
2051         kranal_thread_fini();
2052         return 0;
2053 }