X-Git-Url: https://git.whamcloud.com/?p=fs%2Flustre-release.git;a=blobdiff_plain;f=lustre%2Fmdt%2Fmdt_handler.c;h=d9cc1157c307a5a7fc74f8785351dfd14dbf17a3;hp=47c8a17c3eeec6fc9d8d2d3c614f6a0ea05f7ebe;hb=a079ade7913b923b795ea5c01df4e69bf1a87691;hpb=05ff4729e0a8bb1a67957c20d18e3097ac6003c1;ds=sidebyside

diff --git a/lustre/mdt/mdt_handler.c b/lustre/mdt/mdt_handler.c
index 47c8a17..d9cc115 100644
--- a/lustre/mdt/mdt_handler.c
+++ b/lustre/mdt/mdt_handler.c
@@ -356,8 +356,10 @@ static int mdt_get_root(struct tgt_session_info *tsi)
 	struct mdt_thread_info	*info = tsi2mdt_info(tsi);
 	struct mdt_device	*mdt = info->mti_mdt;
 	struct mdt_body		*repbody;
-	char			*fileset;
+	char			*fileset = NULL, *buffer = NULL;
 	int			 rc;
+	struct obd_export	*exp = info->mti_exp;
+	char			*nodemap_fileset;
 
 	ENTRY;
 
@@ -373,7 +375,27 @@ static int mdt_get_root(struct tgt_session_info *tsi)
 		fileset = req_capsule_client_get(info->mti_pill, &RMF_NAME);
 		if (fileset == NULL)
 			GOTO(out, rc = err_serious(-EFAULT));
+	}
 
+	nodemap_fileset = nodemap_get_fileset(exp->exp_target_data.ted_nodemap);
+	if (nodemap_fileset && nodemap_fileset[0]) {
+		if (fileset) {
+			/* consider fileset from client as a sub-fileset
+			 * of the nodemap one */
+			OBD_ALLOC(buffer, PATH_MAX + 1);
+			if (buffer == NULL)
+				GOTO(out, rc = err_serious(-ENOMEM));
+			if (snprintf(buffer, PATH_MAX + 1, "%s/%s",
+				     nodemap_fileset, fileset) >= PATH_MAX + 1)
+				GOTO(out, rc = err_serious(-EINVAL));
+			fileset = buffer;
+		} else {
+			/* enforce fileset as specified in the nodemap */
+			fileset = nodemap_fileset;
+		}
+	}
+
+	if (fileset) {
 		rc = mdt_lookup_fileset(info, fileset, &repbody->mbo_fid1);
 		if (rc < 0)
 			GOTO(out, rc = err_serious(rc));
@@ -385,6 +407,8 @@ static int mdt_get_root(struct tgt_session_info *tsi)
 	EXIT;
 out:
 	mdt_thread_info_fini(info);
+	if (buffer)
+		OBD_FREE(buffer, PATH_MAX+1);
 	return rc;
 }
 
@@ -501,9 +525,9 @@ int mdt_pack_acl2body(struct mdt_thread_info *info, struct mdt_body *repbody,
 void mdt_pack_attr2body(struct mdt_thread_info *info, struct mdt_body *b,
                         const struct lu_attr *attr, const struct lu_fid *fid)
 {
-	struct md_attr		*ma = &info->mti_attr;
-	struct obd_export	*exp = info->mti_exp;
-	struct lu_nodemap	*nodemap = exp->exp_target_data.ted_nodemap;
+	struct md_attr *ma = &info->mti_attr;
+	struct obd_export *exp = info->mti_exp;
+	struct lu_nodemap *nodemap = NULL;
 
 	LASSERT(ma->ma_valid & MA_INODE);
 
@@ -527,6 +551,11 @@ void mdt_pack_attr2body(struct mdt_thread_info *info, struct mdt_body *b,
 		b->mbo_nlink = attr->la_nlink;
 		b->mbo_valid |= OBD_MD_FLNLINK;
 	}
+	if (attr->la_valid & (LA_UID|LA_GID)) {
+		nodemap = nodemap_get_from_exp(exp);
+		if (IS_ERR(nodemap))
+			goto out;
+	}
 	if (attr->la_valid & LA_UID) {
 		b->mbo_uid = nodemap_map_id(nodemap, NODEMAP_UID,
 					    NODEMAP_FS_TO_CLIENT,
@@ -539,6 +568,7 @@ void mdt_pack_attr2body(struct mdt_thread_info *info, struct mdt_body *b,
 					    attr->la_gid);
 		b->mbo_valid |= OBD_MD_FLGID;
 	}
+
 	b->mbo_mode = attr->la_mode;
 	if (attr->la_valid & LA_MODE)
 		b->mbo_valid |= OBD_MD_FLMODE;
@@ -588,6 +618,10 @@ void mdt_pack_attr2body(struct mdt_thread_info *info, struct mdt_body *b,
 	if (fid != NULL && (b->mbo_valid & OBD_MD_FLSIZE))
 		CDEBUG(D_VFSTRACE, DFID": returning size %llu\n",
 		       PFID(fid), (unsigned long long)b->mbo_size);
+
+out:
+	if (!IS_ERR_OR_NULL(nodemap))
+		nodemap_putref(nodemap);
 }
 
 static inline int mdt_body_has_lov(const struct lu_attr *la,
@@ -915,7 +949,6 @@ static int mdt_getattr_internal(struct mdt_thread_info *info,
 	struct mdt_body		*repbody;
 	struct lu_buf		*buffer = &info->mti_buf;
 	struct obd_export	*exp = info->mti_exp;
-	struct lu_nodemap	*nodemap = exp->exp_target_data.ted_nodemap;
 	int			 rc;
 	int			 is_root;
 	ENTRY;
@@ -1122,8 +1155,14 @@ static int mdt_getattr_internal(struct mdt_thread_info *info,
 	}
 #ifdef CONFIG_FS_POSIX_ACL
 	else if ((exp_connect_flags(req->rq_export) & OBD_CONNECT_ACL) &&
-		 (reqbody->mbo_valid & OBD_MD_FLACL))
+		 (reqbody->mbo_valid & OBD_MD_FLACL)) {
+		struct lu_nodemap *nodemap = nodemap_get_from_exp(exp);
+		if (IS_ERR(nodemap))
+			RETURN(PTR_ERR(nodemap));
+
 		rc = mdt_pack_acl2body(info, repbody, o, nodemap);
+		nodemap_putref(nodemap);
+	}
 #endif
 
 out:
@@ -1422,7 +1461,7 @@ static int mdt_getattr_name_lock(struct mdt_thread_info *info,
 		if (!mdt_object_exists(child)) {
 			LU_OBJECT_DEBUG(D_INFO, info->mti_env,
 					&child->mot_obj,
-					"remote object doesn't exist.\n");
+					"remote object doesn't exist.");
 			mdt_object_unlock(info, child, lhc, 1);
 			RETURN(-ENOENT);
 		}
@@ -1462,7 +1501,7 @@ static int mdt_getattr_name_lock(struct mdt_thread_info *info,
 	if (unlikely(!mdt_object_exists(parent)) && lu_name_is_valid(lname)) {
 		LU_OBJECT_DEBUG(D_INODE, info->mti_env,
 				&parent->mot_obj,
-				"Parent doesn't exist!\n");
+				"Parent doesn't exist!");
 		RETURN(-ESTALE);
 	}
 
@@ -1533,7 +1572,7 @@ static int mdt_getattr_name_lock(struct mdt_thread_info *info,
 	if (!mdt_object_exists(child)) {
 		LU_OBJECT_DEBUG(D_INODE, info->mti_env,
 				&child->mot_obj,
-				"Object doesn't exist!\n");
+				"Object doesn't exist!");
 		GOTO(out_child, rc = -ENOENT);
 	}
 
@@ -2021,7 +2060,7 @@ static int mdt_quotactl(struct tgt_session_info *tsi)
 	int			 id, rc;
 	struct mdt_device	*mdt = mdt_exp2dev(exp);
 	struct lu_device	*qmt = mdt->mdt_qmt_dev;
-	struct lu_nodemap	*nodemap = exp->exp_target_data.ted_nodemap;
+	struct lu_nodemap	*nodemap;
 	ENTRY;
 
 	oqctl = req_capsule_client_get(pill, &RMF_OBD_QUOTACTL);
@@ -2032,23 +2071,27 @@ static int mdt_quotactl(struct tgt_session_info *tsi)
 	if (rc)
 		RETURN(err_serious(rc));
 
+	nodemap = nodemap_get_from_exp(exp);
+	if (IS_ERR(nodemap))
+		RETURN(PTR_ERR(nodemap));
+
 	switch (oqctl->qc_cmd) {
 		/* master quotactl */
 	case Q_SETINFO:
 	case Q_SETQUOTA:
 		if (!nodemap_can_setquota(nodemap))
-			RETURN(-EPERM);
+			GOTO(out_nodemap, rc = -EPERM);
 	case Q_GETINFO:
 	case Q_GETQUOTA:
 		if (qmt == NULL)
-			RETURN(-EOPNOTSUPP);
+			GOTO(out_nodemap, rc = -EOPNOTSUPP);
 		/* slave quotactl */
 	case Q_GETOINFO:
 	case Q_GETOQUOTA:
 		break;
 	default:
 		CERROR("Unsupported quotactl command: %d\n", oqctl->qc_cmd);
-		RETURN(-EFAULT);
+		GOTO(out_nodemap, rc = -EFAULT);
 	}
 
 	/* map uid/gid for remote client */
@@ -2060,7 +2103,7 @@ static int mdt_quotactl(struct tgt_session_info *tsi)
 
 		if (unlikely(oqctl->qc_cmd != Q_GETQUOTA &&
 			     oqctl->qc_cmd != Q_GETINFO))
-			RETURN(-EPERM);
+			GOTO(out_nodemap, rc = -EPERM);
 
 		if (oqctl->qc_type == USRQUOTA)
 			id = lustre_idmap_lookup_uid(NULL, idmap, 0,
@@ -2069,11 +2112,11 @@ static int mdt_quotactl(struct tgt_session_info *tsi)
 			id = lustre_idmap_lookup_gid(NULL, idmap, 0,
 						     oqctl->qc_id);
 		else
-			RETURN(-EINVAL);
+			GOTO(out_nodemap, rc = -EINVAL);
 
 		if (id == CFS_IDMAP_NOTFOUND) {
 			CDEBUG(D_QUOTA, "no mapping for id %u\n", oqctl->qc_id);
-			RETURN(-EACCES);
+			GOTO(out_nodemap, rc = -EACCES);
 		}
 	}
 
@@ -2086,7 +2129,7 @@ static int mdt_quotactl(struct tgt_session_info *tsi)
 
 	repoqc = req_capsule_server_get(pill, &RMF_OBD_QUOTACTL);
 	if (repoqc == NULL)
-		RETURN(err_serious(-EFAULT));
+		GOTO(out_nodemap, rc = err_serious(-EFAULT));
 
 	if (oqctl->qc_id != id)
 		swap(oqctl->qc_id, id);
@@ -2110,14 +2153,20 @@ static int mdt_quotactl(struct tgt_session_info *tsi)
 
 	default:
 		CERROR("Unsupported quotactl command: %d\n", oqctl->qc_cmd);
-		RETURN(-EFAULT);
+		GOTO(out_nodemap, rc = -EFAULT);
 	}
 
 	if (oqctl->qc_id != id)
 		swap(oqctl->qc_id, id);
 
 	*repoqc = *oqctl;
-	RETURN(rc);
+
+	EXIT;
+
+out_nodemap:
+	nodemap_putref(nodemap);
+
+	return rc;
 }
 
 /** clone llog ctxt from child (mdd)
@@ -2381,7 +2430,7 @@ int mdt_remote_blocking_ast(struct ldlm_lock *lock, struct ldlm_lock_desc *desc,
 		}
 		break;
 	case LDLM_CB_CANCELING:
-		LDLM_DEBUG(lock, "Revoke remote lock\n");
+		LDLM_DEBUG(lock, "Revoke remote lock");
 		/* discard slc lock here so that it can be cleaned anytime,
 		 * especially for cleanup_resource() */
 		tgt_discard_slc_lock(lock);
@@ -2408,7 +2457,7 @@ int mdt_check_resent_lock(struct mdt_thread_info *info,
 			/* Lock is pinned by ldlm_handle_enqueue0() as it is
 			 * a resend case, however, it could be already destroyed
 			 * due to client eviction or a raced cancel RPC. */
-			LDLM_DEBUG_NOLOCK("Invalid lock handle "LPX64"\n",
+			LDLM_DEBUG_NOLOCK("Invalid lock handle "LPX64,
 					  lhc->mlh_reg_lh.cookie);
 			RETURN(-ESTALE);
 		}
@@ -3121,7 +3170,7 @@ static int
 mdt_intent_lock_replace(struct mdt_thread_info *info,
 			struct ldlm_lock **lockp,
 			struct mdt_lock_handle *lh,
-			__u64 flags)
+			__u64 flags, int result)
 {
         struct ptlrpc_request  *req = mdt_info_req(info);
         struct ldlm_lock       *lock = *lockp;
@@ -3135,8 +3184,19 @@ mdt_intent_lock_replace(struct mdt_thread_info *info,
                 RETURN(0);
         }
 
-        LASSERTF(new_lock != NULL,
-                 "lockh "LPX64"\n", lh->mlh_reg_lh.cookie);
+	if (new_lock == NULL && (flags & LDLM_FL_RESENT)) {
+		/* Lock is pinned by ldlm_handle_enqueue0() as it is
+		 * a resend case, however, it could be already destroyed
+		 * due to client eviction or a raced cancel RPC. */
+		LDLM_DEBUG_NOLOCK("Invalid lock handle "LPX64"\n",
+				  lh->mlh_reg_lh.cookie);
+		lh->mlh_reg_lh.cookie = 0;
+		RETURN(-ESTALE);
+	}
+
+	LASSERTF(new_lock != NULL,
+		 "lockh "LPX64" flags "LPX64" rc %d\n",
+		 lh->mlh_reg_lh.cookie, flags, result);
 
         /*
          * If we've already given this lock to a client once, then we should
@@ -3270,7 +3330,7 @@ static int mdt_intent_getxattr(enum mdt_it_code opcode,
 
 	grc = mdt_getxattr(info);
 
-	rc = mdt_intent_lock_replace(info, lockp, lhc, flags);
+	rc = mdt_intent_lock_replace(info, lockp, lhc, flags, 0);
 
 	if (mdt_info_req(info)->rq_repmsg != NULL)
 		ldlm_rep = req_capsule_server_get(info->mti_pill, &RMF_DLM_REP);
@@ -3339,7 +3399,7 @@ static int mdt_intent_getattr(enum mdt_it_code opcode,
                 GOTO(out_ucred, rc = ELDLM_LOCK_ABORTED);
         }
 
-	rc = mdt_intent_lock_replace(info, lockp, lhc, flags);
+	rc = mdt_intent_lock_replace(info, lockp, lhc, flags, rc);
         EXIT;
 out_ucred:
         mdt_exit_ucred(info);
@@ -3409,7 +3469,7 @@ out_obj:
 	mdt_object_put(info->mti_env, obj);
 
 	if (rc == 0 && lustre_handle_is_used(&lhc->mlh_reg_lh))
-		rc = mdt_intent_lock_replace(info, lockp, lhc, flags);
+		rc = mdt_intent_lock_replace(info, lockp, lhc, flags, rc);
 
 out:
 	lhc->mlh_reg_lh.cookie = 0;
@@ -3464,7 +3524,7 @@ static int mdt_intent_reint(enum mdt_it_code opcode,
 	if (lustre_handle_is_used(&lhc->mlh_reg_lh) &&
 	    (rc == 0 || rc == -MDT_EREMOTE_OPEN)) {
 		rep->lock_policy_res2 = 0;
-		rc = mdt_intent_lock_replace(info, lockp, lhc, flags);
+		rc = mdt_intent_lock_replace(info, lockp, lhc, flags, rc);
 		RETURN(rc);
 	}
 
@@ -4744,6 +4804,7 @@ static struct cfg_interop_param mdt_interop_param[] = {
 	{ "mdt.group_upcall",	NULL },
 	{ "mdt.quota_type",	NULL },
 	{ "mdd.quota_type",	NULL },
+	{ "mdt.som",		NULL },
 	{ "mdt.rootsquash",	"mdt.root_squash" },
 	{ "mdt.nosquash_nid",	"mdt.nosquash_nids" },
 	{ NULL }
@@ -5849,13 +5910,15 @@ static int mdt_iocontrol(unsigned int cmd, struct obd_export *exp, int len,
         if (rc)
                 RETURN(rc);
 
-        switch (cmd) {
-        case OBD_IOC_SYNC:
-                rc = mdt_device_sync(&env, mdt);
-                break;
-        case OBD_IOC_SET_READONLY:
-                rc = dt->dd_ops->dt_ro(&env, dt);
-                break;
+	switch (cmd) {
+	case OBD_IOC_SYNC:
+		rc = mdt_device_sync(&env, mdt);
+		break;
+	case OBD_IOC_SET_READONLY:
+		rc = dt_sync(&env, dt);
+		if (rc == 0)
+			rc = dt_ro(&env, dt);
+		break;
 	case OBD_IOC_ABORT_RECOVERY:
 		CERROR("%s: Aborting recovery for device\n", mdt_obd_name(mdt));
 		obd->obd_abort_recovery = 1;