X-Git-Url: https://git.whamcloud.com/?a=blobdiff_plain;f=lustre%2Flov%2Flov_io.c;h=e6578ed2238475ab7fd032574cc71d295d0798d0;hb=68234a15f0ca73b035f91e5bcc9399ca0bca1287;hp=bf7a9584d57f0d052b3e13aa9ba8305ffbd58796;hpb=8eaf21f9a9a6264598b9fed93b2c5d423b00bd9a;p=fs%2Flustre-release.git diff --git a/lustre/lov/lov_io.c b/lustre/lov/lov_io.c index bf7a958..e6578ed 100644 --- a/lustre/lov/lov_io.c +++ b/lustre/lov/lov_io.c @@ -1,6 +1,4 @@ -/* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*- - * vim:expandtab:shiftwidth=8:tabstop=8: - * +/* * GPL HEADER START * * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER. @@ -26,8 +24,10 @@ * GPL HEADER END */ /* - * Copyright 2008 Sun Microsystems, Inc. All rights reserved. + * Copyright (c) 2008, 2010, Oracle and/or its affiliates. All rights reserved. * Use is subject to license terms. + * + * Copyright (c) 2011, Whamcloud, Inc. */ /* * This file is part of Lustre, http://www.lustre.org/ @@ -36,6 +36,7 @@ * Implementation of cl_io for LOV layer. * * Author: Nikita Danilov + * Author: Jinshan Xiong */ #define DEBUG_SUBSYSTEM S_LOV @@ -46,24 +47,13 @@ * @{ */ -static void lov_sub_enter(struct lov_io_sub *sub) +static inline void lov_sub_enter(struct lov_io_sub *sub) { - ENTRY; - if (sub->sub_reenter++ == 0) { - sub->sub_cookie = cl_env_reenter(); - cl_env_implant(sub->sub_env, &sub->sub_refcheck2); - } - EXIT; + sub->sub_reenter++; } - -static void lov_sub_exit(struct lov_io_sub *sub) +static inline void lov_sub_exit(struct lov_io_sub *sub) { - ENTRY; - if (--sub->sub_reenter == 0) { - cl_env_unplant(sub->sub_env, &sub->sub_refcheck2); - cl_env_reexit(sub->sub_cookie); - } - EXIT; + sub->sub_reenter--; } static void lov_io_sub_fini(const struct lu_env *env, struct lov_io *lio, @@ -95,16 +85,20 @@ static void lov_io_sub_fini(const struct lu_env *env, struct lov_io *lio, static void lov_io_sub_inherit(struct cl_io *io, struct lov_io *lio, int stripe, loff_t start, loff_t end) { - struct lov_stripe_md *lsm = lov_r0(lio->lis_object)->lo_lsm; + struct lov_stripe_md *lsm = lio->lis_lsm; struct cl_io *parent = lio->lis_cl.cis_io; switch(io->ci_type) { - case CIT_TRUNC: { - loff_t new_size = parent->u.ci_truncate.tr_size; - - new_size = lov_size_to_stripe(lsm, new_size, stripe); - io->u.ci_truncate.tr_capa = parent->u.ci_truncate.tr_capa; - io->u.ci_truncate.tr_size = new_size; + case CIT_SETATTR: { + io->u.ci_setattr.sa_attr = parent->u.ci_setattr.sa_attr; + io->u.ci_setattr.sa_valid = parent->u.ci_setattr.sa_valid; + io->u.ci_setattr.sa_capa = parent->u.ci_setattr.sa_capa; + if (cl_io_is_trunc(io)) { + loff_t new_size = parent->u.ci_setattr.sa_attr.lvb_size; + + new_size = lov_size_to_stripe(lsm, new_size, stripe); + io->u.ci_setattr.sa_attr.lvb_size = new_size; + } break; } case CIT_FAULT: { @@ -116,8 +110,17 @@ static void lov_io_sub_inherit(struct cl_io *io, struct lov_io *lio, io->u.ci_fault.ft_index = cl_index(obj, off); break; } - case CIT_READ: - case CIT_WRITE: { + case CIT_FSYNC: { + io->u.ci_fsync.fi_start = start; + io->u.ci_fsync.fi_end = end; + io->u.ci_fsync.fi_capa = parent->u.ci_fsync.fi_capa; + io->u.ci_fsync.fi_fid = parent->u.ci_fsync.fi_fid; + io->u.ci_fsync.fi_mode = parent->u.ci_fsync.fi_mode; + break; + } + case CIT_READ: + case CIT_WRITE: { + io->u.ci_wr.wr_sync = cl_io_is_sync_write(parent); if (cl_io_is_append(parent)) { io->u.ci_wr.wr_append = 1; } else { @@ -257,7 +260,7 @@ static int lov_page_stripe(const struct cl_page *page) struct lov_io_sub *lov_page_subio(const struct lu_env *env, struct lov_io *lio, const struct cl_page_slice *slice) { - struct lov_stripe_md *lsm = lov_r0(lio->lis_object)->lo_lsm; + struct lov_stripe_md *lsm = lio->lis_lsm; struct cl_page *page = slice->cpl_page; int stripe; @@ -275,8 +278,7 @@ struct lov_io_sub *lov_page_subio(const struct lu_env *env, struct lov_io *lio, static int lov_io_subio_init(const struct lu_env *env, struct lov_io *lio, struct cl_io *io) { - struct lov_object *lov = lio->lis_object; - struct lov_stripe_md *lsm = lov_r0(lov)->lo_lsm; + struct lov_stripe_md *lsm = lio->lis_lsm; int result; LASSERT(lio->lis_object != NULL); @@ -286,8 +288,8 @@ static int lov_io_subio_init(const struct lu_env *env, struct lov_io *lio, * Need to be optimized, we can't afford to allocate a piece of memory * when writing a page. -jay */ - OBD_ALLOC(lio->lis_subs, - lsm->lsm_stripe_count * sizeof lio->lis_subs[0]); + OBD_ALLOC_LARGE(lio->lis_subs, + lsm->lsm_stripe_count * sizeof lio->lis_subs[0]); if (lio->lis_subs != NULL) { lio->lis_nr_subios = lio->lis_stripe_count; lio->lis_single_subio_index = -1; @@ -301,14 +303,13 @@ static int lov_io_subio_init(const struct lu_env *env, struct lov_io *lio, static void lov_io_slice_init(struct lov_io *lio, struct lov_object *obj, struct cl_io *io) { - struct lov_stripe_md *lsm = lov_r0(obj)->lo_lsm; + ENTRY; - LASSERT(lsm != NULL); - ENTRY; + io->ci_result = 0; + lio->lis_object = obj; - io->ci_result = 0; - lio->lis_object = obj; - lio->lis_stripe_count = lsm->lsm_stripe_count; + LASSERT(lio->lis_lsm != NULL); + lio->lis_stripe_count = lio->lis_lsm->lsm_stripe_count; switch (io->ci_type) { case CIT_READ: @@ -323,8 +324,11 @@ static void lov_io_slice_init(struct lov_io *lio, } break; - case CIT_TRUNC: - lio->lis_pos = io->u.ci_truncate.tr_size; + case CIT_SETATTR: + if (cl_io_is_trunc(io)) + lio->lis_pos = io->u.ci_setattr.sa_attr.lvb_size; + else + lio->lis_pos = 0; lio->lis_endpos = OBD_OBJECT_EOF; break; @@ -335,6 +339,12 @@ static void lov_io_slice_init(struct lov_io *lio, break; } + case CIT_FSYNC: { + lio->lis_pos = io->u.ci_fsync.fi_start; + lio->lis_endpos = io->u.ci_fsync.fi_end; + break; + } + case CIT_MISC: lio->lis_pos = 0; lio->lis_endpos = OBD_OBJECT_EOF; @@ -356,11 +366,13 @@ static void lov_io_fini(const struct lu_env *env, const struct cl_io_slice *ios) if (lio->lis_subs != NULL) { for (i = 0; i < lio->lis_nr_subios; i++) lov_io_sub_fini(env, lio, &lio->lis_subs[i]); - OBD_FREE(lio->lis_subs, + OBD_FREE_LARGE(lio->lis_subs, lio->lis_nr_subios * sizeof lio->lis_subs[0]); lio->lis_nr_subios = 0; } - EXIT; + lov_lsm_decref(lio->lis_object, lio->lis_lsm); + lio->lis_lsm = NULL; + EXIT; } static obd_off lov_offset_mod(obd_off val, int delta) @@ -373,8 +385,8 @@ static obd_off lov_offset_mod(obd_off val, int delta) static int lov_io_iter_init(const struct lu_env *env, const struct cl_io_slice *ios) { - struct lov_io *lio = cl2lov_io(env, ios); - struct lov_stripe_md *lsm = lov_r0(lio->lis_object)->lo_lsm; + struct lov_io *lio = cl2lov_io(env, ios); + struct lov_stripe_md *lsm = lio->lis_lsm; struct lov_io_sub *sub; obd_off endpos; obd_off start; @@ -396,7 +408,7 @@ static int lov_io_iter_init(const struct lu_env *env, start, end); rc = cl_io_iter_init(sub->sub_env, sub->sub_io); lov_sub_put(sub); - CDEBUG(D_VFSTRACE, "shrink: %i ["LPU64", "LPU64")\n", + CDEBUG(D_VFSTRACE, "shrink: %d ["LPU64", "LPU64")\n", stripe, start, end); } else rc = PTR_ERR(sub); @@ -412,9 +424,9 @@ static int lov_io_iter_init(const struct lu_env *env, static int lov_io_rw_iter_init(const struct lu_env *env, const struct cl_io_slice *ios) { - struct lov_io *lio = cl2lov_io(env, ios); - struct cl_io *io = ios->cis_io; - struct lov_stripe_md *lsm = lov_r0(cl2lov(ios->cis_obj))->lo_lsm; + struct lov_io *lio = cl2lov_io(env, ios); + struct cl_io *io = ios->cis_io; + struct lov_stripe_md *lsm = lio->lis_lsm; loff_t start = io->u.ci_rw.crw_pos; loff_t next; unsigned long ssize = lsm->lsm_stripe_size; @@ -425,30 +437,31 @@ static int lov_io_rw_iter_init(const struct lu_env *env, /* fast path for common case. */ if (lio->lis_nr_subios != 1 && !cl_io_is_append(io)) { - do_div(start, ssize); - next = (start + 1) * ssize; - if (next <= start * ssize) - next = ~0ull; + lov_do_div64(start, ssize); + next = (start + 1) * ssize; + if (next <= start * ssize) + next = ~0ull; io->ci_continue = next < lio->lis_io_endpos; io->u.ci_rw.crw_count = min_t(loff_t, lio->lis_io_endpos, next) - io->u.ci_rw.crw_pos; lio->lis_pos = io->u.ci_rw.crw_pos; lio->lis_endpos = io->u.ci_rw.crw_pos + io->u.ci_rw.crw_count; - CDEBUG(D_VFSTRACE, "stripe: "LPU64" chunk: ["LPU64", "LPU64") "LPU64"\n", - (__u64)start, lio->lis_pos, lio->lis_endpos, - (__u64)lio->lis_io_endpos); - } - /* - * XXX The following call should be optimized: we know, that - * [lio->lis_pos, lio->lis_endpos) intersects with exactly one stripe. - */ - RETURN(lov_io_iter_init(env, ios)); + CDEBUG(D_VFSTRACE, "stripe: "LPU64" chunk: ["LPU64", "LPU64") " + LPU64"\n", (__u64)start, lio->lis_pos, lio->lis_endpos, + (__u64)lio->lis_io_endpos); + } + /* + * XXX The following call should be optimized: we know, that + * [lio->lis_pos, lio->lis_endpos) intersects with exactly one stripe. + */ + RETURN(lov_io_iter_init(env, ios)); } static int lov_io_call(const struct lu_env *env, struct lov_io *lio, int (*iofunc)(const struct lu_env *, struct cl_io *)) { + struct cl_io *parent = lio->lis_cl.cis_io; struct lov_io_sub *sub; int rc = 0; @@ -459,8 +472,11 @@ static int lov_io_call(const struct lu_env *env, struct lov_io *lio, lov_sub_exit(sub); if (rc) break; - } - RETURN(rc); + + if (parent->ci_result == 0) + parent->ci_result = sub->sub_io->ci_result; + } + RETURN(rc); } static int lov_io_lock(const struct lu_env *env, const struct cl_io_slice *ios) @@ -560,8 +576,7 @@ static struct cl_page_list *lov_io_submit_qin(struct lov_device *ld, */ static int lov_io_submit(const struct lu_env *env, const struct cl_io_slice *ios, - enum cl_req_type crt, struct cl_2queue *queue, - enum cl_req_priority priority) + enum cl_req_type crt, struct cl_2queue *queue) { struct lov_io *lio = cl2lov_io(env, ios); struct lov_object *obj = lio->lis_object; @@ -592,15 +607,15 @@ static int lov_io_submit(const struct lu_env *env, LASSERT(!IS_ERR(sub)); LASSERT(sub->sub_io == &lio->lis_single_subio); rc = cl_io_submit_rw(sub->sub_env, sub->sub_io, - crt, queue, priority); + crt, queue); lov_sub_put(sub); RETURN(rc); } LASSERT(lio->lis_subs != NULL); if (alloc) { - OBD_ALLOC(stripes_qin, - sizeof(*stripes_qin) * lio->lis_nr_subios); + OBD_ALLOC_LARGE(stripes_qin, + sizeof(*stripes_qin) * lio->lis_nr_subios); if (stripes_qin == NULL) RETURN(-ENOMEM); @@ -633,7 +648,7 @@ static int lov_io_submit(const struct lu_env *env, sub = lov_sub_get(env, lio, stripe); if (!IS_ERR(sub)) { rc = cl_io_submit_rw(sub->sub_env, sub->sub_io, - crt, cl2q, priority); + crt, cl2q); lov_sub_put(sub); } else rc = PTR_ERR(sub); @@ -653,7 +668,7 @@ static int lov_io_submit(const struct lu_env *env, } if (alloc) { - OBD_FREE(stripes_qin, + OBD_FREE_LARGE(stripes_qin, sizeof(*stripes_qin) * lio->lis_nr_subios); } else { int i; @@ -730,6 +745,28 @@ static int lov_io_fault_start(const struct lu_env *env, RETURN(lov_io_start(env, ios)); } +static void lov_io_fsync_end(const struct lu_env *env, + const struct cl_io_slice *ios) +{ + struct lov_io *lio = cl2lov_io(env, ios); + struct lov_io_sub *sub; + unsigned int *written = &ios->cis_io->u.ci_fsync.fi_nr_written; + ENTRY; + + *written = 0; + cfs_list_for_each_entry(sub, &lio->lis_active, sub_linkage) { + struct cl_io *subio = sub->sub_io; + + lov_sub_enter(sub); + lov_io_end_wrapper(sub->sub_env, subio); + lov_sub_exit(sub); + + if (subio->ci_result == 0) + *written += subio->u.ci_fsync.fi_nr_written; + } + RETURN_EXIT; +} + static const struct cl_io_operations lov_io_ops = { .op = { [CIT_READ] = { @@ -750,7 +787,7 @@ static const struct cl_io_operations lov_io_ops = { .cio_start = lov_io_start, .cio_end = lov_io_end }, - [CIT_TRUNC] = { + [CIT_SETATTR] = { .cio_fini = lov_io_fini, .cio_iter_init = lov_io_iter_init, .cio_iter_fini = lov_io_iter_fini, @@ -768,6 +805,15 @@ static const struct cl_io_operations lov_io_ops = { .cio_start = lov_io_fault_start, .cio_end = lov_io_end }, + [CIT_FSYNC] = { + .cio_fini = lov_io_fini, + .cio_iter_init = lov_io_iter_init, + .cio_iter_fini = lov_io_iter_fini, + .cio_lock = lov_io_lock, + .cio_unlock = lov_io_unlock, + .cio_start = lov_io_start, + .cio_end = lov_io_fsync_end + }, [CIT_MISC] = { .cio_fini = lov_io_fini } @@ -826,7 +872,7 @@ static const struct cl_io_operations lov_empty_io_ops = { .cio_start = LOV_EMPTY_IMPOSSIBLE, .cio_end = LOV_EMPTY_IMPOSSIBLE }, - [CIT_TRUNC] = { + [CIT_SETATTR] = { .cio_fini = lov_empty_io_fini, .cio_iter_init = LOV_EMPTY_IMPOSSIBLE, .cio_lock = LOV_EMPTY_IMPOSSIBLE, @@ -840,6 +886,9 @@ static const struct cl_io_operations lov_empty_io_ops = { .cio_start = LOV_EMPTY_IMPOSSIBLE, .cio_end = LOV_EMPTY_IMPOSSIBLE }, + [CIT_FSYNC] = { + .cio_fini = lov_empty_io_fini + }, [CIT_MISC] = { .cio_fini = lov_empty_io_fini } @@ -865,7 +914,6 @@ int lov_io_init_raid0(const struct lu_env *env, struct cl_object *obj, CFS_INIT_LIST_HEAD(&lio->lis_active); lov_io_slice_init(lio, lov, io); if (io->ci_result == 0) { - LASSERT(lov_r0(lov)->lo_lsm != NULL); io->ci_result = lov_io_subio_init(env, lio, io); if (io->ci_result == 0) cl_io_slice_add(io, &lio->lis_cl, obj, &lov_io_ops); @@ -876,19 +924,21 @@ int lov_io_init_raid0(const struct lu_env *env, struct cl_object *obj, int lov_io_init_empty(const struct lu_env *env, struct cl_object *obj, struct cl_io *io) { - struct lov_io *lio = lov_env_io(env); - int result; + struct lov_io *lio = lov_env_io(env); + int result; + ENTRY; - ENTRY; - switch (io->ci_type) { - default: - LBUG(); + lio->lis_lsm = NULL; + switch (io->ci_type) { + default: + LBUG(); + case CIT_FSYNC: case CIT_MISC: case CIT_READ: result = 0; break; case CIT_WRITE: - case CIT_TRUNC: + case CIT_SETATTR: result = -EBADF; break; case CIT_FAULT: