/* -*- 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 */
return LNET_MATCHMD_DROP;
}
- list_for_each_entry_safe (me, tmp, &ptl->ptl_ml, 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 */
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;
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;
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]);
}
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 */
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);
}
static void
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}),
+ libcfs_id2str(id),
msg->msg_hdr.msg.put.ptl_index,
msg->msg_hdr.msg.put.match_bits,
msg->msg_hdr.msg.put.offset,
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));
{
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,
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 && !the_lnet.ln_shutdown) {
- 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);
+
+ if (rc == 0 &&
+ !the_lnet.ln_shutdown &&
+ ((ptl->ptl_options & LNET_PTL_LAZY) != 0)) {
+ if (version != ptl->ptl_ml_version)
+ goto again;
- the_lnet.ln_portals[index].ptl_msgq_version++;
+ 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,
hdr->msg.put.offset, rlength);
-
- LNET_UNLOCK();
return 0;
}
/* fall through */
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);
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,
{
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) {
+ 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) {
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 (md == NULL || md->md_threshold == 0 || md->md_me != NULL) {
/* Don't moan; this is expected */
CDEBUG(D_NET,
"%s: Dropping ACK from %s to %s MD "LPX64"."LPX64"\n",
(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));
dest_nid = le64_to_cpu(hdr->dest_nid);
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_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",
libcfs_nid2str(from_nid),
* 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 "
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;
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;
}
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)
return rc;
#endif
}
-
git://git.whamcloud.com / fs / lustre-release.git / blobdiff