echo_client.c

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

/* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
 * vim:expandtab:shiftwidth=8:tabstop=8:
 *
 *  Copyright (c) 2001-2003 Cluster File Systems, Inc.
 *
 *   This file is part of the Lustre file system, http://www.lustre.org
 *   Lustre is a trademark of Cluster File Systems, Inc.
 *
 *   You may have signed or agreed to another license before downloading
 *   this software.  If so, you are bound by the terms and conditions
 *   of that agreement, and the following does not apply to you.  See the
 *   LICENSE file included with this distribution for more information.
 *
 *   If you did not agree to a different license, then this copy of Lustre
 *   is open source 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.
 *
 *   In either case, 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
 *   license text for more details.
 */

#define DEBUG_SUBSYSTEM S_ECHO
#ifdef __KERNEL__
#include <libcfs/libcfs.h>
#else
#include <liblustre.h>
#endif

#include <obd.h>
#include <obd_support.h>
#include <obd_class.h>
#include <obd_echo.h>
#include <lustre_debug.h>
#include <lprocfs_status.h>

static obd_id last_object_id;

#if 0
static void
echo_printk_object (char *msg, struct ec_object *eco)
{
        struct lov_stripe_md *lsm = eco->eco_lsm;
        int                   i;

        printk (KERN_INFO "Lustre: %s: object %p: "LPX64", refs %d%s: "LPX64
                "=%u!%u\n", msg, eco, eco->eco_id, eco->eco_refcount,
                eco->eco_deleted ? "(deleted) " : "",
                lsm->lsm_object_id, lsm->lsm_stripe_size,
                lsm->lsm_stripe_count);

        for (i = 0; i < lsm->lsm_stripe_count; i++)
                printk (KERN_INFO "Lustre:   @%2u:"LPX64"\n",
                        lsm->lsm_oinfo[i].loi_ost_idx,
                        lsm->lsm_oinfo[i].loi_id);
}
#endif

static struct ec_object *
echo_find_object_locked (struct obd_device *obd, obd_id id)
{
        struct echo_client_obd *ec = &obd->u.echo_client;
        struct ec_object       *eco = NULL;
        struct list_head       *el;

        list_for_each (el, &ec->ec_objects) {
                eco = list_entry (el, struct ec_object, eco_obj_chain);

                if (eco->eco_id == id)
                        return (eco);
        }
        return (NULL);
}

static int
echo_copyout_lsm (struct lov_stripe_md *lsm, void *_ulsm, int ulsm_nob)
{
        struct lov_stripe_md *ulsm = _ulsm;
        int nob, i;

        nob = offsetof (struct lov_stripe_md, lsm_oinfo[lsm->lsm_stripe_count]);
        if (nob > ulsm_nob)
                return (-EINVAL);

        if (copy_to_user (ulsm, lsm, sizeof(ulsm)))
                return (-EFAULT);

        for (i = 0; i < lsm->lsm_stripe_count; i++) {
                if (copy_to_user (ulsm->lsm_oinfo[i], lsm->lsm_oinfo[i],
                                  sizeof(lsm->lsm_oinfo[0])))
                        return (-EFAULT);
        }
        return (0);
}

static int
echo_copyin_lsm (struct obd_device *obd, struct lov_stripe_md *lsm,
                 void *ulsm, int ulsm_nob)
{
        struct echo_client_obd *ec = &obd->u.echo_client;
        int                     i;

        if (ulsm_nob < sizeof (*lsm))
                return (-EINVAL);

        if (copy_from_user (lsm, ulsm, sizeof (*lsm)))
                return (-EFAULT);

        if (lsm->lsm_stripe_count > ec->ec_nstripes ||
            lsm->lsm_magic != LOV_MAGIC ||
            (lsm->lsm_stripe_size & (~CFS_PAGE_MASK)) != 0 ||
            ((__u64)lsm->lsm_stripe_size * lsm->lsm_stripe_count > ~0UL))
                return (-EINVAL);

        for (i = 0; i < lsm->lsm_stripe_count; i++) {
                if (copy_from_user(lsm->lsm_oinfo[i],
                                   ((struct lov_stripe_md *)ulsm)->lsm_oinfo[i],
                                   sizeof(lsm->lsm_oinfo[0])))
                        return (-EFAULT);
        }

        return (0);
}

static struct ec_object *
echo_allocate_object (struct obd_device *obd)
{
        struct echo_client_obd *ec = &obd->u.echo_client;
        struct ec_object       *eco;
        int rc;

        OBD_ALLOC(eco, sizeof (*eco));
        if (eco == NULL)
                return NULL;

        rc = obd_alloc_memmd(ec->ec_exp, &eco->eco_lsm);
        if (rc < 0) {
                OBD_FREE(eco, sizeof (*eco));
                return NULL;
        }

        eco->eco_device = obd;
        eco->eco_deleted = 0;
        eco->eco_refcount = 0;
        eco->eco_lsm->lsm_magic = LOV_MAGIC;
        /* leave stripe count 0 by default */

        return (eco);
}

static void
echo_free_object (struct ec_object *eco)
{
        struct obd_device      *obd = eco->eco_device;
        struct echo_client_obd *ec = &obd->u.echo_client;

        LASSERT (eco->eco_refcount == 0);
        if (!eco->eco_lsm) 
                CERROR("No object %s\n", obd->obd_name);
        else 
                obd_free_memmd(ec->ec_exp, &eco->eco_lsm);
        OBD_FREE (eco, sizeof (*eco));
}

static int echo_create_object(struct obd_device *obd, int on_target,
                              struct obdo *oa, void *ulsm, int ulsm_nob,
                              struct obd_trans_info *oti)
{
        struct echo_client_obd *ec = &obd->u.echo_client;
        struct ec_object       *eco2;
        struct ec_object       *eco;
        struct lov_stripe_md   *lsm;
        int                     rc;
        int                     i, idx;

        if ((oa->o_valid & OBD_MD_FLID) == 0 && /* no obj id */
            (on_target ||                       /* set_stripe */
             ec->ec_nstripes != 0)) {           /* LOV */
                CERROR ("No valid oid\n");
                return (-EINVAL);
        }

        if (ulsm != NULL) {
                eco = echo_allocate_object (obd);
                if (eco == NULL) 
                        return (-ENOMEM);

                lsm = eco->eco_lsm;

                rc = echo_copyin_lsm (obd, lsm, ulsm, ulsm_nob);
                if (rc != 0)
                        goto failed;

                /* setup object ID here for !on_target and LOV hint */
                if ((oa->o_valid & OBD_MD_FLID) != 0)
                        eco->eco_id = lsm->lsm_object_id = oa->o_id;

                if (lsm->lsm_stripe_count == 0)
                        lsm->lsm_stripe_count = ec->ec_nstripes;

                if (lsm->lsm_stripe_size == 0)
                        lsm->lsm_stripe_size = CFS_PAGE_SIZE;

                idx = ll_rand();

                /* setup stripes: indices + default ids if required */
                for (i = 0; i < lsm->lsm_stripe_count; i++) {
                        if (lsm->lsm_oinfo[i]->loi_id == 0)
                                lsm->lsm_oinfo[i]->loi_id = lsm->lsm_object_id;

                        lsm->lsm_oinfo[i]->loi_ost_idx =
                                (idx + i) % ec->ec_nstripes;
                }
        } else {
                OBD_ALLOC(eco, sizeof(*eco));
                if (!eco) 
                        return (-ENOMEM);
                eco->eco_device = obd;
                lsm = NULL;
        }

        if (oa->o_id == 0)
                oa->o_id = ++last_object_id;

        if (on_target) {
                oa->o_gr = FILTER_GROUP_ECHO;
                oa->o_valid |= OBD_MD_FLGROUP;

                rc = obd_create(ec->ec_exp, oa, &lsm, oti);
                if (rc != 0)
                        goto failed;

                /* See what object ID we were given */
                eco->eco_id = oa->o_id = lsm->lsm_object_id;
                oa->o_valid |= OBD_MD_FLID;

                LASSERT(eco->eco_lsm == NULL || eco->eco_lsm == lsm);
                eco->eco_lsm = lsm;
        }

        spin_lock (&ec->ec_lock);

        eco2 = echo_find_object_locked (obd, oa->o_id);
        if (eco2 != NULL) {                     /* conflict */
                spin_unlock (&ec->ec_lock);

                CERROR ("Can't create object id "LPX64": id already exists%s\n",
                        oa->o_id, on_target ? " (undoing create)" : "");

                if (on_target)
                        obd_destroy(ec->ec_exp, oa, lsm, oti, NULL);

                rc = -EEXIST;
                goto failed;
        }

        list_add (&eco->eco_obj_chain, &ec->ec_objects);
        spin_unlock (&ec->ec_lock);
        CDEBUG (D_INFO,
                "created %p: "LPX64"=%u#%u@%u refs %d del %d\n",
                eco, eco->eco_id,
                eco->eco_lsm->lsm_stripe_size,
                eco->eco_lsm->lsm_stripe_count,
                eco->eco_lsm->lsm_oinfo[0]->loi_ost_idx,
                eco->eco_refcount, eco->eco_deleted);
        return (0);

 failed:
        echo_free_object (eco);
        if (rc) 
                CERROR("%s: err %d on create\n", obd->obd_name, rc);
        return (rc);
}

static int
echo_get_object (struct ec_object **ecop, struct obd_device *obd,
                 struct obdo *oa)
{
        struct echo_client_obd *ec = &obd->u.echo_client;
        struct ec_object       *eco;
        struct ec_object       *eco2;
        int                     rc;

        if ((oa->o_valid & OBD_MD_FLID) == 0 ||
            oa->o_id == 0)                      /* disallow use of object id 0 */
        {
                CERROR ("No valid oid\n");
                return (-EINVAL);
        }

        spin_lock (&ec->ec_lock);
        eco = echo_find_object_locked (obd, oa->o_id);
        if (eco != NULL) {
                if (eco->eco_deleted) {           /* being deleted */
                        spin_unlock(&ec->ec_lock);/* (see comment in cleanup) */
                        return (-EAGAIN);
                }

                eco->eco_refcount++;
                spin_unlock (&ec->ec_lock);
                *ecop = eco;
                CDEBUG (D_INFO,
                        "found %p: "LPX64"=%u#%u@%u refs %d del %d\n",
                        eco, eco->eco_id,
                        eco->eco_lsm->lsm_stripe_size,
                        eco->eco_lsm->lsm_stripe_count,
                        eco->eco_lsm->lsm_oinfo[0]->loi_ost_idx,
                        eco->eco_refcount, eco->eco_deleted);
                return (0);
        }
        spin_unlock (&ec->ec_lock);

        if (ec->ec_nstripes != 0)               /* striping required */
                return (-ENOENT);

        eco = echo_allocate_object (obd);
        if (eco == NULL)
                return (-ENOMEM);

        eco->eco_id = eco->eco_lsm->lsm_object_id = oa->o_id;

        spin_lock (&ec->ec_lock);

        eco2 = echo_find_object_locked (obd, oa->o_id);
        if (eco2 == NULL) {                     /* didn't race */
                list_add (&eco->eco_obj_chain, &ec->ec_objects);
                spin_unlock (&ec->ec_lock);
                eco->eco_refcount = 1;
                *ecop = eco;
                CDEBUG (D_INFO,
                        "created %p: "LPX64"=%u#%u@%d refs %d del %d\n",
                        eco, eco->eco_id,
                        eco->eco_lsm->lsm_stripe_size,
                        eco->eco_lsm->lsm_stripe_count,
                        eco->eco_lsm->lsm_oinfo[0]->loi_ost_idx,
                        eco->eco_refcount, eco->eco_deleted);
                return (0);
        }

        if (eco2->eco_deleted)
                rc = -EAGAIN;                   /* lose race */
        else {
                eco2->eco_refcount++;           /* take existing */
                *ecop = eco2;
                rc = 0;
                LASSERT (eco2->eco_id == eco2->eco_lsm->lsm_object_id);
                CDEBUG (D_INFO,
                        "found(2) %p: "LPX64"=%u#%u@%d refs %d del %d\n",
                        eco2, eco2->eco_id,
                        eco2->eco_lsm->lsm_stripe_size,
                        eco2->eco_lsm->lsm_stripe_count,
                        eco2->eco_lsm->lsm_oinfo[0]->loi_ost_idx,
                        eco2->eco_refcount, eco2->eco_deleted);
        }

        spin_unlock (&ec->ec_lock);

        echo_free_object (eco);
        return (rc);
}

static void
echo_put_object (struct ec_object *eco)
{
        struct obd_device      *obd = eco->eco_device;
        struct echo_client_obd *ec = &obd->u.echo_client;

        /* Release caller's ref on the object.
         * delete => mark for deletion when last ref goes
         */

        spin_lock (&ec->ec_lock);

        eco->eco_refcount--;
        LASSERT (eco->eco_refcount >= 0);