-/* -*- 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.
* 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, 2013, Intel Corporation.
*/
/*
* This file is part of Lustre, http://www.lustre.org/
* Implementation of cl_io for LOV layer.
*
* Author: Nikita Danilov <nikita.danilov@sun.com>
+ * Author: Jinshan Xiong <jinshan.xiong@whamcloud.com>
*/
#define DEBUG_SUBSYSTEM S_LOV
* @{
*/
-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,
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_TRUNC: {
- size_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: {
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 {
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;
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);
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 = lio->lis_object->lo_lsm;
+ int result;
LASSERT(lio->lis_object != NULL);
ENTRY;
* 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;
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:
}
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;
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;
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))
+ cfs_waitq_broadcast(&lov->lo_waitq);
+ EXIT;
}
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;
start, end);
rc = cl_io_iter_init(sub->sub_env, sub->sub_io);
lov_sub_put(sub);
- CDEBUG(D_VFSTRACE, "shrink: %i [%llu, %llu)\n",
+ CDEBUG(D_VFSTRACE, "shrink: %d ["LPU64", "LPU64")\n",
stripe, start, end);
} else
rc = PTR_ERR(sub);
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;
/* 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: %llu chunk: [%llu, %llu) %llu\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;
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)
*/
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;
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);
* 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);
+ mutex_lock(&ld->ld_mutex);
lio->lis_mem_frozen = 1;
}
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);
}
if (alloc) {
- OBD_FREE(stripes_qin,
+ OBD_FREE_LARGE(stripes_qin,
sizeof(*stripes_qin) * lio->lis_nr_subios);
} else {
int i;
lov_io_sub_fini(env, lio, &lio->lis_subs[i]);
}
lio->lis_mem_frozen = 0;
- cfs_mutex_unlock(&ld->ld_mutex);
+ mutex_unlock(&ld->ld_mutex);
}
RETURN(rc);
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] = {
.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,
.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
}
static void lov_empty_io_fini(const struct lu_env *env,
const struct cl_io_slice *ios)
{
+ struct lov_object *lov = cl2lov(ios->cis_obj);
ENTRY;
+
+ if (cfs_atomic_dec_and_test(&lov->lo_active_ios))
+ cfs_waitq_broadcast(&lov->lo_waitq);
EXIT;
}
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,
.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,
.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
}
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);
}
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_TRUNC:
result = -EBADF;
break;
case CIT_FAULT:
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:
+ case CIT_READ:
+ case CIT_WRITE:
+ case CIT_FAULT:
+ /* TODO: need to restore the file. */
+ result = -EBADF;
+ 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 */
git://git.whamcloud.com / fs / lustre-release.git / blobdiff