It may happen that osc oinfo lvb cache has size < kms.
It occurs if a reply re-ordering happens and an older size is applied
to oinfo unconditionally.
Another possibility is RA, when osc_match_base() attaches the dlm lock
to osc object but does not cache the lvb. The next layout change will
overwrites the lock lvb by the oinfo cache (previous LUS-7731 fix),
presumably smaller values. Therefore, the next lock re-use may run
into a problem with partial page write which thinks the preliminary
read is not needed.
Do not let the cached oinfo lvb size to become less than kms.
Also, cache the lock's lvb in the oinfo on osc_match_base().
Signed-off-by: Vitaly Fertman <c17818@cray.com>
Cray-bug-id: LUS-7731
Change-Id: I50136f57491364146ce7b6a81b814e474e3edb86
Reviewed-on: https://review.whamcloud.com/36200
Tested-by: jenkins <devops@whamcloud.com>
Reviewed-by: Patrick Farrell <pfarrell@whamcloud.com>
Tested-by: Maloo <maloo@whamcloud.com>
Reviewed-by: Mike Pershin <mpershin@whamcloud.com>
Reviewed-by: Oleg Drokin <green@whamcloud.com>
struct ldlm_lock *dlmlock,
struct ost_lvb *lvb);
-static int mdc_set_dom_lock_data(const struct lu_env *env,
- struct ldlm_lock *lock, void *data)
+static int mdc_set_dom_lock_data(struct ldlm_lock *lock, void *data)
{
- struct osc_object *obj = data;
int set = 0;
LASSERT(lock != NULL);
LASSERT(lock->l_glimpse_ast == mdc_ldlm_glimpse_ast);
lock_res_and_lock(lock);
- if (lock->l_ast_data == NULL) {
- lock->l_ast_data = data;
- mdc_lock_lvb_update(env, obj, lock, NULL);
- }
+ if (lock->l_ast_data == NULL)
+ lock->l_ast_data = data;
if (lock->l_ast_data == data)
set = 1;
}
int mdc_dom_lock_match(const struct lu_env *env, struct obd_export *exp,
- struct ldlm_res_id *res_id,
- enum ldlm_type type, union ldlm_policy_data *policy,
- enum ldlm_mode mode, __u64 *flags, void *data,
+ struct ldlm_res_id *res_id, enum ldlm_type type,
+ union ldlm_policy_data *policy, enum ldlm_mode mode,
+ __u64 *flags, struct osc_object *obj,
struct lustre_handle *lockh, int unref)
{
struct obd_device *obd = exp->exp_obd;
if (rc == 0 || lflags & LDLM_FL_TEST_LOCK)
RETURN(rc);
- if (data != NULL) {
+ if (obj != NULL) {
struct ldlm_lock *lock = ldlm_handle2lock(lockh);
LASSERT(lock != NULL);
- if (!mdc_set_dom_lock_data(env, lock, data)) {
+ if (mdc_set_dom_lock_data(lock, obj)) {
+ lock_res_and_lock(lock);
+ if (!ldlm_is_lvb_cached(lock)) {
+ LASSERT(lock->l_ast_data == obj);
+ mdc_lock_lvb_update(env, obj, lock, NULL);
+ ldlm_set_lvb_cached(lock);
+ }
+ unlock_res_and_lock(lock);
+ } else {
ldlm_lock_decref(lockh, rc);
rc = 0;
}
struct cl_attr *attr = &osc_env_info(env)->oti_attr;
unsigned valid = CAT_BLOCKS | CAT_ATIME | CAT_CTIME | CAT_MTIME |
CAT_SIZE;
+ unsigned int setkms = 0;
ENTRY;
size = lvb->lvb_size;
if (size >= oinfo->loi_kms) {
- LDLM_DEBUG(dlmlock, "lock acquired, setting rss=%llu,"
- " kms=%llu", lvb->lvb_size, size);
valid |= CAT_KMS;
attr->cat_kms = size;
- } else {
- LDLM_DEBUG(dlmlock, "lock acquired, setting rss=%llu,"
- " leaving kms=%llu",
- lvb->lvb_size, oinfo->loi_kms);
+ setkms = 1;
}
}
+
+ /* The size should not be less than the kms */
+ if (attr->cat_size < oinfo->loi_kms)
+ attr->cat_size = oinfo->loi_kms;
+
+ LDLM_DEBUG(dlmlock, "acquired size %llu, setting rss=%llu;%s "
+ "kms=%llu, end=%llu", lvb->lvb_size, attr->cat_size,
+ setkms ? "" : " leaving",
+ setkms ? attr->cat_kms : oinfo->loi_kms,
+ dlmlock ? dlmlock->l_policy_data.l_extent.end : -1ull);
+
cl_object_attr_update(env, obj, attr, valid);
cl_object_attr_unlock(obj);
EXIT;
static void mdc_lock_granted(const struct lu_env *env, struct osc_lock *oscl,
struct lustre_handle *lockh)
{
+ struct osc_object *osc = cl2osc(oscl->ols_cl.cls_obj);
struct ldlm_lock *dlmlock;
ENTRY;
/* no lvb update for matched lock */
if (!ldlm_is_lvb_cached(dlmlock)) {
LASSERT(oscl->ols_flags & LDLM_FL_LVB_READY);
- mdc_lock_lvb_update(env, cl2osc(oscl->ols_cl.cls_obj),
- dlmlock, NULL);
+ LASSERT(osc == dlmlock->l_ast_data);
+ mdc_lock_lvb_update(env, osc, dlmlock, NULL);
ldlm_set_lvb_cached(dlmlock);
}
}
if (OBD_FAIL_CHECK(OBD_FAIL_MDC_GLIMPSE_DDOS))
ldlm_set_kms_ignore(matched);
- if (mdc_set_dom_lock_data(env, matched, einfo->ei_cbdata)) {
+ if (mdc_set_dom_lock_data(matched, einfo->ei_cbdata)) {
*flags |= LDLM_FL_LVB_READY;
/* We already have a lock, and it's referenced. */
/* Updates lvb in lock by the cached oinfo */
oinfo = osc->oo_oinfo;
+
+ LDLM_DEBUG(lock, "update lock size %llu blocks %llu [cma]time: "
+ "%llu %llu %llu by oinfo size %llu blocks %llu "
+ "[cma]time %llu %llu %llu", lvb->lvb_size,
+ lvb->lvb_blocks, lvb->lvb_ctime, lvb->lvb_mtime,
+ lvb->lvb_atime, oinfo->loi_lvb.lvb_size,
+ oinfo->loi_lvb.lvb_blocks, oinfo->loi_lvb.lvb_ctime,
+ oinfo->loi_lvb.lvb_mtime, oinfo->loi_lvb.lvb_atime);
+ LASSERT(oinfo->loi_lvb.lvb_size >= oinfo->loi_kms);
+
cl_object_attr_lock(&osc->oo_cl);
memcpy(lvb, &oinfo->loi_lvb, sizeof(oinfo->loi_lvb));
cl_object_attr_unlock(&osc->oo_cl);
-
- LDLM_DEBUG(lock, "update lvb size %llu blocks %llu [cma]time: "
- "%llu %llu %llu", lvb->lvb_size, lvb->lvb_blocks,
- lvb->lvb_ctime, lvb->lvb_mtime, lvb->lvb_atime);
-
ldlm_clear_lvb_cached(lock);
}
RETURN(LDLM_ITER_CONTINUE);
extern struct ptlrpc_request_set *PTLRPCD_SET;
+void osc_lock_lvb_update(const struct lu_env *env,
+ struct osc_object *osc,
+ struct ldlm_lock *dlmlock,
+ struct ost_lvb *lvb);
+
int osc_enqueue_base(struct obd_export *exp, struct ldlm_res_id *res_id,
__u64 *flags, union ldlm_policy_data *policy,
struct ost_lvb *lvb, osc_enqueue_upcall_f upcall,
struct ptlrpc_request_set *rqset, int async,
bool speculative);
-int osc_match_base(struct obd_export *exp, struct ldlm_res_id *res_id,
- enum ldlm_type type, union ldlm_policy_data *policy,
- enum ldlm_mode mode, __u64 *flags, void *data,
+int osc_match_base(const struct lu_env *env, struct obd_export *exp,
+ struct ldlm_res_id *res_id, enum ldlm_type type,
+ union ldlm_policy_data *policy, enum ldlm_mode mode,
+ __u64 *flags, struct osc_object *obj,
struct lustre_handle *lockh, int unref);
int osc_setattr_async(struct obd_export *exp, struct obdo *oa,
*
* Called under lock and resource spin-locks.
*/
-static void osc_lock_lvb_update(const struct lu_env *env,
- struct osc_object *osc,
- struct ldlm_lock *dlmlock,
- struct ost_lvb *lvb)
+void osc_lock_lvb_update(const struct lu_env *env,
+ struct osc_object *osc,
+ struct ldlm_lock *dlmlock,
+ struct ost_lvb *lvb)
{
- struct cl_object *obj = osc2cl(osc);
- struct lov_oinfo *oinfo = osc->oo_oinfo;
- struct cl_attr *attr = &osc_env_info(env)->oti_attr;
- unsigned valid;
+ struct cl_object *obj = osc2cl(osc);
+ struct lov_oinfo *oinfo = osc->oo_oinfo;
+ struct cl_attr *attr = &osc_env_info(env)->oti_attr;
+ unsigned valid, setkms = 0;
ENTRY;
if (size > dlmlock->l_policy_data.l_extent.end)
size = dlmlock->l_policy_data.l_extent.end + 1;
if (size >= oinfo->loi_kms) {
- LDLM_DEBUG(dlmlock, "lock acquired, setting rss=%llu"
- ", kms=%llu", lvb->lvb_size, size);
valid |= CAT_KMS;
attr->cat_kms = size;
- } else {
- LDLM_DEBUG(dlmlock, "lock acquired, setting rss="
- "%llu; leaving kms=%llu, end=%llu",
- lvb->lvb_size, oinfo->loi_kms,
- dlmlock->l_policy_data.l_extent.end);
+ setkms = 1;
}
ldlm_lock_allow_match_locked(dlmlock);
}
+ /* The size should not be less than the kms */
+ if (attr->cat_size < oinfo->loi_kms)
+ attr->cat_size = oinfo->loi_kms;
+
+ LDLM_DEBUG(dlmlock, "acquired size %llu, setting rss=%llu;%s "
+ "kms=%llu, end=%llu", lvb->lvb_size, attr->cat_size,
+ setkms ? "" : " leaving",
+ setkms ? attr->cat_kms : oinfo->loi_kms,
+ dlmlock ? dlmlock->l_policy_data.l_extent.end : -1ull);
+
cl_object_attr_update(env, obj, attr, valid);
cl_object_attr_unlock(obj);
static void osc_lock_granted(const struct lu_env *env, struct osc_lock *oscl,
struct lustre_handle *lockh)
{
+ struct osc_object *osc = cl2osc(oscl->ols_cl.cls_obj);
struct ldlm_lock *dlmlock;
dlmlock = ldlm_handle2lock_long(lockh, 0);
/* no lvb update for matched lock */
if (!ldlm_is_lvb_cached(dlmlock)) {
LASSERT(oscl->ols_flags & LDLM_FL_LVB_READY);
- osc_lock_lvb_update(env, cl2osc(oscl->ols_cl.cls_obj),
- dlmlock, NULL);
+ LASSERT(osc == dlmlock->l_ast_data);
+ osc_lock_lvb_update(env, osc, dlmlock, NULL);
ldlm_set_lvb_cached(dlmlock);
}
LINVRNT(osc_lock_invariant(oscl));
* with a uniq gid and it conflicts with all other lock modes too
*/
again:
- mode = osc_match_base(osc_export(obj), resname, LDLM_EXTENT, policy,
- LCK_PR | LCK_PW | LCK_GROUP, &flags, obj, &lockh,
- dap_flags & OSC_DAP_FL_CANCELING);
+ mode = osc_match_base(env, osc_export(obj), resname, LDLM_EXTENT,
+ policy, LCK_PR | LCK_PW | LCK_GROUP, &flags,
+ obj, &lockh, dap_flags & OSC_DAP_FL_CANCELING);
if (mode != 0) {
lock = ldlm_handle2lock(&lockh);
/* RACE: the lock is cancelled so let's try again */
/* Updates lvb in lock by the cached oinfo */
oinfo = osc->oo_oinfo;
+
+ LDLM_DEBUG(lock, "update lock size %llu blocks %llu [cma]time: "
+ "%llu %llu %llu by oinfo size %llu blocks %llu "
+ "[cma]time %llu %llu %llu", lvb->lvb_size,
+ lvb->lvb_blocks, lvb->lvb_ctime, lvb->lvb_mtime,
+ lvb->lvb_atime, oinfo->loi_lvb.lvb_size,
+ oinfo->loi_lvb.lvb_blocks, oinfo->loi_lvb.lvb_ctime,
+ oinfo->loi_lvb.lvb_mtime, oinfo->loi_lvb.lvb_atime);
+ LASSERT(oinfo->loi_lvb.lvb_size >= oinfo->loi_kms);
+
cl_object_attr_lock(&osc->oo_cl);
memcpy(lvb, &oinfo->loi_lvb, sizeof(oinfo->loi_lvb));
cl_object_attr_unlock(&osc->oo_cl);
-
- LDLM_DEBUG(lock, "update lvb size %llu blocks %llu [cma]time: "
- "%llu %llu %llu", lvb->lvb_size, lvb->lvb_blocks,
- lvb->lvb_ctime, lvb->lvb_mtime, lvb->lvb_atime);
-
ldlm_clear_lvb_cached(lock);
}
RETURN(LDLM_ITER_CONTINUE);
RETURN(rc);
}
-int osc_match_base(struct obd_export *exp, struct ldlm_res_id *res_id,
- enum ldlm_type type, union ldlm_policy_data *policy,
- enum ldlm_mode mode, __u64 *flags, void *data,
+int osc_match_base(const struct lu_env *env, struct obd_export *exp,
+ struct ldlm_res_id *res_id, enum ldlm_type type,
+ union ldlm_policy_data *policy, enum ldlm_mode mode,
+ __u64 *flags, struct osc_object *obj,
struct lustre_handle *lockh, int unref)
{
struct obd_device *obd = exp->exp_obd;
if (rc == 0 || lflags & LDLM_FL_TEST_LOCK)
RETURN(rc);
- if (data != NULL) {
+ if (obj != NULL) {
struct ldlm_lock *lock = ldlm_handle2lock(lockh);
LASSERT(lock != NULL);
- if (!osc_set_lock_data(lock, data)) {
+ if (osc_set_lock_data(lock, obj)) {
+ lock_res_and_lock(lock);
+ if (!ldlm_is_lvb_cached(lock)) {
+ LASSERT(lock->l_ast_data == obj);
+ osc_lock_lvb_update(env, obj, lock, NULL);
+ ldlm_set_lvb_cached(lock);
+ }
+ unlock_res_and_lock(lock);
+ } else {
ldlm_lock_decref(lockh, rc);
rc = 0;
}