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
  20  * http://www.sun.com/software/products/lustre/docs/GPLv2.pdf
  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                 kptllnd_schedule_ptltrace_dump();
 305         }
 306
 307         return 0;
 308 }
 309
 310 int
 311 kptllnd_send(lnet_ni_t *ni, void *private, lnet_msg_t *lntmsg)
 312 {
 313         lnet_hdr_t       *hdr = &lntmsg->msg_hdr;
 314         int               type = lntmsg->msg_type;
 315         lnet_process_id_t target = lntmsg->msg_target;
 316         int               target_is_router = lntmsg->msg_target_is_router;
 317         int               routing = lntmsg->msg_routing;
 318         unsigned int      payload_niov = lntmsg->msg_niov;
 319         struct iovec     *payload_iov = lntmsg->msg_iov;
 320         lnet_kiov_t      *payload_kiov = lntmsg->msg_kiov;
 321         unsigned int      payload_offset = lntmsg->msg_offset;
 322         unsigned int      payload_nob = lntmsg->msg_len;
 323         kptl_peer_t      *peer;
 324         kptl_tx_t        *tx;
 325         int               nob;
 326         int               nfrag;
 327         int               rc;
 328
 329         LASSERT (payload_nob == 0 || payload_niov > 0);
 330         LASSERT (payload_niov <= LNET_MAX_IOV);
 331         LASSERT (payload_niov <= PTL_MD_MAX_IOV); /* !!! */
 332         LASSERT (!(payload_kiov != NULL && payload_iov != NULL));
 333         LASSERT (!in_interrupt());
 334
 335         rc = kptllnd_find_target(&peer, target);
 336         if (rc != 0)
 337                 return rc;
 338
 339         switch (type) {
 340         default:
 341                 LBUG();
 342                 return -EINVAL;
 343
 344         case LNET_MSG_REPLY:
 345         case LNET_MSG_PUT:
 346                 /* Should the payload avoid RDMA? */
 347                 nob = offsetof(kptl_msg_t, ptlm_u.immediate.kptlim_payload[payload_nob]);
 348                 if (payload_kiov == NULL &&
 349                     nob <= peer->peer_max_msg_size)
 350                         break;
 351
 352                 tx = kptllnd_get_idle_tx(TX_TYPE_PUT_REQUEST);
 353                 if (tx == NULL) {
 354                         CERROR("Can't send %s to %s: can't allocate descriptor\n",
 355                                lnet_msgtyp2str(type),
 356                                libcfs_id2str(target));
 357                         rc = -ENOMEM;
 358                         goto out;
 359                 }
 360
 361                 kptllnd_init_rdma_md(tx, payload_niov,
 362                                      payload_iov, payload_kiov,
 363                                      payload_offset, payload_nob);
 364
 365                 tx->tx_lnet_msg = lntmsg;
 366                 tx->tx_msg->ptlm_u.rdma.kptlrm_hdr = *hdr;
 367                 kptllnd_init_msg (tx->tx_msg, PTLLND_MSG_TYPE_PUT,
 368                                   sizeof(kptl_rdma_msg_t));
 369
 370                 CDEBUG(D_NETTRACE, "%s: passive PUT p %d %p\n",
 371                        libcfs_id2str(target),
 372                        le32_to_cpu(lntmsg->msg_hdr.msg.put.ptl_index), tx);
 373
 374                 kptllnd_tx_launch(peer, tx, 0);
 375                 goto out;
 376
 377         case LNET_MSG_GET:
 378                 /* routed gets don't RDMA */
 379                 if (target_is_router || routing)
 380                         break;
 381
 382                 /* Is the payload small enough not to need RDMA? */
 383                 nob = lntmsg->msg_md->md_length;
 384                 nob = offsetof(kptl_msg_t,
 385                                ptlm_u.immediate.kptlim_payload[nob]);
 386                 if (nob <= peer->peer_max_msg_size)
 387                         break;
 388
 389                 tx = kptllnd_get_idle_tx(TX_TYPE_GET_REQUEST);
 390                 if (tx == NULL) {
 391                         CERROR("Can't send GET to %s: can't allocate descriptor\n",
 392                                libcfs_id2str(target));
 393                         rc = -ENOMEM;
 394                         goto out;
 395                 }
 396
 397                 tx->tx_lnet_replymsg =
 398                         lnet_create_reply_msg(kptllnd_data.kptl_ni, lntmsg);
 399                 if (tx->tx_lnet_replymsg == NULL) {
 400                         CERROR("Failed to allocate LNET reply for %s\n",
 401                                libcfs_id2str(target));
 402                         kptllnd_tx_decref(tx);
 403                         rc = -ENOMEM;
 404                         goto out;
 405                 }
 406
 407                 if ((lntmsg->msg_md->md_options & LNET_MD_KIOV) == 0)
 408                         kptllnd_init_rdma_md(tx, lntmsg->msg_md->md_niov,
 409                                              lntmsg->msg_md->md_iov.iov, NULL,
 410                                              0, lntmsg->msg_md->md_length);
 411                 else
 412                         kptllnd_init_rdma_md(tx, lntmsg->msg_md->md_niov,
 413                                              NULL, lntmsg->msg_md->md_iov.kiov,
 414                                              0, lntmsg->msg_md->md_length);
 415
 416                 tx->tx_lnet_msg = lntmsg;
 417                 tx->tx_msg->ptlm_u.rdma.kptlrm_hdr = *hdr;
 418                 kptllnd_init_msg (tx->tx_msg, PTLLND_MSG_TYPE_GET,
 419                                   sizeof(kptl_rdma_msg_t));
 420
 421                 CDEBUG(D_NETTRACE, "%s: passive GET p %d %p\n",
 422                        libcfs_id2str(target),
 423                        le32_to_cpu(lntmsg->msg_hdr.msg.put.ptl_index), tx);
 424
 425                 kptllnd_tx_launch(peer, tx, 0);
 426                 goto out;
 427
 428         case LNET_MSG_ACK:
 429                 CDEBUG(D_NET, "LNET_MSG_ACK\n");
 430                 LASSERT (payload_nob == 0);
 431                 break;
 432         }
 433
 434         /* I don't have to handle kiovs */
 435         LASSERT (payload_nob == 0 || payload_iov != NULL);
 436
 437         tx = kptllnd_get_idle_tx(TX_TYPE_SMALL_MESSAGE);
 438         if (tx == NULL) {
 439                 CERROR("Can't send %s to %s: can't allocate descriptor\n",
 440                        lnet_msgtyp2str(type), libcfs_id2str(target));
 441                 rc = -ENOMEM;
 442                 goto out;
 443         }
 444
 445         tx->tx_lnet_msg = lntmsg;
 446         tx->tx_msg->ptlm_u.immediate.kptlim_hdr = *hdr;
 447
 448         if (payload_nob == 0) {
 449                 nfrag = 0;
 450         } else {
 451                 tx->tx_frags->iov[0].iov_base = tx->tx_msg;
 452                 tx->tx_frags->iov[0].iov_len = offsetof(kptl_msg_t,
 453                                                         ptlm_u.immediate.kptlim_payload);
 454
 455                 /* NB relying on lustre not asking for PTL_MD_MAX_IOV
 456                  * fragments!! */
 457 #ifdef _USING_LUSTRE_PORTALS_
 458                 nfrag = 1 + lnet_extract_iov(PTL_MD_MAX_IOV - 1,
 459                                              &tx->tx_frags->iov[1],
 460                                              payload_niov, payload_iov,
 461                                              payload_offset, payload_nob);
 462 #else
 463                 nfrag = 1 + kptllnd_extract_iov(PTL_MD_MAX_IOV - 1,
 464                                                 &tx->tx_frags->iov[1],
 465                                                 payload_niov, payload_iov,
 466                                                 payload_offset, payload_nob);
 467 #endif
 468         }
 469
 470         nob = offsetof(kptl_immediate_msg_t, kptlim_payload[payload_nob]);
 471         kptllnd_init_msg(tx->tx_msg, PTLLND_MSG_TYPE_IMMEDIATE, nob);
 472
 473         CDEBUG(D_NETTRACE, "%s: immediate %s p %d %p\n",
 474                libcfs_id2str(target),
 475                lnet_msgtyp2str(lntmsg->msg_type),
 476                (le32_to_cpu(lntmsg->msg_type) == LNET_MSG_PUT) ?
 477                le32_to_cpu(lntmsg->msg_hdr.msg.put.ptl_index) :
 478                (le32_to_cpu(lntmsg->msg_type) == LNET_MSG_GET) ?
 479                le32_to_cpu(lntmsg->msg_hdr.msg.get.ptl_index) : -1,
 480                tx);
 481
 482         kptllnd_tx_launch(peer, tx, nfrag);
 483
 484  out:
 485         kptllnd_peer_decref(peer);
 486         return rc;
 487 }
 488
 489 int
 490 kptllnd_eager_recv(struct lnet_ni *ni, void *private,
 491                    lnet_msg_t *msg, void **new_privatep)
 492 {
 493         kptl_rx_t        *rx = private;
 494
 495         CDEBUG(D_NET, "Eager RX=%p RXB=%p\n", rx, rx->rx_rxb);
 496
 497         /* I have to release my ref on rxb (if I have one) to ensure I'm an
 498          * eager receiver, so I copy the incoming request from the buffer it
 499          * landed in, into space reserved in the descriptor... */
 500
 501 #if (PTL_MD_LOCAL_ALIGN8 == 0)
 502         if (rx->rx_rxb == NULL)                 /* already copied */
 503                 return 0;                       /* to fix alignment */
 504 #else
 505         LASSERT(rx->rx_rxb != NULL);
 506 #endif
 507         LASSERT(rx->rx_nob <= *kptllnd_tunables.kptl_max_msg_size);
 508
 509         memcpy(rx->rx_space, rx->rx_msg, rx->rx_nob);
 510         rx->rx_msg = (kptl_msg_t *)rx->rx_space;
 511
 512         kptllnd_rx_buffer_decref(rx->rx_rxb);
 513         rx->rx_rxb = NULL;
 514
 515         return 0;
 516 }
 517
 518
 519 int
 520 kptllnd_recv (lnet_ni_t *ni, void *private, lnet_msg_t *lntmsg, int delayed,
 521               unsigned int niov, struct iovec *iov, lnet_kiov_t *kiov,
 522               unsigned int offset, unsigned int mlen, unsigned int rlen)
 523 {
 524         kptl_rx_t    *rx = private;
 525         kptl_msg_t   *rxmsg = rx->rx_msg;
 526         int           nob;
 527         int           rc;
 528
 529         CDEBUG(D_NET, "%s niov=%d offset=%d mlen=%d rlen=%d\n",
 530                kptllnd_msgtype2str(rxmsg->ptlm_type),
 531                niov, offset, mlen, rlen);
 532
 533         LASSERT (mlen <= rlen);
 534         LASSERT (mlen >= 0);
 535         LASSERT (!in_interrupt());
 536         LASSERT (!(kiov != NULL && iov != NULL)); /* never both */
 537         LASSERT (niov <= PTL_MD_MAX_IOV);       /* !!! */
 538
 539 #ifdef CRAY_XT3
 540         if (lntmsg != NULL &&
 541             rx->rx_uid != 0) {
 542                 /* Set the UID if the sender's uid isn't 0; i.e. non-root
 543                  * running in userspace (e.g. a catamount node; linux kernel
 544                  * senders, including routers have uid 0).  If this is a lustre
 545                  * RPC request, this tells lustre not to trust the creds in the
 546                  * RPC message body. */
 547                 lnet_set_msg_uid(ni, lntmsg, rx->rx_uid);
 548         }
 549 #endif
 550         switch(rxmsg->ptlm_type)
 551         {
 552         default:
 553                 LBUG();
 554                 rc = -EINVAL;
 555                 break;
 556
 557         case PTLLND_MSG_TYPE_IMMEDIATE:
 558                 CDEBUG(D_NET, "PTLLND_MSG_TYPE_IMMEDIATE %d,%d\n", mlen, rlen);
 559
 560                 nob = offsetof(kptl_msg_t, ptlm_u.immediate.kptlim_payload[rlen]);
 561                 if (nob > rx->rx_nob) {
 562                         CERROR ("Immediate message from %s too big: %d(%d)\n",
 563                                 libcfs_id2str(rx->rx_peer->peer_id), nob,
 564                                 rx->rx_nob);
 565                         rc = -EINVAL;
 566                         break;
 567                 }
 568
 569                 if (kiov != NULL)
 570                         lnet_copy_flat2kiov(
 571                                 niov, kiov, offset,
 572                                 *kptllnd_tunables.kptl_max_msg_size,
 573                                 rxmsg->ptlm_u.immediate.kptlim_payload,
 574                                 0,
 575                                 mlen);
 576                 else
 577                         lnet_copy_flat2iov(
 578                                 niov, iov, offset,
 579                                 *kptllnd_tunables.kptl_max_msg_size,
 580                                 rxmsg->ptlm_u.immediate.kptlim_payload,
 581                                 0,
 582                                 mlen);
 583
 584                 lnet_finalize (ni, lntmsg, 0);
 585                 rc = 0;
 586                 break;
 587
 588         case PTLLND_MSG_TYPE_GET:
 589                 CDEBUG(D_NET, "PTLLND_MSG_TYPE_GET %d,%d\n", mlen, rlen);
 590
 591                 /* NB always send RDMA so the peer can complete.  I send
 592                  * success/failure in the portals 'hdr_data' */
 593
 594                 if (lntmsg == NULL)
 595                         rc = kptllnd_active_rdma(rx, NULL,
 596                                                  TX_TYPE_GET_RESPONSE,
 597                                                  0, NULL, NULL, 0, 0);
 598                 else
 599                         rc = kptllnd_active_rdma(rx, lntmsg,
 600                                                  TX_TYPE_GET_RESPONSE,
 601                                                  lntmsg->msg_niov,
 602                                                  lntmsg->msg_iov,
 603                                                  lntmsg->msg_kiov,
 604                                                  lntmsg->msg_offset,
 605                                                  lntmsg->msg_len);
 606                 break;
 607
 608         case PTLLND_MSG_TYPE_PUT:
 609                 CDEBUG(D_NET, "PTLLND_MSG_TYPE_PUT %d,%d\n", mlen, rlen);
 610
 611                 /* NB always send RDMA so the peer can complete; it'll be 0
 612                  * bytes if there was no match (lntmsg == NULL). I have no way
 613                  * to let my peer know this, but she's only interested in when
 614                  * the net has stopped accessing her buffer in any case. */
 615
 616                 rc = kptllnd_active_rdma(rx, lntmsg, TX_TYPE_PUT_RESPONSE,
 617                                          niov, iov, kiov, offset, mlen);
 618                 break;
 619         }
 620
 621         /*
 622          * We're done with the RX
 623          */
 624         kptllnd_rx_done(rx, PTLLND_POSTRX_PEER_CREDIT);
 625         return rc;
 626 }
 627
 628 void
 629 kptllnd_eq_callback(ptl_event_t *ev)
 630 {
 631         kptl_eventarg_t *eva = ev->md.user_ptr;
 632
 633         switch (eva->eva_type) {
 634         default:
 635                 LBUG();
 636
 637         case PTLLND_EVENTARG_TYPE_MSG:
 638         case PTLLND_EVENTARG_TYPE_RDMA:
 639                 kptllnd_tx_callback(ev);
 640                 break;
 641
 642         case PTLLND_EVENTARG_TYPE_BUF:
 643                 kptllnd_rx_buffer_callback(ev);
 644                 break;
 645         }
 646 }
 647
 648 void
 649 kptllnd_thread_fini (void)
 650 {
 651         atomic_dec(&kptllnd_data.kptl_nthreads);
 652 }
 653
 654 int
 655 kptllnd_thread_start (int (*fn)(void *arg), void *arg)
 656 {
 657         long                pid;
 658
 659         atomic_inc(&kptllnd_data.kptl_nthreads);
 660
 661         pid = kernel_thread (fn, arg, 0);
 662         if (pid >= 0)
 663                 return 0;
 664
 665         CERROR("Failed to start kernel_thread: error %d\n", (int)pid);
 666         kptllnd_thread_fini();
 667         return (int)pid;
 668 }
 669
 670 int
 671 kptllnd_watchdog(void *arg)
 672 {
 673         int                 id = (long)arg;
 674         char                name[16];
 675         wait_queue_t        waitlink;
 676         int                 stamp = 0;
 677         int                 peer_index = 0;
 678         unsigned long       deadline = jiffies;
 679         int                 timeout;
 680         int                 i;
 681
 682         snprintf(name, sizeof(name), "kptllnd_wd_%02d", id);
 683         cfs_daemonize(name);
 684         cfs_block_allsigs();
 685
 686         init_waitqueue_entry(&waitlink, current);
 687
 688         /* threads shut down in phase 2 after all peers have been destroyed */
 689         while (kptllnd_data.kptl_shutdown < 2) {
 690
 691                 /* add a check for needs ptltrace
 692                  * yes, this is blatant hijacking of this thread
 693                  * we can't dump directly from tx or rx _callbacks as it deadlocks portals
 694                  * and takes out the node
 695                 */
 696
 697                 if (atomic_read(&kptllnd_data.kptl_needs_ptltrace)) {
 698 #ifdef CRAY_XT3
 699                         kptllnd_dump_ptltrace();
 700                         /* we only dump once, no matter how many pending */
 701                         atomic_set(&kptllnd_data.kptl_needs_ptltrace, 0);
 702 #else
 703                         LBUG();
 704 #endif
 705                 }
 706
 707                 timeout = (int)(deadline - jiffies);
 708
 709                 if (timeout <= 0) {
 710                         const int n = 4;
 711                         const int p = 1;
 712                         int       chunk = kptllnd_data.kptl_peer_hash_size;
 713
 714
 715                         /* Time to check for RDMA timeouts on a few more
 716                          * peers: I do checks every 'p' seconds on a
 717                          * proportion of the peer table and I need to check
 718                          * every connection 'n' times within a timeout
 719                          * interval, to ensure I detect a timeout on any
 720                          * connection within (n+1)/n times the timeout
 721                          * interval. */
 722
 723                         if ((*kptllnd_tunables.kptl_timeout) > n * p)
 724                                 chunk = (chunk * n * p) /
 725                                         (*kptllnd_tunables.kptl_timeout);
 726                         if (chunk == 0)
 727                                 chunk = 1;
 728
 729                         for (i = 0; i < chunk; i++) {
 730                                 kptllnd_peer_check_bucket(peer_index, stamp);
 731                                 peer_index = (peer_index + 1) %
 732                                      kptllnd_data.kptl_peer_hash_size;
 733                         }
 734
 735                         deadline += p * HZ;
 736                         stamp++;
 737                         continue;
 738                 }
 739
 740                 kptllnd_handle_closing_peers();
 741
 742                 set_current_state(TASK_INTERRUPTIBLE);
 743                 add_wait_queue_exclusive(&kptllnd_data.kptl_watchdog_waitq,
 744                                          &waitlink);
 745
 746                 schedule_timeout(timeout);
 747
 748                 set_current_state (TASK_RUNNING);
 749                 remove_wait_queue(&kptllnd_data.kptl_watchdog_waitq, &waitlink);
 750         }
 751
 752         kptllnd_thread_fini();
 753         CDEBUG(D_NET, "<<<\n");
 754         return (0);
 755 };
 756
 757 int
 758 kptllnd_scheduler (void *arg)
 759 {
 760         int                 id = (long)arg;
 761         char                name[16];
 762         wait_queue_t        waitlink;
 763         unsigned long       flags;
 764         int                 did_something;
 765         int                 counter = 0;
 766         kptl_rx_t          *rx;
 767         kptl_rx_buffer_t   *rxb;
 768         kptl_tx_t          *tx;
 769
 770         snprintf(name, sizeof(name), "kptllnd_sd_%02d", id);
 771         cfs_daemonize(name);
 772         cfs_block_allsigs();
 773
 774         init_waitqueue_entry(&waitlink, current);
 775
 776         spin_lock_irqsave(&kptllnd_data.kptl_sched_lock, flags);
 777
 778         /* threads shut down in phase 2 after all peers have been destroyed */
 779         while (kptllnd_data.kptl_shutdown < 2) {
 780
 781                 did_something = 0;
 782
 783                 if (!list_empty(&kptllnd_data.kptl_sched_rxq)) {
 784                         rx = list_entry (kptllnd_data.kptl_sched_rxq.next,
 785                                          kptl_rx_t, rx_list);
 786                         list_del(&rx->rx_list);
 787
 788                         spin_unlock_irqrestore(&kptllnd_data.kptl_sched_lock,
 789                                                flags);
 790
 791                         kptllnd_rx_parse(rx);
 792                         did_something = 1;
 793
 794                         spin_lock_irqsave(&kptllnd_data.kptl_sched_lock, flags);
 795                 }
 796
 797                 if (!list_empty(&kptllnd_data.kptl_sched_rxbq)) {
 798                         rxb = list_entry (kptllnd_data.kptl_sched_rxbq.next,
 799                                           kptl_rx_buffer_t, rxb_repost_list);
 800                         list_del(&rxb->rxb_repost_list);
 801
 802                         spin_unlock_irqrestore(&kptllnd_data.kptl_sched_lock,
 803                                                flags);
 804
 805                         kptllnd_rx_buffer_post(rxb);
 806                         did_something = 1;
 807
 808                         spin_lock_irqsave(&kptllnd_data.kptl_sched_lock, flags);
 809                 }
 810
 811                 if (!list_empty(&kptllnd_data.kptl_sched_txq)) {
 812                         tx = list_entry (kptllnd_data.kptl_sched_txq.next,
 813                                          kptl_tx_t, tx_list);
 814                         list_del_init(&tx->tx_list);
 815
 816                         spin_unlock_irqrestore(&kptllnd_data.kptl_sched_lock, flags);
 817
 818                         kptllnd_tx_fini(tx);
 819                         did_something = 1;
 820
 821                         spin_lock_irqsave(&kptllnd_data.kptl_sched_lock, flags);
 822                 }
 823
 824                 if (did_something) {
 825                         if (++counter != *kptllnd_tunables.kptl_reschedule_loops)
 826                                 continue;
 827                 }
 828
 829                 set_current_state(TASK_INTERRUPTIBLE);
 830                 add_wait_queue_exclusive(&kptllnd_data.kptl_sched_waitq,
 831                                          &waitlink);
 832                 spin_unlock_irqrestore(&kptllnd_data.kptl_sched_lock, flags);
 833
 834                 if (!did_something)
 835                         schedule();
 836                 else
 837                         cond_resched();
 838
 839                 set_current_state(TASK_RUNNING);
 840                 remove_wait_queue(&kptllnd_data.kptl_sched_waitq, &waitlink);
 841
 842                 spin_lock_irqsave(&kptllnd_data.kptl_sched_lock, flags);
 843
 844                 counter = 0;
 845         }
 846
 847         spin_unlock_irqrestore(&kptllnd_data.kptl_sched_lock, flags);
 848
 849         kptllnd_thread_fini();
 850         return 0;
 851 }