[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 (!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->handle;
		epoch->ioepoch = body->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(exp, &op_data->op_fid2, op_data);
		if (rc < 0) {
			CERROR("Can't alloc new fid, rc %d\n", rc);
			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);
34e3ff96	89	if (!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 PT	112	bits = MDS_INODELOCK_UPDATE;
	113	if (op_data->op_attr.ia_valid & (ATTR_MODE\|ATTR_UID\|ATTR_GID))
	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);
d7e09d03 PT	180	epoch->handle = body->handle;
	181	epoch->ioepoch = body->ioepoch;
	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	*/
	217	rc = mdc_fid_alloc(exp, &op_data->op_fid2, op_data);
	218	if (rc < 0) {
	219	CERROR("Can't alloc new fid, rc %d\n", rc);
0a3bdb00	220	return rc;
d7e09d03 PT	221	}
	222	}
	223
	224	rebuild:
	225	count = 0;
	226	if ((op_data->op_flags & MF_MDC_CANCEL_FID1) &&
	227	(fid_is_sane(&op_data->op_fid1)))
	228	count = mdc_resource_get_unused(exp, &op_data->op_fid1,
	229	&cancels, LCK_EX,
	230	MDS_INODELOCK_UPDATE);
	231
	232	req = ptlrpc_request_alloc(class_exp2cliimp(exp),
341f1f0a	233	&RQF_MDS_REINT_CREATE_ACL);
34e3ff96	234	if (!req) {
d7e09d03	235	ldlm_lock_list_put(&cancels, l_bl_ast, count);
0a3bdb00	236	return -ENOMEM;
d7e09d03	237	}
d7e09d03 PT	238	req_capsule_set_size(&req->rq_pill, &RMF_NAME, RCL_CLIENT,
	239	op_data->op_namelen + 1);
	240	req_capsule_set_size(&req->rq_pill, &RMF_EADATA, RCL_CLIENT,
	241	data && datalen ? datalen : 0);
	242
	243	rc = mdc_prep_elc_req(exp, req, MDS_REINT, &cancels, count);
	244	if (rc) {
	245	ptlrpc_request_free(req);
0a3bdb00	246	return rc;
d7e09d03 PT	247	}
	248
	249	/*
	250	* mdc_create_pack() fills msg->bufs[1] with name and msg->bufs[2] with
	251	* tgt, for symlinks or lov MD data.
	252	*/
	253	mdc_create_pack(req, op_data, data, datalen, mode, uid,
	254	gid, cap_effective, rdev);
	255
	256	ptlrpc_request_set_replen(req);
	257
	258	/* ask ptlrpc not to resend on EINPROGRESS since we have our own retry
1df232ee OD	259	* logic here
1df232ee OD	260	*/
d7e09d03 PT	261	req->rq_no_retry_einprogress = 1;
	262
	263	if (resends) {
	264	req->rq_generation_set = 1;
	265	req->rq_import_generation = generation;
219e6de6	266	req->rq_sent = ktime_get_real_seconds() + resends;
d7e09d03 PT	267	}
	268	level = LUSTRE_IMP_FULL;
	269	resend:
	270	rc = mdc_reint(req, exp->exp_obd->u.cli.cl_rpc_lock, level);
	271
	272	/* Resend if we were told to. */
	273	if (rc == -ERESTARTSYS) {
	274	level = LUSTRE_IMP_RECOVER;
	275	goto resend;
	276	} else if (rc == -EINPROGRESS) {
	277	/* Retry create infinitely until succeed or get other
1df232ee OD	278	* error code.
1df232ee OD	279	*/
d7e09d03 PT	280	ptlrpc_req_finished(req);
	281	resends++;
	282
	283	CDEBUG(D_HA, "%s: resend:%d create on "DFID"/"DFID"\n",
	284	exp->exp_obd->obd_name, resends,
	285	PFID(&op_data->op_fid1), PFID(&op_data->op_fid2));
	286
	287	if (generation == import->imp_generation) {
	288	goto rebuild;
	289	} else {
	290	CDEBUG(D_HA, "resend cross eviction\n");
0a3bdb00	291	return -EIO;
d7e09d03	292	}
d7e09d03 PT	293	}
	294
	295	*request = req;
0a3bdb00	296	return rc;
d7e09d03 PT	297	}
	298
	299	int mdc_unlink(struct obd_export exp, struct md_op_data op_data,
	300	struct ptlrpc_request **request)
	301	{
	302	LIST_HEAD(cancels);
	303	struct obd_device *obd = class_exp2obd(exp);
	304	struct ptlrpc_request req = request;
	305	int count = 0, rc;
d7e09d03	306
34e3ff96	307	LASSERT(!req);
d7e09d03 PT	308
	309	if ((op_data->op_flags & MF_MDC_CANCEL_FID1) &&
	310	(fid_is_sane(&op_data->op_fid1)) &&
	311	!OBD_FAIL_CHECK(OBD_FAIL_LDLM_BL_CALLBACK_NET))
	312	count = mdc_resource_get_unused(exp, &op_data->op_fid1,
	313	&cancels, LCK_EX,
	314	MDS_INODELOCK_UPDATE);
	315	if ((op_data->op_flags & MF_MDC_CANCEL_FID3) &&
	316	(fid_is_sane(&op_data->op_fid3)) &&
	317	!OBD_FAIL_CHECK(OBD_FAIL_LDLM_BL_CALLBACK_NET))
	318	count += mdc_resource_get_unused(exp, &op_data->op_fid3,
	319	&cancels, LCK_EX,
	320	MDS_INODELOCK_FULL);
	321	req = ptlrpc_request_alloc(class_exp2cliimp(exp),
	322	&RQF_MDS_REINT_UNLINK);
34e3ff96	323	if (!req) {
d7e09d03	324	ldlm_lock_list_put(&cancels, l_bl_ast, count);
0a3bdb00	325	return -ENOMEM;
d7e09d03	326	}
d7e09d03 PT	327	req_capsule_set_size(&req->rq_pill, &RMF_NAME, RCL_CLIENT,
	328	op_data->op_namelen + 1);
	329
	330	rc = mdc_prep_elc_req(exp, req, MDS_REINT, &cancels, count);
	331	if (rc) {
	332	ptlrpc_request_free(req);
0a3bdb00	333	return rc;
d7e09d03 PT	334	}
	335
	336	mdc_unlink_pack(req, op_data);
	337
	338	req_capsule_set_size(&req->rq_pill, &RMF_MDT_MD, RCL_SERVER,
44779340	339	obd->u.cli.cl_default_mds_easize);
d7e09d03	340	req_capsule_set_size(&req->rq_pill, &RMF_LOGCOOKIES, RCL_SERVER,
44779340	341	obd->u.cli.cl_default_mds_cookiesize);
d7e09d03 PT	342	ptlrpc_request_set_replen(req);
	343
	344	*request = req;
	345
	346	rc = mdc_reint(req, obd->u.cli.cl_rpc_lock, LUSTRE_IMP_FULL);
	347	if (rc == -ERESTARTSYS)
	348	rc = 0;
0a3bdb00	349	return rc;
d7e09d03 PT	350	}
	351
	352	int mdc_link(struct obd_export exp, struct md_op_data op_data,
	353	struct ptlrpc_request **request)
	354	{
	355	LIST_HEAD(cancels);
	356	struct obd_device *obd = exp->exp_obd;
	357	struct ptlrpc_request *req;
	358	int count = 0, rc;
d7e09d03 PT	359
	360	if ((op_data->op_flags & MF_MDC_CANCEL_FID2) &&
	361	(fid_is_sane(&op_data->op_fid2)))
	362	count = mdc_resource_get_unused(exp, &op_data->op_fid2,
	363	&cancels, LCK_EX,
	364	MDS_INODELOCK_UPDATE);
	365	if ((op_data->op_flags & MF_MDC_CANCEL_FID1) &&
	366	(fid_is_sane(&op_data->op_fid1)))
	367	count += mdc_resource_get_unused(exp, &op_data->op_fid1,
	368	&cancels, LCK_EX,
	369	MDS_INODELOCK_UPDATE);
	370
	371	req = ptlrpc_request_alloc(class_exp2cliimp(exp), &RQF_MDS_REINT_LINK);
34e3ff96	372	if (!req) {
d7e09d03	373	ldlm_lock_list_put(&cancels, l_bl_ast, count);
0a3bdb00	374	return -ENOMEM;
d7e09d03	375	}
d7e09d03 PT	376	req_capsule_set_size(&req->rq_pill, &RMF_NAME, RCL_CLIENT,
	377	op_data->op_namelen + 1);
	378
	379	rc = mdc_prep_elc_req(exp, req, MDS_REINT, &cancels, count);
	380	if (rc) {
	381	ptlrpc_request_free(req);
0a3bdb00	382	return rc;
d7e09d03 PT	383	}
	384
	385	mdc_link_pack(req, op_data);
	386	ptlrpc_request_set_replen(req);
	387
	388	rc = mdc_reint(req, obd->u.cli.cl_rpc_lock, LUSTRE_IMP_FULL);
	389	*request = req;
	390	if (rc == -ERESTARTSYS)
	391	rc = 0;
	392
0a3bdb00	393	return rc;
d7e09d03 PT	394	}
	395
	396	int mdc_rename(struct obd_export exp, struct md_op_data op_data,
	397	const char old, int oldlen, const char new, int newlen,
	398	struct ptlrpc_request **request)
	399	{
	400	LIST_HEAD(cancels);
	401	struct obd_device *obd = exp->exp_obd;
	402	struct ptlrpc_request *req;
	403	int count = 0, rc;
d7e09d03 PT	404
	405	if ((op_data->op_flags & MF_MDC_CANCEL_FID1) &&
	406	(fid_is_sane(&op_data->op_fid1)))
	407	count = mdc_resource_get_unused(exp, &op_data->op_fid1,
	408	&cancels, LCK_EX,
	409	MDS_INODELOCK_UPDATE);
	410	if ((op_data->op_flags & MF_MDC_CANCEL_FID2) &&
	411	(fid_is_sane(&op_data->op_fid2)))
	412	count += mdc_resource_get_unused(exp, &op_data->op_fid2,
	413	&cancels, LCK_EX,
	414	MDS_INODELOCK_UPDATE);
	415	if ((op_data->op_flags & MF_MDC_CANCEL_FID3) &&
	416	(fid_is_sane(&op_data->op_fid3)))
	417	count += mdc_resource_get_unused(exp, &op_data->op_fid3,
	418	&cancels, LCK_EX,
	419	MDS_INODELOCK_LOOKUP);
	420	if ((op_data->op_flags & MF_MDC_CANCEL_FID4) &&
22e0bc6a	421	(fid_is_sane(&op_data->op_fid4)))
d7e09d03 PT	422	count += mdc_resource_get_unused(exp, &op_data->op_fid4,
	423	&cancels, LCK_EX,
	424	MDS_INODELOCK_FULL);
	425
	426	req = ptlrpc_request_alloc(class_exp2cliimp(exp),
	427	&RQF_MDS_REINT_RENAME);
34e3ff96	428	if (!req) {
d7e09d03	429	ldlm_lock_list_put(&cancels, l_bl_ast, count);
0a3bdb00	430	return -ENOMEM;
d7e09d03 PT	431	}
d7e09d03 PT	432
d7e09d03 PT	433	req_capsule_set_size(&req->rq_pill, &RMF_NAME, RCL_CLIENT, oldlen + 1);
	434	req_capsule_set_size(&req->rq_pill, &RMF_SYMTGT, RCL_CLIENT, newlen+1);
	435
	436	rc = mdc_prep_elc_req(exp, req, MDS_REINT, &cancels, count);
	437	if (rc) {
	438	ptlrpc_request_free(req);
0a3bdb00	439	return rc;
d7e09d03 PT	440	}
	441
	442	if (exp_connect_cancelset(exp) && req)
	443	ldlm_cli_cancel_list(&cancels, count, req, 0);
	444
	445	mdc_rename_pack(req, op_data, old, oldlen, new, newlen);
	446
	447	req_capsule_set_size(&req->rq_pill, &RMF_MDT_MD, RCL_SERVER,
44779340	448	obd->u.cli.cl_default_mds_easize);
d7e09d03	449	req_capsule_set_size(&req->rq_pill, &RMF_LOGCOOKIES, RCL_SERVER,
44779340	450	obd->u.cli.cl_default_mds_cookiesize);
d7e09d03 PT	451	ptlrpc_request_set_replen(req);
	452
	453	rc = mdc_reint(req, obd->u.cli.cl_rpc_lock, LUSTRE_IMP_FULL);
	454	*request = req;
	455	if (rc == -ERESTARTSYS)
	456	rc = 0;
	457
0a3bdb00	458	return rc;
d7e09d03	459	}