lov_request.c

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

                rc = -EIO;
                /* FIXME update qos data here */
                if (set->set_success)
                        rc = common_attr_done(set);
        }

        if (atomic_dec_and_test(&set->set_refcount))
                lov_finish_set(set);

        RETURN(rc);
}

int lov_update_punch_set(struct lov_request_set *set,
                           struct lov_request *req, int rc)
{
        struct lov_obd *lov = &req->rq_rqset->set_exp->exp_obd->u.lov;
        struct lov_stripe_md *lsm = req->rq_rqset->set_oi->oi_md;
        ENTRY;

        lov_update_set(set, req, rc);

        /* grace error on inactive ost */
        if (rc && !lov->lov_tgts[req->rq_idx]->ltd_active)
                rc = 0;

        if (rc == 0) {
                lov_stripe_lock(lsm);
                if (req->rq_oi.oi_oa->o_valid & OBD_MD_FLBLOCKS) {
                        lsm->lsm_oinfo[req->rq_stripe]->loi_lvb.lvb_blocks =
                                req->rq_oi.oi_oa->o_blocks;
                }

                /* Do we need to update lvb_size here? It needn't because
                 * it have been done in ll_truncate(). -jay */
                lov_stripe_unlock(lsm);
        }

        RETURN(rc);
}

/* The callback for osc_punch that finilizes a request info when a response
 * is recieved. */
static int cb_update_punch(struct obd_info *oinfo, int rc)
{
        struct lov_request *lovreq;
        lovreq = container_of(oinfo, struct lov_request, rq_oi);
        return lov_update_punch_set(lovreq->rq_rqset, lovreq, rc);
}

int lov_prep_punch_set(struct obd_export *exp, struct obd_info *oinfo,
                       struct obd_trans_info *oti,
                       struct lov_request_set **reqset)
{
        struct lov_request_set *set;
        struct lov_oinfo *loi = NULL;
        struct lov_obd *lov = &exp->exp_obd->u.lov;
        int rc = 0, i;
        ENTRY;

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

        set->set_oi = oinfo;
        set->set_exp = exp;

        for (i = 0; i < oinfo->oi_md->lsm_stripe_count; i++) {
                struct lov_request *req;
                obd_off rs, re;

                loi = oinfo->oi_md->lsm_oinfo[i];
                if (!lov->lov_tgts[loi->loi_ost_idx] ||
                    !lov->lov_tgts[loi->loi_ost_idx]->ltd_active) {
                        CDEBUG(D_HA, "lov idx %d inactive\n", loi->loi_ost_idx);
                        continue;
                }

                if (!lov_stripe_intersects(oinfo->oi_md, i,
                                           oinfo->oi_policy.l_extent.start,
                                           oinfo->oi_policy.l_extent.end,
                                           &rs, &re))
                        continue;

                OBD_ALLOC(req, sizeof(*req));
                if (req == NULL)
                        GOTO(out_set, rc = -ENOMEM);
                req->rq_stripe = i;
                req->rq_idx = loi->loi_ost_idx;

                OBDO_ALLOC(req->rq_oi.oi_oa);
                if (req->rq_oi.oi_oa == NULL) {
                        OBD_FREE(req, sizeof(*req));
                        GOTO(out_set, rc = -ENOMEM);
                }
                memcpy(req->rq_oi.oi_oa, oinfo->oi_oa,
                       sizeof(*req->rq_oi.oi_oa));
                req->rq_oi.oi_oa->o_id = loi->loi_id;
                req->rq_oi.oi_oa->o_stripe_idx = i;
                req->rq_oi.oi_cb_up = cb_update_punch;
                req->rq_rqset = set;

                req->rq_oi.oi_policy.l_extent.start = rs;
                req->rq_oi.oi_policy.l_extent.end = re;
                req->rq_oi.oi_policy.l_extent.gid = -1;

                lov_set_add_req(req, set);
        }
        if (!set->set_count)
                GOTO(out_set, rc = -EIO);
        *reqset = set;
        RETURN(rc);
out_set:
        lov_fini_punch_set(set);
        RETURN(rc);
}

int lov_fini_sync_set(struct lov_request_set *set)
{
        int rc = 0;
        ENTRY;

        if (set == NULL)
                RETURN(0);
        LASSERT(set->set_exp);
        if (set->set_completes) {
                if (!set->set_success)
                        rc = -EIO;
                /* FIXME update qos data here */
        }

        if (atomic_dec_and_test(&set->set_refcount))
                lov_finish_set(set);

        RETURN(rc);
}

int lov_prep_sync_set(struct obd_export *exp, struct obd_info *oinfo,
                      struct obdo *src_oa, struct lov_stripe_md *lsm,
                      obd_off start, obd_off end,
                      struct lov_request_set **reqset)
{
        struct lov_request_set *set;
        struct lov_oinfo *loi = NULL;
        struct lov_obd *lov = &exp->exp_obd->u.lov;
        int rc = 0, i;
        ENTRY;

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

        set->set_exp = exp;
        set->set_oi = oinfo;
        set->set_oi->oi_md = lsm;
        set->set_oi->oi_oa = src_oa;

        for (i = 0; i < lsm->lsm_stripe_count; i++) {
                struct lov_request *req;
                obd_off rs, re;

                loi = lsm->lsm_oinfo[i];
                if (!lov->lov_tgts[loi->loi_ost_idx] ||
                    !lov->lov_tgts[loi->loi_ost_idx]->ltd_active) {
                        CDEBUG(D_HA, "lov idx %d inactive\n", loi->loi_ost_idx);
                        continue;
                }

                if (!lov_stripe_intersects(lsm, i, start, end, &rs, &re))
                        continue;

                OBD_ALLOC(req, sizeof(*req));
                if (req == NULL)
                        GOTO(out_set, rc = -ENOMEM);
                req->rq_stripe = i;
                req->rq_idx = loi->loi_ost_idx;

                OBDO_ALLOC(req->rq_oi.oi_oa);
                if (req->rq_oi.oi_oa == NULL) {
                        OBD_FREE(req, sizeof(*req));
                        GOTO(out_set, rc = -ENOMEM);
                }
                memcpy(req->rq_oi.oi_oa, src_oa, sizeof(*req->rq_oi.oi_oa));
                req->rq_oi.oi_oa->o_id = loi->loi_id;
                req->rq_oi.oi_oa->o_stripe_idx = i;

                req->rq_oi.oi_policy.l_extent.start = rs;
                req->rq_oi.oi_policy.l_extent.end = re;
                req->rq_oi.oi_policy.l_extent.gid = -1;

                lov_set_add_req(req, set);
        }
        if (!set->set_count)
                GOTO(out_set, rc = -EIO);
        *reqset = set;
        RETURN(rc);
out_set:
        lov_fini_sync_set(set);
        RETURN(rc);
}

#define LOV_U64_MAX ((__u64)~0ULL)
#define LOV_SUM_MAX(tot, add)                                           \
        do {                                                            \
                if ((tot) + (add) < (tot))                              \
                        (tot) = LOV_U64_MAX;                            \
                else                                                    \
                        (tot) += (add);                                 \
        } while(0)

int lov_fini_statfs(struct obd_device *obd, struct obd_statfs *osfs,int success)
{
        ENTRY;

        if (success) {
                __u32 expected_stripes = lov_get_stripecnt(&obd->u.lov, 0);

                if (osfs->os_files != LOV_U64_MAX)
                        do_div(osfs->os_files, expected_stripes);
                if (osfs->os_ffree != LOV_U64_MAX)
                        do_div(osfs->os_ffree, expected_stripes);

                spin_lock(&obd->obd_osfs_lock);
                memcpy(&obd->obd_osfs, osfs, sizeof(*osfs));
                obd->obd_osfs_age = get_jiffies_64();
                spin_unlock(&obd->obd_osfs_lock);
                RETURN(0);
        }

        RETURN(-EIO);
}

int lov_fini_statfs_set(struct lov_request_set *set)
{
        int rc = 0;
        ENTRY;

        if (set == NULL)
                RETURN(0);

        if (set->set_completes) {
                rc = lov_fini_statfs(set->set_obd, set->set_oi->oi_osfs,
                                     set->set_success);
        }

        if (atomic_dec_and_test(&set->set_refcount))
                lov_finish_set(set);

        RETURN(rc);
}

void lov_update_statfs(struct obd_device *obd, struct obd_statfs *osfs,
                       struct obd_statfs *lov_sfs, int success)
{
        int shift = 0, quit = 0;
        __u64 tmp;
        spin_lock(&obd->obd_osfs_lock);
        memcpy(&obd->obd_osfs, lov_sfs, sizeof(*lov_sfs));
        obd->obd_osfs_age = get_jiffies_64();
        spin_unlock(&obd->obd_osfs_lock);

        if (success == 0) {
                memcpy(osfs, lov_sfs, sizeof(*lov_sfs));
        } else {
                if (osfs->os_bsize != lov_sfs->os_bsize) {
                        /* assume all block sizes are always powers of 2 */
                        /* get the bits difference */
                        tmp = osfs->os_bsize | lov_sfs->os_bsize;
                        for (shift = 0; shift <= 64; ++shift) {
                                if (tmp & 1) {
                                        if (quit)
                                                break;
                                        else
                                                quit = 1;
                                        shift = 0;
                                }
                                tmp >>= 1;
                        }
                }

                if (osfs->os_bsize < lov_sfs->os_bsize) {
                        osfs->os_bsize = lov_sfs->os_bsize;

                        osfs->os_bfree  >>= shift;
                        osfs->os_bavail >>= shift;
                        osfs->os_blocks >>= shift;
                } else if (shift != 0) {
                        lov_sfs->os_bfree  >>= shift;
                        lov_sfs->os_bavail >>= shift;
                        lov_sfs->os_blocks >>= shift;
                }
#ifdef MIN_DF
                /* Sandia requested that df (and so, statfs) only
                   returned minimal available space on
                   a single OST, so people would be able to
                   write this much data guaranteed. */
                if (osfs->os_bavail > lov_sfs->os_bavail) {
                        /* Presumably if new bavail is smaller,
                           new bfree is bigger as well */
                        osfs->os_bfree = lov_sfs->os_bfree;
                        osfs->os_bavail = lov_sfs->os_bavail;
                }
#else
                osfs->os_bfree += lov_sfs->os_bfree;
                osfs->os_bavail += lov_sfs->os_bavail;
#endif
                osfs->os_blocks += lov_sfs->os_blocks;
                /* XXX not sure about this one - depends on policy.
                 *   - could be minimum if we always stripe on all OBDs
                 *     (but that would be wrong for any other policy,
                 *     if one of the OBDs has no more objects left)
                 *   - could be sum if we stripe whole objects
                 *   - could be average, just to give a nice number
                 *
                 * To give a "reasonable" (if not wholly accurate)
                 * number, we divide the total number of free objects
                 * by expected stripe count (watch out for overflow).
                 */
                LOV_SUM_MAX(osfs->os_files, lov_sfs->os_files);
                LOV_SUM_MAX(osfs->os_ffree, lov_sfs->os_ffree);
        }
}

/* The callback for osc_statfs_async that finilizes a request info when a
 * response is recieved. */
static int cb_statfs_update(struct obd_info *oinfo, int rc)
{
        struct lov_request *lovreq;
        struct obd_statfs *osfs, *lov_sfs;
        struct obd_device *obd;
        struct lov_obd *lov;
        int success;
        ENTRY;

        lovreq = container_of(oinfo, struct lov_request, rq_oi);
        lov = &lovreq->rq_rqset->set_obd->u.lov;
        obd = class_exp2obd(lov->lov_tgts[lovreq->rq_idx]->ltd_exp);

        osfs = lovreq->rq_rqset->set_oi->oi_osfs;
        lov_sfs = oinfo->oi_osfs;

        success = lovreq->rq_rqset->set_success;

        /* XXX: the same is done in lov_update_common_set, however
           lovset->set_exp is not initialized. */
        lov_update_set(lovreq->rq_rqset, lovreq, rc);
        if (rc) {
                if (rc && !(lov->lov_tgts[lovreq->rq_idx] &&
                            lov->lov_tgts[lovreq->rq_idx]->ltd_active))
                        rc = 0;
                RETURN(rc);
        }

        lov_update_statfs(obd, osfs, lov_sfs, success);
        qos_update(lov);

        RETURN(0);
}

int lov_prep_statfs_set(struct obd_device *obd, struct obd_info *oinfo,
                        struct lov_request_set **reqset)
{
        struct lov_request_set *set;
        struct lov_obd *lov = &obd->u.lov;
        int rc = 0, i;
        ENTRY;

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

        set->set_obd = obd;
        set->set_oi = oinfo;

        /* We only get block data from the OBD */
        for (i = 0; i < lov->desc.ld_tgt_count; i++) {
                struct lov_request *req;

                if (!lov->lov_tgts[i] || !lov->lov_tgts[i]->ltd_active) {
                        CDEBUG(D_HA, "lov idx %d inactive\n", i);
                        continue;
                }

                OBD_ALLOC(req, sizeof(*req));
                if (req == NULL)
                        GOTO(out_set, rc = -ENOMEM);

                OBD_ALLOC(req->rq_oi.oi_osfs, sizeof(*req->rq_oi.oi_osfs));
                if (req->rq_oi.oi_osfs == NULL) {
                        OBD_FREE(req, sizeof(*req));
                        GOTO(out_set, rc = -ENOMEM);
                }

                req->rq_idx = i;
                req->rq_oi.oi_cb_up = cb_statfs_update;
                req->rq_rqset = set;

                lov_set_add_req(req, set);
        }
        if (!set->set_count)
                GOTO(out_set, rc = -EIO);
        *reqset = set;
        RETURN(rc);
out_set:
        lov_fini_statfs_set(set);
        RETURN(rc);
}