X-Git-Url: https://git.whamcloud.com/?a=blobdiff_plain;ds=sidebyside;f=lustre%2Fmdc%2Fmdc_dev.c;h=036dc9203be6bfa9422c8124a12be16b6d8a4bf7;hb=refs%2Fchanges%2F15%2F35815%2F2;hp=f148f46f04b8616be3e4a73a700e9beffb6b6614;hpb=276221c2a1d26a824032eabcaa7e4accda36f5d1;p=fs%2Flustre-release.git

diff --git a/lustre/mdc/mdc_dev.c b/lustre/mdc/mdc_dev.c
index f148f46..036dc92 100644
--- a/lustre/mdc/mdc_dev.c
+++ b/lustre/mdc/mdc_dev.c
@@ -68,21 +68,17 @@ static void mdc_lock_lvb_update(const struct lu_env *env,
 				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;
 
@@ -92,9 +88,9 @@ static int mdc_set_dom_lock_data(const struct lu_env *env,
 }
 
 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;
@@ -108,11 +104,19 @@ int mdc_dom_lock_match(const struct lu_env *env, struct obd_export *exp,
 	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;
 		}
@@ -294,7 +298,7 @@ void mdc_lock_lockless_cancel(const struct lu_env *env,
  */
 static int mdc_dlm_blocking_ast0(const struct lu_env *env,
 				 struct ldlm_lock *dlmlock,
-				 void *data, int flag)
+				 int flag)
 {
 	struct cl_object *obj = NULL;
 	int result = 0;
@@ -322,7 +326,6 @@ static int mdc_dlm_blocking_ast0(const struct lu_env *env,
 		dlmlock->l_ast_data = NULL;
 		cl_object_get(obj);
 	}
-	ldlm_set_kms_ignore(dlmlock);
 	unlock_res_and_lock(dlmlock);
 
 	/* if l_ast_data is NULL, the dlmlock was enqueued by AGL or
@@ -386,7 +389,7 @@ int mdc_ldlm_blocking_ast(struct ldlm_lock *dlmlock,
 			break;
 		}
 
-		rc = mdc_dlm_blocking_ast0(env, dlmlock, data, flag);
+		rc = mdc_dlm_blocking_ast0(env, dlmlock, flag);
 		cl_env_put(env, &refcheck);
 		break;
 	}
@@ -412,6 +415,7 @@ void mdc_lock_lvb_update(const struct lu_env *env, struct osc_object *osc,
 	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;
 
@@ -429,24 +433,31 @@ void mdc_lock_lvb_update(const struct lu_env *env, struct osc_object *osc,
 		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, bool lvb_update)
+			     struct lustre_handle *lockh)
 {
+	struct osc_object *osc = cl2osc(oscl->ols_cl.cls_obj);
 	struct ldlm_lock *dlmlock;
 
 	ENTRY;
@@ -485,10 +496,11 @@ static void mdc_lock_granted(const struct lu_env *env, struct osc_lock *oscl,
 		descr->cld_end = CL_PAGE_EOF;
 
 		/* no lvb update for matched lock */
-		if (lvb_update) {
+		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);
 		}
 	}
 	unlock_res_and_lock(dlmlock);
@@ -529,7 +541,7 @@ static int mdc_lock_upcall(void *cookie, struct lustre_handle *lockh,
 
 	CDEBUG(D_INODE, "rc %d, err %d\n", rc, errcode);
 	if (rc == 0)
-		mdc_lock_granted(env, oscl, lockh, errcode == ELDLM_OK);
+		mdc_lock_granted(env, oscl, lockh);
 
 	/* Error handling, some errors are tolerable. */
 	if (oscl->ols_locklessable && rc == -EUSERS) {
@@ -703,10 +715,8 @@ int mdc_enqueue_send(const struct lu_env *env, struct obd_export *exp,
 	 * LVB information, e.g. canceled locks or locks of just pruned object,
 	 * such locks should be skipped.
 	 */
-	mode = ldlm_lock_match_with_skip(obd->obd_namespace, match_flags,
-					 LDLM_FL_KMS_IGNORE, res_id,
-					 einfo->ei_type, policy, mode,
-					 &lockh, 0);
+	mode = ldlm_lock_match(obd->obd_namespace, match_flags, res_id,
+			       einfo->ei_type, policy, mode, &lockh, 0);
 	if (mode) {
 		struct ldlm_lock *matched;
 
@@ -714,18 +724,11 @@ int mdc_enqueue_send(const struct lu_env *env, struct obd_export *exp,
 			RETURN(ELDLM_OK);
 
 		matched = ldlm_handle2lock(&lockh);
-		/* this shouldn't happen but this check is kept to make
-		 * related test fail if problem occurs
-		 */
-		if (unlikely(ldlm_is_kms_ignore(matched))) {
-			LDLM_ERROR(matched, "matched lock has KMS ignore flag");
-			goto no_match;
-		}
 
 		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. */
@@ -735,7 +738,6 @@ int mdc_enqueue_send(const struct lu_env *env, struct obd_export *exp,
 			LDLM_LOCK_PUT(matched);
 			RETURN(ELDLM_OK);
 		}
-no_match:
 		ldlm_lock_decref(&lockh, mode);
 		LDLM_LOCK_PUT(matched);
 	}
@@ -779,8 +781,7 @@ no_match:
 		if (!rc) {
 			struct osc_enqueue_args *aa;
 
-			CLASSERT(sizeof(*aa) <= sizeof(req->rq_async_args));
-			aa = ptlrpc_req_async_args(req);
+			aa = ptlrpc_req_async_args(aa, req);
 			aa->oa_exp = exp;
 			aa->oa_mode = einfo->ei_mode;
 			aa->oa_type = einfo->ei_type;
@@ -1249,8 +1250,7 @@ static int mdc_io_data_version_start(const struct lu_env *env,
 	ptlrpc_request_set_replen(req);
 
 	req->rq_interpret_reply = mdc_data_version_interpret;
-	CLASSERT(sizeof(*dva) <= sizeof(req->rq_async_args));
-	dva = ptlrpc_req_async_args(req);
+	dva = ptlrpc_req_async_args(dva, req);
 	dva->dva_oio = oio;
 
 	ptlrpcd_add_req(req);
@@ -1288,13 +1288,13 @@ static void mdc_io_data_version_end(const struct lu_env *env,
 static struct cl_io_operations mdc_io_ops = {
 	.op = {
 		[CIT_READ] = {
-			.cio_iter_init = osc_io_iter_init,
-			.cio_iter_fini = osc_io_iter_fini,
+			.cio_iter_init = osc_io_rw_iter_init,
+			.cio_iter_fini = osc_io_rw_iter_fini,
 			.cio_start     = osc_io_read_start,
 		},
 		[CIT_WRITE] = {
-			.cio_iter_init = osc_io_write_iter_init,
-			.cio_iter_fini = osc_io_write_iter_fini,
+			.cio_iter_init = osc_io_rw_iter_init,
+			.cio_iter_fini = osc_io_rw_iter_fini,
 			.cio_start     = osc_io_write_start,
 			.cio_end       = osc_io_end,
 		},
@@ -1393,11 +1393,35 @@ static int mdc_attr_get(const struct lu_env *env, struct cl_object *obj,
 
 static int mdc_object_ast_clear(struct ldlm_lock *lock, void *data)
 {
+	struct osc_object *osc = (struct osc_object *)data;
+	struct ost_lvb *lvb = &lock->l_ost_lvb;
+	struct lov_oinfo *oinfo;
 	ENTRY;
 
-	if (lock->l_ast_data == data)
+	if (lock->l_ast_data == data) {
 		lock->l_ast_data = NULL;
-	ldlm_set_kms_ignore(lock);
+
+		LASSERT(osc != NULL);
+		LASSERT(osc->oo_oinfo != NULL);
+		LASSERT(lvb != NULL);
+
+		/* 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_clear_lvb_cached(lock);
+	}
 	RETURN(LDLM_ITER_CONTINUE);
 }
 
@@ -1414,6 +1438,12 @@ int mdc_object_prune(const struct lu_env *env, struct cl_object *obj)
 	return 0;
 }
 
+static int mdc_object_flush(const struct lu_env *env, struct cl_object *obj,
+			    struct ldlm_lock *lock)
+{
+	RETURN(mdc_dlm_blocking_ast0(env, lock, LDLM_CB_CANCELING));
+}
+
 static const struct cl_object_operations mdc_ops = {
 	.coo_page_init = osc_page_init,
 	.coo_lock_init = mdc_lock_init,
@@ -1423,6 +1453,7 @@ static const struct cl_object_operations mdc_ops = {
 	.coo_glimpse = osc_object_glimpse,
 	.coo_req_attr_set = mdc_req_attr_set,
 	.coo_prune = mdc_object_prune,
+	.coo_object_flush = mdc_object_flush
 };
 
 static const struct osc_object_operations mdc_object_ops = {