lnet/klnds/ptllnd/ptllnd_cb.c

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