/* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
* vim:expandtab:shiftwidth=8:tabstop=8:
*
- * lib/lib-move.c
- * Data movement routines
+ * GPL HEADER START
+ *
+ * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
*
- * Copyright (c) 2001-2003 Cluster File Systems, Inc.
+ * This program is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License version 2 only,
+ * as published by the Free Software Foundation.
*
- * This file is part of Lustre, http://www.lustre.org
+ * This program is distributed in the hope that it will be useful, but
+ * WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+ * General Public License version 2 for more details (a copy is included
+ * in the LICENSE file that accompanied this code).
*
- * Lustre is free software; you can redistribute it and/or
- * modify it under the terms of version 2 of the GNU General Public
- * License as published by the Free Software Foundation.
+ * You should have received a copy of the GNU General Public License
+ * version 2 along with this program; If not, see
+ * http://www.sun.com/software/products/lustre/docs/GPLv2.pdf
*
- * Lustre is distributed in the hope that it will be useful,
- * but WITHOUT ANY WARRANTY; without even the implied warranty of
- * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
- * GNU General Public License for more details.
+ * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
+ * CA 95054 USA or visit www.sun.com if you need additional information or
+ * have any questions.
*
- * You should have received a copy of the GNU General Public License
- * along with Lustre; if not, write to the Free Software
- * Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
+ * GPL HEADER END
+ */
+/*
+ * Copyright 2008 Sun Microsystems, Inc. All rights reserved
+ * Use is subject to license terms.
+ */
+/*
+ * This file is part of Lustre, http://www.lustre.org/
+ * Lustre is a trademark of Sun Microsystems, Inc.
+ *
+ * lnet/lnet/lib-move.c
+ *
+ * Data movement routines
*/
#define DEBUG_SUBSYSTEM S_LNET
/* forward ref */
static void lnet_commit_md (lnet_libmd_t *md, lnet_msg_t *msg);
-static void lnet_drop_delayed_put(lnet_msg_t *msg, char *reason);
#define LNET_MATCHMD_NONE 0 /* Didn't match */
#define LNET_MATCHMD_OK 1 /* Matched OK */
-#define LNET_MATCHMD_DROP 2 /* Must be disarded */
+#define LNET_MATCHMD_DROP 2 /* Must be discarded */
static int
lnet_try_match_md (int index, int op_mask, lnet_process_id_t src,
- unsigned int rlength, unsigned int roffset,
+ unsigned int rlength, unsigned int roffset,
__u64 match_bits, lnet_libmd_t *md, lnet_msg_t *msg,
unsigned int *mlength_out, unsigned int *offset_out)
{
} else if ((md->md_options & LNET_MD_TRUNCATE) == 0) {
/* this packet _really_ is too big */
CERROR("Matching packet from %s, match "LPU64
- " length %d too big: %d left, %d allowed\n",
+ " length %d too big: %d left, %d allowed\n",
libcfs_id2str(src), match_bits, rlength,
md->md_length - offset, mlength);
lnet_libmd_t **md_out)
{
lnet_portal_t *ptl = &the_lnet.ln_portals[index];
- struct list_head *tmp;
lnet_me_t *me;
+ lnet_me_t *tmp;
lnet_libmd_t *md;
int rc;
return LNET_MATCHMD_DROP;
}
- list_for_each (tmp, &ptl->ptl_ml) {
- me = list_entry(tmp, lnet_me_t, me_list);
+ cfs_list_for_each_entry_safe_typed (me, tmp, &ptl->ptl_ml,
+ lnet_me_t, me_list) {
md = me->me_md;
- /* ME attached but MD not attached yet */
+ /* ME attached but MD not attached yet */
if (md == NULL)
continue;
LASSERT (me == md->md_me);
- rc = lnet_try_match_md(index, op_mask, src, rlength,
+ rc = lnet_try_match_md(index, op_mask, src, rlength,
roffset, match_bits, md, msg,
mlength_out, offset_out);
switch (rc) {
default:
LBUG();
-
+
case LNET_MATCHMD_NONE:
continue;
-
+
case LNET_MATCHMD_OK:
*md_out = md;
return LNET_MATCHMD_OK;
-
+
case LNET_MATCHMD_DROP:
return LNET_MATCHMD_DROP;
}
if (op_mask == LNET_MD_OP_GET ||
(ptl->ptl_options & LNET_PTL_LAZY) == 0)
return LNET_MATCHMD_DROP;
-
+
return LNET_MATCHMD_NONE;
}
struct list_head cull;
LASSERT (the_lnet.ln_init);
-
+
if (threshold != 0) {
/* Adding a new entry */
LIBCFS_ALLOC(tp, sizeof(*tp));
ndiov--;
LASSERT (ndiov > 0);
}
-
+
/* skip complete frags before 'soffset' */
LASSERT (nsiov > 0);
while (soffset >= siov->iov_len) {
ndiov--;
doffset = 0;
}
-
+
if (siov->iov_len > soffset + this_nob) {
soffset += this_nob;
} else {
niov = 1;
for (;;) {
LASSERT (src_niov > 0);
- LASSERT (niov <= dst_niov);
+ LASSERT ((int)niov <= dst_niov);
frag_len = src->iov_len - offset;
dst->iov_base = ((char *)src->iov_base) + offset;
LASSERT (!in_interrupt ());
LASSERT (ndiov > 0);
- while (doffset > diov->kiov_len) {
+ while (doffset >= diov->kiov_len) {
doffset -= diov->kiov_len;
diov++;
ndiov--;
}
LASSERT (nsiov > 0);
- while (soffset > siov->kiov_len) {
+ while (soffset >= siov->kiov_len) {
soffset -= siov->kiov_len;
siov++;
nsiov--;
LASSERT (!in_interrupt ());
LASSERT (niov > 0);
- while (iovoffset > iov->iov_len) {
+ while (iovoffset >= iov->iov_len) {
iovoffset -= iov->iov_len;
iov++;
niov--;
}
LASSERT (nkiov > 0);
- while (kiovoffset > kiov->kiov_len) {
+ while (kiovoffset >= kiov->kiov_len) {
kiovoffset -= kiov->kiov_len;
kiov++;
nkiov--;
LASSERT (!in_interrupt ());
LASSERT (nkiov > 0);
- while (kiovoffset > kiov->kiov_len) {
+ while (kiovoffset >= kiov->kiov_len) {
kiovoffset -= kiov->kiov_len;
kiov++;
nkiov--;
}
LASSERT (niov > 0);
- while (iovoffset > iov->iov_len) {
+ while (iovoffset >= iov->iov_len) {
iovoffset -= iov->iov_len;
iov++;
niov--;
niov = 1;
for (;;) {
LASSERT (src_niov > 0);
- LASSERT (niov <= dst_niov);
+ LASSERT ((int)niov <= dst_niov);
frag_len = src->kiov_len - offset;
dst->kiov_page = src->kiov_page;
LASSERT (!in_interrupt ());
LASSERT (mlen == 0 || msg != NULL);
-
+
if (msg != NULL) {
LASSERT(msg->msg_receiving);
LASSERT(!msg->msg_sending);
niov = msg->msg_niov;
iov = msg->msg_iov;
kiov = msg->msg_kiov;
-
+
LASSERT (niov > 0);
LASSERT ((iov == NULL) != (kiov == NULL));
}
}
-
+
rc = (ni->ni_lnd->lnd_recv)(ni, private, msg, delayed,
niov, iov, kiov, offset, mlen, rlen);
if (rc < 0)
{
if (p1->lp_txqnob < p2->lp_txqnob)
return 1;
-
+
if (p1->lp_txqnob > p2->lp_txqnob)
return -1;
-
+
if (p1->lp_txcredits > p2->lp_txcredits)
return 1;
-
+
if (p1->lp_txcredits < p2->lp_txcredits)
return -1;
-
+
return 0;
}
}
void
-lnet_ni_send(lnet_ni_t *ni, lnet_msg_t *msg)
+lnet_ni_send(lnet_ni_t *ni, lnet_msg_t *msg)
{
void *priv = msg->msg_private;
int rc;
LASSERT (msg->msg_receiving);
LASSERT (!msg->msg_sending);
-
+
peer = msg->msg_rxpeer;
ni = peer->lp_ni;
if (ni->ni_lnd->lnd_eager_recv != NULL) {
LNET_UNLOCK();
-
- rc = (ni->ni_lnd->lnd_eager_recv)(ni, msg->msg_private, msg,
+
+ rc = (ni->ni_lnd->lnd_eager_recv)(ni, msg->msg_private, msg,
&msg->msg_private);
if (rc != 0) {
CERROR("recv from %s / send to %s aborted: "
return rc;
}
+/* NB: caller shall hold a ref on 'lp' as I'd drop LNET_LOCK */
+void
+lnet_ni_peer_alive(lnet_peer_t *lp)
+{
+ time_t last_alive = 0;
+ lnet_ni_t *ni = lp->lp_ni;
+
+ LASSERT (ni != NULL);
+ LASSERT (ni->ni_peertimeout > 0);
+ LASSERT (ni->ni_lnd->lnd_query != NULL);
+
+ LNET_UNLOCK();
+ (ni->ni_lnd->lnd_query)(ni, lp->lp_nid, &last_alive);
+ LNET_LOCK();
+
+ lp->lp_last_query = cfs_time_current_sec();
+
+ if (last_alive != 0) /* NI has updated timestamp */
+ lp->lp_last_alive = last_alive;
+ return;
+}
+
+/* NB: always called with LNET_LOCK held */
+static inline int
+lnet_peer_is_alive (lnet_peer_t *lp, time_t now)
+{
+ lnet_ni_t *ni = lp->lp_ni;
+ time_t deadline;
+ int alive;
+
+ LASSERT (ni != NULL);
+ LASSERT (ni->ni_peertimeout > 0);
+
+ if (!lp->lp_alive && lp->lp_alive_count > 0 &&
+ cfs_time_aftereq(lp->lp_timestamp, lp->lp_last_alive))
+ return 0;
+
+ deadline = cfs_time_add(lp->lp_last_alive, ni->ni_peertimeout);
+ alive = cfs_time_after(deadline, now);
+ if (alive && !lp->lp_alive) /* update obsolete lp_alive */
+ lnet_notify_locked(lp, 0, 1, lp->lp_last_alive);
+
+ return alive;
+}
+
+/* don't query LND about aliveness of a dead peer more frequently than: */
+static int lnet_queryinterval = 1; /* 1 second */
+
+/* NB: returns 1 when alive, 0 when dead, negative when error;
+ * may drop the LNET_LOCK */
+int
+lnet_peer_alive_locked (lnet_peer_t *lp)
+{
+ lnet_ni_t *ni = lp->lp_ni;
+ time_t now = cfs_time_current_sec();
+
+ LASSERT (ni != NULL);
+
+ if (ni->ni_peertimeout <= 0) /* disabled */
+ return -ENODEV;
+
+ if (lnet_peer_is_alive(lp, now))
+ return 1;
+
+ /* peer appears dead, should we query right now? */
+ if (lp->lp_last_query != 0) {
+ time_t deadline =
+ cfs_time_add(lp->lp_last_query,
+ lnet_queryinterval);
+
+ if (cfs_time_before(now, deadline)) {
+ if (lp->lp_alive)
+ CWARN("Unexpected aliveness of peer %s: "
+ "%d < %d (%d/%d)\n",
+ libcfs_nid2str(lp->lp_nid),
+ (int)now, (int)deadline,
+ lnet_queryinterval, ni->ni_peertimeout);
+ return 0;
+ }
+ }
+
+ /* query LND for latest aliveness news */
+ lnet_ni_peer_alive(lp);
+
+ if (lnet_peer_is_alive(lp, now))
+ return 1;
+
+ lnet_notify_locked(lp, 0, 0, lp->lp_last_alive);
+ return 0;
+}
+
int
lnet_post_send_locked (lnet_msg_t *msg, int do_send)
{
/* lnet_send is going to LNET_UNLOCK immediately after this, so it sets
* do_send FALSE and I don't do the unlock/send/lock bit. I return
- * EAGAIN if msg blocked and 0 if sent or OK to send */
+ * EAGAIN if msg blocked, EHOSTUNREACH if msg_txpeer appears dead, and
+ * 0 if sent or OK to send */
lnet_peer_t *lp = msg->msg_txpeer;
lnet_ni_t *ni = lp->lp_ni;
LASSERT (!do_send || msg->msg_delayed);
LASSERT (!msg->msg_receiving);
+ /* NB 'lp' is always the next hop */
+ if ((msg->msg_target.pid & LNET_PID_USERFLAG) == 0 &&
+ lnet_peer_alive_locked(lp) == 0) {
+ LNET_UNLOCK();
+
+ CDEBUG(D_NETERROR, "Dropping message for %s: peer not alive\n",
+ libcfs_id2str(msg->msg_target));
+ if (do_send)
+ lnet_finalize(ni, msg, -EHOSTUNREACH);
+
+ LNET_LOCK();
+ return EHOSTUNREACH;
+ }
+
if (!msg->msg_peertxcredit) {
LASSERT ((lp->lp_txcredits < 0) == !list_empty(&lp->lp_txq));
return EAGAIN;
}
}
-
+
if (!msg->msg_txcredit) {
LASSERT ((ni->ni_txcredits < 0) == !list_empty(&ni->ni_txq));
{
/* ALWAYS called holding the LNET_LOCK */
LASSERT (msg->msg_routing);
-
+
the_lnet.ln_counters.msgs_alloc++;
- if (the_lnet.ln_counters.msgs_alloc >
+ if (the_lnet.ln_counters.msgs_alloc >
the_lnet.ln_counters.msgs_max)
- the_lnet.ln_counters.msgs_max =
+ the_lnet.ln_counters.msgs_max =
the_lnet.ln_counters.msgs_alloc;
the_lnet.ln_counters.route_count++;
the_lnet.ln_counters.route_length += msg->msg_len;
-
+
LASSERT (!msg->msg_onactivelist);
msg->msg_onactivelist = 1;
list_add (&msg->msg_activelist, &the_lnet.ln_active_msgs);
}
lnet_rtrbufpool_t *
-lnet_msg2bufpool(lnet_msg_t *msg)
+lnet_msg2bufpool(lnet_msg_t *msg)
{
lnet_rtrbufpool_t *rbp = &the_lnet.ln_rtrpools[0];
LASSERT (msg->msg_len <= LNET_MTU);
- while (msg->msg_len > rbp->rbp_npages * CFS_PAGE_SIZE) {
+ while (msg->msg_len > (unsigned int)rbp->rbp_npages * CFS_PAGE_SIZE) {
rbp++;
LASSERT (rbp < &the_lnet.ln_rtrpools[LNET_NRBPOOLS]);
}
{
/* lnet_parse is going to LNET_UNLOCK immediately after this, so it
* sets do_recv FALSE and I don't do the unlock/send/lock bit. I
- * return EAGAIN if msg blocked and 0 if sent or OK to send */
+ * return EAGAIN if msg blocked and 0 if received or OK to receive */
lnet_peer_t *lp = msg->msg_rxpeer;
lnet_rtrbufpool_t *rbp;
lnet_rtrbuf_t *rb;
if (!msg->msg_peerrtrcredit) {
LASSERT ((lp->lp_rtrcredits < 0) == !list_empty(&lp->lp_rtrq));
-
+
msg->msg_peerrtrcredit = 1;
lp->lp_rtrcredits--;
if (lp->lp_rtrcredits < lp->lp_minrtrcredits)
lp->lp_minrtrcredits = lp->lp_rtrcredits;
-
+
if (lp->lp_rtrcredits < 0) {
/* must have checked eager_recv before here */
LASSERT (msg->msg_delayed);
return EAGAIN;
}
}
-
+
rbp = lnet_msg2bufpool(msg);
if (!msg->msg_rtrcredit) {
return EAGAIN;
}
}
-
+
LASSERT (!list_empty(&rbp->rbp_bufs));
rb = list_entry(rbp->rbp_bufs.next, lnet_rtrbuf_t, rb_list);
list_del(&rb->rb_list);
-
+
msg->msg_niov = rbp->rbp_npages;
msg->msg_kiov = &rb->rb_kiov[0];
txpeer->lp_txcredits++;
if (txpeer->lp_txcredits <= 0) {
- msg2 = list_entry(txpeer->lp_txq.next,
+ msg2 = list_entry(txpeer->lp_txq.next,
lnet_msg_t, msg_list);
list_del(&msg2->msg_list);
lnet_peer_decref_locked(txpeer);
}
-#ifdef __KERNEL__
+#ifdef __KERNEL__
if (msg->msg_rtrcredit) {
/* give back global router credits */
lnet_rtrbuf_t *rb;
* there until it gets one allocated, or aborts the wait
* itself */
LASSERT (msg->msg_kiov != NULL);
-
+
rb = list_entry(msg->msg_kiov, lnet_rtrbuf_t, rb_kiov[0]);
rbp = rb->rb_pool;
LASSERT (rbp == lnet_msg2bufpool(msg));
msg->msg_kiov = NULL;
msg->msg_rtrcredit = 0;
-
+
LASSERT((rbp->rbp_credits < 0) == !list_empty(&rbp->rbp_msgs));
LASSERT((rbp->rbp_credits > 0) == !list_empty(&rbp->rbp_bufs));
list_add(&rb->rb_list, &rbp->rbp_bufs);
rbp->rbp_credits++;
if (rbp->rbp_credits <= 0) {
- msg2 = list_entry(rbp->rbp_msgs.next,
+ msg2 = list_entry(rbp->rbp_msgs.next,
lnet_msg_t, msg_list);
list_del(&msg2->msg_list);
-
+
(void) lnet_post_routed_recv_locked(msg2, 1);
}
}
-
+
if (msg->msg_peerrtrcredit) {
- /* give pack peer router credits */
+ /* give back peer router credits */
msg->msg_peerrtrcredit = 0;
-
+
LASSERT((rxpeer->lp_rtrcredits < 0) == !list_empty(&rxpeer->lp_rtrq));
rxpeer->lp_rtrcredits++;
msg2 = list_entry(rxpeer->lp_rtrq.next,
lnet_msg_t, msg_list);
list_del(&msg2->msg_list);
-
+
(void) lnet_post_routed_recv_locked(msg2, 1);
}
}
LASSERT (!msg->msg_routing);
}
- /* Is this for someone on a local network? */
+ /* Is this for someone on a local network? */
local_ni = lnet_net2ni_locked(LNET_NIDNET(dst_nid));
if (local_ni != NULL) {
LASSERT (src_nid != LNET_NID_ANY);
- if (!msg->msg_routing) {
- src_nid = lnet_ptlcompat_srcnid(src_nid, dst_nid);
+ if (!msg->msg_routing)
msg->msg_hdr.src_nid = cpu_to_le64(src_nid);
- }
-
+
if (src_ni == the_lnet.ln_loni) {
/* No send credit hassles with LOLND */
LNET_UNLOCK();
lnet_ni_decref(src_ni);
return 0;
}
-
+
rc = lnet_nid2peer_locked(&lp, dst_nid);
lnet_ni_decref_locked(src_ni); /* lp has ref on src_ni; lose mine */
if (rc != 0) {
}
LASSERT (lp->lp_ni == src_ni);
} else {
+#ifndef __KERNEL__
+ LNET_UNLOCK();
+
+ /* NB
+ * - once application finishes computation, check here to update
+ * router states before it waits for pending IO in LNetEQPoll
+ * - recursion breaker: router checker sends no message
+ * to remote networks */
+ if (the_lnet.ln_rc_state == LNET_RC_STATE_RUNNING)
+ lnet_router_checker();
+
+ LNET_LOCK();
+#endif
/* sending to a remote network */
rnet = lnet_find_net_locked(LNET_NIDNET(dst_nid));
if (rnet == NULL) {
if (src_ni != NULL)
lnet_ni_decref_locked(src_ni);
LNET_UNLOCK();
- CERROR("No route to %s (all routers down)\n",
+ CERROR("No route to %s (all routers down)\n",
libcfs_id2str(msg->msg_target));
return -EHOSTUNREACH;
}
if (!msg->msg_routing) {
/* I'm the source and now I know which NI to send on */
- src_nid = lnet_ptlcompat_srcnid(src_nid, dst_nid);
msg->msg_hdr.src_nid = cpu_to_le64(src_nid);
}
rc = lnet_post_send_locked(msg, 0);
LNET_UNLOCK();
+ if (rc == EHOSTUNREACH)
+ return -EHOSTUNREACH;
+
if (rc == 0)
lnet_ni_send(src_ni, msg);
the_lnet.ln_counters.drop_count++;
the_lnet.ln_counters.drop_length += nob;
LNET_UNLOCK();
-
+
lnet_ni_recv(ni, private, NULL, 0, 0, 0, nob);
}
static void
-lnet_drop_delayed_put(lnet_msg_t *msg, char *reason)
+lnet_drop_delayed_put(lnet_msg_t *msg, char *reason)
{
+ lnet_process_id_t id = {0};
+
+ id.nid = msg->msg_hdr.src_nid;
+ id.pid = msg->msg_hdr.src_pid;
+
LASSERT (msg->msg_md == NULL);
LASSERT (msg->msg_delayed);
LASSERT (msg->msg_rxpeer != NULL);
CWARN("Dropping delayed PUT from %s portal %d match "LPU64
" offset %d length %d: %s\n",
- libcfs_id2str((lnet_process_id_t){
- .nid = msg->msg_hdr.src_nid,
- .pid = msg->msg_hdr.src_pid}),
- msg->msg_hdr.msg.put.ptl_index,
- msg->msg_hdr.msg.put.match_bits,
+ libcfs_id2str(id),
+ msg->msg_hdr.msg.put.ptl_index,
+ msg->msg_hdr.msg.put.match_bits,
msg->msg_hdr.msg.put.offset,
msg->msg_hdr.payload_length,
reason);
* called lnet_drop_message(), so I just hang onto msg as well
* until that's done */
- lnet_drop_message(msg->msg_rxpeer->lp_ni,
+ lnet_drop_message(msg->msg_rxpeer->lp_ni,
msg->msg_private, msg->msg_len);
LNET_LOCK();
lnet_peer_decref_locked(msg->msg_rxpeer);
msg->msg_rxpeer = NULL;
-
+
lnet_msg_free(msg);
-
+
LNET_UNLOCK();
}
return 0;
}
- CDEBUG(D_NET, "clearing portal %d lazy\n", portal);
+ if (the_lnet.ln_shutdown)
+ CWARN ("Active lazy portal %d on exit\n", portal);
+ else
+ CDEBUG (D_NET, "clearing portal %d lazy\n", portal);
/* grab all the blocked messages atomically */
list_add(&zombies, &ptl->ptl_msgq);
ptl->ptl_options &= ~LNET_PTL_LAZY;
LNET_UNLOCK();
-
+
while (!list_empty(&zombies)) {
msg = list_entry(zombies.next, lnet_msg_t, msg_list);
list_del(&msg->msg_list);
* it back into the ACK during lnet_finalize() */
msg->msg_ack = (!lnet_is_wire_handle_none(&hdr->msg.put.ack_wmd) &&
(md->md_options & LNET_MD_ACK_DISABLE) == 0);
-
- lnet_ni_recv(msg->msg_rxpeer->lp_ni,
- msg->msg_private,
- msg, delayed, offset, mlength,
+
+ lnet_ni_recv(msg->msg_rxpeer->lp_ni,
+ msg->msg_private,
+ msg, delayed, offset, mlength,
hdr->payload_length);
}
lnet_me_t *me = md->md_me;
lnet_portal_t *ptl = &the_lnet.ln_portals[me->me_portal];
- LASSERT (me->me_portal < the_lnet.ln_nportals);
+ LASSERT (me->me_portal < (unsigned int)the_lnet.ln_nportals);
if ((ptl->ptl_options & LNET_PTL_LAZY) == 0) {
LASSERT (list_empty(&ptl->ptl_msgq));
src.pid = hdr->src_pid;
rc = lnet_try_match_md(index, LNET_MD_OP_PUT, src,
- hdr->payload_length,
- hdr->msg.put.offset,
- hdr->msg.put.match_bits,
+ hdr->payload_length,
+ hdr->msg.put.offset,
+ hdr->msg.put.match_bits,
md, msg, &mlength, &offset);
if (rc == LNET_MATCHMD_NONE)
continue;
-
+
/* Hurrah! This _is_ a match */
- list_del(&msg->msg_list);
+ list_del(&msg->msg_list);
ptl->ptl_msgq_version++;
if (rc == LNET_MATCHMD_OK) {
CDEBUG(D_NET, "Resuming delayed PUT from %s portal %d "
"match "LPU64" offset %d length %d.\n",
libcfs_id2str(src),
- hdr->msg.put.ptl_index,
- hdr->msg.put.match_bits,
+ hdr->msg.put.ptl_index,
+ hdr->msg.put.match_bits,
hdr->msg.put.offset,
hdr->payload_length);
} else {
list_for_each_safe (entry, tmp, &drops) {
msg = list_entry(entry, lnet_msg_t, msg_list);
- list_del(&msg->msg_list);
+ list_del(&msg->msg_list);
lnet_drop_delayed_put(msg, "Bad match");
}
list_for_each_safe (entry, tmp, &matches) {
msg = list_entry(entry, lnet_msg_t, msg_list);
- list_del(&msg->msg_list);
+ list_del(&msg->msg_list);
/* md won't disappear under me, since each msg
* holds a ref on it */
{
int rc;
int index;
+ __u64 version;
lnet_hdr_t *hdr = &msg->msg_hdr;
unsigned int rlength = hdr->payload_length;
unsigned int mlength = 0;
unsigned int offset = 0;
- lnet_process_id_t src = {/* .nid = */ hdr->src_nid,
- /* .pid = */ hdr->src_pid};
+ lnet_process_id_t src= {0};
lnet_libmd_t *md;
+ lnet_portal_t *ptl;
+
+ src.nid = hdr->src_nid;
+ src.pid = hdr->src_pid;
/* Convert put fields to host byte order */
hdr->msg.put.match_bits = le64_to_cpu(hdr->msg.put.match_bits);
hdr->msg.put.offset = le32_to_cpu(hdr->msg.put.offset);
index = hdr->msg.put.ptl_index;
+ ptl = &the_lnet.ln_portals[index];
LNET_LOCK();
+ again:
rc = lnet_match_md(index, LNET_MD_OP_PUT, src,
rlength, hdr->msg.put.offset,
hdr->msg.put.match_bits, msg,
switch (rc) {
default:
LBUG();
-
+
case LNET_MATCHMD_OK:
LNET_UNLOCK();
- lnet_recv_put(md, msg, 0, offset, mlength);
+ lnet_recv_put(md, msg, msg->msg_delayed, offset, mlength);
return 0;
-
+
case LNET_MATCHMD_NONE:
- rc = lnet_eager_recv_locked(msg);
- if (rc == 0) {
- list_add_tail(&msg->msg_list,
- &the_lnet.ln_portals[index].ptl_msgq);
+ version = ptl->ptl_ml_version;
+
+ rc = 0;
+ if (!msg->msg_delayed)
+ rc = lnet_eager_recv_locked(msg);
- the_lnet.ln_portals[index].ptl_msgq_version++;
+ if (rc == 0 &&
+ !the_lnet.ln_shutdown &&
+ ((ptl->ptl_options & LNET_PTL_LAZY) != 0)) {
+ if (version != ptl->ptl_ml_version)
+ goto again;
+
+ list_add_tail(&msg->msg_list, &ptl->ptl_msgq);
+ ptl->ptl_msgq_version++;
+ LNET_UNLOCK();
CDEBUG(D_NET, "Delaying PUT from %s portal %d match "
LPU64" offset %d length %d: no match \n",
- libcfs_id2str(src), index,
- hdr->msg.put.match_bits,
+ libcfs_id2str(src), index,
+ hdr->msg.put.match_bits,
hdr->msg.put.offset, rlength);
-
- LNET_UNLOCK();
return 0;
}
/* fall through */
case LNET_MATCHMD_DROP:
- CWARN("Dropping PUT from %s portal %d match "LPU64
- " offset %d length %d: %d\n",
- libcfs_id2str(src), index,
- hdr->msg.put.match_bits,
- hdr->msg.put.offset, rlength, rc);
+ CDEBUG(D_NETERROR,
+ "Dropping PUT from %s portal %d match "LPU64
+ " offset %d length %d: %d\n",
+ libcfs_id2str(src), index,
+ hdr->msg.put.match_bits,
+ hdr->msg.put.offset, rlength, rc);
LNET_UNLOCK();
return ENOENT; /* +ve: OK but no match */
-
}
}
lnet_hdr_t *hdr = &msg->msg_hdr;
unsigned int mlength = 0;
unsigned int offset = 0;
- lnet_process_id_t src = {/* .nid = */ hdr->src_nid,
- /* .pid = */ hdr->src_pid};
+ lnet_process_id_t src = {0};
lnet_handle_wire_t reply_wmd;
lnet_libmd_t *md;
int rc;
+ src.nid = hdr->src_nid;
+ src.pid = hdr->src_pid;
+
/* Convert get fields to host byte order */
hdr->msg.get.match_bits = le64_to_cpu(hdr->msg.get.match_bits);
hdr->msg.get.ptl_index = le32_to_cpu(hdr->msg.get.ptl_index);
hdr->msg.get.match_bits, msg,
&mlength, &offset, &md);
if (rc == LNET_MATCHMD_DROP) {
- CWARN("Dropping GET from %s portal %d match "LPU64
- " offset %d length %d\n",
- libcfs_id2str(src),
- hdr->msg.get.ptl_index,
- hdr->msg.get.match_bits,
- hdr->msg.get.src_offset,
- hdr->msg.get.sink_length);
+ CDEBUG(D_NETERROR,
+ "Dropping GET from %s portal %d match "LPU64
+ " offset %d length %d\n",
+ libcfs_id2str(src),
+ hdr->msg.get.ptl_index,
+ hdr->msg.get.match_bits,
+ hdr->msg.get.src_offset,
+ hdr->msg.get.sink_length);
LNET_UNLOCK();
return ENOENT; /* +ve: OK but no match */
}
LASSERT (rc == LNET_MATCHMD_OK);
-
+
the_lnet.ln_counters.send_count++;
the_lnet.ln_counters.send_length += mlength;
LNET_UNLOCK();
+ msg->msg_ev.type = LNET_EVENT_GET;
+ msg->msg_ev.target.pid = hdr->dest_pid;
+ msg->msg_ev.target.nid = hdr->dest_nid;
+ msg->msg_ev.hdr_data = 0;
+
reply_wmd = hdr->msg.get.return_wmd;
lnet_prep_send(msg, LNET_MSG_REPLY, src, offset, mlength);
msg->msg_hdr.msg.reply.dst_wmd = reply_wmd;
- msg->msg_ev.type = LNET_EVENT_GET;
- msg->msg_ev.target.pid = hdr->dest_pid;
- msg->msg_ev.target.nid = hdr->dest_nid;
- msg->msg_ev.hdr_data = 0;
-
if (rdma_get) {
/* The LND completes the REPLY from her recv procedure */
lnet_ni_recv(ni, msg->msg_private, msg, 0,
lnet_ni_recv(ni, msg->msg_private, NULL, 0, 0, 0, 0);
msg->msg_receiving = 0;
-
+
rc = lnet_send(ni->ni_nid, msg);
if (rc < 0) {
/* didn't get as far as lnet_ni_send() */
{
void *private = msg->msg_private;
lnet_hdr_t *hdr = &msg->msg_hdr;
- lnet_process_id_t src = {/* .nid = */ hdr->src_nid,
- /* .pid = */ hdr->src_pid};
+ lnet_process_id_t src = {0};
lnet_libmd_t *md;
int rlength;
int mlength;
LNET_LOCK();
+ src.nid = hdr->src_nid;
+ src.pid = hdr->src_pid;
+
/* NB handles only looked up by creator (no flips) */
md = lnet_wire_handle2md(&hdr->msg.reply.dst_wmd);
- if (md == NULL || md->md_threshold == 0) {
- CWARN("%s: Dropping REPLY from %s for %s "
- "MD "LPX64"."LPX64"\n",
- libcfs_nid2str(ni->ni_nid), libcfs_id2str(src),
- (md == NULL) ? "invalid" : "inactive",
- hdr->msg.reply.dst_wmd.wh_interface_cookie,
- hdr->msg.reply.dst_wmd.wh_object_cookie);
+ if (md == NULL || md->md_threshold == 0 || md->md_me != NULL) {
+ CDEBUG(D_NETERROR, "%s: Dropping REPLY from %s for %s "
+ "MD "LPX64"."LPX64"\n",
+ libcfs_nid2str(ni->ni_nid), libcfs_id2str(src),
+ (md == NULL) ? "invalid" : "inactive",
+ hdr->msg.reply.dst_wmd.wh_interface_cookie,
+ hdr->msg.reply.dst_wmd.wh_object_cookie);
+ if (md != NULL && md->md_me != NULL)
+ CERROR("REPLY MD also attached to portal %d\n",
+ md->md_me->me_portal);
LNET_UNLOCK();
return ENOENT; /* +ve: OK but no match */
LASSERT (md->md_offset == 0);
rlength = hdr->payload_length;
- mlength = MIN(rlength, md->md_length);
+ mlength = MIN(rlength, (int)md->md_length);
if (mlength < rlength &&
(md->md_options & LNET_MD_TRUNCATE) == 0) {
- CERROR ("%s: Dropping REPLY from %s length %d "
- "for MD "LPX64" would overflow (%d)\n",
- libcfs_nid2str(ni->ni_nid), libcfs_id2str(src),
+ CDEBUG(D_NETERROR, "%s: Dropping REPLY from %s length %d "
+ "for MD "LPX64" would overflow (%d)\n",
+ libcfs_nid2str(ni->ni_nid), libcfs_id2str(src),
rlength, hdr->msg.reply.dst_wmd.wh_object_cookie,
mlength);
LNET_UNLOCK();
lnet_parse_ack(lnet_ni_t *ni, lnet_msg_t *msg)
{
lnet_hdr_t *hdr = &msg->msg_hdr;
- lnet_process_id_t src = {/* .nid = */ hdr->src_nid,
- /* .pid = */ hdr->src_pid};
- lnet_libmd_t *md;
+ lnet_process_id_t src = {0};
+ lnet_libmd_t *md;
+
+ src.nid = hdr->src_nid;
+ src.pid = hdr->src_pid;
/* Convert ack fields to host byte order */
hdr->msg.ack.match_bits = le64_to_cpu(hdr->msg.ack.match_bits);
/* NB handles only looked up by creator (no flips) */
md = lnet_wire_handle2md(&hdr->msg.ack.dst_wmd);
- if (md == NULL || md->md_threshold == 0) {
-#if 0
+ if (md == NULL || md->md_threshold == 0 || md->md_me != NULL) {
/* Don't moan; this is expected */
- CERROR ("%s: Dropping ACK from %s to %s MD "LPX64"."LPX64"\n",
- libcfs_nid2str(ni->ni_nid), libcfs_id2str(src),
- (md == NULL) ? "invalid" : "inactive",
- hdr->msg.ack.dst_wmd.wh_interface_cookie,
- hdr->msg.ack.dst_wmd.wh_object_cookie);
-#endif
+ CDEBUG(D_NET,
+ "%s: Dropping ACK from %s to %s MD "LPX64"."LPX64"\n",
+ libcfs_nid2str(ni->ni_nid), libcfs_id2str(src),
+ (md == NULL) ? "invalid" : "inactive",
+ hdr->msg.ack.dst_wmd.wh_interface_cookie,
+ hdr->msg.ack.dst_wmd.wh_object_cookie);
+ if (md != NULL && md->md_me != NULL)
+ CERROR("Source MD also attached to portal %d\n",
+ md->md_me->me_portal);
+
LNET_UNLOCK();
return ENOENT; /* +ve! */
}
void
lnet_print_hdr(lnet_hdr_t * hdr)
{
- lnet_process_id_t src = {/* .nid = */ hdr->src_nid,
- /* .pid = */ hdr->src_pid};
- lnet_process_id_t dst = {/* .nid = */ hdr->dest_nid,
- /* .pid = */ hdr->dest_pid};
+ lnet_process_id_t src = {0};
+ lnet_process_id_t dst = {0};
char *type_str = lnet_msgtyp2str (hdr->type);
+ src.nid = hdr->src_nid;
+ src.pid = hdr->src_pid;
+
+ dst.nid = hdr->dest_nid;
+ dst.pid = hdr->dest_pid;
+
CWARN("P3 Header at %p of type %s\n", hdr, type_str);
CWARN(" From %s\n", libcfs_id2str(src));
CWARN(" To %s\n", libcfs_id2str(dst));
int rc = 0;
int for_me;
lnet_msg_t *msg;
+ lnet_pid_t dest_pid;
lnet_nid_t dest_nid;
lnet_nid_t src_nid;
__u32 payload_length;
type = le32_to_cpu(hdr->type);
src_nid = le64_to_cpu(hdr->src_nid);
dest_nid = le64_to_cpu(hdr->dest_nid);
+ dest_pid = le32_to_cpu(hdr->dest_pid);
payload_length = le32_to_cpu(hdr->payload_length);
- for_me = lnet_ptlcompat_matchnid(ni->ni_nid, dest_nid);
+ for_me = (ni->ni_nid == dest_nid);
switch (type) {
case LNET_MSG_ACK:
case LNET_MSG_GET:
if (payload_length > 0) {
CERROR("%s, src %s: bad %s payload %d (0 expected)\n",
- libcfs_nid2str(from_nid),
- libcfs_nid2str(src_nid),
+ libcfs_nid2str(from_nid),
+ libcfs_nid2str(src_nid),
lnet_msgtyp2str(type), payload_length);
return -EPROTO;
}
break;
-
+
case LNET_MSG_PUT:
case LNET_MSG_REPLY:
- if (payload_length > (for_me ? LNET_MAX_PAYLOAD : LNET_MTU)) {
+ if (payload_length > (__u32)(for_me ? LNET_MAX_PAYLOAD : LNET_MTU)) {
CERROR("%s, src %s: bad %s payload %d "
- "(%d max expected)\n",
+ "(%d max expected)\n",
libcfs_nid2str(from_nid),
- libcfs_nid2str(src_nid),
+ libcfs_nid2str(src_nid),
lnet_msgtyp2str(type),
payload_length,
for_me ? LNET_MAX_PAYLOAD : LNET_MTU);
* or malicious so we chop them off at the knees :) */
if (!for_me) {
- if (the_lnet.ln_ptlcompat > 0) {
- /* portals compatibility is single-network */
- CERROR ("%s, src %s: Bad dest nid %s "
- "(routing not supported)\n",
- libcfs_nid2str(from_nid),
- libcfs_nid2str(src_nid),
- libcfs_nid2str(dest_nid));
- return -EPROTO;
- }
-
- if (the_lnet.ln_ptlcompat == 0 &&
- LNET_NIDNET(dest_nid) == LNET_NIDNET(ni->ni_nid)) {
+ if (LNET_NIDNET(dest_nid) == LNET_NIDNET(ni->ni_nid)) {
/* should have gone direct */
CERROR ("%s, src %s: Bad dest nid %s "
"(should have been sent direct)\n",
return -EPROTO;
}
- if (the_lnet.ln_ptlcompat == 0 &&
- lnet_islocalnid(dest_nid)) {
+ if (lnet_islocalnid(dest_nid)) {
/* dest is another local NI; sender should have used
* this node's NID on its own network */
CERROR ("%s, src %s: Bad dest nid %s "
libcfs_nid2str(dest_nid));
return -EPROTO;
}
-
+
if (!the_lnet.ln_routing) {
CERROR ("%s, src %s: Dropping message for %s "
"(routing not enabled)\n",
msg = lnet_msg_alloc();
if (msg == NULL) {
CERROR("%s, src %s: Dropping %s (out of memory)\n",
- libcfs_nid2str(from_nid), libcfs_nid2str(src_nid)
- , lnet_msgtyp2str(type));
+ libcfs_nid2str(from_nid), libcfs_nid2str(src_nid),
+ lnet_msgtyp2str(type));
goto drop;
}
/* msg zeroed in lnet_msg_alloc; i.e. flags all clear, pointers NULL etc */
-
+
msg->msg_type = type;
msg->msg_private = private;
msg->msg_receiving = 1;
LASSERT (for_me);
#else
if (!for_me) {
- msg->msg_target.pid = le32_to_cpu(hdr->dest_pid);
+ msg->msg_target.pid = dest_pid;
msg->msg_target.nid = dest_nid;
msg->msg_routing = 1;
msg->msg_offset = 0;
goto free_drop;
}
}
-
lnet_commit_routedmsg(msg);
rc = lnet_post_routed_recv_locked(msg, 0);
LNET_UNLOCK();
msg->msg_hdr.src_nid = src_nid;
msg->msg_hdr.src_pid = le32_to_cpu(msg->msg_hdr.src_pid);
msg->msg_hdr.dest_nid = dest_nid;
- msg->msg_hdr.dest_pid = le32_to_cpu(msg->msg_hdr.dest_pid);
+ msg->msg_hdr.dest_pid = dest_pid;
msg->msg_hdr.payload_length = payload_length;
msg->msg_ev.sender = from_nid;
-
+
switch (type) {
case LNET_MSG_ACK:
rc = lnet_parse_ack(ni, msg);
if (rc == 0)
return 0;
-
+
LASSERT (rc == ENOENT);
free_drop:
int
LNetPut(lnet_nid_t self, lnet_handle_md_t mdh, lnet_ack_req_t ack,
lnet_process_id_t target, unsigned int portal,
- __u64 match_bits, unsigned int offset,
+ __u64 match_bits, unsigned int offset,
__u64 hdr_data)
{
lnet_msg_t *msg;
LASSERT (the_lnet.ln_init);
LASSERT (the_lnet.ln_refcount > 0);
-
+
if (!list_empty (&the_lnet.ln_test_peers) && /* normally we don't */
fail_peer (target.nid, 1)) /* shall we now? */
{
LNET_LOCK();
md = lnet_handle2md(&mdh);
- if (md == NULL || md->md_threshold == 0) {
+ if (md == NULL || md->md_threshold == 0 || md->md_me != NULL) {
lnet_msg_free(msg);
- LNET_UNLOCK();
- CERROR("Dropping PUT to %s: MD invalid\n",
- libcfs_id2str(target));
+ CERROR("Dropping PUT ("LPU64":%d:%s): MD (%d) invalid\n",
+ match_bits, portal, libcfs_id2str(target),
+ md == NULL ? -1 : md->md_threshold);
+ if (md != NULL && md->md_me != NULL)
+ CERROR("Source MD also attached to portal %d\n",
+ md->md_me->me_portal);
+
+ LNET_UNLOCK();
return -ENOENT;
}
msg->msg_hdr.msg.put.ack_wmd.wh_object_cookie =
md->md_lh.lh_cookie;
} else {
- msg->msg_hdr.msg.put.ack_wmd = LNET_WIRE_HANDLE_NONE;
+ msg->msg_hdr.msg.put.ack_wmd.wh_interface_cookie =
+ LNET_WIRE_HANDLE_COOKIE_NONE;
+ msg->msg_hdr.msg.put.ack_wmd.wh_object_cookie =
+ LNET_WIRE_HANDLE_COOKIE_NONE;
}
msg->msg_ev.type = LNET_EVENT_SEND;
/* NB I trusted my peer to RDMA. If she tells me she's written beyond
* the end of my buffer, I might as well be dead. */
LASSERT (len <= reply->msg_ev.mlength);
-
+
reply->msg_ev.mlength = len;
}
LASSERT (the_lnet.ln_init);
LASSERT (the_lnet.ln_refcount > 0);
-
+
if (!list_empty (&the_lnet.ln_test_peers) && /* normally we don't */
fail_peer (target.nid, 1)) /* shall we now? */
{
LNET_LOCK();
md = lnet_handle2md(&mdh);
- if (md == NULL || md->md_threshold == 0) {
+ if (md == NULL || md->md_threshold == 0 || md->md_me != NULL) {
lnet_msg_free(msg);
- LNET_UNLOCK();
- CERROR("Dropping GET to %s: MD invalid\n",
- libcfs_id2str(target));
+ CERROR("Dropping GET ("LPU64":%d:%s): MD (%d) invalid\n",
+ match_bits, portal, libcfs_id2str(target),
+ md == NULL ? -1 : md->md_threshold);
+ if (md != NULL && md->md_me != NULL)
+ CERROR("REPLY MD also attached to portal %d\n",
+ md->md_me->me_portal);
+
+ LNET_UNLOCK();
return -ENOENT;
}
}
int
-LNetDist (lnet_nid_t dstnid, lnet_nid_t *srcnidp, int *orderp)
+LNetDist (lnet_nid_t dstnid, lnet_nid_t *srcnidp, __u32 *orderp)
{
- struct list_head *e;
+ struct list_head *e;
lnet_ni_t *ni;
lnet_route_t *route;
lnet_remotenet_t *rnet;
__u32 dstnet = LNET_NIDNET(dstnid);
int hops;
- int order = 2;
+ __u32 order = 2;
/* if !local_nid_dist_zero, I don't return a distance of 0 ever
* (when lustre sees a distance of 0, it substitutes 0@lo), so I
list_for_each (e, &the_lnet.ln_nis) {
ni = list_entry(e, lnet_ni_t, ni_list);
-
- if (ni->ni_nid == dstnid ||
- (the_lnet.ln_ptlcompat > 0 &&
- LNET_NIDNET(dstnid) == 0 &&
- LNET_NIDADDR(dstnid) == LNET_NIDADDR(ni->ni_nid) &&
- LNET_NETTYP(LNET_NIDNET(ni->ni_nid)) != LOLND)) {
+
+ if (ni->ni_nid == dstnid) {
if (srcnidp != NULL)
*srcnidp = dstnid;
if (orderp != NULL) {
return local_nid_dist_zero ? 0 : 1;
}
- if (LNET_NIDNET(ni->ni_nid) == dstnet ||
- (the_lnet.ln_ptlcompat > 0 &&
- dstnet == 0 &&
- LNET_NETTYP(LNET_NIDNET(ni->ni_nid)) != LOLND)) {
+ if (LNET_NIDNET(ni->ni_nid) == dstnet) {
if (srcnidp != NULL)
*srcnidp = ni->ni_nid;
if (orderp != NULL)
}
list_for_each (e, &the_lnet.ln_remote_nets) {
- rnet = list_entry(e, lnet_remotenet_t, lrn_list);
+ rnet = list_entry(e, lnet_remotenet_t, lrn_list);
if (rnet->lrn_net == dstnet) {
LASSERT (!list_empty(&rnet->lrn_routes));
int maxnids = 256;
int rc = 0;
int rc2;
-
+
/* Target on a local network? */
-
+
ni = lnet_net2ni(LNET_NIDNET(id.nid));
if (ni != NULL) {
if (ni->ni_lnd->lnd_setasync != NULL)
maxnids *= 2;
goto again;
}
-
+
route = list_entry(tmp, lnet_route_t, lr_list);
nids[nnids++] = route->lr_gateway->lp_nid;
}
ni = lnet_net2ni(LNET_NIDNET(id.nid));
if (ni == NULL)
continue;
-
+
if (ni->ni_lnd->lnd_setasync != NULL) {
rc2 = (ni->ni_lnd->lnd_setasync)(ni, id, nasync);
if (rc2 != 0)
return rc;
#endif
}
-
git://git.whamcloud.com / fs / lustre-release.git / blobdiff