X-Git-Url: https://git.whamcloud.com/?a=blobdiff_plain;f=lustre%2Flov%2Flov_io.c;h=9d62e2c65fac4545f087cae5d5feefe8c1654cff;hb=79e65c81349f539aa2960a1889144ce71d91351e;hp=1f72ad6b5402f44c49023ee83f21f821eae135aa;hpb=48d29ff71de18d8d375b072a7287ba9ecdb6cdce;p=fs%2Flustre-release.git diff --git a/lustre/lov/lov_io.c b/lustre/lov/lov_io.c index 1f72ad6..9d62e2c 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. @@ -28,6 +26,8 @@ /* * Copyright (c) 2008, 2010, Oracle and/or its affiliates. All rights reserved. * Use is subject to license terms. + * + * Copyright (c) 2011, 2013, Intel Corporation. */ /* * 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 @@ -84,8 +85,8 @@ 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 cl_io *parent = lio->lis_cl.cis_io; + struct lov_stripe_md *lsm = lio->lis_object->lo_lsm; + struct cl_io *parent = lio->lis_cl.cis_io; switch(io->ci_type) { case CIT_SETATTR: { @@ -109,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 { @@ -146,7 +156,7 @@ static int lov_io_sub_init(const struct lu_env *env, struct lov_io *lio, sub->sub_borrowed = 0; if (lio->lis_mem_frozen) { - LASSERT(cfs_mutex_is_locked(&ld->ld_mutex)); + LASSERT(mutex_is_locked(&ld->ld_mutex)); sub->sub_io = &ld->ld_emrg[stripe]->emrg_subio; sub->sub_env = ld->ld_emrg[stripe]->emrg_env; sub->sub_borrowed = 1; @@ -187,6 +197,7 @@ static int lov_io_sub_init(const struct lu_env *env, struct lov_io *lio, sub_io->ci_lockreq = io->ci_lockreq; sub_io->ci_type = io->ci_type; sub_io->ci_no_srvlock = io->ci_no_srvlock; + sub_io->ci_noatime = io->ci_noatime; lov_sub_enter(sub); result = cl_io_sub_init(sub->sub_env, sub_io, @@ -235,24 +246,26 @@ void lov_sub_put(struct lov_io_sub *sub) * */ -static int lov_page_stripe(const struct cl_page *page) +int lov_page_stripe(const struct cl_page *page) { - struct lovsub_object *subobj; + struct lovsub_object *subobj; + const struct cl_page_slice *slice; + ENTRY; - ENTRY; - subobj = lu2lovsub( - lu_object_locate(page->cp_child->cp_obj->co_lu.lo_header, - &lovsub_device_type)); - LASSERT(subobj != NULL); - RETURN(subobj->lso_index); + slice = cl_page_at(page, &lovsub_device_type); + LASSERT(slice != NULL); + LASSERT(slice->cpl_obj != NULL); + + subobj = cl2lovsub(slice->cpl_obj); + RETURN(subobj->lso_index); } 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 cl_page *page = slice->cpl_page; - int stripe; + struct lov_stripe_md *lsm = lio->lis_object->lo_lsm; + struct cl_page *page = slice->cpl_page; + int stripe; LASSERT(lio->lis_cl.cis_io != NULL); LASSERT(cl2lov(slice->cpl_obj) == lio->lis_object); @@ -268,19 +281,19 @@ 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; - int result; + struct lov_stripe_md *lsm; + int result; + ENTRY; - LASSERT(lio->lis_object != NULL); - ENTRY; + LASSERT(lio->lis_object != NULL); + lsm = lio->lis_object->lo_lsm; /* * 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; @@ -294,14 +307,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(obj->lo_lsm != NULL); + lio->lis_stripe_count = obj->lo_lsm->lsm_stripe_count; switch (io->ci_type) { case CIT_READ: @@ -331,6 +343,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; @@ -345,18 +363,23 @@ static void lov_io_slice_init(struct lov_io *lio, static void lov_io_fini(const struct lu_env *env, const struct cl_io_slice *ios) { - struct lov_io *lio = cl2lov_io(env, ios); - int i; + struct lov_io *lio = cl2lov_io(env, ios); + struct lov_object *lov = cl2lov(ios->cis_obj); + int i; ENTRY; 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; + + LASSERT(cfs_atomic_read(&lov->lo_active_ios) > 0); + if (cfs_atomic_dec_and_test(&lov->lo_active_ios)) + wake_up_all(&lov->lo_waitq); + EXIT; } static obd_off lov_offset_mod(obd_off val, int delta) @@ -369,8 +392,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_object->lo_lsm; struct lov_io_sub *sub; obd_off endpos; obd_off start; @@ -408,9 +431,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_object->lo_lsm; loff_t start = io->u.ci_rw.crw_pos; loff_t next; unsigned long ssize = lsm->lsm_stripe_size; @@ -421,30 +444,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; @@ -455,8 +479,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) @@ -531,14 +558,6 @@ static void lov_io_unlock(const struct lu_env *env, EXIT; } - -static struct cl_page_list *lov_io_submit_qin(struct lov_device *ld, - struct cl_page_list *qin, - int idx, int alloc) -{ - return alloc ? &qin[idx] : &ld->ld_emrg[idx]->emrg_page_list; -} - /** * lov implementation of cl_operations::cio_submit() method. It takes a list * of pages in \a queue, splits it into per-stripe sub-lists, invokes @@ -555,159 +574,147 @@ static struct cl_page_list *lov_io_submit_qin(struct lov_device *ld, * lov_device::ld_mutex mutex. */ 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) -{ - struct lov_io *lio = cl2lov_io(env, ios); - struct lov_object *obj = lio->lis_object; - struct lov_device *ld = lu2lov_dev(lov2cl(obj)->co_lu.lo_dev); - struct cl_page_list *qin = &queue->c2_qin; - struct cl_2queue *cl2q = &lov_env_info(env)->lti_cl2q; - struct cl_page_list *stripes_qin = NULL; - struct cl_page *page; - struct cl_page *tmp; - int stripe; - -#define QIN(stripe) lov_io_submit_qin(ld, stripes_qin, stripe, alloc) + const struct cl_io_slice *ios, + enum cl_req_type crt, struct cl_2queue *queue) +{ + struct cl_page_list *qin = &queue->c2_qin; + struct lov_io *lio = cl2lov_io(env, ios); + struct lov_io_sub *sub; + struct cl_page_list *plist = &lov_env_info(env)->lti_plist; + struct cl_page *page; + int stripe; + int rc = 0; + ENTRY; - int rc = 0; - int alloc = -#if defined(__KERNEL__) && defined(__linux__) - !(current->flags & PF_MEMALLOC); -#else - 1; -#endif - ENTRY; if (lio->lis_active_subios == 1) { int idx = lio->lis_single_subio_index; - struct lov_io_sub *sub; LASSERT(idx < lio->lis_nr_subios); sub = lov_sub_get(env, lio, idx); 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); - if (stripes_qin == NULL) - RETURN(-ENOMEM); - - for (stripe = 0; stripe < lio->lis_nr_subios; stripe++) - cl_page_list_init(&stripes_qin[stripe]); - } else { - /* - * If we get here, it means pageout & swap doesn't help. - * In order to not make things worse, even don't try to - * allocate the memory with __GFP_NOWARN. -jay - */ - cfs_mutex_lock(&ld->ld_mutex); - lio->lis_mem_frozen = 1; - } - - cl_2queue_init(cl2q); - cl_page_list_for_each_safe(page, tmp, qin) { - stripe = lov_page_stripe(page); - cl_page_list_move(QIN(stripe), qin, page); - } - - for (stripe = 0; stripe < lio->lis_nr_subios; stripe++) { - struct lov_io_sub *sub; - struct cl_page_list *sub_qin = QIN(stripe); - - if (cfs_list_empty(&sub_qin->pl_pages)) - continue; - - cl_page_list_splice(sub_qin, &cl2q->c2_qin); - 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); - lov_sub_put(sub); - } else - rc = PTR_ERR(sub); - cl_page_list_splice(&cl2q->c2_qin, &queue->c2_qin); - cl_page_list_splice(&cl2q->c2_qout, &queue->c2_qout); - if (rc != 0) - break; - } - - for (stripe = 0; stripe < lio->lis_nr_subios; stripe++) { - struct cl_page_list *sub_qin = QIN(stripe); - - if (cfs_list_empty(&sub_qin->pl_pages)) - continue; - - cl_page_list_splice(sub_qin, qin); - } - if (alloc) { - OBD_FREE(stripes_qin, - sizeof(*stripes_qin) * lio->lis_nr_subios); - } else { - int i; + cl_page_list_init(plist); + while (qin->pl_nr > 0) { + struct cl_2queue *cl2q = &lov_env_info(env)->lti_cl2q; - for (i = 0; i < lio->lis_nr_subios; i++) { - struct cl_io *cio = lio->lis_subs[i].sub_io; + cl_2queue_init(cl2q); - if (cio && cio == &ld->ld_emrg[i]->emrg_subio) - lov_io_sub_fini(env, lio, &lio->lis_subs[i]); - } - lio->lis_mem_frozen = 0; - cfs_mutex_unlock(&ld->ld_mutex); - } + page = cl_page_list_first(qin); + cl_page_list_move(&cl2q->c2_qin, qin, page); - RETURN(rc); -#undef QIN -} + stripe = lov_page_stripe(page); + while (qin->pl_nr > 0) { + page = cl_page_list_first(qin); + if (stripe != lov_page_stripe(page)) + break; -static int lov_io_prepare_write(const struct lu_env *env, - const struct cl_io_slice *ios, - const struct cl_page_slice *slice, - unsigned from, unsigned to) -{ - struct lov_io *lio = cl2lov_io(env, ios); - struct cl_page *sub_page = lov_sub_page(slice); - struct lov_io_sub *sub; - int result; + cl_page_list_move(&cl2q->c2_qin, qin, page); + } - ENTRY; - sub = lov_page_subio(env, lio, slice); - if (!IS_ERR(sub)) { - result = cl_io_prepare_write(sub->sub_env, sub->sub_io, - sub_page, from, to); - lov_sub_put(sub); - } else - result = PTR_ERR(sub); - RETURN(result); -} - -static int lov_io_commit_write(const struct lu_env *env, - const struct cl_io_slice *ios, - const struct cl_page_slice *slice, - unsigned from, unsigned to) -{ - struct lov_io *lio = cl2lov_io(env, ios); - struct cl_page *sub_page = lov_sub_page(slice); - struct lov_io_sub *sub; - int result; - - ENTRY; - sub = lov_page_subio(env, lio, slice); - if (!IS_ERR(sub)) { - result = cl_io_commit_write(sub->sub_env, sub->sub_io, - sub_page, from, to); - lov_sub_put(sub); - } else - result = PTR_ERR(sub); - RETURN(result); + sub = lov_sub_get(env, lio, stripe); + if (!IS_ERR(sub)) { + rc = cl_io_submit_rw(sub->sub_env, sub->sub_io, + crt, cl2q); + lov_sub_put(sub); + } else { + rc = PTR_ERR(sub); + } + + cl_page_list_splice(&cl2q->c2_qin, plist); + cl_page_list_splice(&cl2q->c2_qout, &queue->c2_qout); + cl_2queue_fini(env, cl2q); + + if (rc != 0) + break; + } + + cl_page_list_splice(plist, qin); + cl_page_list_fini(env, plist); + + RETURN(rc); +} + +static int lov_io_commit_async(const struct lu_env *env, + const struct cl_io_slice *ios, + struct cl_page_list *queue, int from, int to, + cl_commit_cbt cb) +{ + struct cl_page_list *plist = &lov_env_info(env)->lti_plist; + struct lov_io *lio = cl2lov_io(env, ios); + struct lov_io_sub *sub; + struct cl_page *page; + int rc = 0; + ENTRY; + + if (lio->lis_active_subios == 1) { + int idx = lio->lis_single_subio_index; + + LASSERT(idx < lio->lis_nr_subios); + sub = lov_sub_get(env, lio, idx); + LASSERT(!IS_ERR(sub)); + LASSERT(sub->sub_io == &lio->lis_single_subio); + rc = cl_io_commit_async(sub->sub_env, sub->sub_io, queue, + from, to, cb); + lov_sub_put(sub); + RETURN(rc); + } + + LASSERT(lio->lis_subs != NULL); + + cl_page_list_init(plist); + while (queue->pl_nr > 0) { + int stripe_to = to; + int stripe; + + LASSERT(plist->pl_nr == 0); + page = cl_page_list_first(queue); + cl_page_list_move(plist, queue, page); + + stripe = lov_page_stripe(page); + while (queue->pl_nr > 0) { + page = cl_page_list_first(queue); + if (stripe != lov_page_stripe(page)) + break; + + cl_page_list_move(plist, queue, page); + } + + if (queue->pl_nr > 0) /* still has more pages */ + stripe_to = PAGE_SIZE; + + sub = lov_sub_get(env, lio, stripe); + if (!IS_ERR(sub)) { + rc = cl_io_commit_async(sub->sub_env, sub->sub_io, + plist, from, stripe_to, cb); + lov_sub_put(sub); + } else { + rc = PTR_ERR(sub); + break; + } + + if (plist->pl_nr > 0) /* short write */ + break; + + from = 0; + } + + /* for error case, add the page back into the qin list */ + LASSERT(ergo(rc == 0, plist->pl_nr == 0)); + while (plist->pl_nr > 0) { + /* error occurred, add the uncommitted pages back into queue */ + page = cl_page_list_last(plist); + cl_page_list_move_head(queue, plist, page); + } + + RETURN(rc); } static int lov_io_fault_start(const struct lu_env *env, @@ -726,6 +733,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] = { @@ -764,20 +793,21 @@ static const struct cl_io_operations lov_io_ops = { .cio_start = lov_io_fault_start, .cio_end = lov_io_end }, - [CIT_MISC] = { - .cio_fini = lov_io_fini - } - }, - .req_op = { - [CRT_READ] = { - .cio_submit = lov_io_submit - }, - [CRT_WRITE] = { - .cio_submit = lov_io_submit - } - }, - .cio_prepare_write = lov_io_prepare_write, - .cio_commit_write = lov_io_commit_write + [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 + } + }, + .cio_submit = lov_io_submit, + .cio_commit_async = lov_io_commit_async, }; /***************************************************************************** @@ -789,8 +819,12 @@ static const struct cl_io_operations lov_io_ops = { static void lov_empty_io_fini(const struct lu_env *env, const struct cl_io_slice *ios) { - ENTRY; - EXIT; + struct lov_object *lov = cl2lov(ios->cis_obj); + ENTRY; + + if (cfs_atomic_dec_and_test(&lov->lo_active_ios)) + wake_up_all(&lov->lo_waitq); + EXIT; } static void lov_empty_impossible(const struct lu_env *env, @@ -807,8 +841,8 @@ static void lov_empty_impossible(const struct lu_env *env, static const struct cl_io_operations lov_empty_io_ops = { .op = { [CIT_READ] = { -#if 0 .cio_fini = lov_empty_io_fini, +#if 0 .cio_iter_init = LOV_EMPTY_IMPOSSIBLE, .cio_lock = LOV_EMPTY_IMPOSSIBLE, .cio_start = LOV_EMPTY_IMPOSSIBLE, @@ -836,19 +870,15 @@ static const struct cl_io_operations lov_empty_io_ops = { .cio_start = LOV_EMPTY_IMPOSSIBLE, .cio_end = LOV_EMPTY_IMPOSSIBLE }, - [CIT_MISC] = { - .cio_fini = lov_empty_io_fini - } - }, - .req_op = { - [CRT_READ] = { - .cio_submit = LOV_EMPTY_IMPOSSIBLE - }, - [CRT_WRITE] = { - .cio_submit = LOV_EMPTY_IMPOSSIBLE - } - }, - .cio_commit_write = LOV_EMPTY_IMPOSSIBLE + [CIT_FSYNC] = { + .cio_fini = lov_empty_io_fini + }, + [CIT_MISC] = { + .cio_fini = lov_empty_io_fini + } + }, + .cio_submit = LOV_EMPTY_IMPOSSIBLE, + .cio_commit_async = LOV_EMPTY_IMPOSSIBLE }; int lov_io_init_raid0(const struct lu_env *env, struct cl_object *obj, @@ -861,10 +891,11 @@ 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) + if (io->ci_result == 0) { cl_io_slice_add(io, &lio->lis_cl, obj, &lov_io_ops); + cfs_atomic_inc(&lov->lo_active_ios); + } } RETURN(io->ci_result); } @@ -872,19 +903,24 @@ 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; - - ENTRY; - switch (io->ci_type) { - default: - LBUG(); - case CIT_MISC: - case CIT_READ: - result = 0; - break; + struct lov_object *lov = cl2lov(obj); + struct lov_io *lio = lov_env_io(env); + int result; + ENTRY; + + lio->lis_object = lov; + switch (io->ci_type) { + default: + LBUG(); + case CIT_MISC: + case CIT_READ: + result = 0; + break; + case CIT_FSYNC: + case CIT_SETATTR: + result = +1; + break; case CIT_WRITE: - case CIT_SETATTR: result = -EBADF; break; case CIT_FAULT: @@ -893,10 +929,56 @@ int lov_io_init_empty(const struct lu_env *env, struct cl_object *obj, PFID(lu_object_fid(&obj->co_lu))); break; } - if (result == 0) + if (result == 0) { cl_io_slice_add(io, &lio->lis_cl, obj, &lov_empty_io_ops); - io->ci_result = result; - RETURN(result != 0); + cfs_atomic_inc(&lov->lo_active_ios); + } + + io->ci_result = result < 0 ? result : 0; + RETURN(result != 0); +} + +int lov_io_init_released(const struct lu_env *env, struct cl_object *obj, + struct cl_io *io) +{ + struct lov_object *lov = cl2lov(obj); + struct lov_io *lio = lov_env_io(env); + int result; + ENTRY; + + LASSERT(lov->lo_lsm != NULL); + lio->lis_object = lov; + + switch (io->ci_type) { + default: + LASSERTF(0, "invalid type %d\n", io->ci_type); + case CIT_MISC: + case CIT_FSYNC: + result = 1; + break; + case CIT_SETATTR: + /* the truncate to 0 is managed by MDT: + * - in open, for open O_TRUNC + * - in setattr, for truncate + */ + /* the truncate is for size > 0 so triggers a restore */ + if (cl_io_is_trunc(io)) + io->ci_restore_needed = 1; + result = -ENODATA; + break; + case CIT_READ: + case CIT_WRITE: + case CIT_FAULT: + io->ci_restore_needed = 1; + result = -ENODATA; + break; + } + if (result == 0) { + cl_io_slice_add(io, &lio->lis_cl, obj, &lov_empty_io_ops); + cfs_atomic_inc(&lov->lo_active_ios); + } + + io->ci_result = result < 0 ? result : 0; + RETURN(result != 0); } - /** @} lov */