lnet/klnds/ptllnd/ptllnd_rx_buf.c

   1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
   2  * vim:expandtab:shiftwidth=8:tabstop=8:
   3  *
   4  * Copyright (C) 2005 Cluster File Systems, Inc. All rights reserved.
   5  *   Author: PJ Kirner <pjkirner@clusterfs.com>
   6  *
   7  *   This file is part of the Lustre file system, http://www.lustre.org
   8  *   Lustre is a trademark of Cluster File Systems, Inc.
   9  *
  10  *   This file is confidential source code owned by Cluster File Systems.
  11  *   No viewing, modification, compilation, redistribution, or any other
  12  *   form of use is permitted except through a signed license agreement.
  13  *
  14  *   If you have not signed such an agreement, then you have no rights to
  15  *   this file.  Please destroy it immediately and contact CFS.
  16  *
  17  */
  18
  19  #include "ptllnd.h"
  20
  21 void
  22 kptllnd_rx_buffer_pool_init(kptl_rx_buffer_pool_t *rxbp)
  23 {
  24         memset(rxbp, 0, sizeof(*rxbp));
  25         spin_lock_init(&rxbp->rxbp_lock);
  26         INIT_LIST_HEAD(&rxbp->rxbp_list);
  27 }
  28
  29 void
  30 kptllnd_rx_buffer_destroy(kptl_rx_buffer_t *rxb)
  31 {
  32         kptl_rx_buffer_pool_t *rxbp = rxb->rxb_pool;
  33
  34         LASSERT(rxb->rxb_refcount == 0);
  35         LASSERT(PtlHandleIsEqual(rxb->rxb_mdh, PTL_INVALID_HANDLE));
  36         LASSERT(!rxb->rxb_posted);
  37         LASSERT(rxb->rxb_idle);
  38
  39         list_del(&rxb->rxb_list);
  40         rxbp->rxbp_count--;
  41
  42         LIBCFS_FREE(rxb->rxb_buffer, kptllnd_rx_buffer_size());
  43         LIBCFS_FREE(rxb, sizeof(*rxb));
  44 }
  45
  46 int
  47 kptllnd_rx_buffer_pool_reserve(kptl_rx_buffer_pool_t *rxbp, int count)
  48 {
  49         int               bufsize;
  50         int               msgs_per_buffer;
  51         int               rc;
  52         kptl_rx_buffer_t *rxb;
  53         char             *buffer;
  54         unsigned long     flags;
  55
  56         bufsize = kptllnd_rx_buffer_size();
  57         msgs_per_buffer = bufsize / (*kptllnd_tunables.kptl_max_msg_size);
  58
  59         CDEBUG(D_NET, "kptllnd_rx_buffer_pool_reserve(%d)\n", count);
  60
  61         spin_lock_irqsave(&rxbp->rxbp_lock, flags);
  62
  63         for (;;) {
  64                 if (rxbp->rxbp_shutdown) {
  65                         rc = -ESHUTDOWN;
  66                         break;
  67                 }
  68
  69                 if (rxbp->rxbp_reserved + count <=
  70                     rxbp->rxbp_count * msgs_per_buffer) {
  71                         rc = 0;
  72                         break;
  73                 }
  74
  75                 spin_unlock_irqrestore(&rxbp->rxbp_lock, flags);
  76
  77                 LIBCFS_ALLOC(rxb, sizeof(*rxb));
  78                 LIBCFS_ALLOC(buffer, bufsize);
  79
  80                 if (rxb == NULL || buffer == NULL) {
  81                         CERROR("Failed to allocate rx buffer\n");
  82
  83                         if (rxb != NULL)
  84                                 LIBCFS_FREE(rxb, sizeof(*rxb));
  85                         if (buffer != NULL)
  86                                 LIBCFS_FREE(buffer, bufsize);
  87
  88                         spin_lock_irqsave(&rxbp->rxbp_lock, flags);
  89                         rc = -ENOMEM;
  90                         break;
  91                 }
  92
  93                 memset(rxb, 0, sizeof(*rxb));
  94
  95                 rxb->rxb_eventarg.eva_type = PTLLND_EVENTARG_TYPE_BUF;
  96                 rxb->rxb_refcount = 0;
  97                 rxb->rxb_pool = rxbp;
  98                 rxb->rxb_idle = 0;
  99                 rxb->rxb_posted = 0;
 100                 rxb->rxb_buffer = buffer;
 101                 rxb->rxb_mdh = PTL_INVALID_HANDLE;
 102
 103                 spin_lock_irqsave(&rxbp->rxbp_lock, flags);
 104
 105                 if (rxbp->rxbp_shutdown) {
 106                         spin_unlock_irqrestore(&rxbp->rxbp_lock, flags);
 107
 108                         LIBCFS_FREE(rxb, sizeof(*rxb));
 109                         LIBCFS_FREE(buffer, bufsize);
 110
 111                         spin_lock_irqsave(&rxbp->rxbp_lock, flags);
 112                         rc = -ESHUTDOWN;
 113                         break;
 114                 }
 115
 116                 list_add_tail(&rxb->rxb_list, &rxbp->rxbp_list);
 117                 rxbp->rxbp_count++;
 118
 119                 spin_unlock_irqrestore(&rxbp->rxbp_lock, flags);
 120
 121                 kptllnd_rx_buffer_post(rxb);
 122
 123                 spin_lock_irqsave(&rxbp->rxbp_lock, flags);
 124         }
 125
 126         if (rc == 0)
 127                 rxbp->rxbp_reserved += count;
 128
 129         spin_unlock_irqrestore(&rxbp->rxbp_lock, flags);
 130
 131         return rc;
 132 }
 133
 134 void
 135 kptllnd_rx_buffer_pool_unreserve(kptl_rx_buffer_pool_t *rxbp,
 136                                  int count)
 137 {
 138         unsigned long flags;
 139
 140         spin_lock_irqsave(&rxbp->rxbp_lock, flags);
 141
 142         CDEBUG(D_NET, "kptllnd_rx_buffer_pool_unreserve(%d)\n", count);
 143         rxbp->rxbp_reserved -= count;
 144
 145         spin_unlock_irqrestore(&rxbp->rxbp_lock, flags);
 146 }
 147
 148 void
 149 kptllnd_rx_buffer_pool_fini(kptl_rx_buffer_pool_t *rxbp)
 150 {
 151         kptl_rx_buffer_t       *rxb;
 152         int                     rc;
 153         int                     i;
 154         unsigned long           flags;
 155         struct list_head       *tmp;
 156         struct list_head       *nxt;
 157         ptl_handle_md_t         mdh;
 158
 159         /* CAVEAT EMPTOR: I'm racing with everything here!!!
 160          *
 161          * Buffers can still be posted after I set rxbp_shutdown because I
 162          * can't hold rxbp_lock while I'm posting them.
 163          *
 164          * Calling PtlMDUnlink() here races with auto-unlinks; i.e. a buffer's
 165          * MD handle could become invalid under me.  I am vulnerable to portals
 166          * re-using handles (i.e. make the same handle valid again, but for a
 167          * different MD) from when the MD is actually unlinked, to when the
 168          * event callback tells me it has been unlinked. */
 169
 170         spin_lock_irqsave(&rxbp->rxbp_lock, flags);
 171
 172         rxbp->rxbp_shutdown = 1;
 173
 174         for (i = 9;; i++) {
 175                 list_for_each_safe(tmp, nxt, &rxbp->rxbp_list) {
 176                         rxb = list_entry (tmp, kptl_rx_buffer_t, rxb_list);
 177
 178                         if (rxb->rxb_idle) {
 179                                 spin_unlock_irqrestore(&rxbp->rxbp_lock,
 180                                                        flags);
 181                                 kptllnd_rx_buffer_destroy(rxb);
 182                                 spin_lock_irqsave(&rxbp->rxbp_lock,
 183                                                   flags);
 184                                 continue;
 185                         }
 186
 187                         mdh = rxb->rxb_mdh;
 188                         if (PtlHandleIsEqual(mdh, PTL_INVALID_HANDLE))
 189                                 continue;
 190
 191                         spin_unlock_irqrestore(&rxbp->rxbp_lock, flags);
 192
 193                         rc = PtlMDUnlink(mdh);
 194
 195                         spin_lock_irqsave(&rxbp->rxbp_lock, flags);
 196
 197 #ifdef LUSTRE_PORTALS_UNLINK_SEMANTICS
 198                         /* callback clears rxb_mdh and drops net's ref
 199                          * (which causes repost, but since I set
 200                          * shutdown, it will just set the buffer
 201                          * idle) */
 202 #else
 203                         if (rc == PTL_OK) {
 204                                 rxb->rxb_posted = 0;
 205                                 rxb->rxb_mdh = PTL_INVALID_HANDLE;
 206                                 kptllnd_rx_buffer_decref_locked(rxb);
 207                         }
 208 #endif
 209                 }
 210
 211                 if (list_empty(&rxbp->rxbp_list))
 212                         break;
 213
 214                 spin_unlock_irqrestore(&rxbp->rxbp_lock, flags);
 215
 216                 /* Wait a bit for references to be dropped */
 217                 CDEBUG(((i & (-i)) == i) ? D_WARNING : D_NET, /* power of 2? */
 218                        "Waiting for %d Busy RX Buffers\n",
 219                        rxbp->rxbp_count);
 220
 221                 cfs_pause(cfs_time_seconds(1));
 222
 223                 spin_lock_irqsave(&rxbp->rxbp_lock, flags);
 224         }
 225
 226         spin_unlock_irqrestore(&rxbp->rxbp_lock, flags);
 227 }
 228
 229 void
 230 kptllnd_rx_buffer_post(kptl_rx_buffer_t *rxb)
 231 {
 232         int                     rc;
 233         ptl_md_t                md;
 234         ptl_handle_me_t         meh;
 235         ptl_handle_md_t         mdh;
 236         ptl_process_id_t        any;
 237         kptl_rx_buffer_pool_t  *rxbp = rxb->rxb_pool;
 238         unsigned long           flags;
 239
 240         LASSERT (!in_interrupt());
 241         LASSERT (rxb->rxb_refcount == 0);
 242         LASSERT (!rxb->rxb_idle);
 243         LASSERT (!rxb->rxb_posted);
 244         LASSERT (PtlHandleIsEqual(rxb->rxb_mdh, PTL_INVALID_HANDLE));
 245
 246         any.nid = PTL_NID_ANY;
 247         any.pid = PTL_PID_ANY;
 248
 249         spin_lock_irqsave(&rxbp->rxbp_lock, flags);
 250
 251         if (rxbp->rxbp_shutdown) {
 252                 rxb->rxb_idle = 1;
 253                 spin_unlock_irqrestore(&rxbp->rxbp_lock, flags);
 254                 return;
 255         }
 256
 257         rxb->rxb_refcount = 1;                  /* net's ref */
 258         rxb->rxb_posted = 1;                    /* I'm posting */
 259
 260         spin_unlock_irqrestore(&rxbp->rxbp_lock, flags);
 261
 262         rc = PtlMEAttach(kptllnd_data.kptl_nih,
 263                          *kptllnd_tunables.kptl_portal,
 264                          any,
 265                          LNET_MSG_MATCHBITS,
 266                          0, /* all matchbits are valid - ignore none */
 267                          PTL_UNLINK,
 268                          PTL_INS_AFTER,
 269                          &meh);
 270         if (rc != PTL_OK) {
 271                 CERROR("PtlMeAttach rxb failed %d\n", rc);
 272                 goto failed;
 273         }
 274
 275         /*
 276          * Setup MD
 277          */
 278         md.start = rxb->rxb_buffer;
 279         md.length = PAGE_SIZE * *kptllnd_tunables.kptl_rxb_npages;
 280         md.threshold = PTL_MD_THRESH_INF;
 281         md.options = PTL_MD_OP_PUT |
 282                      PTL_MD_LUSTRE_COMPLETION_SEMANTICS |
 283                      PTL_MD_EVENT_START_DISABLE |
 284                      PTL_MD_MAX_SIZE |
 285                      PTL_MD_LOCAL_ALIGN8;
 286         md.user_ptr = &rxb->rxb_eventarg;
 287         md.max_size = *kptllnd_tunables.kptl_max_msg_size;
 288         md.eq_handle = kptllnd_data.kptl_eqh;
 289
 290         rc = PtlMDAttach(meh, md, PTL_UNLINK, &mdh);
 291         if (rc == PTL_OK) {
 292                 spin_lock_irqsave(&rxbp->rxbp_lock, flags);
 293                 if (rxb->rxb_posted)            /* Not auto-unlinked yet!!! */
 294                         rxb->rxb_mdh = mdh;
 295                 spin_unlock_irqrestore(&rxbp->rxbp_lock, flags);
 296                 return;
 297         }
 298
 299         CERROR("PtlMDAttach rxb failed %d\n", rc);
 300         rc = PtlMEUnlink(meh);
 301         LASSERT(rc == PTL_OK);
 302
 303  failed:
 304         spin_lock_irqsave(&rxbp->rxbp_lock, flags);
 305         rxb->rxb_posted = 0;
 306         /* XXX this will just try again immediately */
 307         kptllnd_rx_buffer_decref_locked(rxb);
 308         spin_unlock_irqrestore(&rxbp->rxbp_lock, flags);
 309 }
 310
 311 kptl_rx_t *
 312 kptllnd_rx_alloc(void)
 313 {
 314         kptl_rx_t* rx;
 315
 316         if (IS_SIMULATION_ENABLED(FAIL_RX_ALLOC)) {
 317                 CERROR ("FAIL_RX_ALLOC SIMULATION triggered\n");
 318                 return NULL;
 319         }
 320
 321         rx = cfs_mem_cache_alloc(kptllnd_data.kptl_rx_cache, CFS_ALLOC_ATOMIC);
 322         if (rx == NULL) {
 323                 CERROR("Failed to allocate rx\n");
 324                 return NULL;
 325         }
 326
 327         memset(rx, 0, sizeof(*rx));
 328         return rx;
 329 }
 330
 331 void
 332 kptllnd_rx_done(kptl_rx_t *rx)
 333 {
 334         kptl_rx_buffer_t *rxb = rx->rx_rxb;
 335         kptl_peer_t      *peer = rx->rx_peer;
 336         unsigned long     flags;
 337
 338         CDEBUG(D_NET, "rx=%p rxb %p peer %p\n", rx, rxb, peer);
 339
 340         if (rxb != NULL)
 341                 kptllnd_rx_buffer_decref(rxb);
 342
 343         if (peer != NULL) {
 344                 /* Update credits (after I've decref-ed the buffer) */
 345                 spin_lock_irqsave(&peer->peer_lock, flags);
 346
 347                 peer->peer_active_rxs--;
 348                 LASSERT (peer->peer_active_rxs >= 0);
 349
 350                 peer->peer_outstanding_credits++;
 351                 LASSERT (peer->peer_outstanding_credits <=
 352                          *kptllnd_tunables.kptl_peercredits);
 353
 354                 CDEBUG(D_NETTRACE, "%s[%d/%d]: rx %p done\n",
 355                        libcfs_id2str(peer->peer_id),
 356                        peer->peer_credits, peer->peer_outstanding_credits, rx);
 357
 358                 spin_unlock_irqrestore(&peer->peer_lock, flags);
 359
 360                 /* I might have to send back credits */
 361                 kptllnd_peer_check_sends(peer);
 362                 kptllnd_peer_decref(peer);
 363         }
 364
 365         cfs_mem_cache_free(kptllnd_data.kptl_rx_cache, rx);
 366 }
 367
 368 void
 369 kptllnd_rx_buffer_callback (ptl_event_t *ev)
 370 {
 371         kptl_eventarg_t        *eva = ev->md.user_ptr;
 372         kptl_rx_buffer_t       *rxb = kptllnd_eventarg2obj(eva);
 373         kptl_rx_buffer_pool_t  *rxbp = rxb->rxb_pool;
 374         kptl_rx_t              *rx;
 375         int                     unlinked;
 376         unsigned long           flags;
 377
 378 #ifdef LUSTRE_PORTALS_UNLINK_SEMANTICS
 379         unlinked = ev->unlinked;
 380 #else
 381         unlinked = ev->type == PTL_EVENT_UNLINK;
 382 #endif
 383
 384         CDEBUG(D_NET, "RXB Callback %s(%d) rxb=%p id=%s unlink=%d rc %d\n",
 385                kptllnd_evtype2str(ev->type), ev->type, rxb,
 386                kptllnd_ptlid2str(ev->initiator),
 387                unlinked, ev->ni_fail_type);
 388
 389         LASSERT (!rxb->rxb_idle);
 390         LASSERT (ev->md.start == rxb->rxb_buffer);
 391         LASSERT (ev->offset + ev->mlength <=
 392                  PAGE_SIZE * *kptllnd_tunables.kptl_rxb_npages);
 393         LASSERT (ev->type == PTL_EVENT_PUT_END ||
 394                  ev->type == PTL_EVENT_UNLINK);
 395         LASSERT (ev->type == PTL_EVENT_UNLINK ||
 396                  ev->match_bits == LNET_MSG_MATCHBITS);
 397
 398         if (ev->ni_fail_type != PTL_NI_OK)
 399                 CERROR("event type %d, status %d from %s\n",
 400                        ev->type, ev->ni_fail_type,
 401                        kptllnd_ptlid2str(ev->initiator));
 402
 403         if (ev->type == PTL_EVENT_PUT_END &&
 404             ev->ni_fail_type == PTL_NI_OK &&
 405             !rxbp->rxbp_shutdown) {
 406
 407                 /* rxbp_shutdown sampled without locking!  I only treat it as a
 408                  * hint since shutdown can start while rx's are queued on
 409                  * kptl_sched_rxq. */
 410 #if (PTL_MD_LOCAL_ALIGN8 == 0)
 411                 /* Portals can't force message alignment - someone sending an
 412                  * odd-length message will misalign subsequent messages and
 413                  * force the fixup below...  */
 414                 if ((ev->mlength & 7) != 0)
 415                         CWARN("Message from %s has odd length %d: "
 416                               "probable version incompatibility\n",
 417                               kptllnd_ptlid2str(ev->initiator),
 418                               ev->mlength);
 419 #endif
 420                 rx = kptllnd_rx_alloc();
 421                 if (rx == NULL) {
 422                         CERROR("Message from %s dropped: ENOMEM",
 423                                kptllnd_ptlid2str(ev->initiator));
 424                 } else {
 425                         if ((ev->offset & 7) == 0) {
 426                                 kptllnd_rx_buffer_addref(rxb);
 427                                 rx->rx_rxb = rxb;
 428                                 rx->rx_nob = ev->mlength;
 429                                 rx->rx_msg = (kptl_msg_t *)
 430                                              (rxb->rxb_buffer + ev->offset);
 431                         } else {
 432 #if (PTL_MD_LOCAL_ALIGN8 == 0)
 433                                 /* Portals can't force alignment - copy into
 434                                  * rx_space (avoiding overflow) to fix */
 435                                 int maxlen = *kptllnd_tunables.kptl_max_msg_size;
 436
 437                                 rx->rx_rxb = NULL;
 438                                 rx->rx_nob = MIN(maxlen, ev->mlength);
 439                                 rx->rx_msg = (kptl_msg_t *)rx->rx_space;
 440                                 memcpy(rx->rx_msg, rxb->rxb_buffer + ev->offset,
 441                                        rx->rx_nob);
 442 #else
 443                                 /* Portals should have forced the alignment */
 444                                 LBUG();
 445 #endif
 446                         }
 447
 448                         rx->rx_initiator = ev->initiator;
 449 #ifdef CRAY_XT3
 450                         rx->rx_uid = ev->uid;
 451 #endif
 452                         /* Queue for attention */
 453                         spin_lock_irqsave(&kptllnd_data.kptl_sched_lock,
 454                                           flags);
 455
 456                         list_add_tail(&rx->rx_list,
 457                                       &kptllnd_data.kptl_sched_rxq);
 458                         wake_up(&kptllnd_data.kptl_sched_waitq);
 459
 460                         spin_unlock_irqrestore(&kptllnd_data.kptl_sched_lock,
 461                                                flags);
 462                 }
 463         }
 464
 465         if (unlinked) {
 466                 spin_lock_irqsave(&rxbp->rxbp_lock, flags);
 467
 468                 rxb->rxb_posted = 0;
 469                 rxb->rxb_mdh = PTL_INVALID_HANDLE;
 470                 kptllnd_rx_buffer_decref_locked(rxb);
 471
 472                 spin_unlock_irqrestore(&rxbp->rxbp_lock, flags);
 473         }
 474 }
 475
 476 void
 477 kptllnd_nak (kptl_rx_t *rx)
 478 {
 479         /* Fire-and-forget a stub message that will let the peer know my
 480          * protocol magic/version and make her drop/refresh any peer state she
 481          * might have with me. */
 482         ptl_md_t md = {
 483                 .start        = kptllnd_data.kptl_nak_msg,
 484                 .length       = kptllnd_data.kptl_nak_msg->ptlm_nob,
 485                 .threshold    = 1,
 486                 .options      = 0,
 487                 .user_ptr     = NULL,
 488                 .eq_handle    = PTL_EQ_NONE};
 489         ptl_handle_md_t   mdh;
 490         int               rc;
 491
 492         rc = PtlMDBind(kptllnd_data.kptl_nih, md, PTL_UNLINK, &mdh);
 493         if (rc != PTL_OK) {
 494                 CWARN("Can't NAK %s: bind failed %d\n",
 495                       kptllnd_ptlid2str(rx->rx_initiator), rc);
 496                 return;
 497         }
 498
 499         rc = PtlPut(mdh, PTL_NOACK_REQ, rx->rx_initiator,
 500                     *kptllnd_tunables.kptl_portal, 0,
 501                     LNET_MSG_MATCHBITS, 0, 0);
 502
 503         if (rc != PTL_OK)
 504                 CWARN("Can't NAK %s: put failed %d\n",
 505                       kptllnd_ptlid2str(rx->rx_initiator), rc);
 506 }
 507
 508 void
 509 kptllnd_rx_parse(kptl_rx_t *rx)
 510 {
 511         kptl_msg_t             *msg = rx->rx_msg;
 512         kptl_peer_t            *peer;
 513         int                     rc;
 514         int                     credits;
 515         unsigned long           flags;
 516         lnet_process_id_t       srcid;
 517
 518         LASSERT (rx->rx_peer == NULL);
 519
 520         if ((rx->rx_nob >= 4 &&
 521              (msg->ptlm_magic == LNET_PROTO_MAGIC ||
 522               msg->ptlm_magic == __swab32(LNET_PROTO_MAGIC))) ||
 523             (rx->rx_nob >= 6 &&
 524              ((msg->ptlm_magic == PTLLND_MSG_MAGIC &&
 525                msg->ptlm_version != PTLLND_MSG_VERSION) ||
 526               (msg->ptlm_magic == __swab32(PTLLND_MSG_MAGIC) &&
 527                msg->ptlm_version != __swab16(PTLLND_MSG_VERSION))))) {
 528                 /* NAK incompatible versions
 529                  * See other LNDs for how to handle this if/when ptllnd begins
 530                  * to allow different versions to co-exist */
 531                 CERROR("Bad version: got %04x expected %04x from %s\n",
 532                        (__u32)(msg->ptlm_magic == PTLLND_MSG_MAGIC ?
 533                                msg->ptlm_version : __swab16(msg->ptlm_version)),
 534                         PTLLND_MSG_VERSION, kptllnd_ptlid2str(rx->rx_initiator));
 535                 kptllnd_nak(rx);
 536                 goto rx_done;
 537         }
 538
 539         rc = kptllnd_msg_unpack(msg, rx->rx_nob);
 540         if (rc != 0) {
 541                 CERROR ("Error %d unpacking rx from %s\n",
 542                         rc, kptllnd_ptlid2str(rx->rx_initiator));
 543                 goto rx_done;
 544         }
 545
 546         srcid.nid = msg->ptlm_srcnid;
 547         srcid.pid = msg->ptlm_srcpid;
 548
 549         CDEBUG(D_NETTRACE, "%s: RX %s c %d %p\n", libcfs_id2str(srcid),
 550                kptllnd_msgtype2str(msg->ptlm_type), msg->ptlm_credits, rx);
 551
 552         if (srcid.nid != kptllnd_ptl2lnetnid(rx->rx_initiator.nid)) {
 553                 CERROR("Bad source id %s from %s\n",
 554                        libcfs_id2str(srcid),
 555                        kptllnd_ptlid2str(rx->rx_initiator));
 556                 goto rx_done;
 557         }
 558
 559         if (msg->ptlm_type == PTLLND_MSG_TYPE_NAK) {
 560                 peer = kptllnd_id2peer(srcid);
 561                 if (peer == NULL)
 562                         goto rx_done;
 563
 564                 CWARN("NAK from %s (%s)\n",
 565                       libcfs_id2str(srcid),
 566                       kptllnd_ptlid2str(rx->rx_initiator));
 567
 568                 rc = -EPROTO;
 569                 goto failed;
 570         }
 571
 572         if (msg->ptlm_dstnid != kptllnd_data.kptl_ni->ni_nid ||
 573             msg->ptlm_dstpid != the_lnet.ln_pid) {
 574                 CERROR("Bad dstid %s (expected %s) from %s\n",
 575                        libcfs_id2str((lnet_process_id_t) {
 576                                .nid = msg->ptlm_dstnid,
 577                                .pid = msg->ptlm_dstpid}),
 578                        libcfs_id2str((lnet_process_id_t) {
 579                                .nid = kptllnd_data.kptl_ni->ni_nid,
 580                                .pid = the_lnet.ln_pid}),
 581                        kptllnd_ptlid2str(rx->rx_initiator));
 582                 goto rx_done;
 583         }
 584
 585         if (msg->ptlm_type == PTLLND_MSG_TYPE_HELLO) {
 586                 peer = kptllnd_peer_handle_hello(rx->rx_initiator, msg);
 587                 if (peer == NULL)
 588                         goto rx_done;
 589         } else {
 590                 peer = kptllnd_id2peer(srcid);
 591                 if (peer == NULL) {
 592                         CWARN("NAK %s: no connection; peer must reconnect\n",
 593                               libcfs_id2str(srcid));
 594                         /* NAK to make the peer reconnect */
 595                         kptllnd_nak(rx);
 596                         goto rx_done;
 597                 }
 598
 599                 /* Ignore anything apart from HELLO while I'm waiting for it and
 600                  * any messages for a previous incarnation of the connection */
 601                 if (peer->peer_state == PEER_STATE_WAITING_HELLO ||
 602                     msg->ptlm_dststamp < peer->peer_myincarnation) {
 603                         kptllnd_peer_decref(peer);
 604                         goto rx_done;
 605                 }
 606
 607                 if (msg->ptlm_srcstamp != peer->peer_incarnation) {
 608                         CERROR("%s: Unexpected srcstamp "LPX64" "
 609                                "("LPX64" expected)\n",
 610                                libcfs_id2str(peer->peer_id),
 611                                msg->ptlm_srcstamp,
 612                                peer->peer_incarnation);
 613                         rc = -EPROTO;
 614                         goto failed;
 615                 }
 616
 617                 if (msg->ptlm_dststamp != peer->peer_myincarnation) {
 618                         CERROR("%s: Unexpected dststamp "LPX64" "
 619                                "("LPX64" expected)\n",
 620                                libcfs_id2str(peer->peer_id), msg->ptlm_dststamp,
 621                                peer->peer_myincarnation);
 622                         rc = -EPROTO;
 623                         goto failed;
 624                 }
 625         }
 626
 627         LASSERT (msg->ptlm_srcnid == peer->peer_id.nid &&
 628                  msg->ptlm_srcpid == peer->peer_id.pid);
 629
 630         spin_lock_irqsave(&peer->peer_lock, flags);
 631
 632         if (peer->peer_active_rxs == *kptllnd_tunables.kptl_peercredits) {
 633                 spin_unlock_irqrestore(&peer->peer_lock, flags);
 634
 635                 CERROR("Message overflow from %s: handling %d already\n",
 636                        libcfs_id2str(peer->peer_id),
 637                        *kptllnd_tunables.kptl_peercredits);
 638                 rc = -EPROTO;
 639                 goto failed;
 640         }
 641
 642         if (msg->ptlm_credits != 0 &&
 643             peer->peer_credits + msg->ptlm_credits >
 644             *kptllnd_tunables.kptl_peercredits) {
 645                 credits = peer->peer_credits;
 646                 spin_unlock_irqrestore(&peer->peer_lock, flags);
 647
 648                 CERROR("Credit overflow from %s: %d + %d > %d\n",
 649                        libcfs_id2str(peer->peer_id),
 650                        credits, msg->ptlm_credits,
 651                        *kptllnd_tunables.kptl_peercredits);
 652                 rc = -EPROTO;
 653                 goto failed;
 654         }
 655
 656         /* ptllnd-level protocol correct: account credits */
 657         peer->peer_credits += msg->ptlm_credits;
 658         peer->peer_active_rxs++;
 659
 660         spin_unlock_irqrestore(&peer->peer_lock, flags);
 661
 662         /* See if something can go out now that credits have come in */
 663         if (msg->ptlm_credits != 0)
 664                 kptllnd_peer_check_sends(peer);
 665
 666         rx->rx_peer = peer;                /* do buffer accounting on rxdone */
 667         kptllnd_peer_alive(peer);
 668
 669         switch (msg->ptlm_type) {
 670         default:
 671                 /* already checked by kptllnd_msg_unpack() */
 672                 LBUG();
 673
 674         case PTLLND_MSG_TYPE_HELLO:
 675                 CDEBUG(D_NET, "PTLLND_MSG_TYPE_HELLO\n");
 676                 goto rx_done;
 677
 678         case PTLLND_MSG_TYPE_NOOP:
 679                 CDEBUG(D_NET, "PTLLND_MSG_TYPE_NOOP\n");
 680                 goto rx_done;
 681
 682         case PTLLND_MSG_TYPE_IMMEDIATE:
 683                 CDEBUG(D_NET, "PTLLND_MSG_TYPE_IMMEDIATE\n");
 684                 rc = lnet_parse(kptllnd_data.kptl_ni,
 685                                 &msg->ptlm_u.immediate.kptlim_hdr,
 686                                 msg->ptlm_srcnid,
 687                                 rx, 0);
 688                 if (rc >= 0)                    /* kptllnd_recv owns 'rx' now */
 689                         return;
 690                 goto failed;
 691
 692         case PTLLND_MSG_TYPE_PUT:
 693         case PTLLND_MSG_TYPE_GET:
 694                 CDEBUG(D_NET, "PTLLND_MSG_TYPE_%s\n",
 695                         msg->ptlm_type == PTLLND_MSG_TYPE_PUT ?
 696                         "PUT" : "GET");
 697
 698                 /* checked in kptllnd_msg_unpack() */
 699                 LASSERT (msg->ptlm_u.rdma.kptlrm_matchbits >=
 700                          PTL_RESERVED_MATCHBITS);
 701
 702                 /* Update last match bits seen */
 703                 spin_lock_irqsave(&peer->peer_lock, flags);
 704
 705                 if (msg->ptlm_u.rdma.kptlrm_matchbits >
 706                     rx->rx_peer->peer_last_matchbits_seen)
 707                         rx->rx_peer->peer_last_matchbits_seen =
 708                                 msg->ptlm_u.rdma.kptlrm_matchbits;
 709
 710                 spin_unlock_irqrestore(&rx->rx_peer->peer_lock, flags);
 711
 712                 rc = lnet_parse(kptllnd_data.kptl_ni,
 713                                 &msg->ptlm_u.rdma.kptlrm_hdr,
 714                                 msg->ptlm_srcnid,
 715                                 rx, 1);
 716                 if (rc >= 0)                    /* kptllnd_recv owns 'rx' now */
 717                         return;
 718                 goto failed;
 719          }
 720
 721  failed:
 722         kptllnd_peer_close(peer, rc);
 723         if (rx->rx_peer == NULL)                /* drop ref on peer */
 724                 kptllnd_peer_decref(peer);      /* unless rx_done will */
 725  rx_done:
 726         kptllnd_rx_done(rx);
 727 }