unsigned long cra_rpc_pages;
/* Release callback. If readahead holds resources underneath, this
* function should be called to release it. */
- void (*cra_release)(const struct lu_env *env, void *cbdata);
+ void (*cra_release)(const struct lu_env *env,
+ struct cl_read_ahead *ra);
+
/* Callback data for cra_release routine */
- void *cra_cbdata;
+ void *cra_dlmlock;
+ void *cra_oio;
+
/* whether lock is in contention */
bool cra_contention;
};
struct cl_read_ahead *ra)
{
if (ra->cra_release != NULL)
- ra->cra_release(env, ra->cra_cbdata);
+ ra->cra_release(env, ra);
memset(ra, 0, sizeof(*ra));
}
int (*cio_read_ahead)(const struct lu_env *env,
const struct cl_io_slice *slice,
pgoff_t start, struct cl_read_ahead *ra);
+ /**
+ *
+ * Reserve LRU slots before IO.
+ */
+ int (*cio_lru_reserve) (const struct lu_env *env,
+ const struct cl_io_slice *slice,
+ loff_t pos, size_t bytes);
/**
* Optional debugging helper. Print given io slice.
*/
struct cl_page_list *queue, int from, int to,
cl_commit_cbt cb);
void cl_io_extent_release (const struct lu_env *env, struct cl_io *io);
+int cl_io_lru_reserve(const struct lu_env *env, struct cl_io *io,
+ loff_t pos, size_t bytes);
int cl_io_read_ahead (const struct lu_env *env, struct cl_io *io,
pgoff_t start, struct cl_read_ahead *ra);
void cl_io_rw_advance (const struct lu_env *env, struct cl_io *io,
/** true if this io is counted as active IO */
oi_is_active:1,
/** true if this io has CAP_SYS_RESOURCE */
- oi_cap_sys_resource:1;
+ oi_cap_sys_resource:1,
+ /** true if this io issued by readahead */
+ oi_is_readahead:1;
/** how many LRU pages are reserved for this IO */
unsigned long oi_lru_reserved;
int osc_io_iter_init(const struct lu_env *env, const struct cl_io_slice *ios);
void osc_io_iter_fini(const struct lu_env *env,
const struct cl_io_slice *ios);
-int osc_io_rw_iter_init(const struct lu_env *env,
- const struct cl_io_slice *ios);
void osc_io_rw_iter_fini(const struct lu_env *env,
const struct cl_io_slice *ios);
int osc_io_fault_start(const struct lu_env *env, const struct cl_io_slice *ios);
struct cl_fsync_io *fio);
void osc_io_fsync_end(const struct lu_env *env,
const struct cl_io_slice *slice);
-void osc_read_ahead_release(const struct lu_env *env, void *cbdata);
+void osc_read_ahead_release(const struct lu_env *env, struct cl_read_ahead *ra);
int osc_io_lseek_start(const struct lu_env *env,
const struct cl_io_slice *slice);
void osc_io_lseek_end(const struct lu_env *env,
const struct cl_io_slice *slice);
+int osc_io_lru_reserve(const struct lu_env *env, const struct cl_io_slice *ios,
+ loff_t pos, size_t count);
/* osc_lock.c */
void osc_lock_to_lockless(const struct lu_env *env, struct osc_lock *ols,
struct vvp_io *vio = vvp_env_io(env);
struct inode *inode = file_inode(file);
struct ll_inode_info *lli = ll_i2info(inode);
+ struct ll_sb_info *sbi = ll_i2sbi(inode);
struct ll_file_data *fd = file->private_data;
struct range_lock range;
struct cl_io *io;
unsigned int retried = 0, dio_lock = 0;
bool is_aio = false;
struct cl_dio_aio *ci_aio = NULL;
+ size_t per_bytes;
+ bool partial_io = false;
+ size_t max_io_pages, max_cached_pages;
ENTRY;
file_dentry(file)->d_name.name,
iot == CIT_READ ? "read" : "write", *ppos, count);
+ max_io_pages = PTLRPC_MAX_BRW_PAGES * OBD_MAX_RIF_DEFAULT;
+ max_cached_pages = sbi->ll_cache->ccc_lru_max;
+ if (max_io_pages > (max_cached_pages >> 2))
+ max_io_pages = max_cached_pages >> 2;
+
io = vvp_env_thread_io(env);
if (file->f_flags & O_DIRECT) {
if (!is_sync_kiocb(args->u.normal.via_iocb))
}
restart:
+ /**
+ * IO block size need be aware of cached page limit, otherwise
+ * if we have small max_cached_mb but large block IO issued, io
+ * could not be finished and blocked whole client.
+ */
+ if (file->f_flags & O_DIRECT)
+ per_bytes = count;
+ else
+ per_bytes = min(max_io_pages << PAGE_SHIFT, count);
+ partial_io = per_bytes < count;
io = vvp_env_thread_io(env);
ll_io_init(io, file, iot, args);
io->ci_aio = ci_aio;
io->ci_dio_lock = dio_lock;
io->ci_ndelay_tried = retried;
- if (cl_io_rw_init(env, io, iot, *ppos, count) == 0) {
+ if (cl_io_rw_init(env, io, iot, *ppos, per_bytes) == 0) {
bool range_locked = false;
if (file->f_flags & O_APPEND)
range_lock_init(&range, 0, LUSTRE_EOF);
else
- range_lock_init(&range, *ppos, *ppos + count - 1);
+ range_lock_init(&range, *ppos, *ppos + per_bytes - 1);
vio->vui_fd = file->private_data;
vio->vui_iter = args->u.normal.via_iter;
/* prepare IO restart */
if (count > 0)
args->u.normal.via_iter = vio->vui_iter;
+
+ if (partial_io) {
+ /**
+ * Reexpand iov count because it was zero
+ * after IO finish.
+ */
+ iov_iter_reexpand(vio->vui_iter, count);
+ if (per_bytes == io->ci_nob)
+ io->ci_need_restart = 1;
+ }
}
out:
cl_io_fini(env, io);
unsigned long pages,
unsigned long pages_min)
{
- struct ll_ra_info *ra = &sbi->ll_ra_info;
- long ret;
+ struct ll_ra_info *ra = &sbi->ll_ra_info;
+ long ret;
+
ENTRY;
- /* If read-ahead pages left are less than 1M, do not do read-ahead,
- * otherwise it will form small read RPC(< 1M), which hurt server
- * performance a lot. */
+ /**
+ * Don't try readahead agreesively if we are limited
+ * LRU pages, otherwise, it could cause deadlock.
+ */
+ pages = min(sbi->ll_cache->ccc_lru_max >> 2, pages);
+
+ /*
+ * If read-ahead pages left are less than 1M, do not do read-ahead,
+ * otherwise it will form small read RPC(< 1M), which hurt server
+ * performance a lot.
+ */
ret = min(ra->ra_max_pages - atomic_read(&ra->ra_cur_pages),
pages);
- if (ret < 0 || ret < min_t(long, PTLRPC_MAX_BRW_PAGES, pages))
- GOTO(out, ret = 0);
+ if (ret < 0 || ret < min_t(long, PTLRPC_MAX_BRW_PAGES, pages))
+ GOTO(out, ret = 0);
if (atomic_add_return(ret, &ra->ra_cur_pages) > ra->ra_max_pages) {
atomic_sub(ret, &ra->ra_cur_pages);
struct cl_object *clob;
int ret = 0;
__u64 kms;
+ struct ll_sb_info *sbi;
+ struct ll_ra_info *ra;
+
ENTRY;
+ ENTRY;
+
clob = io->ci_obj;
inode = vvp_object_inode(clob);
+ sbi = ll_i2sbi(inode);
+ ra = &sbi->ll_ra_info;
+
+ /**
+ * In case we have a limited max_cached_mb, readahead
+ * should be stopped if it have run out of all LRU slots.
+ */
+ if (atomic_read(&ra->ra_cur_pages) >= sbi->ll_cache->ccc_lru_max) {
+ ll_ra_stats_inc(inode, RA_STAT_MAX_IN_FLIGHT);
+ RETURN(0);
+ }
memset(ria, 0, sizeof(*ria));
ret = ll_readahead_file_kms(env, io, &kms);
pgoff_t start_idx = ras_align(ras, ras->ras_next_readahead_idx);
pgoff_t end_idx = start_idx + pages - 1;
+ /**
+ * In case we have a limited max_cached_mb, readahead
+ * should be stopped if it have run out of all LRU slots.
+ */
+ if (atomic_read(&ra->ra_cur_pages) >= sbi->ll_cache->ccc_lru_max) {
+ ll_ra_stats_inc(inode, RA_STAT_MAX_IN_FLIGHT);
+ return 0;
+ }
+
throttle = min(ra->ra_async_pages_per_file_threshold,
ra->ra_max_pages_per_file);
/*
if (!can_populate_pages(env, io, inode))
RETURN(0);
+ if (!(file->f_flags & O_DIRECT)) {
+ result = cl_io_lru_reserve(env, io, pos, cnt);
+ if (result)
+ RETURN(result);
+ }
+
/* Unless this is reading a sparse file, otherwise the lock has already
* been acquired so vvp_prep_size() is an empty op. */
result = vvp_prep_size(env, obj, io, pos, cnt, &exceed);
if (OBD_FAIL_CHECK(OBD_FAIL_LLITE_IMUTEX_NOSEC) && lock_inode)
RETURN(-EINVAL);
+ if (!(file->f_flags & O_DIRECT)) {
+ result = cl_io_lru_reserve(env, io, pos, cnt);
+ if (result)
+ RETURN(result);
+ }
+
if (vio->vui_iter == NULL) {
/* from a temp io in ll_cl_init(). */
result = 0;
RETURN(0);
}
+int lov_io_lru_reserve(const struct lu_env *env,
+ const struct cl_io_slice *ios, loff_t pos, size_t bytes)
+{
+ struct lov_io *lio = cl2lov_io(env, ios);
+ struct lov_stripe_md *lsm = lio->lis_object->lo_lsm;
+ struct lov_io_sub *sub;
+ struct lu_extent ext;
+ int index;
+ int rc = 0;
+
+ ENTRY;
+
+ ext.e_start = pos;
+ ext.e_end = pos + bytes;
+ lov_foreach_io_layout(index, lio, &ext) {
+ struct lov_layout_entry *le = lov_entry(lio->lis_object, index);
+ struct lov_layout_raid0 *r0 = &le->lle_raid0;
+ u64 start;
+ u64 end;
+ int stripe;
+
+ if (!lsm_entry_inited(lsm, index))
+ continue;
+
+ if (!le->lle_valid && !ios->cis_io->ci_designated_mirror) {
+ CERROR(DFID": I/O to invalid component: %d, mirror: %d\n",
+ PFID(lu_object_fid(lov2lu(lio->lis_object))),
+ index, lio->lis_mirror_index);
+ RETURN(-EIO);
+ }
+
+ for (stripe = 0; stripe < r0->lo_nr; stripe++) {
+ if (!lov_stripe_intersects(lsm, index, stripe,
+ &ext, &start, &end))
+ continue;
+
+ if (unlikely(!r0->lo_sub[stripe]))
+ RETURN(-EIO);
+
+ sub = lov_sub_get(env, lio,
+ lov_comp_index(index, stripe));
+ if (IS_ERR(sub))
+ return PTR_ERR(sub);
+
+ rc = cl_io_lru_reserve(sub->sub_env, &sub->sub_io, start,
+ end - start + 1);
+ if (rc != 0)
+ RETURN(rc);
+ }
+ }
+
+ RETURN(0);
+}
+
/**
* 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
}
},
.cio_read_ahead = lov_io_read_ahead,
+ .cio_lru_reserve = lov_io_lru_reserve,
.cio_submit = lov_io_submit,
.cio_commit_async = lov_io_commit_async,
};
pgoff_t start, struct cl_read_ahead *ra)
{
struct osc_object *osc = cl2osc(ios->cis_obj);
+ struct osc_io *oio = cl2osc_io(env, ios);
struct ldlm_lock *dlmlock;
ENTRY;
if (dlmlock == NULL)
RETURN(-ENODATA);
+ oio->oi_is_readahead = 1;
if (dlmlock->l_req_mode != LCK_PR) {
struct lustre_handle lockh;
ra->cra_rpc_pages = osc_cli(osc)->cl_max_pages_per_rpc;
ra->cra_end_idx = CL_PAGE_EOF;
ra->cra_release = osc_read_ahead_release;
- ra->cra_cbdata = dlmlock;
+ ra->cra_dlmlock = dlmlock;
+ ra->cra_oio = oio;
RETURN(0);
}
static const struct cl_io_operations mdc_io_ops = {
.op = {
[CIT_READ] = {
- .cio_iter_init = osc_io_rw_iter_init,
+ .cio_iter_init = osc_io_iter_init,
.cio_iter_fini = osc_io_rw_iter_fini,
.cio_start = osc_io_read_start,
},
[CIT_WRITE] = {
- .cio_iter_init = osc_io_rw_iter_init,
+ .cio_iter_init = osc_io_iter_init,
.cio_iter_fini = osc_io_rw_iter_fini,
.cio_start = osc_io_write_start,
.cio_end = osc_io_end,
},
},
.cio_read_ahead = mdc_io_read_ahead,
+ .cio_lru_reserve = osc_io_lru_reserve,
.cio_submit = osc_io_submit,
.cio_commit_async = osc_io_commit_async,
.cio_extent_release = osc_io_extent_release,
EXPORT_SYMBOL(cl_io_read_ahead);
/**
+ * Called before io start, to reserve enough LRU slots to avoid
+ * deadlock.
+ *
+ * \see cl_io_operations::cio_lru_reserve()
+ */
+int cl_io_lru_reserve(const struct lu_env *env, struct cl_io *io,
+ loff_t pos, size_t bytes)
+{
+ const struct cl_io_slice *scan;
+ int result = 0;
+
+ LINVRNT(io->ci_type == CIT_READ || io->ci_type == CIT_WRITE);
+ LINVRNT(cl_io_invariant(io));
+ ENTRY;
+
+ list_for_each_entry(scan, &io->ci_layers, cis_linkage) {
+ if (scan->cis_iop->cio_lru_reserve) {
+ result = scan->cis_iop->cio_lru_reserve(env, scan,
+ pos, bytes);
+ if (result)
+ break;
+ }
+ }
+
+ RETURN(result);
+}
+EXPORT_SYMBOL(cl_io_lru_reserve);
+
+/**
* Commit a list of contiguous pages into writeback cache.
*
* \returns 0 if all pages committed, or errcode if error occurred.
{
}
-void osc_read_ahead_release(const struct lu_env *env, void *cbdata)
+void osc_read_ahead_release(const struct lu_env *env, struct cl_read_ahead *ra)
{
- struct ldlm_lock *dlmlock = cbdata;
+ struct ldlm_lock *dlmlock = ra->cra_dlmlock;
+ struct osc_io *oio = ra->cra_oio;
struct lustre_handle lockh;
+ oio->oi_is_readahead = 0;
ldlm_lock2handle(dlmlock, &lockh);
ldlm_lock_decref(&lockh, LCK_PR);
LDLM_LOCK_PUT(dlmlock);
const struct cl_io_slice *ios,
pgoff_t start, struct cl_read_ahead *ra)
{
- struct osc_object *osc = cl2osc(ios->cis_obj);
- struct ldlm_lock *dlmlock;
- int result = -ENODATA;
+ struct osc_object *osc = cl2osc(ios->cis_obj);
+ struct osc_io *oio = cl2osc_io(env, ios);
+ struct ldlm_lock *dlmlock;
+ int result = -ENODATA;
+
ENTRY;
+ oio->oi_is_readahead = true;
dlmlock = osc_dlmlock_at_pgoff(env, osc, start, 0);
if (dlmlock != NULL) {
LASSERT(dlmlock->l_ast_data == osc);
ra->cra_end_idx = cl_index(osc2cl(osc),
dlmlock->l_policy_data.l_extent.end);
ra->cra_release = osc_read_ahead_release;
- ra->cra_cbdata = dlmlock;
+ ra->cra_dlmlock = dlmlock;
+ ra->cra_oio = oio;
if (ra->cra_end_idx != CL_PAGE_EOF)
ra->cra_contention = true;
result = 0;
struct obd_import *imp = osc_cli(osc)->cl_import;
struct osc_io *oio = osc_env_io(env);
int rc = -EIO;
+
ENTRY;
spin_lock(&imp->imp_lock);
}
EXPORT_SYMBOL(osc_io_iter_init);
-int osc_io_rw_iter_init(const struct lu_env *env,
- const struct cl_io_slice *ios)
-{
- struct cl_io *io = ios->cis_io;
- struct osc_io *oio = osc_env_io(env);
- struct osc_object *osc = cl2osc(ios->cis_obj);
- unsigned long npages;
- ENTRY;
-
- if (cl_io_is_append(io))
- RETURN(osc_io_iter_init(env, ios));
-
- npages = io->u.ci_rw.crw_count >> PAGE_SHIFT;
- if (io->u.ci_rw.crw_pos & ~PAGE_MASK)
- ++npages;
-
- oio->oi_lru_reserved = osc_lru_reserve(osc_cli(osc), npages);
-
- RETURN(osc_io_iter_init(env, ios));
-}
-EXPORT_SYMBOL(osc_io_rw_iter_init);
-
void osc_io_iter_fini(const struct lu_env *env,
const struct cl_io_slice *ios)
{
}
EXPORT_SYMBOL(osc_io_lseek_end);
+int osc_io_lru_reserve(const struct lu_env *env,
+ const struct cl_io_slice *ios,
+ loff_t pos, size_t bytes)
+{
+ struct osc_object *osc = cl2osc(ios->cis_obj);
+ struct osc_io *oio = osc_env_io(env);
+ unsigned long npages = 0;
+ size_t page_offset;
+
+ ENTRY;
+
+ page_offset = pos & ~PAGE_MASK;
+ if (page_offset) {
+ ++npages;
+ if (bytes > PAGE_SIZE - page_offset)
+ bytes -= (PAGE_SIZE - page_offset);
+ else
+ bytes = 0;
+ }
+ npages += (bytes + PAGE_SIZE - 1) >> PAGE_SHIFT;
+ oio->oi_lru_reserved = osc_lru_reserve(osc_cli(osc), npages);
+
+ RETURN(0);
+}
+EXPORT_SYMBOL(osc_io_lru_reserve);
+
static const struct cl_io_operations osc_io_ops = {
.op = {
[CIT_READ] = {
- .cio_iter_init = osc_io_rw_iter_init,
+ .cio_iter_init = osc_io_iter_init,
.cio_iter_fini = osc_io_rw_iter_fini,
.cio_start = osc_io_read_start,
.cio_fini = osc_io_fini
},
[CIT_WRITE] = {
- .cio_iter_init = osc_io_rw_iter_init,
+ .cio_iter_init = osc_io_iter_init,
.cio_iter_fini = osc_io_rw_iter_fini,
.cio_start = osc_io_write_start,
.cio_end = osc_io_end,
}
},
.cio_read_ahead = osc_io_read_ahead,
+ .cio_lru_reserve = osc_io_lru_reserve,
.cio_submit = osc_io_submit,
.cio_commit_async = osc_io_commit_async,
.cio_extent_release = osc_io_extent_release
break;
if (rc > 0)
continue;
+ /* IO issued by readahead, don't try hard */
+ if (oio->oi_is_readahead) {
+ if (atomic_long_read(cli->cl_lru_left) > 0)
+ continue;
+ rc = -EBUSY;
+ break;
+ }
cond_resched();
rc = l_wait_event_abortable(
unsigned long reserved = 0;
unsigned long max_pages;
unsigned long c;
+ int rc;
- /* reserve a full RPC window at most to avoid that a thread accidentally
- * consumes too many LRU slots */
- max_pages = cli->cl_max_pages_per_rpc * cli->cl_max_rpcs_in_flight;
- if (npages > max_pages)
- npages = max_pages;
-
+again:
c = atomic_long_read(cli->cl_lru_left);
if (c < npages && osc_lru_reclaim(cli, npages) > 0)
c = atomic_long_read(cli->cl_lru_left);
+
+ if (c < npages) {
+ /*
+ * Trigger writeback in the hope some LRU slot could
+ * be freed.
+ */
+ rc = ptlrpcd_queue_work(cli->cl_writeback_work);
+ if (rc)
+ return 0;
+ }
+
while (c >= npages) {
if (c == atomic_long_cmpxchg(cli->cl_lru_left, c, c - npages)) {
reserved = npages;
}
c = atomic_long_read(cli->cl_lru_left);
}
+
+ if (reserved != npages) {
+ cond_resched();
+ rc = l_wait_event_abortable(
+ osc_lru_waitq,
+ atomic_long_read(cli->cl_lru_left) > 0);
+ goto again;
+ }
+
+ max_pages = cli->cl_max_pages_per_rpc * cli->cl_max_rpcs_in_flight;
if (atomic_long_read(cli->cl_lru_left) < max_pages) {
/* If there aren't enough pages in the per-OSC LRU then
* wake up the LRU thread to try and clear out space, so
}
run_test 427 "Failed DNE2 update request shouldn't corrupt updatelog"
+test_428() {
+ [ $PARALLEL == "yes" ] && skip "skip parallel run"
+ local cache_limit=$CACHE_MAX
+
+ stack_trap "$LCTL set_param -n llite.*.max_cached_mb=$cache_limit"
+ $LCTL set_param -n llite.*.max_cached_mb=64
+
+ mkdir $DIR/$tdir
+ $LFS setstripe -c 1 $DIR/$tdir
+ eval touch $DIR/$tdir/$tfile.{1..$OSTCOUNT}
+ stack_trap "rm -f $DIR/$tdir/$tfile.*"
+ #test write
+ for f in $(seq 4); do
+ dd if=/dev/zero of=$DIR/$tdir/$tfile.$f bs=128M count=1 &
+ done
+ wait
+
+ cancel_lru_locks osc
+ # Test read
+ for f in $(seq 4); do
+ dd if=$DIR/$tdir/$tfile.$f of=/dev/null bs=128M count=1 &
+ done
+ wait
+}
+run_test 428 "large block size IO should not hang"
+
lseek_test_430() {
local offset
local file=$1
git://git.whamcloud.com / fs / lustre-release.git / commitdiff