lnet/klnds/ptllnd/ptllnd_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/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         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_peer_t      *peer;
 323         kptl_tx_t        *tx;
 324         int               nob;
 325         int               nfrag;
 326         int               rc;
 327
 328         LASSERT (payload_nob == 0 || payload_niov > 0);
 329         LASSERT (payload_niov <= LNET_MAX_IOV);
 330         LASSERT (payload_niov <= PTL_MD_MAX_IOV); /* !!! */
 331         LASSERT (!(payload_kiov != NULL && payload_iov != NULL));
 332         LASSERT (!in_interrupt());
 333
 334         rc = kptllnd_find_target(&peer, target);
 335         if (rc != 0)
 336                 return rc;
 337
 338         switch (type) {
 339         default:
 340                 LBUG();
 341                 return -EINVAL;
 342
 343         case LNET_MSG_REPLY:
 344         case LNET_MSG_PUT:
 345                 /* Should the payload avoid RDMA? */
 346                 nob = offsetof(kptl_msg_t, ptlm_u.immediate.kptlim_payload[payload_nob]);
 347                 if (payload_kiov == NULL &&
 348                     nob <= peer->peer_max_msg_size)
 349                         break;
 350
 351                 tx = kptllnd_get_idle_tx(TX_TYPE_PUT_REQUEST);
 352                 if (tx == NULL) {
 353                         CERROR("Can't send %s to %s: can't allocate descriptor\n",
 354                                lnet_msgtyp2str(type),
 355                                libcfs_id2str(target));
 356                         rc = -ENOMEM;
 357                         goto out;
 358                 }
 359
 360                 kptllnd_init_rdma_md(tx, payload_niov,
 361                                      payload_iov, payload_kiov,
 362                                      payload_offset, payload_nob);
 363
 364                 tx->tx_lnet_msg = lntmsg;
 365                 tx->tx_msg->ptlm_u.rdma.kptlrm_hdr = *hdr;
 366                 kptllnd_init_msg (tx->tx_msg, PTLLND_MSG_TYPE_PUT,
 367                                   sizeof(kptl_rdma_msg_t));
 368
 369                 CDEBUG(D_NETTRACE, "%s: passive PUT p %d %p\n",
 370                        libcfs_id2str(target),
 371                        le32_to_cpu(lntmsg->msg_hdr.msg.put.ptl_index), tx);
 372
 373                 kptllnd_tx_launch(peer, tx, 0);
 374                 goto out;
 375
 376         case LNET_MSG_GET:
 377                 /* routed gets don't RDMA */
 378                 if (target_is_router || routing)
 379                         break;
 380
 381                 /* Is the payload small enough not to need RDMA? */
 382                 nob = lntmsg->msg_md->md_length;
 383                 nob = offsetof(kptl_msg_t,
 384                                ptlm_u.immediate.kptlim_payload[nob]);
 385                 if (nob <= peer->peer_max_msg_size)
 386                         break;
 387
 388                 tx = kptllnd_get_idle_tx(TX_TYPE_GET_REQUEST);
 389                 if (tx == NULL) {
 390                         CERROR("Can't send GET to %s: can't allocate descriptor\n",
 391                                libcfs_id2str(target));
 392                         rc = -ENOMEM;
 393                         goto out;
 394                 }
 395
 396                 tx->tx_lnet_replymsg =
 397                         lnet_create_reply_msg(kptllnd_data.kptl_ni, lntmsg);
 398                 if (tx->tx_lnet_replymsg == NULL) {
 399                         CERROR("Failed to allocate LNET reply for %s\n",
 400                                libcfs_id2str(target));
 401                         kptllnd_tx_decref(tx);
 402                         rc = -ENOMEM;
 403                         goto out;
 404                 }
 405
 406                 if ((lntmsg->msg_md->md_options & LNET_MD_KIOV) == 0)
 407                         kptllnd_init_rdma_md(tx, lntmsg->msg_md->md_niov,
 408                                              lntmsg->msg_md->md_iov.iov, NULL,
 409                                              0, lntmsg->msg_md->md_length);
 410                 else
 411                         kptllnd_init_rdma_md(tx, lntmsg->msg_md->md_niov,
 412                                              NULL, lntmsg->msg_md->md_iov.kiov,
 413                                              0, lntmsg->msg_md->md_length);
 414
 415                 tx->tx_lnet_msg = lntmsg;
 416                 tx->tx_msg->ptlm_u.rdma.kptlrm_hdr = *hdr;
 417                 kptllnd_init_msg (tx->tx_msg, PTLLND_MSG_TYPE_GET,
 418                                   sizeof(kptl_rdma_msg_t));
 419
 420                 CDEBUG(D_NETTRACE, "%s: passive GET p %d %p\n",
 421                        libcfs_id2str(target),
 422                        le32_to_cpu(lntmsg->msg_hdr.msg.put.ptl_index), tx);
 423
 424                 kptllnd_tx_launch(peer, tx, 0);
 425                 goto out;
 426
 427         case LNET_MSG_ACK:
 428                 CDEBUG(D_NET, "LNET_MSG_ACK\n");
 429                 LASSERT (payload_nob == 0);
 430                 break;
 431         }
 432
 433         /* I don't have to handle kiovs */
 434         LASSERT (payload_nob == 0 || payload_iov != NULL);
 435
 436         tx = kptllnd_get_idle_tx(TX_TYPE_SMALL_MESSAGE);
 437         if (tx == NULL) {
 438                 CERROR("Can't send %s to %s: can't allocate descriptor\n",
 439                        lnet_msgtyp2str(type), libcfs_id2str(target));
 440                 rc = -ENOMEM;
 441                 goto out;
 442         }
 443
 444         tx->tx_lnet_msg = lntmsg;
 445         tx->tx_msg->ptlm_u.immediate.kptlim_hdr = *hdr;
 446
 447         if (payload_nob == 0) {
 448                 nfrag = 0;
 449         } else {
 450                 tx->tx_frags->iov[0].iov_base = tx->tx_msg;
 451                 tx->tx_frags->iov[0].iov_len = offsetof(kptl_msg_t,
 452                                                         ptlm_u.immediate.kptlim_payload);
 453
 454                 /* NB relying on lustre not asking for PTL_MD_MAX_IOV
 455                  * fragments!! */
 456 #ifdef _USING_LUSTRE_PORTALS_
 457                 nfrag = 1 + lnet_extract_iov(PTL_MD_MAX_IOV - 1,
 458                                              &tx->tx_frags->iov[1],
 459                                              payload_niov, payload_iov,
 460                                              payload_offset, payload_nob);
 461 #else
 462                 nfrag = 1 + kptllnd_extract_iov(PTL_MD_MAX_IOV - 1,
 463                                                 &tx->tx_frags->iov[1],
 464                                                 payload_niov, payload_iov,
 465                                                 payload_offset, payload_nob);
 466 #endif
 467         }
 468
 469         nob = offsetof(kptl_immediate_msg_t, kptlim_payload[payload_nob]);
 470         kptllnd_init_msg(tx->tx_msg, PTLLND_MSG_TYPE_IMMEDIATE, nob);
 471
 472         CDEBUG(D_NETTRACE, "%s: immediate %s p %d %p\n",
 473                libcfs_id2str(target),
 474                lnet_msgtyp2str(lntmsg->msg_type),
 475                (le32_to_cpu(lntmsg->msg_type) == LNET_MSG_PUT) ?
 476                le32_to_cpu(lntmsg->msg_hdr.msg.put.ptl_index) :
 477                (le32_to_cpu(lntmsg->msg_type) == LNET_MSG_GET) ?
 478                le32_to_cpu(lntmsg->msg_hdr.msg.get.ptl_index) : -1,
 479                tx);
 480
 481         kptllnd_tx_launch(peer, tx, nfrag);
 482
 483  out:
 484         kptllnd_peer_decref(peer);
 485         return rc;
 486 }
 487
 488 int
 489 kptllnd_eager_recv(struct lnet_ni *ni, void *private,
 490                    lnet_msg_t *msg, void **new_privatep)
 491 {
 492         kptl_rx_t        *rx = private;
 493
 494         CDEBUG(D_NET, "Eager RX=%p RXB=%p\n", rx, rx->rx_rxb);
 495
 496         /* I have to release my ref on rxb (if I have one) to ensure I'm an
 497          * eager receiver, so I copy the incoming request from the buffer it
 498          * landed in, into space reserved in the descriptor... */
 499
 500 #if (PTL_MD_LOCAL_ALIGN8 == 0)
 501         if (rx->rx_rxb == NULL)                 /* already copied */
 502                 return 0;                       /* to fix alignment */
 503 #else
 504         LASSERT(rx->rx_rxb != NULL);
 505 #endif
 506         LASSERT(rx->rx_nob <= *kptllnd_tunables.kptl_max_msg_size);
 507
 508         memcpy(rx->rx_space, rx->rx_msg, rx->rx_nob);
 509         rx->rx_msg = (kptl_msg_t *)rx->rx_space;
 510
 511         kptllnd_rx_buffer_decref(rx->rx_rxb);
 512         rx->rx_rxb = NULL;
 513
 514         return 0;
 515 }
 516
 517
 518 int
 519 kptllnd_recv (lnet_ni_t *ni, void *private, lnet_msg_t *lntmsg, int delayed,
 520               unsigned int niov, struct iovec *iov, lnet_kiov_t *kiov,
 521               unsigned int offset, unsigned int mlen, unsigned int rlen)
 522 {
 523         kptl_rx_t    *rx = private;
 524         kptl_msg_t   *rxmsg = rx->rx_msg;
 525         int           nob;
 526         int           rc;
 527
 528         CDEBUG(D_NET, "%s niov=%d offset=%d mlen=%d rlen=%d\n",
 529                kptllnd_msgtype2str(rxmsg->ptlm_type),
 530                niov, offset, mlen, rlen);
 531
 532         LASSERT (mlen <= rlen);
 533         LASSERT (mlen >= 0);
 534         LASSERT (!in_interrupt());
 535         LASSERT (!(kiov != NULL && iov != NULL)); /* never both */
 536         LASSERT (niov <= PTL_MD_MAX_IOV);       /* !!! */
 537
 538 #ifdef CRAY_XT3
 539         if (lntmsg != NULL &&
 540             rx->rx_uid != 0) {
 541                 /* Set the UID if the sender's uid isn't 0; i.e. non-root
 542                  * running in userspace (e.g. a catamount node; linux kernel
 543                  * senders, including routers have uid 0).  If this is a lustre
 544                  * RPC request, this tells lustre not to trust the creds in the
 545                  * RPC message body. */
 546                 lnet_set_msg_uid(ni, lntmsg, rx->rx_uid);
 547         }
 548 #endif
 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         atomic_dec(&kptllnd_data.kptl_nthreads);
 651 }
 652
 653 int
 654 kptllnd_thread_start (int (*fn)(void *arg), void *arg)
 655 {
 656         long                pid;
 657
 658         atomic_inc(&kptllnd_data.kptl_nthreads);
 659
 660         pid = kernel_thread (fn, arg, 0);
 661         if (pid >= 0)
 662                 return 0;
 663
 664         CERROR("Failed to start kernel_thread: error %d\n", (int)pid);
 665         kptllnd_thread_fini();
 666         return (int)pid;
 667 }
 668
 669 int
 670 kptllnd_watchdog(void *arg)
 671 {
 672         int                 id = (long)arg;
 673         char                name[16];
 674         wait_queue_t        waitlink;
 675         int                 stamp = 0;
 676         int                 peer_index = 0;
 677         unsigned long       deadline = jiffies;
 678         int                 timeout;
 679         int                 i;
 680
 681         snprintf(name, sizeof(name), "kptllnd_wd_%02d", id);
 682         cfs_daemonize(name);
 683         cfs_block_allsigs();
 684
 685         init_waitqueue_entry(&waitlink, current);
 686
 687         /* threads shut down in phase 2 after all peers have been destroyed */
 688         while (kptllnd_data.kptl_shutdown < 2) {
 689
 690                 timeout = (int)(deadline - jiffies);
 691
 692                 if (timeout <= 0) {
 693                         const int n = 4;
 694                         const int p = 1;
 695                         int       chunk = kptllnd_data.kptl_peer_hash_size;
 696
 697
 698                         /* Time to check for RDMA timeouts on a few more
 699                          * peers: I do checks every 'p' seconds on a
 700                          * proportion of the peer table and I need to check
 701                          * every connection 'n' times within a timeout
 702                          * interval, to ensure I detect a timeout on any
 703                          * connection within (n+1)/n times the timeout
 704                          * interval. */
 705
 706                         if ((*kptllnd_tunables.kptl_timeout) > n * p)
 707                                 chunk = (chunk * n * p) /
 708                                         (*kptllnd_tunables.kptl_timeout);
 709                         if (chunk == 0)
 710                                 chunk = 1;
 711
 712                         for (i = 0; i < chunk; i++) {
 713                                 kptllnd_peer_check_bucket(peer_index, stamp);
 714                                 peer_index = (peer_index + 1) %
 715                                      kptllnd_data.kptl_peer_hash_size;
 716                         }
 717
 718                         deadline += p * HZ;
 719                         stamp++;
 720                         continue;
 721                 }
 722
 723                 kptllnd_handle_closing_peers();
 724
 725                 set_current_state(TASK_INTERRUPTIBLE);
 726                 add_wait_queue_exclusive(&kptllnd_data.kptl_watchdog_waitq,
 727                                          &waitlink);
 728
 729                 schedule_timeout(timeout);
 730
 731                 set_current_state (TASK_RUNNING);
 732                 remove_wait_queue(&kptllnd_data.kptl_watchdog_waitq, &waitlink);
 733         }
 734
 735         kptllnd_thread_fini();
 736         CDEBUG(D_NET, "<<<\n");
 737         return (0);
 738 };
 739
 740 int
 741 kptllnd_scheduler (void *arg)
 742 {
 743         int                 id = (long)arg;
 744         char                name[16];
 745         wait_queue_t        waitlink;
 746         unsigned long       flags;
 747         int                 did_something;
 748         int                 counter = 0;
 749         kptl_rx_t          *rx;
 750         kptl_rx_buffer_t   *rxb;
 751         kptl_tx_t          *tx;
 752
 753         snprintf(name, sizeof(name), "kptllnd_sd_%02d", id);
 754         cfs_daemonize(name);
 755         cfs_block_allsigs();
 756
 757         init_waitqueue_entry(&waitlink, current);
 758
 759         spin_lock_irqsave(&kptllnd_data.kptl_sched_lock, flags);
 760
 761         /* threads shut down in phase 2 after all peers have been destroyed */
 762         while (kptllnd_data.kptl_shutdown < 2) {
 763
 764                 did_something = 0;
 765
 766                 if (!list_empty(&kptllnd_data.kptl_sched_rxq)) {
 767                         rx = list_entry (kptllnd_data.kptl_sched_rxq.next,
 768                                          kptl_rx_t, rx_list);
 769                         list_del(&rx->rx_list);
 770
 771                         spin_unlock_irqrestore(&kptllnd_data.kptl_sched_lock,
 772                                                flags);
 773
 774                         kptllnd_rx_parse(rx);
 775                         did_something = 1;
 776
 777                         spin_lock_irqsave(&kptllnd_data.kptl_sched_lock, flags);
 778                 }
 779
 780                 if (!list_empty(&kptllnd_data.kptl_sched_rxbq)) {
 781                         rxb = list_entry (kptllnd_data.kptl_sched_rxbq.next,
 782                                           kptl_rx_buffer_t, rxb_repost_list);
 783                         list_del(&rxb->rxb_repost_list);
 784
 785                         spin_unlock_irqrestore(&kptllnd_data.kptl_sched_lock,
 786                                                flags);
 787
 788                         kptllnd_rx_buffer_post(rxb);
 789                         did_something = 1;
 790
 791                         spin_lock_irqsave(&kptllnd_data.kptl_sched_lock, flags);
 792                 }
 793
 794                 if (!list_empty(&kptllnd_data.kptl_sched_txq)) {
 795                         tx = list_entry (kptllnd_data.kptl_sched_txq.next,
 796                                          kptl_tx_t, tx_list);
 797                         list_del_init(&tx->tx_list);
 798
 799                         spin_unlock_irqrestore(&kptllnd_data.kptl_sched_lock, flags);
 800
 801                         kptllnd_tx_fini(tx);
 802                         did_something = 1;
 803
 804                         spin_lock_irqsave(&kptllnd_data.kptl_sched_lock, flags);
 805                 }
 806
 807                 if (did_something) {
 808                         if (++counter != *kptllnd_tunables.kptl_reschedule_loops)
 809                                 continue;
 810                 }
 811
 812                 set_current_state(TASK_INTERRUPTIBLE);
 813                 add_wait_queue_exclusive(&kptllnd_data.kptl_sched_waitq,
 814                                          &waitlink);
 815                 spin_unlock_irqrestore(&kptllnd_data.kptl_sched_lock, flags);
 816
 817                 if (!did_something)
 818                         schedule();
 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 }