X-Git-Url: https://git.whamcloud.com/?a=blobdiff_plain;f=lustre%2Flov%2Flov_page.c;h=77cae6f4e18d873f2b40277adc610fc93513ac87;hb=94a6bc2a947359a9b03b039a2fb4bb49842666e5;hp=1af2c6221576f06e0c8622726e17137cd987f1ec;hpb=98060d83459ba10409f295898f0ec917f938b4d3;p=fs%2Flustre-release.git diff --git a/lustre/lov/lov_page.c b/lustre/lov/lov_page.c index 1af2c62..77cae6f 100644 --- a/lustre/lov/lov_page.c +++ b/lustre/lov/lov_page.c @@ -36,6 +36,7 @@ * Implementation of cl_page for LOV layer. * * Author: Nikita Danilov + * Author: Jinshan Xiong */ #define DEBUG_SUBSYSTEM S_LOV @@ -52,183 +53,137 @@ * */ -static int lov_page_invariant(const struct cl_page_slice *slice) +/** + * Adjust the stripe index by layout of raid0. @max_index is the maximum + * page index covered by an underlying DLM lock. + * This function converts max_index from stripe level to file level, and make + * sure it's not beyond one stripe. + */ +static int lov_raid0_page_is_under_lock(const struct lu_env *env, + const struct cl_page_slice *slice, + struct cl_io *unused, + pgoff_t *max_index) { - const struct cl_page *page = slice->cpl_page; - const struct cl_page *sub = lov_sub_page(slice); + struct lov_object *loo = cl2lov(slice->cpl_obj); + struct lov_layout_raid0 *r0 = lov_r0(loo); + pgoff_t index = *max_index; + unsigned int pps; /* pages per stripe */ + ENTRY; - return ergo(sub != NULL, - page->cp_child == sub && - sub->cp_parent == page && - page->cp_state == sub->cp_state); -} + CDEBUG(D_READA, DFID "*max_index = %lu, nr = %d\n", + PFID(lu_object_fid(lov2lu(loo))), index, r0->lo_nr); -static void lov_page_fini(const struct lu_env *env, - struct cl_page_slice *slice) -{ - struct cl_page *sub = lov_sub_page(slice); - - LINVRNT(lov_page_invariant(slice)); - ENTRY; - - if (sub != NULL) { - LASSERT(sub->cp_state == CPS_FREEING); - lu_ref_del(&sub->cp_reference, "lov", sub->cp_parent); - sub->cp_parent = NULL; - slice->cpl_page->cp_child = NULL; - cl_page_put(env, sub); - } - EXIT; -} + if (index == 0) /* the page is not covered by any lock */ + RETURN(0); -static int lov_page_own(const struct lu_env *env, - const struct cl_page_slice *slice, struct cl_io *io, - int nonblock) -{ - struct lov_io *lio = lov_env_io(env); - struct lov_io_sub *sub; - - LINVRNT(lov_page_invariant(slice)); - LINVRNT(!cl2lov_page(slice)->lps_invalid); - ENTRY; - - sub = lov_page_subio(env, lio, slice); - if (!IS_ERR(sub)) { - lov_sub_page(slice)->cp_owner = sub->sub_io; - lov_sub_put(sub); - } else - LBUG(); /* Arrgh */ - RETURN(0); -} + if (r0->lo_nr == 1) /* single stripe file */ + RETURN(0); -static void lov_page_assume(const struct lu_env *env, - const struct cl_page_slice *slice, struct cl_io *io) -{ - lov_page_own(env, slice, io, 0); -} + /* max_index is stripe level, convert it into file level */ + if (index != CL_PAGE_EOF) { + int stripeno = lov_page_stripe(slice->cpl_page); + *max_index = lov_stripe_pgoff(loo->lo_lsm, index, stripeno); + } -static int lov_page_cache_add(const struct lu_env *env, - const struct cl_page_slice *slice, - struct cl_io *io) -{ - struct lov_io *lio = lov_env_io(env); - struct lov_io_sub *sub; - int rc = 0; + /* calculate the end of current stripe */ + pps = loo->lo_lsm->lsm_stripe_size >> PAGE_CACHE_SHIFT; + index = slice->cpl_index + pps - slice->cpl_index % pps - 1; - LINVRNT(lov_page_invariant(slice)); - LINVRNT(!cl2lov_page(slice)->lps_invalid); - ENTRY; + CDEBUG(D_READA, DFID "*max_index = %lu, index = %lu, pps = %u, " + "stripe_size = %u, stripe no = %u, page index = %lu\n", + PFID(lu_object_fid(lov2lu(loo))), *max_index, index, pps, + loo->lo_lsm->lsm_stripe_size, lov_page_stripe(slice->cpl_page), + slice->cpl_index); - sub = lov_page_subio(env, lio, slice); - if (!IS_ERR(sub)) { - rc = cl_page_cache_add(sub->sub_env, sub->sub_io, - slice->cpl_page->cp_child, CRT_WRITE); - lov_sub_put(sub); - } else { - rc = PTR_ERR(sub); - CL_PAGE_DEBUG(D_ERROR, env, slice->cpl_page, "rc = %d\n", rc); - } - RETURN(rc); + /* never exceed the end of the stripe */ + *max_index = min_t(pgoff_t, *max_index, index); + RETURN(0); } -static int lov_page_print(const struct lu_env *env, - const struct cl_page_slice *slice, - void *cookie, lu_printer_t printer) +static int lov_raid0_page_print(const struct lu_env *env, + const struct cl_page_slice *slice, + void *cookie, lu_printer_t printer) { - struct lov_page *lp = cl2lov_page(slice); + struct lov_page *lp = cl2lov_page(slice); - return (*printer)(env, cookie, LUSTRE_LOV_NAME"-page@%p\n", lp); + return (*printer)(env, cookie, LUSTRE_LOV_NAME"-page@%p, raid0\n", lp); } -static const struct cl_page_operations lov_page_ops = { - .cpo_fini = lov_page_fini, - .cpo_own = lov_page_own, - .cpo_assume = lov_page_assume, - .io = { - [CRT_WRITE] = { - .cpo_cache_add = lov_page_cache_add - } - }, - .cpo_print = lov_page_print +static const struct cl_page_operations lov_raid0_page_ops = { + .cpo_is_under_lock = lov_raid0_page_is_under_lock, + .cpo_print = lov_raid0_page_print }; -static void lov_empty_page_fini(const struct lu_env *env, - struct cl_page_slice *slice) -{ - LASSERT(slice->cpl_page->cp_child == NULL); -} - int lov_page_init_raid0(const struct lu_env *env, struct cl_object *obj, - struct cl_page *page, cfs_page_t *vmpage) + struct cl_page *page, pgoff_t index) { - struct lov_object *loo = cl2lov(obj); - struct lov_layout_raid0 *r0 = lov_r0(loo); - struct lov_io *lio = lov_env_io(env); - struct cl_page *subpage; - struct cl_object *subobj; - struct lov_io_sub *sub; - struct lov_page *lpg = cl_object_page_slice(obj, page); - loff_t offset; - obd_off suboff; - int stripe; - int rc; - ENTRY; - - offset = cl_offset(obj, page->cp_index); + struct lov_object *loo = cl2lov(obj); + struct lov_layout_raid0 *r0 = lov_r0(loo); + struct lov_io *lio = lov_env_io(env); + struct cl_object *subobj; + struct cl_object *o; + struct lov_io_sub *sub; + struct lov_page *lpg = cl_object_page_slice(obj, page); + loff_t offset; + obd_off suboff; + int stripe; + int rc; + ENTRY; + + offset = cl_offset(obj, index); stripe = lov_stripe_number(loo->lo_lsm, offset); LASSERT(stripe < r0->lo_nr); rc = lov_stripe_offset(loo->lo_lsm, offset, stripe, - &suboff); - LASSERT(rc == 0); - - lpg->lps_invalid = 1; - cl_page_slice_add(page, &lpg->lps_cl, obj, &lov_page_ops); - - sub = lov_sub_get(env, lio, stripe); - if (IS_ERR(sub)) - GOTO(out, rc = PTR_ERR(sub)); - - subobj = lovsub2cl(r0->lo_sub[stripe]); - subpage = cl_page_find_sub(sub->sub_env, subobj, - cl_index(subobj, suboff), vmpage, page); - lov_sub_put(sub); - if (IS_ERR(subpage)) - GOTO(out, rc = PTR_ERR(subpage)); - - if (likely(subpage->cp_parent == page)) { - lu_ref_add(&subpage->cp_reference, "lov", page); - lpg->lps_invalid = 0; - rc = 0; - } else { - CL_PAGE_DEBUG(D_ERROR, env, page, "parent page\n"); - CL_PAGE_DEBUG(D_ERROR, env, subpage, "child page\n"); - LASSERT(0); - } - - EXIT; -out: - return rc; + &suboff); + LASSERT(rc == 0); + + lpg->lps_stripe = stripe; + cl_page_slice_add(page, &lpg->lps_cl, obj, index, &lov_raid0_page_ops); + + sub = lov_sub_get(env, lio, stripe); + if (IS_ERR(sub)) + RETURN(PTR_ERR(sub)); + + subobj = lovsub2cl(r0->lo_sub[stripe]); + list_for_each_entry(o, &subobj->co_lu.lo_header->loh_layers, + co_lu.lo_linkage) { + if (o->co_ops->coo_page_init != NULL) { + rc = o->co_ops->coo_page_init(sub->sub_env, o, page, + cl_index(subobj, suboff)); + if (rc != 0) + break; + } + } + lov_sub_put(sub); + RETURN(rc); } +static int lov_empty_page_print(const struct lu_env *env, + const struct cl_page_slice *slice, + void *cookie, lu_printer_t printer) +{ + struct lov_page *lp = cl2lov_page(slice); + + return (*printer)(env, cookie, LUSTRE_LOV_NAME"-page@%p, empty.\n", lp); +} static const struct cl_page_operations lov_empty_page_ops = { - .cpo_fini = lov_empty_page_fini, - .cpo_print = lov_page_print + .cpo_print = lov_empty_page_print }; int lov_page_init_empty(const struct lu_env *env, struct cl_object *obj, - struct cl_page *page, cfs_page_t *vmpage) + struct cl_page *page, pgoff_t index) { - struct lov_page *lpg = cl_object_page_slice(obj, page); + struct lov_page *lpg = cl_object_page_slice(obj, page); void *addr; - ENTRY; + ENTRY; - cl_page_slice_add(page, &lpg->lps_cl, obj, &lov_empty_page_ops); - addr = cfs_kmap(vmpage); + cl_page_slice_add(page, &lpg->lps_cl, obj, index, &lov_empty_page_ops); + addr = kmap(page->cp_vmpage); memset(addr, 0, cl_page_size(obj)); - cfs_kunmap(vmpage); + kunmap(page->cp_vmpage); cl_page_export(env, page, 1); - RETURN(0); + RETURN(0); }