[deliverable/linux.git] / drivers / staging / lustre / lustre / mdc / mdc_reint.c

/*
 * GPL HEADER START
 *
 * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
 *
 * This program is free software; you can redistribute it and/or modify
 * it under the terms of the GNU General Public License version 2 only,
 * as published by the Free Software Foundation.
 *
 * This program is distributed in the hope that it will be useful, but
 * WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 * General Public License version 2 for more details (a copy is included
 * in the LICENSE file that accompanied this code).
 *
 * You should have received a copy of the GNU General Public License
 * version 2 along with this program; If not, see
 * http://www.gnu.org/licenses/gpl-2.0.html
 *
 * GPL HEADER END
 */
/*
 * Copyright (c) 2002, 2010, Oracle and/or its affiliates. All rights reserved.
 * Use is subject to license terms.
 *
 * Copyright (c) 2011, 2015, Intel Corporation.
 */
/*
 * This file is part of Lustre, http://www.lustre.org/
 * Lustre is a trademark of Sun Microsystems, Inc.
 */

#define DEBUG_SUBSYSTEM S_MDC

# include <linux/module.h>
# include <linux/kernel.h>

#include "../include/obd_class.h"
#include "mdc_internal.h"
#include "../include/lustre_fid.h"

/* mdc_setattr does its own semaphore handling */
static int mdc_reint(struct ptlrpc_request *request,
		     struct mdc_rpc_lock *rpc_lock,
		     int level)
{
	int rc;

	request->rq_send_state = level;

	mdc_get_rpc_lock(rpc_lock, NULL);
	rc = ptlrpc_queue_wait(request);
	mdc_put_rpc_lock(rpc_lock, NULL);
	if (rc)
		CDEBUG(D_INFO, "error in handling %d\n", rc);
	else if (!req_capsule_server_get(&request->rq_pill, &RMF_MDT_BODY))
		rc = -EPROTO;

	return rc;
}

/* Find and cancel locally locks matched by inode @bits & @mode in the resource
 * found by @fid. Found locks are added into @cancel list. Returns the amount of
 * locks added to @cancels list.
 */
int mdc_resource_get_unused(struct obd_export *exp, const struct lu_fid *fid,
			    struct list_head *cancels, enum ldlm_mode mode,
			    __u64 bits)
{
	struct ldlm_namespace *ns = exp->exp_obd->obd_namespace;
	ldlm_policy_data_t policy = {};
	struct ldlm_res_id res_id;
	struct ldlm_resource *res;
	int count;

	/* Return, i.e. cancel nothing, only if ELC is supported (flag in
	 * export) but disabled through procfs (flag in NS).
	 *
	 * This distinguishes from a case when ELC is not supported originally,
	 * when we still want to cancel locks in advance and just cancel them
	 * locally, without sending any RPC.
	 */
	if (exp_connect_cancelset(exp) && !ns_connect_cancelset(ns))
		return 0;

	fid_build_reg_res_name(fid, &res_id);
	res = ldlm_resource_get(exp->exp_obd->obd_namespace,
				NULL, &res_id, 0, 0);
	if (IS_ERR(res))
		return 0;
	LDLM_RESOURCE_ADDREF(res);
	/* Initialize ibits lock policy. */
	policy.l_inodebits.bits = bits;
	count = ldlm_cancel_resource_local(res, cancels, &policy,
					   mode, 0, 0, NULL);
	LDLM_RESOURCE_DELREF(res);
	ldlm_resource_putref(res);
	return count;
}

int mdc_setattr(struct obd_export *exp, struct md_op_data *op_data,
		void *ea, int ealen, void *ea2, int ea2len,
		struct ptlrpc_request **request, struct md_open_data **mod)
{
	LIST_HEAD(cancels);
	struct ptlrpc_request *req;
	struct mdc_rpc_lock *rpc_lock;
	struct obd_device *obd = exp->exp_obd;
	int count = 0, rc;
	__u64 bits;

	bits = MDS_INODELOCK_UPDATE;
	if (op_data->op_attr.ia_valid & (ATTR_MODE | ATTR_UID | ATTR_GID))
		bits |= MDS_INODELOCK_LOOKUP;
	if ((op_data->op_flags & MF_MDC_CANCEL_FID1) &&
	    (fid_is_sane(&op_data->op_fid1)) &&
	    !OBD_FAIL_CHECK(OBD_FAIL_LDLM_BL_CALLBACK_NET))
		count = mdc_resource_get_unused(exp, &op_data->op_fid1,
						&cancels, LCK_EX, bits);
	req = ptlrpc_request_alloc(class_exp2cliimp(exp),
				   &RQF_MDS_REINT_SETATTR);
	if (!req) {
		ldlm_lock_list_put(&cancels, l_bl_ast, count);
		return -ENOMEM;
	}
	if ((op_data->op_flags & (MF_SOM_CHANGE | MF_EPOCH_OPEN)) == 0)
		req_capsule_set_size(&req->rq_pill, &RMF_MDT_EPOCH, RCL_CLIENT,
				     0);
	req_capsule_set_size(&req->rq_pill, &RMF_EADATA, RCL_CLIENT, ealen);
	req_capsule_set_size(&req->rq_pill, &RMF_LOGCOOKIES, RCL_CLIENT,
			     ea2len);

	rc = mdc_prep_elc_req(exp, req, MDS_REINT, &cancels, count);
	if (rc) {
		ptlrpc_request_free(req);
		return rc;
	}

	rpc_lock = obd->u.cli.cl_rpc_lock;

	if (op_data->op_attr.ia_valid & (ATTR_MTIME | ATTR_CTIME))
		CDEBUG(D_INODE, "setting mtime %ld, ctime %ld\n",
		       LTIME_S(op_data->op_attr.ia_mtime),
		       LTIME_S(op_data->op_attr.ia_ctime));
	mdc_setattr_pack(req, op_data, ea, ealen, ea2, ea2len);

	ptlrpc_request_set_replen(req);
	if (mod && (op_data->op_flags & MF_EPOCH_OPEN) &&
	    req->rq_import->imp_replayable) {
		LASSERT(!*mod);

		*mod = obd_mod_alloc();
		if (!*mod) {
			DEBUG_REQ(D_ERROR, req, "Can't allocate md_open_data");
		} else {
			req->rq_replay = 1;
			req->rq_cb_data = *mod;
			(*mod)->mod_open_req = req;
			req->rq_commit_cb = mdc_commit_open;
			(*mod)->mod_is_create = true;
			/**
			 * Take an extra reference on \var mod, it protects \var
			 * mod from being freed on eviction (commit callback is
			 * called despite rq_replay flag).
			 * Will be put on mdc_done_writing().
			 */
			obd_mod_get(*mod);
		}
	}

	rc = mdc_reint(req, rpc_lock, LUSTRE_IMP_FULL);

	/* Save the obtained info in the original RPC for the replay case. */
	if (rc == 0 && (op_data->op_flags & MF_EPOCH_OPEN)) {
		struct mdt_ioepoch *epoch;
		struct mdt_body  *body;

		epoch = req_capsule_client_get(&req->rq_pill, &RMF_MDT_EPOCH);
		body = req_capsule_server_get(&req->rq_pill, &RMF_MDT_BODY);
		epoch->handle = body->mbo_handle;
		epoch->ioepoch = body->mbo_ioepoch;
		req->rq_replay_cb = mdc_replay_open;
	/** bug 3633, open may be committed and estale answer is not error */
	} else if (rc == -ESTALE && (op_data->op_flags & MF_SOM_CHANGE)) {
		rc = 0;
	} else if (rc == -ERESTARTSYS) {
		rc = 0;
	}
	*request = req;
	if (rc && req->rq_commit_cb) {
		/* Put an extra reference on \var mod on error case. */
		if (mod && *mod)
			obd_mod_put(*mod);
		req->rq_commit_cb(req);
	}
	return rc;
}

int mdc_create(struct obd_export *exp, struct md_op_data *op_data,
	       const void *data, int datalen, int mode, __u32 uid, __u32 gid,
	       cfs_cap_t cap_effective, __u64 rdev,
	       struct ptlrpc_request **request)
{
	struct ptlrpc_request *req;
	int level, rc;
	int count, resends = 0;
	struct obd_import *import = exp->exp_obd->u.cli.cl_import;
	int generation = import->imp_generation;
	LIST_HEAD(cancels);

	/* For case if upper layer did not alloc fid, do it now. */
	if (!fid_is_sane(&op_data->op_fid2)) {
		/*
		 * mdc_fid_alloc() may return errno 1 in case of switch to new
		 * sequence, handle this.
		 */
		rc = mdc_fid_alloc(NULL, exp, &op_data->op_fid2, op_data);
		if (rc < 0)
			return rc;
	}

rebuild:
	count = 0;
	if ((op_data->op_flags & MF_MDC_CANCEL_FID1) &&
	    (fid_is_sane(&op_data->op_fid1)))
		count = mdc_resource_get_unused(exp, &op_data->op_fid1,
						&cancels, LCK_EX,
						MDS_INODELOCK_UPDATE);

	req = ptlrpc_request_alloc(class_exp2cliimp(exp),
				   &RQF_MDS_REINT_CREATE_ACL);
	if (!req) {
		ldlm_lock_list_put(&cancels, l_bl_ast, count);
		return -ENOMEM;
	}
	req_capsule_set_size(&req->rq_pill, &RMF_NAME, RCL_CLIENT,
			     op_data->op_namelen + 1);
	req_capsule_set_size(&req->rq_pill, &RMF_EADATA, RCL_CLIENT,
			     data && datalen ? datalen : 0);

	rc = mdc_prep_elc_req(exp, req, MDS_REINT, &cancels, count);
	if (rc) {
		ptlrpc_request_free(req);
		return rc;
	}

	/*
	 * mdc_create_pack() fills msg->bufs[1] with name and msg->bufs[2] with
	 * tgt, for symlinks or lov MD data.
	 */
	mdc_create_pack(req, op_data, data, datalen, mode, uid,
			gid, cap_effective, rdev);

	ptlrpc_request_set_replen(req);

	/* ask ptlrpc not to resend on EINPROGRESS since we have our own retry
	 * logic here
	 */
	req->rq_no_retry_einprogress = 1;

	if (resends) {
		req->rq_generation_set = 1;
		req->rq_import_generation = generation;
		req->rq_sent = ktime_get_real_seconds() + resends;
	}
	level = LUSTRE_IMP_FULL;
 resend:
	rc = mdc_reint(req, exp->exp_obd->u.cli.cl_rpc_lock, level);

	/* Resend if we were told to. */
	if (rc == -ERESTARTSYS) {
		level = LUSTRE_IMP_RECOVER;
		goto resend;
	} else if (rc == -EINPROGRESS) {
		/* Retry create infinitely until succeed or get other
		 * error code.
		 */
		ptlrpc_req_finished(req);
		resends++;

		CDEBUG(D_HA, "%s: resend:%d create on "DFID"/"DFID"\n",
		       exp->exp_obd->obd_name, resends,
		       PFID(&op_data->op_fid1), PFID(&op_data->op_fid2));

		if (generation == import->imp_generation) {
			goto rebuild;
		} else {
			CDEBUG(D_HA, "resend cross eviction\n");
			return -EIO;
		}
	}

	*request = req;
	return rc;
}

int mdc_unlink(struct obd_export *exp, struct md_op_data *op_data,
	       struct ptlrpc_request **request)
{
	LIST_HEAD(cancels);
	struct obd_device *obd = class_exp2obd(exp);
	struct ptlrpc_request *req = *request;
	int count = 0, rc;

	LASSERT(!req);

	if ((op_data->op_flags & MF_MDC_CANCEL_FID1) &&
	    (fid_is_sane(&op_data->op_fid1)) &&
	    !OBD_FAIL_CHECK(OBD_FAIL_LDLM_BL_CALLBACK_NET))
		count = mdc_resource_get_unused(exp, &op_data->op_fid1,
						&cancels, LCK_EX,
						MDS_INODELOCK_UPDATE);
	if ((op_data->op_flags & MF_MDC_CANCEL_FID3) &&
	    (fid_is_sane(&op_data->op_fid3)) &&
	    !OBD_FAIL_CHECK(OBD_FAIL_LDLM_BL_CALLBACK_NET))
		count += mdc_resource_get_unused(exp, &op_data->op_fid3,
						 &cancels, LCK_EX,
						 MDS_INODELOCK_FULL);
	req = ptlrpc_request_alloc(class_exp2cliimp(exp),
				   &RQF_MDS_REINT_UNLINK);
	if (!req) {
		ldlm_lock_list_put(&cancels, l_bl_ast, count);
		return -ENOMEM;
	}
	req_capsule_set_size(&req->rq_pill, &RMF_NAME, RCL_CLIENT,
			     op_data->op_namelen + 1);

	rc = mdc_prep_elc_req(exp, req, MDS_REINT, &cancels, count);
	if (rc) {
		ptlrpc_request_free(req);
		return rc;
	}

	mdc_unlink_pack(req, op_data);

	req_capsule_set_size(&req->rq_pill, &RMF_MDT_MD, RCL_SERVER,
			     obd->u.cli.cl_default_mds_easize);
	req_capsule_set_size(&req->rq_pill, &RMF_LOGCOOKIES, RCL_SERVER,
			     obd->u.cli.cl_default_mds_cookiesize);
	ptlrpc_request_set_replen(req);

	*request = req;

	rc = mdc_reint(req, obd->u.cli.cl_rpc_lock, LUSTRE_IMP_FULL);
	if (rc == -ERESTARTSYS)
		rc = 0;
	return rc;
}

int mdc_link(struct obd_export *exp, struct md_op_data *op_data,
	     struct ptlrpc_request **request)
{
	LIST_HEAD(cancels);
	struct obd_device *obd = exp->exp_obd;
	struct ptlrpc_request *req;
	int count = 0, rc;

	if ((op_data->op_flags & MF_MDC_CANCEL_FID2) &&
	    (fid_is_sane(&op_data->op_fid2)))
		count = mdc_resource_get_unused(exp, &op_data->op_fid2,
						&cancels, LCK_EX,
						MDS_INODELOCK_UPDATE);
	if ((op_data->op_flags & MF_MDC_CANCEL_FID1) &&
	    (fid_is_sane(&op_data->op_fid1)))
		count += mdc_resource_get_unused(exp, &op_data->op_fid1,
						 &cancels, LCK_EX,
						 MDS_INODELOCK_UPDATE);

	req = ptlrpc_request_alloc(class_exp2cliimp(exp), &RQF_MDS_REINT_LINK);
	if (!req) {
		ldlm_lock_list_put(&cancels, l_bl_ast, count);
		return -ENOMEM;
	}
	req_capsule_set_size(&req->rq_pill, &RMF_NAME, RCL_CLIENT,
			     op_data->op_namelen + 1);

	rc = mdc_prep_elc_req(exp, req, MDS_REINT, &cancels, count);
	if (rc) {
		ptlrpc_request_free(req);
		return rc;
	}

	mdc_link_pack(req, op_data);
	ptlrpc_request_set_replen(req);

	rc = mdc_reint(req, obd->u.cli.cl_rpc_lock, LUSTRE_IMP_FULL);
	*request = req;
	if (rc == -ERESTARTSYS)
		rc = 0;

	return rc;
}

int mdc_rename(struct obd_export *exp, struct md_op_data *op_data,
	       const char *old, int oldlen, const char *new, int newlen,
	       struct ptlrpc_request **request)
{
	LIST_HEAD(cancels);
	struct obd_device *obd = exp->exp_obd;
	struct ptlrpc_request *req;
	int count = 0, rc;

	if ((op_data->op_flags & MF_MDC_CANCEL_FID1) &&
	    (fid_is_sane(&op_data->op_fid1)))
		count = mdc_resource_get_unused(exp, &op_data->op_fid1,
						&cancels, LCK_EX,
						MDS_INODELOCK_UPDATE);
	if ((op_data->op_flags & MF_MDC_CANCEL_FID2) &&
	    (fid_is_sane(&op_data->op_fid2)))
		count += mdc_resource_get_unused(exp, &op_data->op_fid2,
						 &cancels, LCK_EX,
						 MDS_INODELOCK_UPDATE);
	if ((op_data->op_flags & MF_MDC_CANCEL_FID3) &&
	    (fid_is_sane(&op_data->op_fid3)))
		count += mdc_resource_get_unused(exp, &op_data->op_fid3,
						 &cancels, LCK_EX,
						 MDS_INODELOCK_LOOKUP);
	if ((op_data->op_flags & MF_MDC_CANCEL_FID4) &&
	    (fid_is_sane(&op_data->op_fid4)))
		count += mdc_resource_get_unused(exp, &op_data->op_fid4,
						 &cancels, LCK_EX,
						 MDS_INODELOCK_FULL);

	req = ptlrpc_request_alloc(class_exp2cliimp(exp),
				   &RQF_MDS_REINT_RENAME);
	if (!req) {
		ldlm_lock_list_put(&cancels, l_bl_ast, count);
		return -ENOMEM;
	}

	req_capsule_set_size(&req->rq_pill, &RMF_NAME, RCL_CLIENT, oldlen + 1);
	req_capsule_set_size(&req->rq_pill, &RMF_SYMTGT, RCL_CLIENT,
			     newlen + 1);

	rc = mdc_prep_elc_req(exp, req, MDS_REINT, &cancels, count);
	if (rc) {
		ptlrpc_request_free(req);
		return rc;
	}

	if (exp_connect_cancelset(exp) && req)
		ldlm_cli_cancel_list(&cancels, count, req, 0);

	mdc_rename_pack(req, op_data, old, oldlen, new, newlen);

	req_capsule_set_size(&req->rq_pill, &RMF_MDT_MD, RCL_SERVER,
			     obd->u.cli.cl_default_mds_easize);
	req_capsule_set_size(&req->rq_pill, &RMF_LOGCOOKIES, RCL_SERVER,
			     obd->u.cli.cl_default_mds_cookiesize);
	ptlrpc_request_set_replen(req);

	rc = mdc_reint(req, obd->u.cli.cl_rpc_lock, LUSTRE_IMP_FULL);
	*request = req;
	if (rc == -ERESTARTSYS)
		rc = 0;

	return rc;
}
Commit	Line	Data
d7e09d03 PT	1	/*
	2	* GPL HEADER START
	3	*
	4	* DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
	5	*
	6	* This program is free software; you can redistribute it and/or modify
	7	* it under the terms of the GNU General Public License version 2 only,
	8	* as published by the Free Software Foundation.
	9	*
	10	* This program is distributed in the hope that it will be useful, but
	11	* WITHOUT ANY WARRANTY; without even the implied warranty of
	12	* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
	13	* General Public License version 2 for more details (a copy is included
	14	* in the LICENSE file that accompanied this code).
	15	*
	16	* You should have received a copy of the GNU General Public License
	17	* version 2 along with this program; If not, see
6a5b99a4	18	* http://www.gnu.org/licenses/gpl-2.0.html
d7e09d03	19	*
d7e09d03 PT	20	* GPL HEADER END
	21	*/
	22	/*
	23	* Copyright (c) 2002, 2010, Oracle and/or its affiliates. All rights reserved.
	24	* Use is subject to license terms.
	25	*
1dc563a6	26	* Copyright (c) 2011, 2015, Intel Corporation.
d7e09d03 PT	27	*/
	28	/*
	29	* This file is part of Lustre, http://www.lustre.org/
	30	* Lustre is a trademark of Sun Microsystems, Inc.
	31	*/
	32
	33	#define DEBUG_SUBSYSTEM S_MDC
	34
	35	# include <linux/module.h>
	36	# include <linux/kernel.h>
	37
05932307	38	#include "../include/obd_class.h"
d7e09d03	39	#include "mdc_internal.h"
05932307	40	#include "../include/lustre_fid.h"
d7e09d03 PT	41
	42	/* mdc_setattr does its own semaphore handling */
	43	static int mdc_reint(struct ptlrpc_request *request,
	44	struct mdc_rpc_lock *rpc_lock,
	45	int level)
	46	{
	47	int rc;
	48
	49	request->rq_send_state = level;
	50
	51	mdc_get_rpc_lock(rpc_lock, NULL);
	52	rc = ptlrpc_queue_wait(request);
	53	mdc_put_rpc_lock(rpc_lock, NULL);
	54	if (rc)
	55	CDEBUG(D_INFO, "error in handling %d\n", rc);
22403d9e	56	else if (!req_capsule_server_get(&request->rq_pill, &RMF_MDT_BODY))
d7e09d03	57	rc = -EPROTO;
22403d9e	58
d7e09d03 PT	59	return rc;
	60	}
	61
	62	/* Find and cancel locally locks matched by inode @bits & @mode in the resource
	63	* found by @fid. Found locks are added into @cancel list. Returns the amount of
1df232ee OD	64	* locks added to @cancels list.
1df232ee OD	65	*/
e93a3082	66	int mdc_resource_get_unused(struct obd_export exp, const struct lu_fid fid,
52ee0d20	67	struct list_head *cancels, enum ldlm_mode mode,
d7e09d03 PT	68	__u64 bits)
	69	{
	70	struct ldlm_namespace *ns = exp->exp_obd->obd_namespace;
301af906	71	ldlm_policy_data_t policy = {};
d7e09d03 PT	72	struct ldlm_res_id res_id;
	73	struct ldlm_resource *res;
	74	int count;
d7e09d03 PT	75
	76	/* Return, i.e. cancel nothing, only if ELC is supported (flag in
	77	* export) but disabled through procfs (flag in NS).
	78	*
	79	* This distinguishes from a case when ELC is not supported originally,
	80	* when we still want to cancel locks in advance and just cancel them
1df232ee OD	81	* locally, without sending any RPC.
1df232ee OD	82	*/
d7e09d03	83	if (exp_connect_cancelset(exp) && !ns_connect_cancelset(ns))
0a3bdb00	84	return 0;
d7e09d03 PT	85
	86	fid_build_reg_res_name(fid, &res_id);
	87	res = ldlm_resource_get(exp->exp_obd->obd_namespace,
	88	NULL, &res_id, 0, 0);
099d5adf	89	if (IS_ERR(res))
0a3bdb00	90	return 0;
d7e09d03 PT	91	LDLM_RESOURCE_ADDREF(res);
	92	/* Initialize ibits lock policy. */
	93	policy.l_inodebits.bits = bits;
	94	count = ldlm_cancel_resource_local(res, cancels, &policy,
	95	mode, 0, 0, NULL);
	96	LDLM_RESOURCE_DELREF(res);
	97	ldlm_resource_putref(res);
0a3bdb00	98	return count;
d7e09d03 PT	99	}
	100
	101	int mdc_setattr(struct obd_export exp, struct md_op_data op_data,
	102	void ea, int ealen, void ea2, int ea2len,
	103	struct ptlrpc_request request, struct md_open_data mod)
	104	{
	105	LIST_HEAD(cancels);
	106	struct ptlrpc_request *req;
	107	struct mdc_rpc_lock *rpc_lock;
	108	struct obd_device *obd = exp->exp_obd;
	109	int count = 0, rc;
	110	__u64 bits;
d7e09d03	111
d7e09d03	112	bits = MDS_INODELOCK_UPDATE;
cd94f231	113	if (op_data->op_attr.ia_valid & (ATTR_MODE \| ATTR_UID \| ATTR_GID))
d7e09d03 PT	114	bits \|= MDS_INODELOCK_LOOKUP;
	115	if ((op_data->op_flags & MF_MDC_CANCEL_FID1) &&
	116	(fid_is_sane(&op_data->op_fid1)) &&
	117	!OBD_FAIL_CHECK(OBD_FAIL_LDLM_BL_CALLBACK_NET))
	118	count = mdc_resource_get_unused(exp, &op_data->op_fid1,
	119	&cancels, LCK_EX, bits);
	120	req = ptlrpc_request_alloc(class_exp2cliimp(exp),
	121	&RQF_MDS_REINT_SETATTR);
34e3ff96	122	if (!req) {
d7e09d03	123	ldlm_lock_list_put(&cancels, l_bl_ast, count);
0a3bdb00	124	return -ENOMEM;
d7e09d03	125	}
d7e09d03 PT	126	if ((op_data->op_flags & (MF_SOM_CHANGE \| MF_EPOCH_OPEN)) == 0)
	127	req_capsule_set_size(&req->rq_pill, &RMF_MDT_EPOCH, RCL_CLIENT,
	128	0);
	129	req_capsule_set_size(&req->rq_pill, &RMF_EADATA, RCL_CLIENT, ealen);
	130	req_capsule_set_size(&req->rq_pill, &RMF_LOGCOOKIES, RCL_CLIENT,
	131	ea2len);
	132
	133	rc = mdc_prep_elc_req(exp, req, MDS_REINT, &cancels, count);
	134	if (rc) {
	135	ptlrpc_request_free(req);
0a3bdb00	136	return rc;
d7e09d03 PT	137	}
	138
	139	rpc_lock = obd->u.cli.cl_rpc_lock;
	140
	141	if (op_data->op_attr.ia_valid & (ATTR_MTIME \| ATTR_CTIME))
93d3a405	142	CDEBUG(D_INODE, "setting mtime %ld, ctime %ld\n",
d7e09d03 PT	143	LTIME_S(op_data->op_attr.ia_mtime),
	144	LTIME_S(op_data->op_attr.ia_ctime));
	145	mdc_setattr_pack(req, op_data, ea, ealen, ea2, ea2len);
	146
	147	ptlrpc_request_set_replen(req);
	148	if (mod && (op_data->op_flags & MF_EPOCH_OPEN) &&
1a4cd3e9	149	req->rq_import->imp_replayable) {
34e3ff96	150	LASSERT(!*mod);
d7e09d03 PT	151
d7e09d03 PT	152	*mod = obd_mod_alloc();
34e3ff96	153	if (!*mod) {
ee990b33	154	DEBUG_REQ(D_ERROR, req, "Can't allocate md_open_data");
d7e09d03 PT	155	} else {
	156	req->rq_replay = 1;
	157	req->rq_cb_data = *mod;
	158	(*mod)->mod_open_req = req;
	159	req->rq_commit_cb = mdc_commit_open;
63d42578	160	(*mod)->mod_is_create = true;
d7e09d03 PT	161	/**
	162	* Take an extra reference on \var mod, it protects \var
	163	* mod from being freed on eviction (commit callback is
	164	* called despite rq_replay flag).
	165	* Will be put on mdc_done_writing().
	166	*/
	167	obd_mod_get(*mod);
	168	}
	169	}
	170
	171	rc = mdc_reint(req, rpc_lock, LUSTRE_IMP_FULL);
	172
	173	/* Save the obtained info in the original RPC for the replay case. */
	174	if (rc == 0 && (op_data->op_flags & MF_EPOCH_OPEN)) {
	175	struct mdt_ioepoch *epoch;
	176	struct mdt_body *body;
	177
	178	epoch = req_capsule_client_get(&req->rq_pill, &RMF_MDT_EPOCH);
	179	body = req_capsule_server_get(&req->rq_pill, &RMF_MDT_BODY);
2e1b5b8b JH	180	epoch->handle = body->mbo_handle;
2e1b5b8b JH	181	epoch->ioepoch = body->mbo_ioepoch;
d7e09d03 PT	182	req->rq_replay_cb = mdc_replay_open;
	183	/** bug 3633, open may be committed and estale answer is not error */
	184	} else if (rc == -ESTALE && (op_data->op_flags & MF_SOM_CHANGE)) {
	185	rc = 0;
	186	} else if (rc == -ERESTARTSYS) {
	187	rc = 0;
	188	}
	189	*request = req;
	190	if (rc && req->rq_commit_cb) {
	191	/* Put an extra reference on \var mod on error case. */
34e3ff96	192	if (mod && *mod)
ba0e1b53	193	obd_mod_put(*mod);
d7e09d03 PT	194	req->rq_commit_cb(req);
d7e09d03 PT	195	}
0a3bdb00	196	return rc;
d7e09d03 PT	197	}
	198
	199	int mdc_create(struct obd_export exp, struct md_op_data op_data,
	200	const void *data, int datalen, int mode, __u32 uid, __u32 gid,
	201	cfs_cap_t cap_effective, __u64 rdev,
	202	struct ptlrpc_request **request)
	203	{
	204	struct ptlrpc_request *req;
	205	int level, rc;
	206	int count, resends = 0;
	207	struct obd_import *import = exp->exp_obd->u.cli.cl_import;
	208	int generation = import->imp_generation;
	209	LIST_HEAD(cancels);
d7e09d03 PT	210
	211	/* For case if upper layer did not alloc fid, do it now. */
	212	if (!fid_is_sane(&op_data->op_fid2)) {
	213	/*
	214	* mdc_fid_alloc() may return errno 1 in case of switch to new
	215	* sequence, handle this.
	216	*/
8f18c8a4	217	rc = mdc_fid_alloc(NULL, exp, &op_data->op_fid2, op_data);
8f18c8a4	218	if (rc < 0)
0a3bdb00	219	return rc;
d7e09d03 PT	220	}
	221
	222	rebuild:
	223	count = 0;
	224	if ((op_data->op_flags & MF_MDC_CANCEL_FID1) &&
	225	(fid_is_sane(&op_data->op_fid1)))
	226	count = mdc_resource_get_unused(exp, &op_data->op_fid1,
	227	&cancels, LCK_EX,
	228	MDS_INODELOCK_UPDATE);
	229
	230	req = ptlrpc_request_alloc(class_exp2cliimp(exp),
341f1f0a	231	&RQF_MDS_REINT_CREATE_ACL);
34e3ff96	232	if (!req) {
d7e09d03	233	ldlm_lock_list_put(&cancels, l_bl_ast, count);
0a3bdb00	234	return -ENOMEM;
d7e09d03	235	}
d7e09d03 PT	236	req_capsule_set_size(&req->rq_pill, &RMF_NAME, RCL_CLIENT,
	237	op_data->op_namelen + 1);
	238	req_capsule_set_size(&req->rq_pill, &RMF_EADATA, RCL_CLIENT,
	239	data && datalen ? datalen : 0);
	240
	241	rc = mdc_prep_elc_req(exp, req, MDS_REINT, &cancels, count);
	242	if (rc) {
	243	ptlrpc_request_free(req);
0a3bdb00	244	return rc;
d7e09d03 PT	245	}
	246
	247	/*
	248	* mdc_create_pack() fills msg->bufs[1] with name and msg->bufs[2] with
	249	* tgt, for symlinks or lov MD data.
	250	*/
	251	mdc_create_pack(req, op_data, data, datalen, mode, uid,
	252	gid, cap_effective, rdev);
	253
	254	ptlrpc_request_set_replen(req);
	255
	256	/* ask ptlrpc not to resend on EINPROGRESS since we have our own retry
1df232ee OD	257	* logic here
1df232ee OD	258	*/
d7e09d03 PT	259	req->rq_no_retry_einprogress = 1;
	260
	261	if (resends) {
	262	req->rq_generation_set = 1;
	263	req->rq_import_generation = generation;
219e6de6	264	req->rq_sent = ktime_get_real_seconds() + resends;
d7e09d03 PT	265	}
	266	level = LUSTRE_IMP_FULL;
	267	resend:
	268	rc = mdc_reint(req, exp->exp_obd->u.cli.cl_rpc_lock, level);
	269
	270	/* Resend if we were told to. */
	271	if (rc == -ERESTARTSYS) {
	272	level = LUSTRE_IMP_RECOVER;
	273	goto resend;
	274	} else if (rc == -EINPROGRESS) {
	275	/* Retry create infinitely until succeed or get other
1df232ee OD	276	* error code.
1df232ee OD	277	*/
d7e09d03 PT	278	ptlrpc_req_finished(req);
	279	resends++;
	280
	281	CDEBUG(D_HA, "%s: resend:%d create on "DFID"/"DFID"\n",
	282	exp->exp_obd->obd_name, resends,
	283	PFID(&op_data->op_fid1), PFID(&op_data->op_fid2));
	284
	285	if (generation == import->imp_generation) {
	286	goto rebuild;
	287	} else {
	288	CDEBUG(D_HA, "resend cross eviction\n");
0a3bdb00	289	return -EIO;
d7e09d03	290	}
d7e09d03 PT	291	}
	292
	293	*request = req;
0a3bdb00	294	return rc;
d7e09d03 PT	295	}
	296
	297	int mdc_unlink(struct obd_export exp, struct md_op_data op_data,
	298	struct ptlrpc_request **request)
	299	{
	300	LIST_HEAD(cancels);
	301	struct obd_device *obd = class_exp2obd(exp);
	302	struct ptlrpc_request req = request;
	303	int count = 0, rc;
d7e09d03	304
34e3ff96	305	LASSERT(!req);
d7e09d03 PT	306
	307	if ((op_data->op_flags & MF_MDC_CANCEL_FID1) &&
	308	(fid_is_sane(&op_data->op_fid1)) &&
	309	!OBD_FAIL_CHECK(OBD_FAIL_LDLM_BL_CALLBACK_NET))
	310	count = mdc_resource_get_unused(exp, &op_data->op_fid1,
	311	&cancels, LCK_EX,
	312	MDS_INODELOCK_UPDATE);
	313	if ((op_data->op_flags & MF_MDC_CANCEL_FID3) &&
	314	(fid_is_sane(&op_data->op_fid3)) &&
	315	!OBD_FAIL_CHECK(OBD_FAIL_LDLM_BL_CALLBACK_NET))
	316	count += mdc_resource_get_unused(exp, &op_data->op_fid3,
	317	&cancels, LCK_EX,
	318	MDS_INODELOCK_FULL);
	319	req = ptlrpc_request_alloc(class_exp2cliimp(exp),
	320	&RQF_MDS_REINT_UNLINK);
34e3ff96	321	if (!req) {
d7e09d03	322	ldlm_lock_list_put(&cancels, l_bl_ast, count);
0a3bdb00	323	return -ENOMEM;
d7e09d03	324	}
d7e09d03 PT	325	req_capsule_set_size(&req->rq_pill, &RMF_NAME, RCL_CLIENT,
	326	op_data->op_namelen + 1);
	327
	328	rc = mdc_prep_elc_req(exp, req, MDS_REINT, &cancels, count);
	329	if (rc) {
	330	ptlrpc_request_free(req);
0a3bdb00	331	return rc;
d7e09d03 PT	332	}
	333
	334	mdc_unlink_pack(req, op_data);
	335
	336	req_capsule_set_size(&req->rq_pill, &RMF_MDT_MD, RCL_SERVER,
44779340	337	obd->u.cli.cl_default_mds_easize);
d7e09d03	338	req_capsule_set_size(&req->rq_pill, &RMF_LOGCOOKIES, RCL_SERVER,
44779340	339	obd->u.cli.cl_default_mds_cookiesize);
d7e09d03 PT	340	ptlrpc_request_set_replen(req);
	341
	342	*request = req;
	343
	344	rc = mdc_reint(req, obd->u.cli.cl_rpc_lock, LUSTRE_IMP_FULL);
	345	if (rc == -ERESTARTSYS)
	346	rc = 0;
0a3bdb00	347	return rc;
d7e09d03 PT	348	}
	349
	350	int mdc_link(struct obd_export exp, struct md_op_data op_data,
	351	struct ptlrpc_request **request)
	352	{
	353	LIST_HEAD(cancels);
	354	struct obd_device *obd = exp->exp_obd;
	355	struct ptlrpc_request *req;
	356	int count = 0, rc;
d7e09d03 PT	357
	358	if ((op_data->op_flags & MF_MDC_CANCEL_FID2) &&
	359	(fid_is_sane(&op_data->op_fid2)))
	360	count = mdc_resource_get_unused(exp, &op_data->op_fid2,
	361	&cancels, LCK_EX,
	362	MDS_INODELOCK_UPDATE);
	363	if ((op_data->op_flags & MF_MDC_CANCEL_FID1) &&
	364	(fid_is_sane(&op_data->op_fid1)))
	365	count += mdc_resource_get_unused(exp, &op_data->op_fid1,
	366	&cancels, LCK_EX,
	367	MDS_INODELOCK_UPDATE);
	368
	369	req = ptlrpc_request_alloc(class_exp2cliimp(exp), &RQF_MDS_REINT_LINK);
34e3ff96	370	if (!req) {
d7e09d03	371	ldlm_lock_list_put(&cancels, l_bl_ast, count);
0a3bdb00	372	return -ENOMEM;
d7e09d03	373	}
d7e09d03 PT	374	req_capsule_set_size(&req->rq_pill, &RMF_NAME, RCL_CLIENT,
	375	op_data->op_namelen + 1);
	376
	377	rc = mdc_prep_elc_req(exp, req, MDS_REINT, &cancels, count);
	378	if (rc) {
	379	ptlrpc_request_free(req);
0a3bdb00	380	return rc;
d7e09d03 PT	381	}
	382
	383	mdc_link_pack(req, op_data);
	384	ptlrpc_request_set_replen(req);
	385
	386	rc = mdc_reint(req, obd->u.cli.cl_rpc_lock, LUSTRE_IMP_FULL);
	387	*request = req;
	388	if (rc == -ERESTARTSYS)
	389	rc = 0;
	390
0a3bdb00	391	return rc;
d7e09d03 PT	392	}
	393
	394	int mdc_rename(struct obd_export exp, struct md_op_data op_data,
	395	const char old, int oldlen, const char new, int newlen,
	396	struct ptlrpc_request **request)
	397	{
	398	LIST_HEAD(cancels);
	399	struct obd_device *obd = exp->exp_obd;
	400	struct ptlrpc_request *req;
	401	int count = 0, rc;
d7e09d03 PT	402
	403	if ((op_data->op_flags & MF_MDC_CANCEL_FID1) &&
	404	(fid_is_sane(&op_data->op_fid1)))
	405	count = mdc_resource_get_unused(exp, &op_data->op_fid1,
	406	&cancels, LCK_EX,
	407	MDS_INODELOCK_UPDATE);
	408	if ((op_data->op_flags & MF_MDC_CANCEL_FID2) &&
	409	(fid_is_sane(&op_data->op_fid2)))
	410	count += mdc_resource_get_unused(exp, &op_data->op_fid2,
	411	&cancels, LCK_EX,
	412	MDS_INODELOCK_UPDATE);
	413	if ((op_data->op_flags & MF_MDC_CANCEL_FID3) &&
	414	(fid_is_sane(&op_data->op_fid3)))
	415	count += mdc_resource_get_unused(exp, &op_data->op_fid3,
	416	&cancels, LCK_EX,
	417	MDS_INODELOCK_LOOKUP);
	418	if ((op_data->op_flags & MF_MDC_CANCEL_FID4) &&
22e0bc6a	419	(fid_is_sane(&op_data->op_fid4)))
d7e09d03 PT	420	count += mdc_resource_get_unused(exp, &op_data->op_fid4,
	421	&cancels, LCK_EX,
	422	MDS_INODELOCK_FULL);
	423
	424	req = ptlrpc_request_alloc(class_exp2cliimp(exp),
	425	&RQF_MDS_REINT_RENAME);
34e3ff96	426	if (!req) {
d7e09d03	427	ldlm_lock_list_put(&cancels, l_bl_ast, count);
0a3bdb00	428	return -ENOMEM;
d7e09d03 PT	429	}
d7e09d03 PT	430
d7e09d03	431	req_capsule_set_size(&req->rq_pill, &RMF_NAME, RCL_CLIENT, oldlen + 1);
cd94f231 OD	432	req_capsule_set_size(&req->rq_pill, &RMF_SYMTGT, RCL_CLIENT,
cd94f231 OD	433	newlen + 1);
d7e09d03 PT	434
	435	rc = mdc_prep_elc_req(exp, req, MDS_REINT, &cancels, count);
	436	if (rc) {
	437	ptlrpc_request_free(req);
0a3bdb00	438	return rc;
d7e09d03 PT	439	}
	440
	441	if (exp_connect_cancelset(exp) && req)
	442	ldlm_cli_cancel_list(&cancels, count, req, 0);
	443
	444	mdc_rename_pack(req, op_data, old, oldlen, new, newlen);
	445
	446	req_capsule_set_size(&req->rq_pill, &RMF_MDT_MD, RCL_SERVER,
44779340	447	obd->u.cli.cl_default_mds_easize);
d7e09d03	448	req_capsule_set_size(&req->rq_pill, &RMF_LOGCOOKIES, RCL_SERVER,
44779340	449	obd->u.cli.cl_default_mds_cookiesize);
d7e09d03 PT	450	ptlrpc_request_set_replen(req);
	451
	452	rc = mdc_reint(req, obd->u.cli.cl_rpc_lock, LUSTRE_IMP_FULL);
	453	*request = req;
	454	if (rc == -ERESTARTSYS)
	455	rc = 0;
	456
0a3bdb00	457	return rc;
d7e09d03	458	}