lov_obd.c

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

                lovset->set_completes = 0;
        err = lov_fini_punch_set(lovset);
        RETURN(rc ? rc : err);
}

/* FIXME: maybe we'll just make one node the authoritative attribute node, then
 * we can send this 'punch' to just the authoritative node and the nodes
 * that the punch will affect. */
static int lov_punch(struct obd_export *exp, struct obd_info *oinfo,
                     struct obd_trans_info *oti,
                     struct ptlrpc_request_set *rqset)
{
        struct lov_request_set *set;
        struct lov_obd *lov;
        struct list_head *pos;
        struct lov_request *req;
        int rc = 0;
        ENTRY;

        LASSERT(oinfo);
        ASSERT_LSM_MAGIC(oinfo->oi_md);

        if (!exp || !exp->exp_obd)
                RETURN(-ENODEV);

        lov = &exp->exp_obd->u.lov;
        rc = lov_prep_punch_set(exp, oinfo, oti, &set);
        if (rc)
                RETURN(rc);

        list_for_each (pos, &set->set_list) {
                req = list_entry(pos, struct lov_request, rq_link);

                rc = obd_punch(lov->lov_tgts[req->rq_idx]->ltd_exp,
                               &req->rq_oi, NULL, rqset);
                if (rc) {
                        CERROR("error: punch objid "LPX64" subobj "LPX64
                               " on OST idx %d: rc = %d\n",
                               set->set_oi->oi_oa->o_id,
                               req->rq_oi.oi_oa->o_id, req->rq_idx, rc);
                        break;
                }
        }

        if (rc || list_empty(&rqset->set_requests)) {
                int err;
                err = lov_fini_punch_set(set);
                RETURN(rc ? rc : err);
        }

        LASSERT(rqset->set_interpret == NULL);
        rqset->set_interpret = lov_punch_interpret;
        rqset->set_arg = (void *)set;

        RETURN(0);
}

static int lov_sync(struct obd_export *exp, struct obdo *oa,
                    struct lov_stripe_md *lsm, obd_off start, obd_off end)
{
        struct lov_request_set *set;
        struct obd_info oinfo;
        struct lov_obd *lov;
        struct list_head *pos;
        struct lov_request *req;
        int err = 0, rc = 0;
        ENTRY;

        ASSERT_LSM_MAGIC(lsm);

        if (!exp->exp_obd)
                RETURN(-ENODEV);

        lov = &exp->exp_obd->u.lov;
        rc = lov_prep_sync_set(exp, &oinfo, oa, lsm, start, end, &set);
        if (rc)
                RETURN(rc);

        list_for_each (pos, &set->set_list) {
                req = list_entry(pos, struct lov_request, rq_link);

                rc = obd_sync(lov->lov_tgts[req->rq_idx]->ltd_exp, 
                              req->rq_oi.oi_oa, NULL, 
                              req->rq_oi.oi_policy.l_extent.start,
                              req->rq_oi.oi_policy.l_extent.end);
                err = lov_update_common_set(set, req, rc);
                if (err) {
                        CERROR("error: fsync objid "LPX64" subobj "LPX64
                               " on OST idx %d: rc = %d\n",
                               set->set_oi->oi_oa->o_id,
                               req->rq_oi.oi_oa->o_id, req->rq_idx, rc);
                        if (!rc)
                                rc = err;
                }
        }
        err = lov_fini_sync_set(set);
        if (!rc)
                rc = err;
        RETURN(rc);
}

static int lov_brw_check(struct lov_obd *lov, struct obd_info *lov_oinfo,
                         obd_count oa_bufs, struct brw_page *pga)
{
        struct obd_info oinfo = { { { 0 } } };
        int i, rc = 0;

        oinfo.oi_oa = lov_oinfo->oi_oa;

        /* The caller just wants to know if there's a chance that this
         * I/O can succeed */
        for (i = 0; i < oa_bufs; i++) {
                int stripe = lov_stripe_number(lov_oinfo->oi_md, pga[i].off);
                int ost = lov_oinfo->oi_md->lsm_oinfo[stripe]->loi_ost_idx;
                obd_off start, end;

                if (!lov_stripe_intersects(lov_oinfo->oi_md, i, pga[i].off,
                                           pga[i].off + pga[i].count,
                                           &start, &end))
                        continue;

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

                rc = obd_brw(OBD_BRW_CHECK, lov->lov_tgts[ost]->ltd_exp, &oinfo,
                             1, &pga[i], NULL);
                if (rc)
                        break;
        }
        return rc;
}

static int lov_brw(int cmd, struct obd_export *exp, struct obd_info *oinfo,
                   obd_count oa_bufs, struct brw_page *pga,
                   struct obd_trans_info *oti)
{
        struct lov_request_set *set;
        struct lov_request *req;
        struct list_head *pos;
        struct lov_obd *lov = &exp->exp_obd->u.lov;
        int err, rc = 0;
        ENTRY;

        ASSERT_LSM_MAGIC(oinfo->oi_md);

        if (cmd == OBD_BRW_CHECK) {
                rc = lov_brw_check(lov, oinfo, oa_bufs, pga);
                RETURN(rc);
        }

        rc = lov_prep_brw_set(exp, oinfo, oa_bufs, pga, oti, &set);
        if (rc)
                RETURN(rc);

        list_for_each (pos, &set->set_list) {
                struct obd_export *sub_exp;
                struct brw_page *sub_pga;
                req = list_entry(pos, struct lov_request, rq_link);

                sub_exp = lov->lov_tgts[req->rq_idx]->ltd_exp;
                sub_pga = set->set_pga + req->rq_pgaidx;
                rc = obd_brw(cmd, sub_exp, &req->rq_oi, req->rq_oabufs,
                             sub_pga, oti);
                if (rc)
                        break;
                lov_update_common_set(set, req, rc);
        }

        err = lov_fini_brw_set(set);
        if (!rc)
                rc = err;
        RETURN(rc);
}

static int lov_brw_interpret(struct ptlrpc_request_set *reqset, void *data,
                             int rc)
{
        struct lov_request_set *lovset = (struct lov_request_set *)data;
        ENTRY;

        if (rc) {
                lovset->set_completes = 0;
                lov_fini_brw_set(lovset);
        } else {
                rc = lov_fini_brw_set(lovset);
        }

        RETURN(rc);
}

static int lov_brw_async(int cmd, struct obd_export *exp,
                         struct obd_info *oinfo, obd_count oa_bufs,
                         struct brw_page *pga, struct obd_trans_info *oti,
                         struct ptlrpc_request_set *set)
{
        struct lov_request_set *lovset;
        struct lov_request *req;
        struct list_head *pos;
        struct lov_obd *lov = &exp->exp_obd->u.lov;
        int rc = 0;
        ENTRY;

        LASSERT(oinfo);
        ASSERT_LSM_MAGIC(oinfo->oi_md);

        if (cmd == OBD_BRW_CHECK) {
                rc = lov_brw_check(lov, oinfo, oa_bufs, pga);
                RETURN(rc);
        }

        rc = lov_prep_brw_set(exp, oinfo, oa_bufs, pga, oti, &lovset);
        if (rc)
                RETURN(rc);

        list_for_each (pos, &lovset->set_list) {
                struct obd_export *sub_exp;
                struct brw_page *sub_pga;
                req = list_entry(pos, struct lov_request, rq_link);

                sub_exp = lov->lov_tgts[req->rq_idx]->ltd_exp;
                sub_pga = lovset->set_pga + req->rq_pgaidx;
                rc = obd_brw_async(cmd, sub_exp, &req->rq_oi, req->rq_oabufs,
                                   sub_pga, oti, set);
                if (rc)
                        GOTO(out, rc);
                lov_update_common_set(lovset, req, rc);
        }
        LASSERT(rc == 0);
        LASSERT(set->set_interpret == NULL);
        LASSERT(set->set_arg == NULL);
        rc = ptlrpc_set_add_cb(set, lov_brw_interpret, lovset);
        if (rc)
                GOTO(out, rc);

        RETURN(rc);
out:
        lov_fini_brw_set(lovset);
        RETURN(rc);
}

static int lov_ap_make_ready(void *data, int cmd)
{
        struct lov_async_page *lap = LAP_FROM_COOKIE(data);

        return lap->lap_caller_ops->ap_make_ready(lap->lap_caller_data, cmd);
}

static int lov_ap_refresh_count(void *data, int cmd)
{
        struct lov_async_page *lap = LAP_FROM_COOKIE(data);

        return lap->lap_caller_ops->ap_refresh_count(lap->lap_caller_data,
                                                     cmd);
}

static void lov_ap_fill_obdo(void *data, int cmd, struct obdo *oa)
{
        struct lov_async_page *lap = LAP_FROM_COOKIE(data);

        lap->lap_caller_ops->ap_fill_obdo(lap->lap_caller_data, cmd, oa);
        /* XXX woah, shouldn't we be altering more here?  size? */
        oa->o_id = lap->lap_loi_id;
        oa->o_stripe_idx = lap->lap_stripe;
}

static void lov_ap_update_obdo(void *data, int cmd, struct obdo *oa,
                               obd_valid valid)
{
        struct lov_async_page *lap = LAP_FROM_COOKIE(data);

        lap->lap_caller_ops->ap_update_obdo(lap->lap_caller_data, cmd,oa,valid);
}

static int lov_ap_completion(void *data, int cmd, struct obdo *oa, int rc)
{
        struct lov_async_page *lap = LAP_FROM_COOKIE(data);

        /* in a raid1 regime this would down a count of many ios
         * in flight, onl calling the caller_ops completion when all
         * the raid1 ios are complete */
        rc = lap->lap_caller_ops->ap_completion(lap->lap_caller_data,cmd,oa,rc);
        return rc;
}

static struct obd_async_page_ops lov_async_page_ops = {
        .ap_make_ready =        lov_ap_make_ready,
        .ap_refresh_count =     lov_ap_refresh_count,
        .ap_fill_obdo =         lov_ap_fill_obdo,
        .ap_update_obdo =       lov_ap_update_obdo,
        .ap_completion =        lov_ap_completion,
};

int lov_prep_async_page(struct obd_export *exp, struct lov_stripe_md *lsm,
                           struct lov_oinfo *loi, cfs_page_t *page,
                           obd_off offset, struct obd_async_page_ops *ops,
                           void *data, void **res, int nocache,
                           struct lustre_handle *lockh)
{
        struct lov_obd *lov = &exp->exp_obd->u.lov;
        struct lov_async_page *lap;
        struct lov_lock_handles *lov_lockh = NULL;
        int rc = 0;
        ENTRY;

        if (!page) {
                int i = 0;
                /* Find an existing osc so we can get it's stupid sizeof(*oap).
                   Only because of this layering limitation will a client 
                   mount with no osts fail */
                while (!lov->lov_tgts || !lov->lov_tgts[i] || 
                       !lov->lov_tgts[i]->ltd_exp) {
                        i++;
                        if (i >= lov->desc.ld_tgt_count) 
                                RETURN(-ENOMEDIUM);
                }
                rc = size_round(sizeof(*lap)) +
                        obd_prep_async_page(lov->lov_tgts[i]->ltd_exp, NULL,
                                            NULL, NULL, 0, NULL, NULL, NULL, 0,
                                            NULL);
                RETURN(rc);
        }
        ASSERT_LSM_MAGIC(lsm);
        LASSERT(loi == NULL);

        lap = *res;
        lap->lap_magic = LOV_AP_MAGIC;
        lap->lap_caller_ops = ops;
        lap->lap_caller_data = data;

        /* for now only raid 0 which passes through */
        lap->lap_stripe = lov_stripe_number(lsm, offset);
        lov_stripe_offset(lsm, offset, lap->lap_stripe, &lap->lap_sub_offset);
        loi = lsm->lsm_oinfo[lap->lap_stripe];

        /* so the callback doesn't need the lsm */
        lap->lap_loi_id = loi->loi_id;

        lap->lap_sub_cookie = (void *)lap + size_round(sizeof(*lap));

        if (lockh) {
                lov_lockh = lov_handle2llh(lockh);
                if (lov_lockh) {
                        lockh = lov_lockh->llh_handles + lap->lap_stripe;
                }
        }

        rc = obd_prep_async_page(lov->lov_tgts[loi->loi_ost_idx]->ltd_exp,
                                 lsm, loi, page, lap->lap_sub_offset,
                                 &lov_async_page_ops, lap,
                                 &lap->lap_sub_cookie, nocache, lockh);
        if (lov_lockh)
                lov_llh_put(lov_lockh);
        if (rc)
                RETURN(rc);
        CDEBUG(D_CACHE, "lap %p page %p cookie %p off "LPU64"\n", lap, page,
               lap->lap_sub_cookie, offset);
        RETURN(0);
}

static int lov_queue_async_io(struct obd_export *exp,
                              struct lov_stripe_md *lsm,
                              struct lov_oinfo *loi, void *cookie,
                              int cmd, obd_off off, int count,
                              obd_flag brw_flags, obd_flag async_flags)
{
        struct lov_obd *lov = &exp->exp_obd->u.lov;
        struct lov_async_page *lap;