lib-move.c

来自「lustre 1.6.5 source code」· C语言 代码 · 共 1,970 行 · 第 1/5 页

/* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
 * vim:expandtab:shiftwidth=8:tabstop=8:
 *
 * lib/lib-move.c
 * Data movement routines
 *
 *  Copyright (c) 2001-2003 Cluster File Systems, Inc.
 *
 *   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
 *   License as published by the Free Software Foundation.
 *
 *   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.
 *
 *   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.
 */

#define DEBUG_SUBSYSTEM S_LNET

#include <lnet/lib-lnet.h>

static int local_nid_dist_zero = 1;
CFS_MODULE_PARM(local_nid_dist_zero, "i", int, 0444,
                "Reserved");

/* 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 discarded */

static int
lnet_try_match_md (int index, int op_mask, lnet_process_id_t src,
                   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)
{
        /* ALWAYS called holding the LNET_LOCK, and can't LNET_UNLOCK;
         * lnet_match_blocked_msg() relies on this to avoid races */
        unsigned int  offset;
        unsigned int  mlength;
        lnet_me_t    *me = md->md_me;

        /* mismatched MD op */
        if ((md->md_options & op_mask) == 0)
                return LNET_MATCHMD_NONE;

        /* MD exhausted */
        if (lnet_md_exhausted(md))
                return LNET_MATCHMD_NONE;

        /* mismatched ME nid/pid? */
        if (me->me_match_id.nid != LNET_NID_ANY &&
            me->me_match_id.nid != src.nid)
                return LNET_MATCHMD_NONE;

        if (me->me_match_id.pid != LNET_PID_ANY &&
            me->me_match_id.pid != src.pid)
                return LNET_MATCHMD_NONE;

        /* mismatched ME matchbits? */
        if (((me->me_match_bits ^ match_bits) & ~me->me_ignore_bits) != 0)
                return LNET_MATCHMD_NONE;

        /* Hurrah! This _is_ a match; check it out... */

        if ((md->md_options & LNET_MD_MANAGE_REMOTE) == 0)
                offset = md->md_offset;
        else
                offset = roffset;

        if ((md->md_options & LNET_MD_MAX_SIZE) != 0) {
                mlength = md->md_max_size;
                LASSERT (md->md_offset + mlength <= md->md_length);
        } else {
                mlength = md->md_length - offset;
        }

        if (rlength <= mlength) {        /* fits in allowed space */
                mlength = rlength;
        } 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", 
                       libcfs_id2str(src), match_bits, rlength,
                       md->md_length - offset, mlength);

                return LNET_MATCHMD_DROP;
        }

        /* Commit to this ME/MD */
        CDEBUG(D_NET, "Incoming %s index %x from %s of "
               "length %d/%d into md "LPX64" [%d] + %d\n",
               (op_mask == LNET_MD_OP_PUT) ? "put" : "get",
               index, libcfs_id2str(src), mlength, rlength,
               md->md_lh.lh_cookie, md->md_niov, offset);

        lnet_commit_md(md, msg);
        md->md_offset = offset + mlength;

        /* NB Caller will set ev.type and ev.hdr_data */
        msg->msg_ev.initiator = src;
        msg->msg_ev.pt_index = index;
        msg->msg_ev.match_bits = match_bits;
        msg->msg_ev.rlength = rlength;
        msg->msg_ev.mlength = mlength;
        msg->msg_ev.offset = offset;

        lnet_md_deconstruct(md, &msg->msg_ev.md);
        lnet_md2handle(&msg->msg_ev.md_handle, md);

        *offset_out = offset;
        *mlength_out = mlength;

        /* Auto-unlink NOW, so the ME gets unlinked if required.
         * We bumped md->md_refcount above so the MD just gets flagged
         * for unlink when it is finalized. */
        if ((md->md_flags & LNET_MD_FLAG_AUTO_UNLINK) != 0 &&
            lnet_md_exhausted(md)) {
                lnet_md_unlink(md);
        }

        return LNET_MATCHMD_OK;
}

static int
lnet_match_md(int index, int op_mask, lnet_process_id_t src,
              unsigned int rlength, unsigned int roffset,
              __u64 match_bits, lnet_msg_t *msg,
              unsigned int *mlength_out, unsigned int *offset_out,
              lnet_libmd_t **md_out)
{
        lnet_portal_t    *ptl = &the_lnet.ln_portals[index];
        lnet_me_t        *me;
        lnet_me_t        *tmp;
        lnet_libmd_t     *md;
        int               rc;

        CDEBUG (D_NET, "Request from %s of length %d into portal %d "
                "MB="LPX64"\n", libcfs_id2str(src), rlength, index, match_bits);

        if (index < 0 || index >= the_lnet.ln_nportals) {
                CERROR("Invalid portal %d not in [0-%d]\n",
                       index, the_lnet.ln_nportals);
                return LNET_MATCHMD_DROP;
        }

        list_for_each_entry_safe (me, tmp, &ptl->ptl_ml, me_list) {
                md = me->me_md;

                /* 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, 
                                       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;
                }
                /* not reached */
        }

        if (op_mask == LNET_MD_OP_GET ||
            (ptl->ptl_options & LNET_PTL_LAZY) == 0)
                return LNET_MATCHMD_DROP;
        
        return LNET_MATCHMD_NONE;
}

int
lnet_fail_nid (lnet_nid_t nid, unsigned int threshold)
{
        lnet_test_peer_t   *tp;
        struct list_head  *el;
        struct list_head  *next;
        struct list_head   cull;

        LASSERT (the_lnet.ln_init);
        
        if (threshold != 0) {
                /* Adding a new entry */
                LIBCFS_ALLOC(tp, sizeof(*tp));
                if (tp == NULL)
                        return -ENOMEM;

                tp->tp_nid = nid;
                tp->tp_threshold = threshold;

                LNET_LOCK();
                list_add_tail (&tp->tp_list, &the_lnet.ln_test_peers);
                LNET_UNLOCK();
                return 0;
        }

        /* removing entries */
        CFS_INIT_LIST_HEAD (&cull);

        LNET_LOCK();

        list_for_each_safe (el, next, &the_lnet.ln_test_peers) {
                tp = list_entry (el, lnet_test_peer_t, tp_list);

                if (tp->tp_threshold == 0 ||    /* needs culling anyway */
                    nid == LNET_NID_ANY ||       /* removing all entries */
                    tp->tp_nid == nid)          /* matched this one */
                {
                        list_del (&tp->tp_list);
                        list_add (&tp->tp_list, &cull);
                }
        }

        LNET_UNLOCK();

        while (!list_empty (&cull)) {
                tp = list_entry (cull.next, lnet_test_peer_t, tp_list);

                list_del (&tp->tp_list);
                LIBCFS_FREE(tp, sizeof (*tp));
        }
        return 0;
}

static int
fail_peer (lnet_nid_t nid, int outgoing)
{
        lnet_test_peer_t  *tp;
        struct list_head *el;
        struct list_head *next;
        struct list_head  cull;
        int               fail = 0;

        CFS_INIT_LIST_HEAD (&cull);

        LNET_LOCK();

        list_for_each_safe (el, next, &the_lnet.ln_test_peers) {
                tp = list_entry (el, lnet_test_peer_t, tp_list);

                if (tp->tp_threshold == 0) {
                        /* zombie entry */
                        if (outgoing) {
                                /* only cull zombies on outgoing tests,
                                 * since we may be at interrupt priority on
                                 * incoming messages. */
                                list_del (&tp->tp_list);
                                list_add (&tp->tp_list, &cull);
                        }
                        continue;
                }

                if (tp->tp_nid == LNET_NID_ANY || /* fail every peer */
                    nid == tp->tp_nid) {        /* fail this peer */
                        fail = 1;

                        if (tp->tp_threshold != LNET_MD_THRESH_INF) {
                                tp->tp_threshold--;
                                if (outgoing &&
                                    tp->tp_threshold == 0) {
                                        /* see above */
                                        list_del (&tp->tp_list);
                                        list_add (&tp->tp_list, &cull);
                                }
                        }
                        break;
                }
        }

        LNET_UNLOCK ();

        while (!list_empty (&cull)) {
                tp = list_entry (cull.next, lnet_test_peer_t, tp_list);
                list_del (&tp->tp_list);

                LIBCFS_FREE(tp, sizeof (*tp));
        }

        return (fail);
}

unsigned int
lnet_iov_nob (unsigned int niov, struct iovec *iov)
{
        unsigned int nob = 0;

        while (niov-- > 0)
                nob += (iov++)->iov_len;

        return (nob);
}

void
lnet_copy_iov2iov (unsigned int ndiov, struct iovec *diov, unsigned int doffset,
                   unsigned int nsiov, struct iovec *siov, unsigned int soffset,
                   unsigned int nob)
{
        /* NB diov, siov are READ-ONLY */
        unsigned int  this_nob;

        if (nob == 0)
                return;

        /* skip complete frags before 'doffset' */
        LASSERT (ndiov > 0);
        while (doffset >= diov->iov_len) {
                doffset -= diov->iov_len;
                diov++;
                ndiov--;
                LASSERT (ndiov > 0);
        }
        
        /* skip complete frags before 'soffset' */
        LASSERT (nsiov > 0);
        while (soffset >= siov->iov_len) {
                soffset -= siov->iov_len;
                siov++;
                nsiov--;
                LASSERT (nsiov > 0);
        }

        do {
                LASSERT (ndiov > 0);
                LASSERT (nsiov > 0);
                this_nob = MIN(diov->iov_len - doffset,
                               siov->iov_len - soffset);
                this_nob = MIN(this_nob, nob);

                memcpy ((char *)diov->iov_base + doffset,
                        (char *)siov->iov_base + soffset, this_nob);
                nob -= this_nob;

                if (diov->iov_len > doffset + this_nob) {
                        doffset += this_nob;
                } else {
                        diov++;
                        ndiov--;
                        doffset = 0;
                }
                
                if (siov->iov_len > soffset + this_nob) {
                        soffset += this_nob;
                } else {
                        siov++;
                        nsiov--;
                        soffset = 0;
                }
        } while (nob > 0);
}

int
lnet_extract_iov (int dst_niov, struct iovec *dst,
                  int src_niov, struct iovec *src,
                  unsigned int offset, unsigned int len)
{
        /* Initialise 'dst' to the subset of 'src' starting at 'offset',
         * for exactly 'len' bytes, and return the number of entries.
         * NB not destructive to 'src' */
        unsigned int    frag_len;
        unsigned int    niov;

        if (len == 0)                           /* no data => */
                return (0);                     /* no frags */

        LASSERT (src_niov > 0);
        while (offset >= src->iov_len) {      /* skip initial frags */
                offset -= src->iov_len;
                src_niov--;
                src++;
                LASSERT (src_niov > 0);
        }

        niov = 1;