lov_obd.c

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

        if (tgt->ltd_exp)
                lov_disconnect_obd(obd, index);

        /* XXX - right now there is a dependency on ld_tgt_count being the
         * maximum tgt index for computing the mds_max_easize. So we can't
         * shrink it. */

        lov->lov_tgts[index] = NULL;
        OBD_FREE_PTR(tgt);        

        /* Manual cleanup - no cleanup logs to clean up the osc's.  We must
           do it ourselves. And we can't do it from lov_cleanup,
           because we just lost our only reference to it. */
        if (osc_obd) {
                /* Use lov's force/fail flags. */
                osc_obd->obd_force = obd->obd_force;
                osc_obd->obd_fail = obd->obd_fail;
                class_manual_cleanup(osc_obd);
        }
}

void lov_fix_desc_stripe_size(__u64 *val)
{
        if (*val < PTLRPC_MAX_BRW_SIZE) {
                LCONSOLE_WARN("Increasing default stripe size to min %u\n",
                              PTLRPC_MAX_BRW_SIZE);
                *val = PTLRPC_MAX_BRW_SIZE;
        } else if (*val & (LOV_MIN_STRIPE_SIZE - 1)) {
                *val &= ~(LOV_MIN_STRIPE_SIZE - 1);
                LCONSOLE_WARN("Changing default stripe size to "LPU64" (a "
                              "multiple of %u)\n",
                              *val, LOV_MIN_STRIPE_SIZE);
        }
}

void lov_fix_desc_stripe_count(__u32 *val)
{
        if (*val == 0)
                *val = 1;
}

void lov_fix_desc_pattern(__u32 *val)
{
        /* from lov_setstripe */
        if ((*val != 0) && (*val != LOV_PATTERN_RAID0)) {
                LCONSOLE_WARN("Unknown stripe pattern: %#x\n", *val);
                *val = 0;
        }
}

void lov_fix_desc_qos_maxage(__u32 *val)
{
        /* fix qos_maxage */
        if (*val == 0)
                *val = QOS_DEFAULT_MAXAGE;
}

void lov_fix_desc(struct lov_desc *desc)
{
        lov_fix_desc_stripe_size(&desc->ld_default_stripe_size);
        lov_fix_desc_stripe_count(&desc->ld_default_stripe_count);
        lov_fix_desc_pattern(&desc->ld_pattern);
        lov_fix_desc_qos_maxage(&desc->ld_qos_maxage);
}

static int lov_setup(struct obd_device *obd, obd_count len, void *buf)
{
        struct lprocfs_static_vars lvars = { 0 };
        struct lustre_cfg *lcfg = buf;
        struct lov_desc *desc;
        struct lov_obd *lov = &obd->u.lov;
        int count;
        ENTRY;

        if (LUSTRE_CFG_BUFLEN(lcfg, 1) < 1) {
                CERROR("LOV setup requires a descriptor\n");
                RETURN(-EINVAL);
        }

        desc = (struct lov_desc *)lustre_cfg_buf(lcfg, 1);

        if (sizeof(*desc) > LUSTRE_CFG_BUFLEN(lcfg, 1)) {
                CERROR("descriptor size wrong: %d > %d\n",
                       (int)sizeof(*desc), LUSTRE_CFG_BUFLEN(lcfg, 1));
                RETURN(-EINVAL);
        }

        if (desc->ld_magic != LOV_DESC_MAGIC) {
                if (desc->ld_magic == __swab32(LOV_DESC_MAGIC)) {
                            CDEBUG(D_OTHER, "%s: Swabbing lov desc %p\n",
                                   obd->obd_name, desc);
                            lustre_swab_lov_desc(desc);
                } else {
                        CERROR("%s: Bad lov desc magic: %#x\n",
                               obd->obd_name, desc->ld_magic);
                        RETURN(-EINVAL);
                }
        }

        lov_fix_desc(desc);

        /* Because of 64-bit divide/mod operations only work with a 32-bit
         * divisor in a 32-bit kernel, we cannot support a stripe width
         * of 4GB or larger on 32-bit CPUs. */
        count = desc->ld_default_stripe_count;
        if ((count > 0 ? count : desc->ld_tgt_count) *
            desc->ld_default_stripe_size > 0xffffffff) {
                CERROR("LOV: stripe width "LPU64"x%u > 4294967295 bytes\n",
                       desc->ld_default_stripe_size, count);
                RETURN(-EINVAL);
        }

        desc->ld_active_tgt_count = 0;
        lov->desc = *desc;
        lov->lov_tgt_size = 0;
        sema_init(&lov->lov_lock, 1);
        atomic_set(&lov->lov_refcount, 0);
        CFS_INIT_LIST_HEAD(&lov->lov_qos.lq_oss_list);
        init_rwsem(&lov->lov_qos.lq_rw_sem);
        lov->lov_qos.lq_dirty = 1;
        lov->lov_qos.lq_dirty_rr = 1;
        lov->lov_qos.lq_reset = 1;
        /* Default priority is toward free space balance */
        lov->lov_qos.lq_prio_free = 232;

        lprocfs_lov_init_vars(&lvars);
        lprocfs_obd_setup(obd, lvars.obd_vars);
#ifdef LPROCFS
        {
                cfs_proc_dir_entry_t *entry;

                entry = create_proc_entry("target_obd", 0444,
                                          obd->obd_proc_entry);
                if (entry != NULL) {
                        entry->proc_fops = &lov_proc_target_fops;
                        entry->data = obd;
                }
        }
#endif

        RETURN(0);
}

static int lov_precleanup(struct obd_device *obd, enum obd_cleanup_stage stage)
{
        int rc = 0;
        ENTRY;

        switch (stage) {
        case OBD_CLEANUP_EARLY: {
                struct lov_obd *lov = &obd->u.lov;
                int i;
                for (i = 0; i < lov->desc.ld_tgt_count; i++) {
                        if (!lov->lov_tgts[i] || !lov->lov_tgts[i]->ltd_active)
                                continue;
                        obd_precleanup(class_exp2obd(lov->lov_tgts[i]->ltd_exp),
                                       OBD_CLEANUP_EARLY);
                }
                break;
        }
        case OBD_CLEANUP_EXPORTS:
                break;
        case OBD_CLEANUP_SELF_EXP:
                rc = obd_llog_finish(obd, 0);
                if (rc != 0)
                        CERROR("failed to cleanup llogging subsystems\n");
                break;
        case OBD_CLEANUP_OBD:
                break;
        }
        RETURN(rc);
}

static int lov_cleanup(struct obd_device *obd)
{
        struct lov_obd *lov = &obd->u.lov;

        lprocfs_obd_cleanup(obd);
        if (lov->lov_tgts) {
                int i;
                for (i = 0; i < lov->desc.ld_tgt_count; i++) {
                        if (lov->lov_tgts[i]) {
                                /* Inactive targets may never have connected */
                                if (lov->lov_tgts[i]->ltd_active ||
                                    atomic_read(&lov->lov_refcount)) 
                                        /* We should never get here - these 
                                           should have been removed in the 
                                           disconnect. */
                                        CERROR("lov tgt %d not cleaned!"
                                               " deathrow=%d, lovrc=%d\n",
                                               i, lov->lov_death_row, 
                                               atomic_read(&lov->lov_refcount));
                                lov_del_target(obd, i, 0, 0);
                        }
                }
                OBD_FREE(lov->lov_tgts, sizeof(*lov->lov_tgts) * 
                         lov->lov_tgt_size);
                lov->lov_tgt_size = 0;
        }
        
        if (lov->lov_qos.lq_rr_size) 
                OBD_FREE(lov->lov_qos.lq_rr_array, lov->lov_qos.lq_rr_size);

        RETURN(0);
}

static int lov_process_config(struct obd_device *obd, obd_count len, void *buf)
{
        struct lustre_cfg *lcfg = buf;
        struct obd_uuid obd_uuid;
        int cmd;
        int rc = 0;
        ENTRY;

        switch(cmd = lcfg->lcfg_command) {
        case LCFG_LOV_ADD_OBD:
        case LCFG_LOV_ADD_INA:
        case LCFG_LOV_DEL_OBD: {
                __u32 index;
                int gen;
                /* lov_modify_tgts add  0:lov_mdsA  1:ost1_UUID  2:0  3:1 */
                if (LUSTRE_CFG_BUFLEN(lcfg, 1) > sizeof(obd_uuid.uuid))
                        GOTO(out, rc = -EINVAL);

                obd_str2uuid(&obd_uuid,  lustre_cfg_buf(lcfg, 1));

                if (sscanf(lustre_cfg_buf(lcfg, 2), "%d", &index) != 1)
                        GOTO(out, rc = -EINVAL);
                if (sscanf(lustre_cfg_buf(lcfg, 3), "%d", &gen) != 1)
                        GOTO(out, rc = -EINVAL);
                if (cmd == LCFG_LOV_ADD_OBD)
                        rc = lov_add_target(obd, &obd_uuid, index, gen, 1);
                else if (cmd == LCFG_LOV_ADD_INA)
                        rc = lov_add_target(obd, &obd_uuid, index, gen, 0);
                else
                        rc = lov_del_target(obd, index, &obd_uuid, gen);
                GOTO(out, rc);
        }
        case LCFG_PARAM: {
                struct lprocfs_static_vars lvars = { 0 };
                struct lov_desc *desc = &(obd->u.lov.desc);
                
                if (!desc)
                        GOTO(out, rc = -EINVAL);
                
                lprocfs_lov_init_vars(&lvars);
                
                rc = class_process_proc_param(PARAM_LOV, lvars.obd_vars,
                                              lcfg, obd);
                GOTO(out, rc);
        }
        default: {
                CERROR("Unknown command: %d\n", lcfg->lcfg_command);
                GOTO(out, rc = -EINVAL);

        }
        }
out:
        RETURN(rc);
}

#ifndef log2
#define log2(n) ffz(~(n))
#endif

static int lov_clear_orphans(struct obd_export *export, struct obdo *src_oa,
                             struct lov_stripe_md **ea,
                             struct obd_trans_info *oti)
{
        struct lov_obd *lov;
        struct obdo *tmp_oa;
        struct obd_uuid *ost_uuid = NULL;
        int rc = 0, i;
        ENTRY;

        LASSERT(src_oa->o_valid & OBD_MD_FLFLAGS &&
                src_oa->o_flags == OBD_FL_DELORPHAN);

        lov = &export->exp_obd->u.lov;

        OBDO_ALLOC(tmp_oa);
        if (tmp_oa == NULL)
                RETURN(-ENOMEM);

        if (oti->oti_ost_uuid) {
                ost_uuid = oti->oti_ost_uuid;
                CDEBUG(D_HA, "clearing orphans only for %s\n",
                       ost_uuid->uuid);
        }

        lov_getref(export->exp_obd);
        for (i = 0; i < lov->desc.ld_tgt_count; i++) {
                struct lov_stripe_md obj_md;
                struct lov_stripe_md *obj_mdp = &obj_md;
                struct lov_tgt_desc *tgt;
                int err;

                tgt = lov->lov_tgts[i];
                if (!tgt)
                        continue;

                /* if called for a specific target, we don't
                   care if it is not active. */
                if (!lov->lov_tgts[i]->ltd_active && ost_uuid == NULL) {
                        CDEBUG(D_HA, "lov idx %d inactive\n", i);
                        continue;
                }

                if (ost_uuid && !obd_uuid_equals(ost_uuid, &tgt->ltd_uuid))
                        continue;

                CDEBUG(D_CONFIG,"Clear orphans for %d:%s\n", i, 
                       obd_uuid2str(ost_uuid));

                memcpy(tmp_oa, src_oa, sizeof(*tmp_oa));

                LASSERT(lov->lov_tgts[i]->ltd_exp);
                /* XXX: LOV STACKING: use real "obj_mdp" sub-data */
                err = obd_create(lov->lov_tgts[i]->ltd_exp, 
                                 tmp_oa, &obj_mdp, oti);
                if (err)
                        /* This export will be disabled until it is recovered,
                           and then orphan recovery will be completed. */
                        CERROR("error in orphan recovery on OST idx %d/%d: "
                               "rc = %d\n", i, lov->desc.ld_tgt_count, err);

                if (ost_uuid)
                        break;
        }
        lov_putref(export->exp_obd);

        OBDO_FREE(tmp_oa);
        RETURN(rc);
}

static int lov_recreate(struct obd_export *exp, struct obdo *src_oa,
                        struct lov_stripe_md **ea, struct obd_trans_info *oti)
{
        struct lov_stripe_md *obj_mdp, *lsm;
        struct lov_obd *lov = &exp->exp_obd->u.lov;
        unsigned ost_idx;
        int rc, i;
        ENTRY;

        LASSERT(src_oa->o_valid & OBD_MD_FLFLAGS &&
                src_oa->o_flags & OBD_FL_RECREATE_OBJS);

        OBD_ALLOC(obj_mdp, sizeof(*obj_mdp));
        if (obj_mdp == NULL)
                RETURN(-ENOMEM);

        ost_idx = src_oa->o_nlink;
        lsm = *ea;
        if (lsm == NULL)
                GOTO(out, rc = -EINVAL);
        if (ost_idx >= lov->desc.ld_tgt_count ||
            !lov->lov_tgts[ost_idx])
                GOTO(out, rc = -EINVAL);

        for (i = 0; i < lsm->lsm_stripe_count; i++) {
                if (lsm->lsm_oinfo[i]->loi_ost_idx == ost_idx) {
                        if (lsm->lsm_oinfo[i]->loi_id != src_oa->o_id)
                                GOTO(out, rc = -EINVAL);
                        break;
                }
        }
        if (i == lsm->lsm_stripe_count)
                GOTO(out, rc = -EINVAL);

        rc = obd_create(lov->lov_tgts[ost_idx]->ltd_exp, src_oa, &obj_mdp, oti);
out:
        OBD_FREE(obj_mdp, sizeof(*obj_mdp));
        RETURN(rc);