lov_request.c

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

{
        struct {
                obd_count       index;
                obd_count       count;
                obd_count       off;
        } *info = NULL;
        struct lov_request_set *set;
        struct lov_oinfo *loi = NULL;
        struct lov_obd *lov = &exp->exp_obd->u.lov;
        int rc = 0, i, shift;
        ENTRY;

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

        set->set_exp = exp;
        set->set_oti = oti;
        set->set_oi = oinfo;
        set->set_oabufs = oa_bufs;
        OBD_ALLOC(set->set_pga, oa_bufs * sizeof(*set->set_pga));
        if (!set->set_pga)
                GOTO(out, rc = -ENOMEM);

        OBD_ALLOC(info, sizeof(*info) * oinfo->oi_md->lsm_stripe_count);
        if (!info)
                GOTO(out, rc = -ENOMEM);

        /* calculate the page count for each stripe */
        for (i = 0; i < oa_bufs; i++) {
                int stripe = lov_stripe_number(oinfo->oi_md, pga[i].off);
                info[stripe].count++;
        }

        /* alloc and initialize lov request */
        shift = 0;
        for (i = 0 ; i < oinfo->oi_md->lsm_stripe_count; i++){
                struct lov_request *req;

                if (info[i].count == 0)
                        continue;

                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);
                        GOTO(out, rc = -EIO);
                }

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

                OBDO_ALLOC(req->rq_oi.oi_oa);
                if (req->rq_oi.oi_oa == NULL) {
                        OBD_FREE(req, sizeof(*req));
                        GOTO(out, rc = -ENOMEM);
                }

                if (oinfo->oi_oa) {
                        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_buflen = sizeof(*req->rq_oi.oi_md);
                OBD_ALLOC(req->rq_oi.oi_md, req->rq_buflen);
                if (req->rq_oi.oi_md == NULL) {
                        OBDO_FREE(req->rq_oi.oi_oa);
                        OBD_FREE(req, sizeof(*req));
                        GOTO(out, rc = -ENOMEM);
                }

                req->rq_idx = loi->loi_ost_idx;
                req->rq_stripe = i;

                /* XXX LOV STACKING */
                req->rq_oi.oi_md->lsm_object_id = loi->loi_id;
                req->rq_oi.oi_md->lsm_object_gr = oinfo->oi_md->lsm_object_gr;
                req->rq_oabufs = info[i].count;
                req->rq_pgaidx = shift;
                shift += req->rq_oabufs;

                /* remember the index for sort brw_page array */
                info[i].index = req->rq_pgaidx;

                lov_set_add_req(req, set);
        }
        if (!set->set_count)
                GOTO(out, rc = -EIO);

        /* rotate & sort the brw_page array */
        for (i = 0; i < oa_bufs; i++) {
                int stripe = lov_stripe_number(oinfo->oi_md, pga[i].off);

                shift = info[stripe].index + info[stripe].off;
                LASSERT(shift < oa_bufs);
                set->set_pga[shift] = pga[i];
                lov_stripe_offset(oinfo->oi_md, pga[i].off, stripe,
                                  &set->set_pga[shift].off);
                info[stripe].off++;
        }
out:
        if (info)
                OBD_FREE(info, sizeof(*info) * oinfo->oi_md->lsm_stripe_count);

        if (rc == 0)
                *reqset = set;
        else
                lov_fini_brw_set(set);

        RETURN(rc);
}

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

        if (set == NULL)
                RETURN(0);
        LASSERT(set->set_exp);
        if (set->set_completes)
                rc = common_attr_done(set);

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

        RETURN(rc);
}

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

int lov_prep_getattr_set(struct obd_export *exp, struct obd_info *oinfo,
                         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;

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

                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;
                }

                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_cb_up = cb_getattr_update;

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

int lov_fini_destroy_set(struct lov_request_set *set)
{
        ENTRY;

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

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

        RETURN(0);
}

int lov_prep_destroy_set(struct obd_export *exp, struct obd_info *oinfo,
                         struct obdo *src_oa, struct lov_stripe_md *lsm,
                         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_exp = exp;
        set->set_oi = oinfo;
        set->set_oi->oi_md = lsm;
        set->set_oi->oi_oa = src_oa;
        set->set_oti = oti;
        if (oti != NULL && src_oa->o_valid & OBD_MD_FLCOOKIE)
                set->set_cookies = oti->oti_logcookies;

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

                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;
                }

                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;
                lov_set_add_req(req, set);
        }
        if (!set->set_count)
                GOTO(out_set, rc = -EIO);
        *reqset = set;
        RETURN(rc);
out_set:
        lov_fini_destroy_set(set);
        RETURN(rc);
}

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

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

        if (atomic_dec_and_test(&set->set_refcount))
                lov_finish_set(set);
        RETURN(rc);
}

int lov_update_setattr_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] && 
                    lov->lov_tgts[req->rq_idx]->ltd_active))
                rc = 0;

        if (rc == 0) {
                if (req->rq_oi.oi_oa->o_valid & OBD_MD_FLCTIME)
                        lsm->lsm_oinfo[req->rq_stripe]->loi_lvb.lvb_ctime =
                                req->rq_oi.oi_oa->o_ctime;
                if (req->rq_oi.oi_oa->o_valid & OBD_MD_FLMTIME)
                        lsm->lsm_oinfo[req->rq_stripe]->loi_lvb.lvb_mtime =
                                req->rq_oi.oi_oa->o_mtime;
                if (req->rq_oi.oi_oa->o_valid & OBD_MD_FLATIME)
                        lsm->lsm_oinfo[req->rq_stripe]->loi_lvb.lvb_atime =
                                req->rq_oi.oi_oa->o_atime;
        }

        RETURN(rc);
}

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

int lov_prep_setattr_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_exp = exp;
        set->set_oti = oti;
        set->set_oi = oinfo;
        if (oti != NULL && oinfo->oi_oa->o_valid & OBD_MD_FLCOOKIE)
                set->set_cookies = oti->oti_logcookies;

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

                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;
                }

                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_setattr_update;
                req->rq_rqset = set;

                if (oinfo->oi_oa->o_valid & OBD_MD_FLSIZE) {
                        int off = lov_stripe_offset(oinfo->oi_md,
                                                    oinfo->oi_oa->o_size, i,
                                                    &req->rq_oi.oi_oa->o_size);

                        if (off < 0 && req->rq_oi.oi_oa->o_size)
                                req->rq_oi.oi_oa->o_size--;

                        CDEBUG(D_INODE, "stripe %d has size "LPU64"/"LPU64"\n",
                               i, req->rq_oi.oi_oa->o_size,
                               oinfo->oi_oa->o_size);
                }
                lov_set_add_req(req, set);
        }
        if (!set->set_count)
                GOTO(out_set, rc = -EIO);
        *reqset = set;
        RETURN(rc);
out_set:
        lov_fini_setattr_set(set);
        RETURN(rc);
}

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

        if (set == NULL)
                RETURN(0);
        LASSERT(set->set_exp);
        if (set->set_completes) {