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