lnet/klnds/ptllnd/ptllnd_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) 2007, 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/ptllnd/ptllnd_cb.c
  37  *
  38  * Author: PJ Kirner <pjkirner@clusterfs.com>
  39  */
  40
  41 #include "ptllnd.h"
  42
  43 #ifndef _USING_LUSTRE_PORTALS_
  44 int
  45 kptllnd_extract_iov (int dst_niov, ptl_md_iovec_t *dst,
  46                      int src_niov, struct iovec *src,
  47                      unsigned int offset, unsigned int len)
  48 {
  49         /* Initialise 'dst' to the subset of 'src' starting at 'offset',
  50          * for exactly 'len' bytes, and return the number of entries.
  51          * NB not destructive to 'src' */
  52         unsigned int    frag_len;
  53         unsigned int    niov;
  54
  55         if (len == 0)                           /* no data => */
  56                 return (0);                     /* no frags */
  57
  58         LASSERT (src_niov > 0);
  59         while (offset >= src->iov_len) {      /* skip initial frags */
  60                 offset -= src->iov_len;
  61                 src_niov--;
  62                 src++;
  63                 LASSERT (src_niov > 0);
  64         }
  65
  66         niov = 1;
  67         for (;;) {
  68                 LASSERT (src_niov > 0);
  69                 LASSERT (niov <= dst_niov);
  70
  71                 frag_len = src->iov_len - offset;
  72                 dst->iov_base = ((char *)src->iov_base) + offset;
  73
  74                 if (len <= frag_len) {
  75                         dst->iov_len = len;
  76                         return (niov);
  77                 }
  78
  79                 dst->iov_len = frag_len;
  80
  81                 len -= frag_len;
  82                 dst++;
  83                 src++;
  84                 niov++;
  85                 src_niov--;
  86                 offset = 0;
  87         }
  88 }
  89
  90 int
  91 kptllnd_extract_phys (int dst_niov, ptl_md_iovec_t *dst,
  92                       int src_niov, lnet_kiov_t *src,
  93                       unsigned int offset, unsigned int len)
  94 {
  95         /* Initialise 'dst' to the physical addresses of the subset of 'src'
  96          * starting at 'offset', for exactly 'len' bytes, and return the number
  97          * of entries.  NB not destructive to 'src' */
  98         unsigned int    frag_len;
  99         unsigned int    niov;
 100         __u64           phys_page;
 101         __u64           phys;
 102
 103         if (len == 0)                           /* no data => */
 104                 return (0);                     /* no frags */
 105
 106         LASSERT (src_niov > 0);
 107         while (offset >= src->kiov_len) {      /* skip initial frags */
 108                 offset -= src->kiov_len;
 109                 src_niov--;
 110                 src++;
 111                 LASSERT (src_niov > 0);
 112         }
 113
 114         niov = 1;
 115         for (;;) {
 116                 LASSERT (src_niov > 0);
 117                 LASSERT (niov <= dst_niov);
 118
 119                 frag_len = min(src->kiov_len - offset, len);
 120                 phys_page = lnet_page2phys(src->kiov_page);
 121                 phys = phys_page + src->kiov_offset + offset;
 122
 123                 LASSERT (sizeof(void *) > 4 ||
 124                          (phys <= 0xffffffffULL &&
 125                           phys + (frag_len - 1) <= 0xffffffffULL));
 126
 127                 dst->iov_base = (void *)((unsigned long)phys);
 128                 dst->iov_len = frag_len;
 129
 130                 if (frag_len == len)
 131                         return niov;
 132
 133                 len -= frag_len;
 134                 dst++;
 135                 src++;
 136                 niov++;
 137                 src_niov--;
 138                 offset = 0;
 139         }
 140 }
 141 #endif
 142
 143 void
 144 kptllnd_init_rdma_md(kptl_tx_t *tx, unsigned int niov,
 145                      struct iovec *iov, lnet_kiov_t *kiov,
 146                      unsigned int offset, unsigned int nob)
 147 {
 148         LASSERT (iov == NULL || kiov == NULL);
 149
 150         memset(&tx->tx_rdma_md, 0, sizeof(tx->tx_rdma_md));
 151
 152         tx->tx_rdma_md.start     = tx->tx_frags;
 153         tx->tx_rdma_md.user_ptr  = &tx->tx_rdma_eventarg;
 154         tx->tx_rdma_md.eq_handle = kptllnd_data.kptl_eqh;
 155         tx->tx_rdma_md.options   = PTL_MD_LUSTRE_COMPLETION_SEMANTICS |
 156                                    PTL_MD_EVENT_START_DISABLE;
 157         switch (tx->tx_type) {
 158         default:
 159                 LBUG();
 160
 161         case TX_TYPE_PUT_REQUEST:               /* passive: peer gets */
 162                 tx->tx_rdma_md.threshold = 1;   /* GET event */
 163                 tx->tx_rdma_md.options |= PTL_MD_OP_GET;
 164                 break;
 165
 166         case TX_TYPE_GET_REQUEST:               /* passive: peer puts */
 167                 tx->tx_rdma_md.threshold = 1;   /* PUT event */
 168                 tx->tx_rdma_md.options |= PTL_MD_OP_PUT;
 169                 break;
 170
 171         case TX_TYPE_PUT_RESPONSE:              /* active: I get */
 172                 tx->tx_rdma_md.threshold = 2;   /* SEND + REPLY */
 173                 break;
 174
 175         case TX_TYPE_GET_RESPONSE:              /* active: I put */
 176                 tx->tx_rdma_md.threshold = tx->tx_acked ? 2 : 1;   /* SEND + ACK? */
 177                 break;
 178         }
 179
 180         if (nob == 0) {
 181                 tx->tx_rdma_md.length = 0;
 182                 return;
 183         }
 184
 185 #ifdef _USING_LUSTRE_PORTALS_
 186         if (iov != NULL) {
 187                 tx->tx_rdma_md.options |= PTL_MD_IOVEC;
 188                 tx->tx_rdma_md.length =
 189                         lnet_extract_iov(PTL_MD_MAX_IOV, tx->tx_frags->iov,
 190                                          niov, iov, offset, nob);
 191                 return;
 192         }
 193
 194         /* Cheating OK since ptl_kiov_t == lnet_kiov_t */
 195         CLASSERT(sizeof(ptl_kiov_t) == sizeof(lnet_kiov_t));
 196         CLASSERT(offsetof(ptl_kiov_t, kiov_offset) ==
 197                  offsetof(lnet_kiov_t, kiov_offset));
 198         CLASSERT(offsetof(ptl_kiov_t, kiov_page) ==
 199                  offsetof(lnet_kiov_t, kiov_page));
 200         CLASSERT(offsetof(ptl_kiov_t, kiov_len) ==
 201                  offsetof(lnet_kiov_t, kiov_len));
 202
 203         tx->tx_rdma_md.options |= PTL_MD_KIOV;
 204         tx->tx_rdma_md.length =
 205                 lnet_extract_kiov(PTL_MD_MAX_IOV, tx->tx_frags->kiov,
 206                                   niov, kiov, offset, nob);
 207 #else
 208         if (iov != NULL) {
 209                 tx->tx_rdma_md.options |= PTL_MD_IOVEC;
 210                 tx->tx_rdma_md.length =
 211                         kptllnd_extract_iov(PTL_MD_MAX_IOV, tx->tx_frags->iov,
 212                                             niov, iov, offset, nob);
 213                 return;
 214         }
 215
 216         tx->tx_rdma_md.options |= PTL_MD_IOVEC | PTL_MD_PHYS;
 217         tx->tx_rdma_md.length =
 218                 kptllnd_extract_phys(PTL_MD_MAX_IOV, tx->tx_frags->iov,
 219                                      niov, kiov, offset, nob);
 220 #endif
 221 }
 222
 223 int
 224 kptllnd_active_rdma(kptl_rx_t *rx, lnet_msg_t *lntmsg, int type,
 225                     unsigned int niov, struct iovec *iov, lnet_kiov_t *kiov,
 226                     unsigned int offset, int nob)
 227 {
 228         kptl_tx_t       *tx;
 229         ptl_err_t        ptlrc;
 230         kptl_msg_t      *rxmsg = rx->rx_msg;
 231         kptl_peer_t     *peer = rx->rx_peer;
 232         unsigned long    flags;
 233         ptl_handle_md_t  mdh;
 234
 235         LASSERT (type == TX_TYPE_PUT_RESPONSE ||
 236                  type == TX_TYPE_GET_RESPONSE);
 237
 238         tx = kptllnd_get_idle_tx(type);
 239         if (tx == NULL) {
 240                 CERROR ("Can't do %s rdma to %s: can't allocate descriptor\n",
 241                         type == TX_TYPE_PUT_RESPONSE ? "GET" : "PUT",
 242                         libcfs_id2str(peer->peer_id));
 243                 return -ENOMEM;
 244         }
 245
 246         kptllnd_set_tx_peer(tx, peer);
 247         kptllnd_init_rdma_md(tx, niov, iov, kiov, offset, nob);
 248
 249         ptlrc = PtlMDBind(kptllnd_data.kptl_nih, tx->tx_rdma_md,
 250                           PTL_UNLINK, &mdh);
 251         if (ptlrc != PTL_OK) {
 252                 CERROR("PtlMDBind(%s) failed: %s(%d)\n",
 253                        libcfs_id2str(peer->peer_id),
 254                        kptllnd_errtype2str(ptlrc), ptlrc);
 255                 tx->tx_status = -EIO;
 256                 kptllnd_tx_decref(tx);
 257                 return -EIO;
 258         }
 259
 260         spin_lock_irqsave(&peer->peer_lock, flags);
 261
 262         tx->tx_lnet_msg = lntmsg;
 263         /* lnet_finalize() will be called when tx is torn down, so I must
 264          * return success from here on... */
 265
 266         tx->tx_deadline = jiffies + (*kptllnd_tunables.kptl_timeout * HZ);
 267         tx->tx_rdma_mdh = mdh;
 268         tx->tx_active = 1;
 269         cfs_list_add_tail(&tx->tx_list, &peer->peer_activeq);
 270
 271         /* peer has now got my ref on 'tx' */
 272
 273         spin_unlock_irqrestore(&peer->peer_lock, flags);
 274
 275         tx->tx_tposted = jiffies;
 276
 277         if (type == TX_TYPE_GET_RESPONSE)
 278                 ptlrc = PtlPut(mdh,
 279                                tx->tx_acked ? PTL_ACK_REQ : PTL_NOACK_REQ,
 280                                rx->rx_initiator,
 281                                *kptllnd_tunables.kptl_portal,
 282                                0,                     /* acl cookie */
 283                                rxmsg->ptlm_u.rdma.kptlrm_matchbits,
 284                                0,                     /* offset */
 285                                (lntmsg != NULL) ?     /* header data */
 286                                PTLLND_RDMA_OK :
 287                                PTLLND_RDMA_FAIL);
 288         else
 289                 ptlrc = PtlGet(mdh,
 290                                rx->rx_initiator,
 291                                *kptllnd_tunables.kptl_portal,
 292                                0,                     /* acl cookie */
 293                                rxmsg->ptlm_u.rdma.kptlrm_matchbits,
 294                                0);                    /* offset */
 295
 296         if (ptlrc != PTL_OK) {
 297                 CERROR("Ptl%s failed: %s(%d)\n",
 298                        (type == TX_TYPE_GET_RESPONSE) ? "Put" : "Get",
 299                        kptllnd_errtype2str(ptlrc), ptlrc);
 300
 301                 kptllnd_peer_close(peer, -EIO);
 302                 /* Everything (including this RDMA) queued on the peer will
 303                  * be completed with failure */
 304         }
 305
 306         return 0;
 307 }
 308
 309 int
 310 kptllnd_send(lnet_ni_t *ni, void *private, lnet_msg_t *lntmsg)
 311 {
 312         lnet_hdr_t       *hdr = &lntmsg->msg_hdr;
 313         int               type = lntmsg->msg_type;
 314         lnet_process_id_t target = lntmsg->msg_target;
 315         int               target_is_router = lntmsg->msg_target_is_router;
 316         int               routing = lntmsg->msg_routing;
 317         unsigned int      payload_niov = lntmsg->msg_niov;
 318         struct iovec     *payload_iov = lntmsg->msg_iov;
 319         lnet_kiov_t      *payload_kiov = lntmsg->msg_kiov;
 320         unsigned int      payload_offset = lntmsg->msg_offset;
 321         unsigned int      payload_nob = lntmsg->msg_len;
 322         kptl_net_t       *net = ni->ni_data;
 323         kptl_peer_t      *peer = NULL;
 324         int               mpflag = 0;
 325         kptl_tx_t        *tx;
 326         int               nob;
 327         int               nfrag;
 328         int               rc;
 329
 330         LASSERT (net->net_ni == ni);
 331         LASSERT (!net->net_shutdown);
 332         LASSERT (payload_nob == 0 || payload_niov > 0);
 333         LASSERT (payload_niov <= LNET_MAX_IOV);
 334         LASSERT (payload_niov <= PTL_MD_MAX_IOV); /* !!! */
 335         LASSERT (!(payload_kiov != NULL && payload_iov != NULL));
 336         LASSERT (!in_interrupt());
 337
 338         if (lntmsg->msg_vmflush)
 339                 mpflag = cfs_memory_pressure_get_and_set();
 340
 341         rc = kptllnd_find_target(net, target, &peer);
 342         if (rc != 0)
 343                 goto out;
 344
 345         /* NB peer->peer_id does NOT always equal target, be careful with
 346          * which one to use */
 347         switch (type) {
 348         default:
 349                 LBUG();
 350                 return -EINVAL;
 351
 352         case LNET_MSG_REPLY:
 353         case LNET_MSG_PUT:
 354                 /* Should the payload avoid RDMA? */
 355                 nob = offsetof(kptl_msg_t, ptlm_u.immediate.kptlim_payload[payload_nob]);
 356                 if (payload_kiov == NULL &&
 357                     nob <= peer->peer_max_msg_size)
 358                         break;
 359
 360                 tx = kptllnd_get_idle_tx(TX_TYPE_PUT_REQUEST);
 361                 if (tx == NULL) {
 362                         CERROR("Can't send %s to %s: can't allocate descriptor\n",
 363                                lnet_msgtyp2str(type),
 364                                libcfs_id2str(target));
 365                         rc = -ENOMEM;
 366                         goto out;
 367                 }
 368
 369                 kptllnd_init_rdma_md(tx, payload_niov,
 370                                      payload_iov, payload_kiov,
 371                                      payload_offset, payload_nob);
 372
 373                 tx->tx_lnet_msg = lntmsg;
 374                 tx->tx_msg->ptlm_u.rdma.kptlrm_hdr = *hdr;
 375                 kptllnd_init_msg (tx->tx_msg, PTLLND_MSG_TYPE_PUT,
 376                                  target, sizeof(kptl_rdma_msg_t));
 377
 378                 CDEBUG(D_NETTRACE, "%s: passive PUT p %d %p\n",
 379                        libcfs_id2str(target),
 380                        le32_to_cpu(lntmsg->msg_hdr.msg.put.ptl_index), tx);
 381
 382                 kptllnd_tx_launch(peer, tx, 0);
 383                 goto out;
 384
 385         case LNET_MSG_GET:
 386                 /* routed gets don't RDMA */
 387                 if (target_is_router || routing)
 388                         break;
 389
 390                 /* Is the payload small enough not to need RDMA? */
 391                 nob = lntmsg->msg_md->md_length;
 392                 nob = offsetof(kptl_msg_t,
 393                                ptlm_u.immediate.kptlim_payload[nob]);
 394                 if (nob <= peer->peer_max_msg_size)
 395                         break;
 396
 397                 tx = kptllnd_get_idle_tx(TX_TYPE_GET_REQUEST);
 398                 if (tx == NULL) {
 399                         CERROR("Can't send GET to %s: can't allocate descriptor\n",
 400                                libcfs_id2str(target));
 401                         rc = -ENOMEM;
 402                         goto out;
 403                 }
 404
 405                 tx->tx_lnet_replymsg = lnet_create_reply_msg(ni, lntmsg);
 406                 if (tx->tx_lnet_replymsg == NULL) {
 407                         CERROR("Failed to allocate LNET reply for %s\n",
 408                                libcfs_id2str(target));
 409                         kptllnd_tx_decref(tx);
 410                         rc = -ENOMEM;
 411                         goto out;
 412                 }
 413
 414                 if ((lntmsg->msg_md->md_options & LNET_MD_KIOV) == 0)
 415                         kptllnd_init_rdma_md(tx, lntmsg->msg_md->md_niov,
 416                                              lntmsg->msg_md->md_iov.iov, NULL,
 417                                              0, lntmsg->msg_md->md_length);
 418                 else
 419                         kptllnd_init_rdma_md(tx, lntmsg->msg_md->md_niov,
 420                                              NULL, lntmsg->msg_md->md_iov.kiov,
 421                                              0, lntmsg->msg_md->md_length);
 422
 423                 tx->tx_lnet_msg = lntmsg;
 424                 tx->tx_msg->ptlm_u.rdma.kptlrm_hdr = *hdr;
 425                 kptllnd_init_msg (tx->tx_msg, PTLLND_MSG_TYPE_GET,
 426                                  target, sizeof(kptl_rdma_msg_t));
 427
 428                 CDEBUG(D_NETTRACE, "%s: passive GET p %d %p\n",
 429                        libcfs_id2str(target),
 430                        le32_to_cpu(lntmsg->msg_hdr.msg.put.ptl_index), tx);
 431
 432                 kptllnd_tx_launch(peer, tx, 0);
 433                 goto out;
 434
 435         case LNET_MSG_ACK:
 436                 CDEBUG(D_NET, "LNET_MSG_ACK\n");
 437                 LASSERT (payload_nob == 0);
 438                 break;
 439         }
 440
 441         /* I don't have to handle kiovs */
 442         LASSERT (payload_nob == 0 || payload_iov != NULL);
 443
 444         tx = kptllnd_get_idle_tx(TX_TYPE_SMALL_MESSAGE);
 445         if (tx == NULL) {
 446                 CERROR("Can't send %s to %s: can't allocate descriptor\n",
 447                        lnet_msgtyp2str(type), libcfs_id2str(target));
 448                 rc = -ENOMEM;
 449                 goto out;
 450         }
 451
 452         tx->tx_lnet_msg = lntmsg;
 453         tx->tx_msg->ptlm_u.immediate.kptlim_hdr = *hdr;
 454
 455         if (payload_nob == 0) {
 456                 nfrag = 0;
 457         } else {
 458                 tx->tx_frags->iov[0].iov_base = tx->tx_msg;
 459                 tx->tx_frags->iov[0].iov_len = offsetof(kptl_msg_t,
 460                                                         ptlm_u.immediate.kptlim_payload);
 461
 462                 /* NB relying on lustre not asking for PTL_MD_MAX_IOV
 463                  * fragments!! */
 464 #ifdef _USING_LUSTRE_PORTALS_
 465                 nfrag = 1 + lnet_extract_iov(PTL_MD_MAX_IOV - 1,
 466                                              &tx->tx_frags->iov[1],
 467                                              payload_niov, payload_iov,
 468                                              payload_offset, payload_nob);
 469 #else
 470                 nfrag = 1 + kptllnd_extract_iov(PTL_MD_MAX_IOV - 1,
 471                                                 &tx->tx_frags->iov[1],
 472                                                 payload_niov, payload_iov,
 473                                                 payload_offset, payload_nob);
 474 #endif
 475         }
 476
 477         nob = offsetof(kptl_immediate_msg_t, kptlim_payload[payload_nob]);
 478         kptllnd_init_msg(tx->tx_msg, PTLLND_MSG_TYPE_IMMEDIATE, target, nob);
 479
 480         CDEBUG(D_NETTRACE, "%s: immediate %s p %d %p\n",
 481                libcfs_id2str(target),
 482                lnet_msgtyp2str(lntmsg->msg_type),
 483                (le32_to_cpu(lntmsg->msg_type) == LNET_MSG_PUT) ?
 484                le32_to_cpu(lntmsg->msg_hdr.msg.put.ptl_index) :
 485                (le32_to_cpu(lntmsg->msg_type) == LNET_MSG_GET) ?
 486                le32_to_cpu(lntmsg->msg_hdr.msg.get.ptl_index) : -1,
 487                tx);
 488
 489         kptllnd_tx_launch(peer, tx, nfrag);
 490
 491  out:
 492         if (lntmsg->msg_vmflush)
 493                 cfs_memory_pressure_restore(mpflag);
 494         if (peer)
 495                 kptllnd_peer_decref(peer);
 496         return rc;
 497 }
 498
 499 int
 500 kptllnd_eager_recv(struct lnet_ni *ni, void *private,
 501                    lnet_msg_t *msg, void **new_privatep)
 502 {
 503         kptl_rx_t        *rx = private;
 504
 505         CDEBUG(D_NET, "Eager RX=%p RXB=%p\n", rx, rx->rx_rxb);
 506
 507         /* I have to release my ref on rxb (if I have one) to ensure I'm an
 508          * eager receiver, so I copy the incoming request from the buffer it
 509          * landed in, into space reserved in the descriptor... */
 510
 511 #if (PTL_MD_LOCAL_ALIGN8 == 0)
 512         if (rx->rx_rxb == NULL)                 /* already copied */
 513                 return 0;                       /* to fix alignment */
 514 #else
 515         LASSERT(rx->rx_rxb != NULL);
 516 #endif
 517         LASSERT(rx->rx_nob <= *kptllnd_tunables.kptl_max_msg_size);
 518
 519         memcpy(rx->rx_space, rx->rx_msg, rx->rx_nob);
 520         rx->rx_msg = (kptl_msg_t *)rx->rx_space;
 521
 522         kptllnd_rx_buffer_decref(rx->rx_rxb);
 523         rx->rx_rxb = NULL;
 524
 525         return 0;
 526 }
 527
 528
 529 int
 530 kptllnd_recv (lnet_ni_t *ni, void *private, lnet_msg_t *lntmsg, int delayed,
 531               unsigned int niov, struct iovec *iov, lnet_kiov_t *kiov,
 532               unsigned int offset, unsigned int mlen, unsigned int rlen)
 533 {
 534         kptl_rx_t    *rx = private;
 535         kptl_msg_t   *rxmsg = rx->rx_msg;
 536         int           nob;
 537         int           rc;
 538
 539         CDEBUG(D_NET, "%s niov=%d offset=%d mlen=%d rlen=%d\n",
 540                kptllnd_msgtype2str(rxmsg->ptlm_type),
 541                niov, offset, mlen, rlen);
 542
 543         LASSERT (mlen <= rlen);
 544         LASSERT (mlen >= 0);
 545         LASSERT (!in_interrupt());
 546         LASSERT (!(kiov != NULL && iov != NULL)); /* never both */
 547         LASSERT (niov <= PTL_MD_MAX_IOV);       /* !!! */
 548
 549         switch(rxmsg->ptlm_type)
 550         {
 551         default:
 552                 LBUG();
 553                 rc = -EINVAL;
 554                 break;
 555
 556         case PTLLND_MSG_TYPE_IMMEDIATE:
 557                 CDEBUG(D_NET, "PTLLND_MSG_TYPE_IMMEDIATE %d,%d\n", mlen, rlen);
 558
 559                 nob = offsetof(kptl_msg_t, ptlm_u.immediate.kptlim_payload[rlen]);
 560                 if (nob > rx->rx_nob) {
 561                         CERROR ("Immediate message from %s too big: %d(%d)\n",
 562                                 libcfs_id2str(rx->rx_peer->peer_id), nob,
 563                                 rx->rx_nob);
 564                         rc = -EINVAL;
 565                         break;
 566                 }
 567
 568                 if (kiov != NULL)
 569                         lnet_copy_flat2kiov(
 570                                 niov, kiov, offset,
 571                                 *kptllnd_tunables.kptl_max_msg_size,
 572                                 rxmsg->ptlm_u.immediate.kptlim_payload,
 573                                 0,
 574                                 mlen);
 575                 else
 576                         lnet_copy_flat2iov(
 577                                 niov, iov, offset,
 578                                 *kptllnd_tunables.kptl_max_msg_size,
 579                                 rxmsg->ptlm_u.immediate.kptlim_payload,
 580                                 0,
 581                                 mlen);
 582
 583                 lnet_finalize (ni, lntmsg, 0);
 584                 rc = 0;
 585                 break;
 586
 587         case PTLLND_MSG_TYPE_GET:
 588                 CDEBUG(D_NET, "PTLLND_MSG_TYPE_GET %d,%d\n", mlen, rlen);
 589
 590                 /* NB always send RDMA so the peer can complete.  I send
 591                  * success/failure in the portals 'hdr_data' */
 592
 593                 if (lntmsg == NULL)
 594                         rc = kptllnd_active_rdma(rx, NULL,
 595                                                  TX_TYPE_GET_RESPONSE,
 596                                                  0, NULL, NULL, 0, 0);
 597                 else
 598                         rc = kptllnd_active_rdma(rx, lntmsg,
 599                                                  TX_TYPE_GET_RESPONSE,
 600                                                  lntmsg->msg_niov,
 601                                                  lntmsg->msg_iov,
 602                                                  lntmsg->msg_kiov,
 603                                                  lntmsg->msg_offset,
 604                                                  lntmsg->msg_len);
 605                 break;
 606
 607         case PTLLND_MSG_TYPE_PUT:
 608                 CDEBUG(D_NET, "PTLLND_MSG_TYPE_PUT %d,%d\n", mlen, rlen);
 609
 610                 /* NB always send RDMA so the peer can complete; it'll be 0
 611                  * bytes if there was no match (lntmsg == NULL). I have no way
 612                  * to let my peer know this, but she's only interested in when
 613                  * the net has stopped accessing her buffer in any case. */
 614
 615                 rc = kptllnd_active_rdma(rx, lntmsg, TX_TYPE_PUT_RESPONSE,
 616                                          niov, iov, kiov, offset, mlen);
 617                 break;
 618         }
 619
 620         /*
 621          * We're done with the RX
 622          */
 623         kptllnd_rx_done(rx, PTLLND_POSTRX_PEER_CREDIT);
 624         return rc;
 625 }
 626
 627 void
 628 kptllnd_eq_callback(ptl_event_t *ev)
 629 {
 630         kptl_eventarg_t *eva = ev->md.user_ptr;
 631
 632         switch (eva->eva_type) {
 633         default:
 634                 LBUG();
 635
 636         case PTLLND_EVENTARG_TYPE_MSG:
 637         case PTLLND_EVENTARG_TYPE_RDMA:
 638                 kptllnd_tx_callback(ev);
 639                 break;
 640
 641         case PTLLND_EVENTARG_TYPE_BUF:
 642                 kptllnd_rx_buffer_callback(ev);
 643                 break;
 644         }
 645 }
 646
 647 void
 648 kptllnd_thread_fini (void)
 649 {
 650         cfs_atomic_dec(&kptllnd_data.kptl_nthreads);
 651 }
 652
 653 int
 654 kptllnd_thread_start(int (*fn)(void *arg), void *arg, char *name)
 655 {
 656         struct task_struct *task;
 657
 658         cfs_atomic_inc(&kptllnd_data.kptl_nthreads);
 659
 660         task = kthread_run(fn, arg, name);
 661         if (IS_ERR(task)) {
 662                 CERROR("Failed to start thread: error %ld\n", PTR_ERR(task));
 663                 kptllnd_thread_fini();
 664         }
 665         return PTR_ERR(task);
 666 }
 667
 668 int
 669 kptllnd_watchdog(void *arg)
 670 {
 671         int                 id = (long)arg;
 672         wait_queue_t        waitlink;
 673         int                 stamp = 0;
 674         int                 peer_index = 0;
 675         unsigned long       deadline = jiffies;
 676         int                 timeout;
 677         int                 i;
 678
 679         cfs_block_allsigs();
 680
 681         init_waitqueue_entry_current(&waitlink);
 682
 683         /* threads shut down in phase 2 after all peers have been destroyed */
 684         while (kptllnd_data.kptl_shutdown < 2) {
 685
 686                 timeout = (int)(deadline - jiffies);
 687                 if (timeout <= 0) {
 688                         const int n = 4;
 689                         const int p = 1;
 690                         int       chunk = kptllnd_data.kptl_peer_hash_size;
 691
 692
 693                         /* Time to check for RDMA timeouts on a few more
 694                          * peers: I do checks every 'p' seconds on a
 695                          * proportion of the peer table and I need to check
 696                          * every connection 'n' times within a timeout
 697                          * interval, to ensure I detect a timeout on any
 698                          * connection within (n+1)/n times the timeout
 699                          * interval. */
 700
 701                         if ((*kptllnd_tunables.kptl_timeout) > n * p)
 702                                 chunk = (chunk * n * p) /
 703                                         (*kptllnd_tunables.kptl_timeout);
 704                         if (chunk == 0)
 705                                 chunk = 1;
 706
 707                         for (i = 0; i < chunk; i++) {
 708                                 kptllnd_peer_check_bucket(peer_index, stamp);
 709                                 peer_index = (peer_index + 1) %
 710                                      kptllnd_data.kptl_peer_hash_size;
 711                         }
 712
 713                         deadline += p * HZ;
 714                         stamp++;
 715                         continue;
 716                 }
 717
 718                 kptllnd_handle_closing_peers();
 719
 720                 set_current_state(TASK_INTERRUPTIBLE);
 721                 add_wait_queue_exclusive(&kptllnd_data.kptl_watchdog_waitq,
 722                                         &waitlink);
 723
 724                 waitq_timedwait(&waitlink, TASK_INTERRUPTIBLE, timeout);
 725
 726                 set_current_state (TASK_RUNNING);
 727                 remove_wait_queue(&kptllnd_data.kptl_watchdog_waitq, &waitlink);
 728         }
 729
 730         kptllnd_thread_fini();
 731         CDEBUG(D_NET, "<<<\n");
 732         return (0);
 733 };
 734
 735 int
 736 kptllnd_scheduler (void *arg)
 737 {
 738         int                 id = (long)arg;
 739         wait_queue_t        waitlink;
 740         unsigned long       flags;
 741         int                 did_something;
 742         int                 counter = 0;
 743         kptl_rx_t          *rx;
 744         kptl_rx_buffer_t   *rxb;
 745         kptl_tx_t          *tx;
 746
 747         cfs_block_allsigs();
 748
 749         init_waitqueue_entry_current(&waitlink);
 750
 751         spin_lock_irqsave(&kptllnd_data.kptl_sched_lock, flags);
 752
 753         /* threads shut down in phase 2 after all peers have been destroyed */
 754         while (kptllnd_data.kptl_shutdown < 2) {
 755
 756                 did_something = 0;
 757
 758                 if (!cfs_list_empty(&kptllnd_data.kptl_sched_rxq)) {
 759                         rx = cfs_list_entry (kptllnd_data.kptl_sched_rxq.next,
 760                                              kptl_rx_t, rx_list);
 761                         cfs_list_del(&rx->rx_list);
 762
 763                         spin_unlock_irqrestore(&kptllnd_data. \
 764                                                    kptl_sched_lock,
 765                                                    flags);
 766
 767                         kptllnd_rx_parse(rx);
 768                         did_something = 1;
 769
 770                         spin_lock_irqsave(&kptllnd_data.kptl_sched_lock,
 771                                               flags);
 772                 }
 773
 774                 if (!cfs_list_empty(&kptllnd_data.kptl_sched_rxbq)) {
 775                         rxb = cfs_list_entry (kptllnd_data.kptl_sched_rxbq.next,
 776                                               kptl_rx_buffer_t,
 777                                               rxb_repost_list);
 778                         cfs_list_del(&rxb->rxb_repost_list);
 779
 780                         spin_unlock_irqrestore(&kptllnd_data. \
 781                                                    kptl_sched_lock,
 782                                                    flags);
 783
 784                         kptllnd_rx_buffer_post(rxb);
 785                         did_something = 1;
 786
 787                         spin_lock_irqsave(&kptllnd_data.kptl_sched_lock,
 788                                               flags);
 789                 }
 790
 791                 if (!cfs_list_empty(&kptllnd_data.kptl_sched_txq)) {
 792                         tx = cfs_list_entry (kptllnd_data.kptl_sched_txq.next,
 793                                              kptl_tx_t, tx_list);
 794                         cfs_list_del_init(&tx->tx_list);
 795
 796                         spin_unlock_irqrestore(&kptllnd_data. \
 797                                                    kptl_sched_lock, flags);
 798
 799                         kptllnd_tx_fini(tx);
 800                         did_something = 1;
 801
 802                         spin_lock_irqsave(&kptllnd_data.kptl_sched_lock,
 803                                               flags);
 804                 }
 805
 806                 if (did_something) {
 807                         if (++counter != *kptllnd_tunables.kptl_reschedule_loops)
 808                                 continue;
 809                 }
 810
 811                 set_current_state(TASK_INTERRUPTIBLE);
 812                 add_wait_queue_exclusive(&kptllnd_data.kptl_sched_waitq,
 813                                         &waitlink);
 814                 spin_unlock_irqrestore(&kptllnd_data.kptl_sched_lock,
 815                                            flags);
 816
 817                 if (!did_something)
 818                         waitq_wait(&waitlink, TASK_INTERRUPTIBLE);
 819                 else
 820                         cond_resched();
 821
 822                 set_current_state(TASK_RUNNING);
 823                 remove_wait_queue(&kptllnd_data.kptl_sched_waitq, &waitlink);
 824
 825                 spin_lock_irqsave(&kptllnd_data.kptl_sched_lock, flags);
 826
 827                 counter = 0;
 828         }
 829
 830         spin_unlock_irqrestore(&kptllnd_data.kptl_sched_lock, flags);
 831
 832         kptllnd_thread_fini();
 833         return 0;
 834 }