X-Git-Url: https://git.whamcloud.com/?a=blobdiff_plain;f=lnet%2Flnet%2Flib-move.c;h=aef73d60d54bb45968153aee604b0dc757640075;hb=5c44cc125af564e8c0ab75b620e9713c3c840519;hp=13451d93e59eef397ec5c37567f5443bebc60693;hpb=191061ee668400324f4505cf498f1ee2d57e4962;p=fs%2Flustre-release.git diff --git a/lnet/lnet/lib-move.c b/lnet/lnet/lib-move.c index 13451d9..aef73d6 100644 --- a/lnet/lnet/lib-move.c +++ b/lnet/lnet/lib-move.c @@ -5,9 +5,8 @@ * Data movement routines * * Copyright (c) 2001-2003 Cluster File Systems, Inc. - * Copyright (c) 2001-2002 Sandia National Laboratories * - * This file is part of Lustre, http://www.sf.net/projects/lustre/ + * This file is part of Lustre, http://www.lustre.org * * Lustre is free software; you can redistribute it and/or * modify it under the terms of version 2 of the GNU General Public @@ -23,21 +22,24 @@ * Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA. */ +#define DEBUG_SUBSYSTEM S_PORTALS + #ifndef __KERNEL__ # include #else -# define DEBUG_SUBSYSTEM S_PORTALS -# include +# include #endif #include #include /* forward ref */ static void lib_commit_md (lib_nal_t *nal, lib_md_t *md, lib_msg_t *msg); +static ptl_err_t do_lib_parse(lib_nal_t *nal, ptl_hdr_t *hdr, + void *private, int loopback); static lib_md_t * -lib_match_md(lib_nal_t *nal, int index, int op_mask, - ptl_nid_t src_nid, ptl_pid_t src_pid, +lib_match_md(lib_nal_t *nal, int index, int op_mask, + ptl_nid_t src_nid, ptl_pid_t src_pid, ptl_size_t rlength, ptl_size_t roffset, ptl_match_bits_t match_bits, lib_msg_t *msg, ptl_size_t *mlength_out, ptl_size_t *offset_out) @@ -82,8 +84,9 @@ lib_match_md(lib_nal_t *nal, int index, int op_mask, if (me->match_id.nid != PTL_NID_ANY && me->match_id.nid != src_nid) continue; - - CDEBUG(D_NET,"match_id.pid [%x], src_pid [%x]\n", me->match_id.pid, src_pid); + + CDEBUG(D_NET, "match_id.pid [%x], src_pid [%x]\n", + me->match_id.pid, src_pid); if (me->match_id.pid != PTL_PID_ANY && me->match_id.pid != src_pid) @@ -119,9 +122,9 @@ lib_match_md(lib_nal_t *nal, int index, int op_mask, /* Commit to this ME/MD */ CDEBUG(D_NET, "Incoming %s index %x from "LPU64"/%u of " - "length %d/%d into md "LPX64" [%d] + %d\n", + "length %d/%d into md "LPX64" [%d] + %d\n", (op_mask == PTL_MD_OP_PUT) ? "put" : "get", - index, src_nid, src_pid, mlength, rlength, + index, src_nid, src_pid, mlength, rlength, md->md_lh.lh_cookie, md->md_niov, offset); lib_commit_md(nal, md, msg); @@ -168,30 +171,30 @@ int lib_api_fail_nid (nal_t *apinal, ptl_nid_t nid, unsigned int threshold) struct list_head *el; struct list_head *next; struct list_head cull; - + if (threshold != 0) { /* Adding a new entry */ PORTAL_ALLOC(tp, sizeof(*tp)); if (tp == NULL) return PTL_NO_SPACE; - + tp->tp_nid = nid; tp->tp_threshold = threshold; - + LIB_LOCK(nal, flags); list_add_tail (&tp->tp_list, &nal->libnal_ni.ni_test_peers); LIB_UNLOCK(nal, flags); return PTL_OK; } - + /* removing entries */ - INIT_LIST_HEAD (&cull); - + CFS_INIT_LIST_HEAD (&cull); + LIB_LOCK(nal, flags); list_for_each_safe (el, next, &nal->libnal_ni.ni_test_peers) { tp = list_entry (el, lib_test_peer_t, tp_list); - + if (tp->tp_threshold == 0 || /* needs culling anyway */ nid == PTL_NID_ANY || /* removing all entries */ tp->tp_nid == nid) /* matched this one */ @@ -200,9 +203,9 @@ int lib_api_fail_nid (nal_t *apinal, ptl_nid_t nid, unsigned int threshold) list_add (&tp->tp_list, &cull); } } - + LIB_UNLOCK(nal, flags); - + while (!list_empty (&cull)) { tp = list_entry (cull.next, lib_test_peer_t, tp_list); @@ -212,8 +215,21 @@ int lib_api_fail_nid (nal_t *apinal, ptl_nid_t nid, unsigned int threshold) return PTL_OK; } +int +lib_api_loopback (nal_t *apinal, int set, int *enabled) +{ + lib_nal_t *nal = apinal->nal_data; + + if (set) + nal->libnal_ni.ni_loopback = *enabled; + else + *enabled = nal->libnal_ni.ni_loopback; + + return PTL_OK; +} + static int -fail_peer (lib_nal_t *nal, ptl_nid_t nid, int outgoing) +fail_peer (lib_nal_t *nal, ptl_nid_t nid, int outgoing) { lib_test_peer_t *tp; struct list_head *el; @@ -222,8 +238,8 @@ fail_peer (lib_nal_t *nal, ptl_nid_t nid, int outgoing) struct list_head cull; int fail = 0; - INIT_LIST_HEAD (&cull); - + CFS_INIT_LIST_HEAD (&cull); + LIB_LOCK (nal, flags); list_for_each_safe (el, next, &nal->libnal_ni.ni_test_peers) { @@ -240,11 +256,11 @@ fail_peer (lib_nal_t *nal, ptl_nid_t nid, int outgoing) } continue; } - + if (tp->tp_nid == PTL_NID_ANY || /* fail every peer */ nid == tp->tp_nid) { /* fail this peer */ fail = 1; - + if (tp->tp_threshold != PTL_MD_THRESH_INF) { tp->tp_threshold--; if (outgoing && @@ -257,13 +273,13 @@ fail_peer (lib_nal_t *nal, ptl_nid_t nid, int outgoing) break; } } - + LIB_UNLOCK (nal, flags); while (!list_empty (&cull)) { tp = list_entry (cull.next, lib_test_peer_t, tp_list); list_del (&tp->tp_list); - + PORTAL_FREE(tp, sizeof (*tp)); } @@ -274,22 +290,22 @@ ptl_size_t lib_iov_nob (int niov, struct iovec *iov) { ptl_size_t nob = 0; - + while (niov-- > 0) nob += (iov++)->iov_len; - + return (nob); } void -lib_copy_iov2buf (char *dest, int niov, struct iovec *iov, +lib_copy_iov2buf (char *dest, int niov, struct iovec *iov, ptl_size_t offset, ptl_size_t len) { ptl_size_t nob; if (len == 0) return; - + /* skip complete frags before 'offset' */ LASSERT (niov > 0); while (offset >= iov->iov_len) { @@ -298,7 +314,7 @@ lib_copy_iov2buf (char *dest, int niov, struct iovec *iov, niov--; LASSERT (niov > 0); } - + do { LASSERT (niov > 0); nob = MIN (iov->iov_len - offset, len); @@ -313,7 +329,7 @@ lib_copy_iov2buf (char *dest, int niov, struct iovec *iov, } void -lib_copy_buf2iov (int niov, struct iovec *iov, ptl_size_t offset, +lib_copy_buf2iov (int niov, struct iovec *iov, ptl_size_t offset, char *src, ptl_size_t len) { ptl_size_t nob; @@ -329,12 +345,12 @@ lib_copy_buf2iov (int niov, struct iovec *iov, ptl_size_t offset, niov--; LASSERT (niov > 0); } - + do { LASSERT (niov > 0); nob = MIN (iov->iov_len - offset, len); memcpy (iov->iov_base + offset, src, nob); - + len -= nob; src += nob; niov--; @@ -369,7 +385,7 @@ lib_extract_iov (int dst_niov, struct iovec *dst, for (;;) { LASSERT (src_niov > 0); LASSERT (niov <= dst_niov); - + frag_len = src->iov_len - offset; dst->iov_base = ((char *)src->iov_base) + offset; @@ -377,7 +393,7 @@ lib_extract_iov (int dst_niov, struct iovec *dst, dst->iov_len = len; return (niov); } - + dst->iov_len = frag_len; len -= frag_len; @@ -391,14 +407,14 @@ lib_extract_iov (int dst_niov, struct iovec *dst, #ifndef __KERNEL__ ptl_size_t -lib_kiov_nob (int niov, ptl_kiov_t *kiov) +lib_kiov_nob (int niov, ptl_kiov_t *kiov) { LASSERT (0); return (0); } void -lib_copy_kiov2buf (char *dest, int niov, ptl_kiov_t *kiov, +lib_copy_kiov2buf (char *dest, int niov, ptl_kiov_t *kiov, ptl_size_t offset, ptl_size_t len) { LASSERT (0); @@ -412,17 +428,34 @@ lib_copy_buf2kiov (int niov, ptl_kiov_t *kiov, ptl_size_t offset, } int -lib_extract_kiov (int dst_niov, ptl_kiov_t *dst, +lib_extract_kiov (int dst_niov, ptl_kiov_t *dst, int src_niov, ptl_kiov_t *src, ptl_size_t offset, ptl_size_t len) { LASSERT (0); } -#else +ptl_err_t +lib_lo_rxkiov(lib_nal_t *nal, void *private, lib_msg_t *libmsg, + unsigned int niov, ptl_kiov_t *kiov, + size_t offset, size_t mlen, size_t rlen) +{ + LASSERT (0); +} + +ptl_err_t +lib_lo_txkiov (lib_nal_t *nal, void *private, lib_msg_t *libmsg, + ptl_hdr_t *hdr, int type, ptl_nid_t nid, ptl_pid_t pid, + unsigned int payload_niov, ptl_kiov_t *payload_kiov, + size_t payload_offset, size_t payload_nob) +{ + LASSERT (0); +} + +#else /* __KERNEL__ */ ptl_size_t -lib_kiov_nob (int niov, ptl_kiov_t *kiov) +lib_kiov_nob (int niov, ptl_kiov_t *kiov) { ptl_size_t nob = 0; @@ -433,7 +466,7 @@ lib_kiov_nob (int niov, ptl_kiov_t *kiov) } void -lib_copy_kiov2buf (char *dest, int niov, ptl_kiov_t *kiov, +lib_copy_kiov2buf (char *dest, int niov, ptl_kiov_t *kiov, ptl_size_t offset, ptl_size_t len) { ptl_size_t nob; @@ -441,7 +474,7 @@ lib_copy_kiov2buf (char *dest, int niov, ptl_kiov_t *kiov, if (len == 0) return; - + LASSERT (!in_interrupt ()); LASSERT (niov > 0); @@ -451,15 +484,16 @@ lib_copy_kiov2buf (char *dest, int niov, ptl_kiov_t *kiov, niov--; LASSERT (niov > 0); } - - do{ + + do { LASSERT (niov > 0); nob = MIN (kiov->kiov_len - offset, len); - - addr = ((char *)kmap (kiov->kiov_page)) + kiov->kiov_offset + offset; + + addr = ((char *)cfs_kmap(kiov->kiov_page)) + kiov->kiov_offset + + offset; memcpy (dest, addr, nob); - kunmap (kiov->kiov_page); - + cfs_kunmap (kiov->kiov_page); + len -= nob; dest += nob; niov--; @@ -487,15 +521,16 @@ lib_copy_buf2kiov (int niov, ptl_kiov_t *kiov, ptl_size_t offset, niov--; LASSERT (niov > 0); } - + do { LASSERT (niov > 0); nob = MIN (kiov->kiov_len - offset, len); - - addr = ((char *)kmap (kiov->kiov_page)) + kiov->kiov_offset + offset; + + addr = ((char *)cfs_kmap(kiov->kiov_page)) + kiov->kiov_offset + + offset; memcpy (addr, src, nob); - kunmap (kiov->kiov_page); - + cfs_kunmap (kiov->kiov_page); + len -= nob; src += nob; niov--; @@ -505,7 +540,7 @@ lib_copy_buf2kiov (int niov, ptl_kiov_t *kiov, ptl_size_t offset, } int -lib_extract_kiov (int dst_niov, ptl_kiov_t *dst, +lib_extract_kiov (int dst_niov, ptl_kiov_t *dst, int src_niov, ptl_kiov_t *src, ptl_size_t offset, ptl_size_t len) { @@ -530,7 +565,7 @@ lib_extract_kiov (int dst_niov, ptl_kiov_t *dst, for (;;) { LASSERT (src_niov > 0); LASSERT (niov <= dst_niov); - + frag_len = src->kiov_len - offset; dst->kiov_page = src->kiov_page; dst->kiov_offset = src->kiov_offset + offset; @@ -552,9 +587,249 @@ lib_extract_kiov (int dst_niov, ptl_kiov_t *dst, offset = 0; } } + +ptl_err_t +lib_lo_rxkiov(lib_nal_t *nal, + void *private, + lib_msg_t *libmsg, + unsigned int niov, + ptl_kiov_t *kiov, + size_t offset, + size_t mlen, + size_t rlen) +{ + void *srcaddr = NULL; + void *dstaddr = NULL; + unsigned long srcfrag = 0; + unsigned long dstfrag = 0; + unsigned long fraglen; + lo_desc_t *lod = (lo_desc_t *)private; + + /* I only handle unmapped->unmapped matches */ + LASSERT(lod->lod_type == LOD_KIOV); + + if (mlen == 0) + return PTL_OK; + + while (offset >= kiov->kiov_len) { + offset -= kiov->kiov_len; + kiov++; + niov--; + LASSERT(niov > 0); + } + + while (lod->lod_offset >= lod->lod_iov.kiov->kiov_len) { + lod->lod_offset -= lod->lod_iov.kiov->kiov_len; + lod->lod_iov.kiov++; + lod->lod_niov--; + LASSERT(lod->lod_niov > 0); + } + + do { + /* CAVEAT EMPTOR: + * I kmap 2 pages at once == slight risk of deadlock */ + LASSERT(niov > 0); + if (dstaddr == NULL) { + dstaddr = (void *) + ((unsigned long)cfs_kmap(kiov->kiov_page) + + kiov->kiov_offset + offset); + dstfrag = kiov->kiov_len - offset; + } + + LASSERT(lod->lod_niov > 0); + if (srcaddr == NULL) { + srcaddr = (void *) + ((unsigned long)cfs_kmap(lod->lod_iov.kiov->kiov_page)+ + lod->lod_iov.kiov->kiov_offset + lod->lod_offset); + srcfrag = lod->lod_iov.kiov->kiov_len - lod->lod_offset; + } + + fraglen = MIN(srcfrag, dstfrag); + if (fraglen > mlen) + fraglen = mlen; + + memcpy(dstaddr, srcaddr, fraglen); + + if (fraglen < dstfrag) { + dstfrag -= fraglen; + dstaddr = (void *)((unsigned long)dstaddr + fraglen); + } else { + cfs_kunmap(kiov->kiov_page); + dstaddr = NULL; + offset = 0; + kiov++; + niov--; + } + + if (fraglen < srcfrag) { + srcfrag -= fraglen; + srcaddr = (void *)((unsigned long)srcaddr + fraglen); + } else { + cfs_kunmap(lod->lod_iov.kiov->kiov_page); + srcaddr = NULL; + lod->lod_offset = 0; + lod->lod_iov.kiov++; + lod->lod_niov--; + } + + mlen -= fraglen; + } while (mlen > 0); + + if (dstaddr != NULL) + cfs_kunmap(kiov->kiov_page); + + if (srcaddr != NULL) + cfs_kunmap(lod->lod_iov.kiov->kiov_page); + + lib_finalize(nal, private, libmsg, PTL_OK); + return PTL_OK; +} + +ptl_err_t +lib_lo_txkiov (lib_nal_t *nal, + void *private, + lib_msg_t *libmsg, + ptl_hdr_t *hdr, + int type, + ptl_nid_t nid, + ptl_pid_t pid, + unsigned int payload_niov, + ptl_kiov_t *payload_kiov, + size_t payload_offset, + size_t payload_nob) +{ + lo_desc_t lod = { + .lod_type = LOD_KIOV, + .lod_niov = payload_niov, + .lod_offset = payload_offset, + .lod_nob = payload_nob, + .lod_iov = { .kiov = payload_kiov } }; + ptl_err_t rc; + + rc = do_lib_parse(nal, hdr, &lod, 1); + if (rc == PTL_OK) + lib_finalize(nal, private, libmsg, PTL_OK); + + return rc; +} #endif ptl_err_t +lib_lo_rxiov(lib_nal_t *nal, + void *private, + lib_msg_t *libmsg, + unsigned int niov, + struct iovec *iov, + size_t offset, + size_t mlen, + size_t rlen) +{ + lo_desc_t *lod = (lo_desc_t *)private; + + /* I only handle mapped->mapped matches */ + LASSERT(lod->lod_type == LOD_IOV); + LASSERT(mlen > 0); + + while (offset >= iov->iov_len) { + offset -= iov->iov_len; + iov++; + niov--; + LASSERT(niov > 0); + } + + while (lod->lod_offset >= lod->lod_iov.iov->iov_len) { + lod->lod_offset -= lod->lod_iov.iov->iov_len; + lod->lod_iov.iov++; + lod->lod_niov--; + LASSERT(lod->lod_niov > 0); + } + + do { + int fraglen = MIN(iov->iov_len - offset, + lod->lod_iov.iov->iov_len - lod->lod_offset); + + LASSERT(niov > 0); + LASSERT(lod->lod_niov > 0); + + if (fraglen > mlen) + fraglen = mlen; + + memcpy((void *)((unsigned long)iov->iov_base + offset), + (void *)((unsigned long)lod->lod_iov.iov->iov_base + + lod->lod_offset), + fraglen); + + if (offset + fraglen < iov->iov_len) { + offset += fraglen; + } else { + offset = 0; + iov++; + niov--; + } + + if (lod->lod_offset + fraglen < lod->lod_iov.iov->iov_len ) { + lod->lod_offset += fraglen; + } else { + lod->lod_offset = 0; + lod->lod_iov.iov++; + lod->lod_niov--; + } + + mlen -= fraglen; + } while (mlen > 0); + + lib_finalize(nal, private, libmsg, PTL_OK); + return PTL_OK; +} + +ptl_err_t +lib_lo_txiov (lib_nal_t *nal, + void *private, + lib_msg_t *libmsg, + ptl_hdr_t *hdr, + int type, + ptl_nid_t nid, + ptl_pid_t pid, + unsigned int payload_niov, + struct iovec *payload_iov, + size_t payload_offset, + size_t payload_nob) +{ + lo_desc_t lod = { + .lod_type = LOD_IOV, + .lod_niov = payload_niov, + .lod_offset = payload_offset, + .lod_nob = payload_nob, + .lod_iov = { .iov = payload_iov } }; + ptl_err_t rc; + + rc = do_lib_parse(nal, hdr, &lod, 1); + if (rc == PTL_OK) + lib_finalize(nal, private, libmsg, PTL_OK); + + return rc; +} + +ptl_err_t +lib_lo_recv (lib_nal_t *nal, void *private, lib_msg_t *msg, lib_md_t *md, + ptl_size_t offset, ptl_size_t mlen, ptl_size_t rlen) +{ + if (mlen == 0) { + lib_finalize(nal, private, msg, PTL_OK); + return PTL_OK; + } + + if ((md->options & PTL_MD_KIOV) == 0) + return lib_lo_rxiov(nal, private, msg, + md->md_niov, md->md_iov.iov, + offset, mlen, rlen); + + return lib_lo_rxkiov(nal, private, msg, + md->md_niov, md->md_iov.kiov, + offset, mlen, rlen); +} + +ptl_err_t lib_recv (lib_nal_t *nal, void *private, lib_msg_t *msg, lib_md_t *md, ptl_size_t offset, ptl_size_t mlen, ptl_size_t rlen) { @@ -565,10 +840,10 @@ lib_recv (lib_nal_t *nal, void *private, lib_msg_t *msg, lib_md_t *md, if ((md->options & PTL_MD_KIOV) == 0) return (nal->libnal_recv(nal, private, msg, - md->md_niov, md->md_iov.iov, + md->md_niov, md->md_iov.iov, offset, mlen, rlen)); - return (nal->libnal_recv_pages(nal, private, msg, + return (nal->libnal_recv_pages(nal, private, msg, md->md_niov, md->md_iov.kiov, offset, mlen, rlen)); } @@ -576,24 +851,47 @@ lib_recv (lib_nal_t *nal, void *private, lib_msg_t *msg, lib_md_t *md, ptl_err_t lib_send (lib_nal_t *nal, void *private, lib_msg_t *msg, ptl_hdr_t *hdr, int type, ptl_nid_t nid, ptl_pid_t pid, - lib_md_t *md, ptl_size_t offset, ptl_size_t len) + lib_md_t *md, ptl_size_t offset, ptl_size_t len) { - if (len == 0) - return (nal->libnal_send(nal, private, msg, - hdr, type, nid, pid, - 0, NULL, - offset, len)); - - if ((md->options & PTL_MD_KIOV) == 0) - return (nal->libnal_send(nal, private, msg, - hdr, type, nid, pid, - md->md_niov, md->md_iov.iov, - offset, len)); + int loopback = (nal->libnal_ni.ni_loopback && + (nid == nal->libnal_ni.ni_pid.nid)); + + if (len == 0) { + if (loopback) + return lib_lo_txiov(nal, private, msg, + hdr, type, nid, pid, + 0, NULL, + offset, len); + else + return nal->libnal_send(nal, private, msg, + hdr, type, nid, pid, + 0, NULL, + offset, len); + } - return (nal->libnal_send_pages(nal, private, msg, - hdr, type, nid, pid, - md->md_niov, md->md_iov.kiov, - offset, len)); + if ((md->options & PTL_MD_KIOV) == 0) { + if (loopback) + return lib_lo_txiov(nal, private, msg, + hdr, type, nid, pid, + md->md_niov, md->md_iov.iov, + offset, len); + else + return nal->libnal_send(nal, private, msg, + hdr, type, nid, pid, + md->md_niov, md->md_iov.iov, + offset, len); + } + + if (loopback) + return lib_lo_txkiov(nal, private, msg, + hdr, type, nid, pid, + md->md_niov, md->md_iov.kiov, + offset, len); + else + return nal->libnal_send_pages(nal, private, msg, + hdr, type, nid, pid, + md->md_niov, md->md_iov.kiov, + offset, len); } static void @@ -606,7 +904,7 @@ lib_commit_md (lib_nal_t *nal, lib_md_t *md, lib_msg_t *msg) * decrementing its threshold. Come what may, the network "owns" * the MD until a call to lib_finalize() signals completion. */ msg->md = md; - + md->pending++; if (md->threshold != PTL_MD_THRESH_INF) { LASSERT (md->threshold > 0); @@ -621,21 +919,23 @@ lib_commit_md (lib_nal_t *nal, lib_md_t *md, lib_msg_t *msg) } static void -lib_drop_message (lib_nal_t *nal, void *private, ptl_hdr_t *hdr) +lib_drop_message (lib_nal_t *nal, void *private, ptl_hdr_t *hdr, int loopback) { unsigned long flags; /* CAVEAT EMPTOR: this only drops messages that we've not committed * to receive (init_msg() not called) and therefore can't cause an * event. */ - + LIB_LOCK(nal, flags); nal->libnal_ni.ni_counters.drop_count++; nal->libnal_ni.ni_counters.drop_length += hdr->payload_length; LIB_UNLOCK(nal, flags); /* NULL msg => if NAL calls lib_finalize it will be a noop */ - (void) lib_recv(nal, private, NULL, NULL, 0, 0, hdr->payload_length); + if (!loopback) + (void) lib_recv(nal, private, NULL, NULL, 0, 0, + hdr->payload_length); } /* @@ -646,7 +946,8 @@ lib_drop_message (lib_nal_t *nal, void *private, ptl_hdr_t *hdr) * */ static ptl_err_t -parse_put(lib_nal_t *nal, ptl_hdr_t *hdr, void *private, lib_msg_t *msg) +parse_put(lib_nal_t *nal, ptl_hdr_t *hdr, void *private, + lib_msg_t *msg, int loopback) { lib_ni_t *ni = &nal->libnal_ni; ptl_size_t mlength = 0; @@ -654,7 +955,7 @@ parse_put(lib_nal_t *nal, ptl_hdr_t *hdr, void *private, lib_msg_t *msg) ptl_err_t rc; lib_md_t *md; unsigned long flags; - + /* Convert put fields to host byte order */ hdr->msg.put.match_bits = le64_to_cpu(hdr->msg.put.match_bits); hdr->msg.put.ptl_index = le32_to_cpu(hdr->msg.put.ptl_index); @@ -685,8 +986,13 @@ parse_put(lib_nal_t *nal, ptl_hdr_t *hdr, void *private, lib_msg_t *msg) LIB_UNLOCK(nal, flags); - rc = lib_recv(nal, private, msg, md, offset, mlength, - hdr->payload_length); + if (loopback) + rc = lib_lo_recv(nal, private, msg, md, offset, mlength, + hdr->payload_length); + else + rc = lib_recv(nal, private, msg, md, offset, mlength, + hdr->payload_length); + if (rc != PTL_OK) CERROR(LPU64": error on receiving PUT from "LPU64": %d\n", ni->ni_pid.nid, hdr->src_nid, rc); @@ -695,7 +1001,8 @@ parse_put(lib_nal_t *nal, ptl_hdr_t *hdr, void *private, lib_msg_t *msg) } static ptl_err_t -parse_get(lib_nal_t *nal, ptl_hdr_t *hdr, void *private, lib_msg_t *msg) +parse_get(lib_nal_t *nal, ptl_hdr_t *hdr, void *private, + lib_msg_t *msg, int loopback) { lib_ni_t *ni = &nal->libnal_ni; ptl_size_t mlength = 0; @@ -744,20 +1051,23 @@ parse_get(lib_nal_t *nal, ptl_hdr_t *hdr, void *private, lib_msg_t *msg) /* NB call lib_send() _BEFORE_ lib_recv() completes the incoming * message. Some NALs _require_ this to implement optimized GET */ - rc = lib_send (nal, private, msg, &reply, PTL_MSG_REPLY, + rc = lib_send (nal, private, msg, &reply, PTL_MSG_REPLY, hdr->src_nid, hdr->src_pid, md, offset, mlength); if (rc != PTL_OK) CERROR(LPU64": Unable to send REPLY for GET from "LPU64": %d\n", ni->ni_pid.nid, hdr->src_nid, rc); /* Discard any junk after the hdr */ - (void) lib_recv(nal, private, NULL, NULL, 0, 0, hdr->payload_length); + if (!loopback) + (void) lib_recv(nal, private, NULL, NULL, 0, 0, + hdr->payload_length); return (rc); } static ptl_err_t -parse_reply(lib_nal_t *nal, ptl_hdr_t *hdr, void *private, lib_msg_t *msg) +parse_reply(lib_nal_t *nal, ptl_hdr_t *hdr, void *private, + lib_msg_t *msg, int loopback) { lib_ni_t *ni = &nal->libnal_ni; lib_md_t *md; @@ -799,7 +1109,7 @@ parse_reply(lib_nal_t *nal, ptl_hdr_t *hdr, void *private, lib_msg_t *msg) } CDEBUG(D_NET, "Reply from "LPU64" of length %d/%d into md "LPX64"\n", - hdr->src_nid, length, rlength, + hdr->src_nid, length, rlength, hdr->msg.reply.dst_wmd.wh_object_cookie); lib_commit_md(nal, md, msg); @@ -819,7 +1129,11 @@ parse_reply(lib_nal_t *nal, ptl_hdr_t *hdr, void *private, lib_msg_t *msg) LIB_UNLOCK(nal, flags); - rc = lib_recv(nal, private, msg, md, 0, length, rlength); + if (loopback) + rc = lib_lo_recv(nal, private, msg, md, 0, length, rlength); + else + rc = lib_recv(nal, private, msg, md, 0, length, rlength); + if (rc != PTL_OK) CERROR(LPU64": error on receiving REPLY from "LPU64": %d\n", ni->ni_pid.nid, hdr->src_nid, rc); @@ -828,7 +1142,8 @@ parse_reply(lib_nal_t *nal, ptl_hdr_t *hdr, void *private, lib_msg_t *msg) } static ptl_err_t -parse_ack(lib_nal_t *nal, ptl_hdr_t *hdr, void *private, lib_msg_t *msg) +parse_ack(lib_nal_t *nal, ptl_hdr_t *hdr, void *private, + lib_msg_t *msg, int loopback) { lib_ni_t *ni = &nal->libnal_ni; lib_md_t *md; @@ -844,7 +1159,7 @@ parse_ack(lib_nal_t *nal, ptl_hdr_t *hdr, void *private, lib_msg_t *msg) md = ptl_wire_handle2md(&hdr->msg.ack.dst_wmd, nal); if (md == NULL || md->threshold == 0) { CDEBUG(D_INFO, LPU64": Dropping ACK from "LPU64" to %s MD " - LPX64"."LPX64"\n", ni->ni_pid.nid, hdr->src_nid, + LPX64"."LPX64"\n", ni->ni_pid.nid, hdr->src_nid, (md == NULL) ? "invalid" : "inactive", hdr->msg.ack.dst_wmd.wh_interface_cookie, hdr->msg.ack.dst_wmd.wh_object_cookie); @@ -854,7 +1169,7 @@ parse_ack(lib_nal_t *nal, ptl_hdr_t *hdr, void *private, lib_msg_t *msg) } CDEBUG(D_NET, LPU64": ACK from "LPU64" into md "LPX64"\n", - ni->ni_pid.nid, hdr->src_nid, + ni->ni_pid.nid, hdr->src_nid, hdr->msg.ack.dst_wmd.wh_object_cookie); lib_commit_md(nal, md, msg); @@ -871,14 +1186,16 @@ parse_ack(lib_nal_t *nal, ptl_hdr_t *hdr, void *private, lib_msg_t *msg) ni->ni_counters.recv_count++; LIB_UNLOCK(nal, flags); - + /* We have received and matched up the ack OK, create the * completion event now... */ lib_finalize(nal, private, msg, PTL_OK); /* ...and now discard any junk after the hdr */ - (void) lib_recv(nal, private, NULL, NULL, 0, 0, hdr->payload_length); - + if (!loopback) + (void) lib_recv(nal, private, NULL, NULL, 0, 0, + hdr->payload_length); + return (PTL_OK); } @@ -958,6 +1275,12 @@ void print_hdr(lib_nal_t *nal, ptl_hdr_t * hdr) ptl_err_t lib_parse(lib_nal_t *nal, ptl_hdr_t *hdr, void *private) { + return do_lib_parse(nal, hdr, private, 0); +} + +ptl_err_t +do_lib_parse(lib_nal_t *nal, ptl_hdr_t *hdr, void *private, int loopback) +{ unsigned long flags; ptl_err_t rc; lib_msg_t *msg; @@ -965,7 +1288,7 @@ lib_parse(lib_nal_t *nal, ptl_hdr_t *hdr, void *private) /* NB we return PTL_OK if we manage to parse the header and believe * it looks OK. Anything that goes wrong with receiving the * message after that point is the responsibility of the NAL */ - + /* convert common fields to host byte order */ hdr->type = le32_to_cpu(hdr->type); hdr->src_nid = le64_to_cpu(hdr->src_nid); @@ -987,19 +1310,19 @@ lib_parse(lib_nal_t *nal, ptl_hdr_t *hdr, void *private) mv->version_minor == PORTALS_PROTO_VERSION_MINOR) { CWARN (LPU64": Dropping unexpected HELLO message: " "magic %d, version %d.%d from "LPD64"\n", - nal->libnal_ni.ni_pid.nid, mv->magic, + nal->libnal_ni.ni_pid.nid, mv->magic, mv->version_major, mv->version_minor, hdr->src_nid); /* it's good but we don't want it */ - lib_drop_message(nal, private, hdr); + lib_drop_message(nal, private, hdr, loopback); return PTL_OK; } /* we got garbage */ CERROR (LPU64": Bad HELLO message: " "magic %d, version %d.%d from "LPD64"\n", - nal->libnal_ni.ni_pid.nid, mv->magic, + nal->libnal_ni.ni_pid.nid, mv->magic, mv->version_major, mv->version_minor, hdr->src_nid); return PTL_FAIL; @@ -1012,7 +1335,7 @@ lib_parse(lib_nal_t *nal, ptl_hdr_t *hdr, void *private) hdr->dest_nid = le64_to_cpu(hdr->dest_nid); if (hdr->dest_nid != nal->libnal_ni.ni_pid.nid) { CERROR(LPU64": BAD dest NID in %s message from" - LPU64" to "LPU64" (not me)\n", + LPU64" to "LPU64" (not me)\n", nal->libnal_ni.ni_pid.nid, hdr_type_string (hdr), hdr->src_nid, hdr->dest_nid); return PTL_FAIL; @@ -1033,9 +1356,9 @@ lib_parse(lib_nal_t *nal, ptl_hdr_t *hdr, void *private) { CERROR(LPU64": Dropping incoming %s from "LPU64 ": simulated failure\n", - nal->libnal_ni.ni_pid.nid, hdr_type_string (hdr), + nal->libnal_ni.ni_pid.nid, hdr_type_string (hdr), hdr->src_nid); - lib_drop_message(nal, private, hdr); + lib_drop_message(nal, private, hdr, loopback); return PTL_OK; } @@ -1043,31 +1366,31 @@ lib_parse(lib_nal_t *nal, ptl_hdr_t *hdr, void *private) if (msg == NULL) { CERROR(LPU64": Dropping incoming %s from "LPU64 ": can't allocate a lib_msg_t\n", - nal->libnal_ni.ni_pid.nid, hdr_type_string (hdr), + nal->libnal_ni.ni_pid.nid, hdr_type_string (hdr), hdr->src_nid); - lib_drop_message(nal, private, hdr); + lib_drop_message(nal, private, hdr, loopback); return PTL_OK; } switch (hdr->type) { case PTL_MSG_ACK: - rc = parse_ack(nal, hdr, private, msg); + rc = parse_ack(nal, hdr, private, msg, loopback); break; case PTL_MSG_PUT: - rc = parse_put(nal, hdr, private, msg); + rc = parse_put(nal, hdr, private, msg, loopback); break; case PTL_MSG_GET: - rc = parse_get(nal, hdr, private, msg); + rc = parse_get(nal, hdr, private, msg, loopback); break; case PTL_MSG_REPLY: - rc = parse_reply(nal, hdr, private, msg); + rc = parse_reply(nal, hdr, private, msg, loopback); break; default: LASSERT(0); rc = PTL_FAIL; /* no compiler warning please */ break; } - + if (rc != PTL_OK) { if (msg->md != NULL) { /* committed... */ @@ -1077,7 +1400,7 @@ lib_parse(lib_nal_t *nal, ptl_hdr_t *hdr, void *private) lib_msg_free(nal, msg); /* expects LIB_LOCK held */ LIB_UNLOCK(nal, flags); - lib_drop_message(nal, private, hdr); + lib_drop_message(nal, private, hdr, loopback); } } @@ -1085,11 +1408,11 @@ lib_parse(lib_nal_t *nal, ptl_hdr_t *hdr, void *private) /* That's "OK I can parse it", not "OK I like it" :) */ } -int -lib_api_put(nal_t *apinal, ptl_handle_md_t *mdh, +int +lib_api_put(nal_t *apinal, ptl_handle_md_t *mdh, ptl_ack_req_t ack, ptl_process_id_t *id, ptl_pt_index_t portal, ptl_ac_index_t ac, - ptl_match_bits_t match_bits, + ptl_match_bits_t match_bits, ptl_size_t offset, ptl_hdr_data_t hdr_data) { lib_nal_t *nal = apinal->nal_data; @@ -1099,7 +1422,7 @@ lib_api_put(nal_t *apinal, ptl_handle_md_t *mdh, lib_md_t *md; unsigned long flags; int rc; - + if (!list_empty (&ni->ni_test_peers) && /* normally we don't */ fail_peer (nal, id->nid, 1)) /* shall we now? */ { @@ -1121,7 +1444,7 @@ lib_api_put(nal_t *apinal, ptl_handle_md_t *mdh, if (md == NULL || md->threshold == 0) { lib_msg_free(nal, msg); LIB_UNLOCK(nal, flags); - + return PTL_MD_INVALID; } @@ -1149,7 +1472,7 @@ lib_api_put(nal_t *apinal, ptl_handle_md_t *mdh, hdr.msg.put.hdr_data = hdr_data; lib_commit_md(nal, md, msg); - + msg->ev.type = PTL_EVENT_SEND_END; msg->ev.initiator.nid = ni->ni_pid.nid; msg->ev.initiator.pid = ni->ni_pid.pid; @@ -1167,7 +1490,7 @@ lib_api_put(nal_t *apinal, ptl_handle_md_t *mdh, ni->ni_counters.send_length += md->length; LIB_UNLOCK(nal, flags); - + rc = lib_send (nal, NULL, msg, &hdr, PTL_MSG_PUT, id->nid, id->pid, md, 0, md->length); if (rc != PTL_OK) { @@ -1175,12 +1498,12 @@ lib_api_put(nal_t *apinal, ptl_handle_md_t *mdh, id->nid, rc); lib_finalize (nal, NULL, msg, rc); } - + /* completion will be signalled by an event */ return PTL_OK; } -lib_msg_t * +lib_msg_t * lib_create_reply_msg (lib_nal_t *nal, ptl_nid_t peer_nid, lib_msg_t *getmsg) { /* The NAL can DMA direct to the GET md (i.e. no REPLY msg). This @@ -1244,7 +1567,7 @@ lib_create_reply_msg (lib_nal_t *nal, ptl_nid_t peer_nid, lib_msg_t *getmsg) return NULL; } -int +int lib_api_get(nal_t *apinal, ptl_handle_md_t *mdh, ptl_process_id_t *id, ptl_pt_index_t portal, ptl_ac_index_t ac, ptl_match_bits_t match_bits, ptl_size_t offset) @@ -1256,7 +1579,7 @@ lib_api_get(nal_t *apinal, ptl_handle_md_t *mdh, ptl_process_id_t *id, lib_md_t *md; unsigned long flags; int rc; - + if (!list_empty (&ni->ni_test_peers) && /* normally we don't */ fail_peer (nal, id->nid, 1)) /* shall we now? */ { @@ -1327,7 +1650,7 @@ lib_api_get(nal_t *apinal, ptl_handle_md_t *mdh, ptl_process_id_t *id, ni->ni_pid.nid, id->nid, rc); lib_finalize (nal, NULL, msg, rc); } - + /* completion will be signalled by an event */ return PTL_OK; }