1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
2 * vim:expandtab:shiftwidth=8:tabstop=8:
6 * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
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.
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).
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
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
29 * Copyright 2008 Sun Microsystems, Inc. All rights reserved
30 * Use is subject to license terms.
33 * This file is part of Lustre, http://www.lustre.org/
34 * Lustre is a trademark of Sun Microsystems, Inc.
36 * lnet/klnds/ptllnd/ptllnd_rx_buf.c
38 * Author: PJ Kirner <pjkirner@clusterfs.com>
44 kptllnd_rx_buffer_pool_init(kptl_rx_buffer_pool_t *rxbp)
46 memset(rxbp, 0, sizeof(*rxbp));
47 cfs_spin_lock_init(&rxbp->rxbp_lock);
48 CFS_INIT_LIST_HEAD(&rxbp->rxbp_list);
52 kptllnd_rx_buffer_destroy(kptl_rx_buffer_t *rxb)
54 kptl_rx_buffer_pool_t *rxbp = rxb->rxb_pool;
56 LASSERT(rxb->rxb_refcount == 0);
57 LASSERT(PtlHandleIsEqual(rxb->rxb_mdh, PTL_INVALID_HANDLE));
58 LASSERT(!rxb->rxb_posted);
59 LASSERT(rxb->rxb_idle);
61 cfs_list_del(&rxb->rxb_list);
64 LIBCFS_FREE(rxb->rxb_buffer, kptllnd_rx_buffer_size());
65 LIBCFS_FREE(rxb, sizeof(*rxb));
69 kptllnd_rx_buffer_pool_reserve(kptl_rx_buffer_pool_t *rxbp, int count)
74 kptl_rx_buffer_t *rxb;
78 bufsize = kptllnd_rx_buffer_size();
79 msgs_per_buffer = bufsize / (*kptllnd_tunables.kptl_max_msg_size);
81 CDEBUG(D_NET, "kptllnd_rx_buffer_pool_reserve(%d)\n", count);
83 cfs_spin_lock_irqsave(&rxbp->rxbp_lock, flags);
86 if (rxbp->rxbp_shutdown) {
91 if (rxbp->rxbp_reserved + count <=
92 rxbp->rxbp_count * msgs_per_buffer) {
97 cfs_spin_unlock_irqrestore(&rxbp->rxbp_lock, flags);
99 LIBCFS_ALLOC(rxb, sizeof(*rxb));
100 LIBCFS_ALLOC(buffer, bufsize);
102 if (rxb == NULL || buffer == NULL) {
103 CERROR("Failed to allocate rx buffer\n");
106 LIBCFS_FREE(rxb, sizeof(*rxb));
108 LIBCFS_FREE(buffer, bufsize);
110 cfs_spin_lock_irqsave(&rxbp->rxbp_lock, flags);
115 memset(rxb, 0, sizeof(*rxb));
117 rxb->rxb_eventarg.eva_type = PTLLND_EVENTARG_TYPE_BUF;
118 rxb->rxb_refcount = 0;
119 rxb->rxb_pool = rxbp;
122 rxb->rxb_buffer = buffer;
123 rxb->rxb_mdh = PTL_INVALID_HANDLE;
125 cfs_spin_lock_irqsave(&rxbp->rxbp_lock, flags);
127 if (rxbp->rxbp_shutdown) {
128 cfs_spin_unlock_irqrestore(&rxbp->rxbp_lock, flags);
130 LIBCFS_FREE(rxb, sizeof(*rxb));
131 LIBCFS_FREE(buffer, bufsize);
133 cfs_spin_lock_irqsave(&rxbp->rxbp_lock, flags);
138 cfs_list_add_tail(&rxb->rxb_list, &rxbp->rxbp_list);
141 cfs_spin_unlock_irqrestore(&rxbp->rxbp_lock, flags);
143 kptllnd_rx_buffer_post(rxb);
145 cfs_spin_lock_irqsave(&rxbp->rxbp_lock, flags);
149 rxbp->rxbp_reserved += count;
151 cfs_spin_unlock_irqrestore(&rxbp->rxbp_lock, flags);
157 kptllnd_rx_buffer_pool_unreserve(kptl_rx_buffer_pool_t *rxbp,
162 cfs_spin_lock_irqsave(&rxbp->rxbp_lock, flags);
164 CDEBUG(D_NET, "kptllnd_rx_buffer_pool_unreserve(%d)\n", count);
165 rxbp->rxbp_reserved -= count;
167 cfs_spin_unlock_irqrestore(&rxbp->rxbp_lock, flags);
171 kptllnd_rx_buffer_pool_fini(kptl_rx_buffer_pool_t *rxbp)
173 kptl_rx_buffer_t *rxb;
181 /* CAVEAT EMPTOR: I'm racing with everything here!!!
183 * Buffers can still be posted after I set rxbp_shutdown because I
184 * can't hold rxbp_lock while I'm posting them.
186 * Calling PtlMDUnlink() here races with auto-unlinks; i.e. a buffer's
187 * MD handle could become invalid under me. I am vulnerable to portals
188 * re-using handles (i.e. make the same handle valid again, but for a
189 * different MD) from when the MD is actually unlinked, to when the
190 * event callback tells me it has been unlinked. */
192 cfs_spin_lock_irqsave(&rxbp->rxbp_lock, flags);
194 rxbp->rxbp_shutdown = 1;
197 cfs_list_for_each_safe(tmp, nxt, &rxbp->rxbp_list) {
198 rxb = cfs_list_entry (tmp, kptl_rx_buffer_t, rxb_list);
201 cfs_spin_unlock_irqrestore(&rxbp->rxbp_lock,
203 kptllnd_rx_buffer_destroy(rxb);
204 cfs_spin_lock_irqsave(&rxbp->rxbp_lock,
210 if (PtlHandleIsEqual(mdh, PTL_INVALID_HANDLE))
213 cfs_spin_unlock_irqrestore(&rxbp->rxbp_lock, flags);
215 rc = PtlMDUnlink(mdh);
217 cfs_spin_lock_irqsave(&rxbp->rxbp_lock, flags);
219 #ifdef LUSTRE_PORTALS_UNLINK_SEMANTICS
220 /* callback clears rxb_mdh and drops net's ref
221 * (which causes repost, but since I set
222 * shutdown, it will just set the buffer
227 rxb->rxb_mdh = PTL_INVALID_HANDLE;
228 kptllnd_rx_buffer_decref_locked(rxb);
233 if (cfs_list_empty(&rxbp->rxbp_list))
236 cfs_spin_unlock_irqrestore(&rxbp->rxbp_lock, flags);
238 /* Wait a bit for references to be dropped */
239 CDEBUG(((i & (-i)) == i) ? D_WARNING : D_NET, /* power of 2? */
240 "Waiting for %d Busy RX Buffers\n",
243 cfs_pause(cfs_time_seconds(1));
245 cfs_spin_lock_irqsave(&rxbp->rxbp_lock, flags);
248 cfs_spin_unlock_irqrestore(&rxbp->rxbp_lock, flags);
252 kptllnd_rx_buffer_post(kptl_rx_buffer_t *rxb)
258 ptl_process_id_t any;
259 kptl_rx_buffer_pool_t *rxbp = rxb->rxb_pool;
262 LASSERT (!cfs_in_interrupt());
263 LASSERT (rxb->rxb_refcount == 0);
264 LASSERT (!rxb->rxb_idle);
265 LASSERT (!rxb->rxb_posted);
266 LASSERT (PtlHandleIsEqual(rxb->rxb_mdh, PTL_INVALID_HANDLE));
268 any.nid = PTL_NID_ANY;
269 any.pid = PTL_PID_ANY;
271 cfs_spin_lock_irqsave(&rxbp->rxbp_lock, flags);
273 if (rxbp->rxbp_shutdown) {
275 cfs_spin_unlock_irqrestore(&rxbp->rxbp_lock, flags);
279 rxb->rxb_refcount = 1; /* net's ref */
280 rxb->rxb_posted = 1; /* I'm posting */
282 cfs_spin_unlock_irqrestore(&rxbp->rxbp_lock, flags);
284 rc = PtlMEAttach(kptllnd_data.kptl_nih,
285 *kptllnd_tunables.kptl_portal,
288 0, /* all matchbits are valid - ignore none */
293 CERROR("PtlMeAttach rxb failed %s(%d)\n",
294 kptllnd_errtype2str(rc), rc);
301 md.start = rxb->rxb_buffer;
302 md.length = kptllnd_rx_buffer_size();
303 md.threshold = PTL_MD_THRESH_INF;
304 md.options = PTL_MD_OP_PUT |
305 PTL_MD_LUSTRE_COMPLETION_SEMANTICS |
306 PTL_MD_EVENT_START_DISABLE |
309 md.user_ptr = &rxb->rxb_eventarg;
310 md.max_size = *kptllnd_tunables.kptl_max_msg_size;
311 md.eq_handle = kptllnd_data.kptl_eqh;
313 rc = PtlMDAttach(meh, md, PTL_UNLINK, &mdh);
315 cfs_spin_lock_irqsave(&rxbp->rxbp_lock, flags);
316 if (rxb->rxb_posted) /* Not auto-unlinked yet!!! */
318 cfs_spin_unlock_irqrestore(&rxbp->rxbp_lock, flags);
322 CERROR("PtlMDAttach rxb failed %s(%d)\n",
323 kptllnd_errtype2str(rc), rc);
324 rc = PtlMEUnlink(meh);
325 LASSERT(rc == PTL_OK);
328 cfs_spin_lock_irqsave(&rxbp->rxbp_lock, flags);
330 /* XXX this will just try again immediately */
331 kptllnd_rx_buffer_decref_locked(rxb);
332 cfs_spin_unlock_irqrestore(&rxbp->rxbp_lock, flags);
336 kptllnd_rx_alloc(void)
340 if (IS_SIMULATION_ENABLED(FAIL_RX_ALLOC)) {
341 CERROR ("FAIL_RX_ALLOC SIMULATION triggered\n");
345 rx = cfs_mem_cache_alloc(kptllnd_data.kptl_rx_cache, CFS_ALLOC_ATOMIC);
347 CERROR("Failed to allocate rx\n");
351 memset(rx, 0, sizeof(*rx));
356 kptllnd_rx_done(kptl_rx_t *rx, int post_credit)
358 kptl_rx_buffer_t *rxb = rx->rx_rxb;
359 kptl_peer_t *peer = rx->rx_peer;
362 LASSERT (post_credit == PTLLND_POSTRX_NO_CREDIT ||
363 post_credit == PTLLND_POSTRX_PEER_CREDIT);
365 CDEBUG(D_NET, "rx=%p rxb %p peer %p\n", rx, rxb, peer);
368 kptllnd_rx_buffer_decref(rxb);
371 /* Update credits (after I've decref-ed the buffer) */
372 cfs_spin_lock_irqsave(&peer->peer_lock, flags);
374 if (post_credit == PTLLND_POSTRX_PEER_CREDIT)
375 peer->peer_outstanding_credits++;
377 LASSERT (peer->peer_outstanding_credits +
378 peer->peer_sent_credits <=
379 *kptllnd_tunables.kptl_peertxcredits);
381 CDEBUG(D_NETTRACE, "%s[%d/%d+%d]: rx %p done\n",
382 libcfs_id2str(peer->peer_id), peer->peer_credits,
383 peer->peer_outstanding_credits, peer->peer_sent_credits,
386 cfs_spin_unlock_irqrestore(&peer->peer_lock, flags);
388 /* I might have to send back credits */
389 kptllnd_peer_check_sends(peer);
390 kptllnd_peer_decref(peer);
393 cfs_mem_cache_free(kptllnd_data.kptl_rx_cache, rx);
397 kptllnd_rx_buffer_callback (ptl_event_t *ev)
399 kptl_eventarg_t *eva = ev->md.user_ptr;
400 kptl_rx_buffer_t *rxb = kptllnd_eventarg2obj(eva);
401 kptl_rx_buffer_pool_t *rxbp = rxb->rxb_pool;
406 #ifdef LUSTRE_PORTALS_UNLINK_SEMANTICS
407 unlinked = ev->unlinked;
409 unlinked = ev->type == PTL_EVENT_UNLINK;
412 CDEBUG(D_NET, "%s: %s(%d) rxb=%p fail=%s(%d) unlink=%d\n",
413 kptllnd_ptlid2str(ev->initiator),
414 kptllnd_evtype2str(ev->type), ev->type, rxb,
415 kptllnd_errtype2str(ev->ni_fail_type), ev->ni_fail_type,
418 LASSERT (!rxb->rxb_idle);
419 LASSERT (ev->md.start == rxb->rxb_buffer);
420 LASSERT (ev->offset + ev->mlength <=
421 PAGE_SIZE * *kptllnd_tunables.kptl_rxb_npages);
422 LASSERT (ev->type == PTL_EVENT_PUT_END ||
423 ev->type == PTL_EVENT_UNLINK);
424 LASSERT (ev->type == PTL_EVENT_UNLINK ||
425 ev->match_bits == LNET_MSG_MATCHBITS);
427 if (ev->ni_fail_type != PTL_NI_OK) {
428 CERROR("Portals error from %s: %s(%d) rxb=%p fail=%s(%d) unlink=%dn",
429 kptllnd_ptlid2str(ev->initiator),
430 kptllnd_evtype2str(ev->type), ev->type, rxb,
431 kptllnd_errtype2str(ev->ni_fail_type),
432 ev->ni_fail_type, unlinked);
433 kptllnd_schedule_ptltrace_dump();
434 } else if (ev->type == PTL_EVENT_PUT_END &&
435 !rxbp->rxbp_shutdown) {
437 /* rxbp_shutdown sampled without locking! I only treat it as a
438 * hint since shutdown can start while rx's are queued on
440 #if (PTL_MD_LOCAL_ALIGN8 == 0)
441 /* Portals can't force message alignment - someone sending an
442 * odd-length message will misalign subsequent messages and
443 * force the fixup below... */
444 if ((ev->mlength & 7) != 0)
445 CWARN("Message from %s has odd length "LPU64": "
446 "probable version incompatibility\n",
447 kptllnd_ptlid2str(ev->initiator),
450 rx = kptllnd_rx_alloc();
452 CERROR("Message from %s dropped: ENOMEM",
453 kptllnd_ptlid2str(ev->initiator));
455 if ((ev->offset & 7) == 0) {
456 kptllnd_rx_buffer_addref(rxb);
458 rx->rx_nob = ev->mlength;
459 rx->rx_msg = (kptl_msg_t *)
460 (rxb->rxb_buffer + ev->offset);
462 #if (PTL_MD_LOCAL_ALIGN8 == 0)
463 /* Portals can't force alignment - copy into
464 * rx_space (avoiding overflow) to fix */
465 int maxlen = *kptllnd_tunables.kptl_max_msg_size;
468 rx->rx_nob = MIN(maxlen, ev->mlength);
469 rx->rx_msg = (kptl_msg_t *)rx->rx_space;
470 memcpy(rx->rx_msg, rxb->rxb_buffer + ev->offset,
473 /* Portals should have forced the alignment */
478 rx->rx_initiator = ev->initiator;
479 rx->rx_treceived = jiffies;
481 rx->rx_uid = ev->uid;
483 /* Queue for attention */
484 cfs_spin_lock_irqsave(&kptllnd_data.kptl_sched_lock,
487 cfs_list_add_tail(&rx->rx_list,
488 &kptllnd_data.kptl_sched_rxq);
489 cfs_waitq_signal(&kptllnd_data.kptl_sched_waitq);
491 cfs_spin_unlock_irqrestore(&kptllnd_data. \
492 kptl_sched_lock, flags);
497 cfs_spin_lock_irqsave(&rxbp->rxbp_lock, flags);
500 rxb->rxb_mdh = PTL_INVALID_HANDLE;
501 kptllnd_rx_buffer_decref_locked(rxb);
503 cfs_spin_unlock_irqrestore(&rxbp->rxbp_lock, flags);
508 kptllnd_nak (ptl_process_id_t dest)
510 /* Fire-and-forget a stub message that will let the peer know my
511 * protocol magic/version and make her drop/refresh any peer state she
512 * might have with me. */
514 .start = kptllnd_data.kptl_nak_msg,
515 .length = kptllnd_data.kptl_nak_msg->ptlm_nob,
519 .eq_handle = PTL_EQ_NONE};
523 rc = PtlMDBind(kptllnd_data.kptl_nih, md, PTL_UNLINK, &mdh);
525 CWARN("Can't NAK %s: bind failed %s(%d)\n",
526 kptllnd_ptlid2str(dest), kptllnd_errtype2str(rc), rc);
530 rc = PtlPut(mdh, PTL_NOACK_REQ, dest,
531 *kptllnd_tunables.kptl_portal, 0,
532 LNET_MSG_MATCHBITS, 0, 0);
534 CWARN("Can't NAK %s: put failed %s(%d)\n",
535 kptllnd_ptlid2str(dest), kptllnd_errtype2str(rc), rc);
536 kptllnd_schedule_ptltrace_dump();
541 kptllnd_find_net (lnet_nid_t nid)
545 cfs_read_lock(&kptllnd_data.kptl_net_rw_lock);
546 cfs_list_for_each_entry (net, &kptllnd_data.kptl_nets, net_list) {
547 LASSERT (!net->net_shutdown);
549 if (net->net_ni->ni_nid == nid) {
550 kptllnd_net_addref(net);
551 cfs_read_unlock(&kptllnd_data.kptl_net_rw_lock);
555 cfs_read_unlock(&kptllnd_data.kptl_net_rw_lock);
561 kptllnd_rx_parse(kptl_rx_t *rx)
563 kptl_msg_t *msg = rx->rx_msg;
565 int post_credit = PTLLND_POSTRX_PEER_CREDIT;
566 kptl_net_t *net = NULL;
570 lnet_process_id_t srcid;
572 LASSERT (!cfs_in_interrupt());
573 LASSERT (rx->rx_peer == NULL);
575 CFS_INIT_LIST_HEAD(&txs);
577 if ((rx->rx_nob >= 4 &&
578 (msg->ptlm_magic == LNET_PROTO_MAGIC ||
579 msg->ptlm_magic == __swab32(LNET_PROTO_MAGIC))) ||
581 ((msg->ptlm_magic == PTLLND_MSG_MAGIC &&
582 msg->ptlm_version != PTLLND_MSG_VERSION) ||
583 (msg->ptlm_magic == __swab32(PTLLND_MSG_MAGIC) &&
584 msg->ptlm_version != __swab16(PTLLND_MSG_VERSION))))) {
585 /* NAK incompatible versions
586 * See other LNDs for how to handle this if/when ptllnd begins
587 * to allow different versions to co-exist */
588 CERROR("Bad version: got %04x expected %04x from %s\n",
589 (__u32)(msg->ptlm_magic == PTLLND_MSG_MAGIC ?
590 msg->ptlm_version : __swab16(msg->ptlm_version)),
591 PTLLND_MSG_VERSION, kptllnd_ptlid2str(rx->rx_initiator));
592 /* NB backward compatibility */
593 kptllnd_nak(rx->rx_initiator);
597 rc = kptllnd_msg_unpack(msg, rx->rx_nob);
599 CERROR ("Error %d unpacking rx from %s\n",
600 rc, kptllnd_ptlid2str(rx->rx_initiator));
604 srcid.nid = msg->ptlm_srcnid;
605 srcid.pid = msg->ptlm_srcpid;
607 CDEBUG(D_NETTRACE, "%s: RX %s c %d %p rxb %p queued %lu ticks (%ld s)\n",
608 libcfs_id2str(srcid), kptllnd_msgtype2str(msg->ptlm_type),
609 msg->ptlm_credits, rx, rx->rx_rxb,
610 jiffies - rx->rx_treceived,
611 cfs_duration_sec(jiffies - rx->rx_treceived));
613 if (kptllnd_lnet2ptlnid(srcid.nid) != rx->rx_initiator.nid) {
614 CERROR("Bad source nid %s from %s\n",
615 libcfs_id2str(srcid),
616 kptllnd_ptlid2str(rx->rx_initiator));
620 if (msg->ptlm_type == PTLLND_MSG_TYPE_NAK) {
621 peer = kptllnd_id2peer(srcid);
625 CWARN("NAK from %s (%d:%s)\n",
626 libcfs_id2str(srcid), peer->peer_state,
627 kptllnd_ptlid2str(rx->rx_initiator));
629 /* NB can't nuke new peer - bug 17546 comment 31 */
630 if (peer->peer_state == PEER_STATE_WAITING_HELLO) {
631 CDEBUG(D_NET, "Stale NAK from %s(%s): WAITING_HELLO\n",
632 libcfs_id2str(srcid),
633 kptllnd_ptlid2str(rx->rx_initiator));
634 kptllnd_peer_decref(peer);
642 net = kptllnd_find_net(msg->ptlm_dstnid);
643 if (net == NULL || msg->ptlm_dstpid != the_lnet.ln_pid) {
644 CERROR("Bad dstid %s from %s\n",
645 libcfs_id2str((lnet_process_id_t) {
646 .nid = msg->ptlm_dstnid,
647 .pid = msg->ptlm_dstpid}),
648 kptllnd_ptlid2str(rx->rx_initiator));
652 if (LNET_NIDNET(srcid.nid) != LNET_NIDNET(net->net_ni->ni_nid)) {
653 lnet_nid_t nid = LNET_MKNID(LNET_NIDNET(net->net_ni->ni_nid),
654 LNET_NIDADDR(srcid.nid));
655 CERROR("Bad source nid %s from %s, %s expected.\n",
656 libcfs_id2str(srcid),
657 kptllnd_ptlid2str(rx->rx_initiator),
658 libcfs_nid2str(nid));
662 if (msg->ptlm_type == PTLLND_MSG_TYPE_HELLO) {
663 peer = kptllnd_peer_handle_hello(net, rx->rx_initiator, msg);
667 peer = kptllnd_id2peer(srcid);
669 CWARN("NAK %s: no connection, %s must reconnect\n",
670 kptllnd_msgtype2str(msg->ptlm_type),
671 libcfs_id2str(srcid));
672 /* NAK to make the peer reconnect */
673 kptllnd_nak(rx->rx_initiator);
677 /* Ignore any messages for a previous incarnation of me */
678 if (msg->ptlm_dststamp < peer->peer_myincarnation) {
679 kptllnd_peer_decref(peer);
683 if (msg->ptlm_dststamp != peer->peer_myincarnation) {
684 CERROR("%s: Unexpected dststamp "LPX64" "
685 "("LPX64" expected)\n",
686 libcfs_id2str(peer->peer_id), msg->ptlm_dststamp,
687 peer->peer_myincarnation);
692 if (peer->peer_state == PEER_STATE_WAITING_HELLO) {
693 /* recoverable error - restart txs */
694 cfs_spin_lock_irqsave(&peer->peer_lock, flags);
695 kptllnd_cancel_txlist(&peer->peer_sendq, &txs);
696 cfs_spin_unlock_irqrestore(&peer->peer_lock, flags);
698 CWARN("NAK %s: Unexpected %s message\n",
699 libcfs_id2str(srcid),
700 kptllnd_msgtype2str(msg->ptlm_type));
701 kptllnd_nak(rx->rx_initiator);
706 if (msg->ptlm_srcstamp != peer->peer_incarnation) {
707 CERROR("%s: Unexpected srcstamp "LPX64" "
708 "("LPX64" expected)\n",
709 libcfs_id2str(srcid),
711 peer->peer_incarnation);
717 LASSERTF (LNET_NIDADDR(msg->ptlm_srcnid) ==
718 LNET_NIDADDR(peer->peer_id.nid), "m %s p %s\n",
719 libcfs_nid2str(msg->ptlm_srcnid),
720 libcfs_nid2str(peer->peer_id.nid));
721 LASSERTF (msg->ptlm_srcpid == peer->peer_id.pid, "m %u p %u\n",
722 msg->ptlm_srcpid, peer->peer_id.pid);
724 cfs_spin_lock_irqsave(&peer->peer_lock, flags);
726 /* Check peer only sends when I've sent her credits */
727 if (peer->peer_sent_credits == 0) {
728 int c = peer->peer_credits;
729 int oc = peer->peer_outstanding_credits;
730 int sc = peer->peer_sent_credits;
732 cfs_spin_unlock_irqrestore(&peer->peer_lock, flags);
734 CERROR("%s: buffer overrun [%d/%d+%d]\n",
735 libcfs_id2str(peer->peer_id), c, sc, oc);
739 peer->peer_sent_credits--;
741 /* No check for credit overflow - the peer may post new
742 * buffers after the startup handshake. */
743 peer->peer_credits += msg->ptlm_credits;
745 /* This ensures the credit taken by NOOP can be returned */
746 if (msg->ptlm_type == PTLLND_MSG_TYPE_NOOP) {
747 peer->peer_outstanding_credits++;
748 post_credit = PTLLND_POSTRX_NO_CREDIT;
751 cfs_spin_unlock_irqrestore(&peer->peer_lock, flags);
753 /* See if something can go out now that credits have come in */
754 if (msg->ptlm_credits != 0)
755 kptllnd_peer_check_sends(peer);
757 /* ptllnd-level protocol correct - rx takes my ref on peer and increments
758 * peer_outstanding_credits when it completes */
760 kptllnd_peer_alive(peer);
762 switch (msg->ptlm_type) {
764 /* already checked by kptllnd_msg_unpack() */
767 case PTLLND_MSG_TYPE_HELLO:
768 CDEBUG(D_NET, "PTLLND_MSG_TYPE_HELLO\n");
771 case PTLLND_MSG_TYPE_NOOP:
772 CDEBUG(D_NET, "PTLLND_MSG_TYPE_NOOP\n");
775 case PTLLND_MSG_TYPE_IMMEDIATE:
776 CDEBUG(D_NET, "PTLLND_MSG_TYPE_IMMEDIATE\n");
777 rc = lnet_parse(net->net_ni,
778 &msg->ptlm_u.immediate.kptlim_hdr,
781 if (rc >= 0) { /* kptllnd_recv owns 'rx' now */
782 kptllnd_net_decref(net);
787 case PTLLND_MSG_TYPE_PUT:
788 case PTLLND_MSG_TYPE_GET:
789 CDEBUG(D_NET, "PTLLND_MSG_TYPE_%s\n",
790 msg->ptlm_type == PTLLND_MSG_TYPE_PUT ?
793 /* checked in kptllnd_msg_unpack() */
794 LASSERT (msg->ptlm_u.rdma.kptlrm_matchbits >=
795 PTL_RESERVED_MATCHBITS);
797 /* Update last match bits seen */
798 cfs_spin_lock_irqsave(&peer->peer_lock, flags);
800 if (msg->ptlm_u.rdma.kptlrm_matchbits >
801 rx->rx_peer->peer_last_matchbits_seen)
802 rx->rx_peer->peer_last_matchbits_seen =
803 msg->ptlm_u.rdma.kptlrm_matchbits;
805 cfs_spin_unlock_irqrestore(&rx->rx_peer->peer_lock, flags);
807 rc = lnet_parse(net->net_ni,
808 &msg->ptlm_u.rdma.kptlrm_hdr,
811 if (rc >= 0) { /* kptllnd_recv owns 'rx' now */
812 kptllnd_net_decref(net);
820 kptllnd_peer_close(peer, rc);
821 if (rx->rx_peer == NULL) /* drop ref on peer */
822 kptllnd_peer_decref(peer); /* unless rx_done will */
823 if (!cfs_list_empty(&txs)) {
824 LASSERT (net != NULL);
825 kptllnd_restart_txs(net, srcid, &txs);
829 kptllnd_net_decref(net);
830 kptllnd_rx_done(rx, post_credit);