/**
 * Add new item for parent FID verification.
 *
 * Prepare new verification item and pass it to the dedicated
 * verification thread for further processing.
 *
 * \param[in] env	execution environment
 * \param[in] fo	OFD object
 * \param[in] oa	OBDO structure with PFID
 */
static void ofd_add_inconsistency_item(const struct lu_env *env,
				       struct ofd_object *fo, struct obdo *oa)
{
	struct ofd_device *ofd = ofd_obj2dev(fo);
	struct ofd_inconsistency_item *oii;
	struct filter_fid *ff;
	bool wakeup = false;

	OBD_ALLOC_PTR(oii);
	if (oii == NULL)
		return;

	INIT_LIST_HEAD(&oii->oii_list);
	lu_object_get(&fo->ofo_obj.do_lu);
	oii->oii_obj = fo;
	ff = &oii->oii_ff;
	ff->ff_parent.f_seq = oa->o_parent_seq;
	ff->ff_parent.f_oid = oa->o_parent_oid;
	ff->ff_parent.f_stripe_idx = oa->o_stripe_idx;
	ff->ff_layout = oa->o_layout;

	spin_lock(&ofd->ofd_inconsistency_lock);
	if (fo->ofo_pfid_checking || fo->ofo_pfid_verified) {
		spin_unlock(&ofd->ofd_inconsistency_lock);
		OBD_FREE_PTR(oii);

		return;
	}

	fo->ofo_pfid_checking = 1;
	if (list_empty(&ofd->ofd_inconsistency_list))
		wakeup = true;
	list_add_tail(&oii->oii_list, &ofd->ofd_inconsistency_list);
	spin_unlock(&ofd->ofd_inconsistency_lock);
	if (wakeup)
		wake_up_all(&ofd->ofd_inconsistency_thread.t_ctl_waitq);

	/* XXX: When the found inconsistency exceeds some threshold,
	 *	we can trigger the LFSCK to scan part of the system
	 *	or the whole system, which depends on how to define
	 *	the threshold, a simple way maybe like that: define
	 *	the absolute value of how many inconsisteny allowed
	 *	to be repaired via self detect/repair mechanism, if
	 *	exceeded, then trigger the LFSCK to scan the layout
	 *	inconsistency within the whole system. */
}

/*
 * Free llog handle and header data if exists. Used in llog_close() only
 */
void llog_free_handle(struct llog_handle *loghandle)
{
	LASSERT(loghandle != NULL);

	/* failed llog_init_handle */
	if (!loghandle->lgh_hdr)
		goto out;

	if (loghandle->lgh_hdr->llh_flags & LLOG_F_IS_PLAIN)
		LASSERT(list_empty(&loghandle->u.phd.phd_entry));
	else if (loghandle->lgh_hdr->llh_flags & LLOG_F_IS_CAT)
		LASSERT(list_empty(&loghandle->u.chd.chd_head));
	LASSERT(sizeof(*(loghandle->lgh_hdr)) == LLOG_CHUNK_SIZE);
	OBD_FREE(loghandle->lgh_hdr, LLOG_CHUNK_SIZE);
out:
	OBD_FREE_PTR(loghandle);
}

/*
 * Connect a quota master to the backend OSD device.
 *
 * \param env - is the environment passed by the caller
 * \param qmt - is the quota master target to be connected
 * \param cfg - is the configuration log record from which we need to extract
 *              the service name of the backend OSD device to connect to.
 *
 * \retval - 0 on success, appropriate error on failure
 */
static int qmt_connect_to_osd(const struct lu_env *env, struct qmt_device *qmt,
			      struct lustre_cfg *cfg)
{
	struct obd_connect_data	*data = NULL;
	struct obd_device	*obd;
	struct lu_device	*ld = qmt2lu_dev(qmt);
	int			 rc;
	ENTRY;

	LASSERT(qmt->qmt_child_exp == NULL);

	OBD_ALLOC_PTR(data);
	if (data == NULL)
		GOTO(out, rc = -ENOMEM);

	/* look-up OBD device associated with the backend OSD device.
	 * The MDT is kind enough to pass the OBD name in QMT configuration */
	obd = class_name2obd(lustre_cfg_string(cfg, 3));
	if (obd == NULL) {
		CERROR("%s: can't locate backend osd device: %s\n",
		       qmt->qmt_svname, lustre_cfg_string(cfg, 3));
		GOTO(out, rc = -ENOTCONN);
	}

	data->ocd_connect_flags = OBD_CONNECT_VERSION;
	data->ocd_version = LUSTRE_VERSION_CODE;

	/* connect to OSD device */
	rc = obd_connect(NULL, &qmt->qmt_child_exp, obd, &obd->obd_uuid, data,
			 NULL);
	if (rc) {
		CERROR("%s: cannot connect to osd dev %s (%d)\n",
		       qmt->qmt_svname, obd->obd_name, rc);
		GOTO(out, rc);
	}

	/* initialize site (although it isn't used anywhere) and lu_device
	 * pointer to next device */
	qmt->qmt_child = lu2dt_dev(qmt->qmt_child_exp->exp_obd->obd_lu_dev);
	ld->ld_site = qmt->qmt_child_exp->exp_obd->obd_lu_dev->ld_site;
	EXIT;
out:
	if (data)
		OBD_FREE_PTR(data);
	return rc;
}

/*
 * caller should make sure no race with other threads
 */
static
void ctx_clear_timer_kr(struct ptlrpc_cli_ctx *ctx)
{
        struct gss_cli_ctx_keyring *gctx_kr = ctx2gctx_keyring(ctx);
        struct timer_list          *timer = gctx_kr->gck_timer;

        if (timer == NULL)
                return;

        CDEBUG(D_SEC, "ctx %p, key %p\n", ctx, gctx_kr->gck_key);

        gctx_kr->gck_timer = NULL;

        del_singleshot_timer_sync(timer);

        OBD_FREE_PTR(timer);
}

/** Get a config log from the MGS and process it.
 * This func is called for both clients and servers.
 * Continue to process new statements appended to the logs
 * (whenever the config lock is revoked) until lustre_end_log
 * is called.
 * @param sb The superblock is used by the MGC to write to the local copy of
 *   the config log
 * @param logname The name of the llog to replicate from the MGS
 * @param cfg Since the same mgc may be used to follow multiple config logs
 *   (e.g. ost1, ost2, client), the config_llog_instance keeps the state for
 *   this log, and is added to the mgc's list of logs to follow.
 */
int lustre_process_log(struct super_block *sb, char *logname,
		     struct config_llog_instance *cfg)
{
	struct lustre_cfg *lcfg;
	struct lustre_cfg_bufs *bufs;
	struct lustre_sb_info *lsi = s2lsi(sb);
	struct obd_device *mgc = lsi->lsi_mgc;
	int rc;

	LASSERT(mgc);
	LASSERT(cfg);

	OBD_ALLOC_PTR(bufs);
	if (bufs == NULL)
		return -ENOMEM;

	/* mgc_process_config */
	lustre_cfg_bufs_reset(bufs, mgc->obd_name);
	lustre_cfg_bufs_set_string(bufs, 1, logname);
	lustre_cfg_bufs_set(bufs, 2, cfg, sizeof(*cfg));
	lustre_cfg_bufs_set(bufs, 3, &sb, sizeof(sb));
	lcfg = lustre_cfg_new(LCFG_LOG_START, bufs);
	rc = obd_process_config(mgc, sizeof(*lcfg), lcfg);
	lustre_cfg_free(lcfg);

	OBD_FREE_PTR(bufs);

	if (rc == -EINVAL)
		LCONSOLE_ERROR_MSG(0x15b, "%s: The configuration from log '%s'"
				   "failed from the MGS (%d).  Make sure this "
				   "client and the MGS are running compatible "
				   "versions of Lustre.\n",
				   mgc->obd_name, logname, rc);

	if (rc)
		LCONSOLE_ERROR_MSG(0x15c, "%s: The configuration from log '%s' "
				   "failed (%d). This may be the result of "
				   "communication errors between this node and "
				   "the MGS, a bad configuration, or other "
				   "errors. See the syslog for more "
				   "information.\n", mgc->obd_name, logname,
				   rc);

	/* class_obd_list(); */
	return rc;
}

static void lov_emerg_free(struct lov_device_emerg **emrg, int nr)
{
	int i;

	for (i = 0; i < nr; ++i) {
		struct lov_device_emerg *em;

		em = emrg[i];
		if (em != NULL) {
			LASSERT(em->emrg_page_list.pl_nr == 0);
			if (em->emrg_env != NULL)
				cl_env_put(em->emrg_env, &em->emrg_refcheck);
			OBD_FREE_PTR(em);
		}
	}
	OBD_FREE(emrg, nr * sizeof(emrg[0]));
}

int llog_process_or_fork(const struct lu_env *env,
			 struct llog_handle *loghandle,
			 llog_cb_t cb, void *data, void *catdata, bool fork)
{
        struct llog_process_info *lpi;
        int                      rc;

        ENTRY;

        OBD_ALLOC_PTR(lpi);
        if (lpi == NULL) {
                CERROR("cannot alloc pointer\n");
                RETURN(-ENOMEM);
        }
        lpi->lpi_loghandle = loghandle;
        lpi->lpi_cb        = cb;
        lpi->lpi_cbdata    = data;
        lpi->lpi_catdata   = catdata;

	if (fork) {
		struct task_struct *task;

		/* The new thread can't use parent env,
		 * init the new one in llog_process_thread_daemonize. */
		lpi->lpi_env = NULL;
		init_completion(&lpi->lpi_completion);
		task = kthread_run(llog_process_thread_daemonize, lpi,
				   "llog_process_thread");
		if (IS_ERR(task)) {
			rc = PTR_ERR(task);
			CERROR("%s: cannot start thread: rc = %d\n",
			       loghandle->lgh_ctxt->loc_obd->obd_name, rc);
			GOTO(out_lpi, rc);
		}
		wait_for_completion(&lpi->lpi_completion);
	} else {
		lpi->lpi_env = env;
		llog_process_thread(lpi);
	}
	rc = lpi->lpi_rc;

out_lpi:
	OBD_FREE_PTR(lpi);
	RETURN(rc);
}

void osd_compat_fini(struct osd_device *dev)
{
        int i;

        ENTRY;

        if (dev->od_ost_map == NULL)
                RETURN_EXIT;

        for (i = 0; i < MAX_OBJID_GROUP; i++)
                osd_compat_seq_fini(dev, i);

        dput(dev->od_ost_map->root);
        OBD_FREE_PTR(dev->od_ost_map);
        dev->od_ost_map = NULL;

        EXIT;
}

static int mdd_changelog_user_register(const struct lu_env *env,
				       struct mdd_device *mdd, int *id)
{
        struct llog_ctxt *ctxt;
        struct llog_changelog_user_rec *rec;
        int rc;
        ENTRY;

        ctxt = llog_get_context(mdd2obd_dev(mdd),
				LLOG_CHANGELOG_USER_ORIG_CTXT);
        if (ctxt == NULL)
                RETURN(-ENXIO);

        OBD_ALLOC_PTR(rec);
        if (rec == NULL) {
                llog_ctxt_put(ctxt);
                RETURN(-ENOMEM);
        }

        /* Assume we want it on since somebody registered */
        rc = mdd_changelog_on(env, mdd, 1);
        if (rc)
                GOTO(out, rc);

        rec->cur_hdr.lrh_len = sizeof(*rec);
        rec->cur_hdr.lrh_type = CHANGELOG_USER_REC;
	spin_lock(&mdd->mdd_cl.mc_user_lock);
	if (mdd->mdd_cl.mc_lastuser == (unsigned int)(-1)) {
		spin_unlock(&mdd->mdd_cl.mc_user_lock);
		CERROR("Maximum number of changelog users exceeded!\n");
		GOTO(out, rc = -EOVERFLOW);
	}
	*id = rec->cur_id = ++mdd->mdd_cl.mc_lastuser;
	rec->cur_endrec = mdd->mdd_cl.mc_index;
	spin_unlock(&mdd->mdd_cl.mc_user_lock);

	rc = llog_cat_add(env, ctxt->loc_handle, &rec->cur_hdr, NULL);

        CDEBUG(D_IOCTL, "Registered changelog user %d\n", *id);
out:
        OBD_FREE_PTR(rec);
        llog_ctxt_put(ctxt);
        RETURN(rc);
}

struct ptlrpc_connection *
ptlrpc_connection_get(struct lnet_process_id peer, lnet_nid_t self,
		      struct obd_uuid *uuid)
{
	struct ptlrpc_connection *conn, *conn2;
	ENTRY;

	peer.nid = LNetPrimaryNID(peer.nid);
	conn = cfs_hash_lookup(conn_hash, &peer);
	if (conn)
		GOTO(out, conn);

	OBD_ALLOC_PTR(conn);
	if (!conn)
		RETURN(NULL);

	conn->c_peer = peer;
	conn->c_self = self;
	INIT_HLIST_NODE(&conn->c_hash);
	atomic_set(&conn->c_refcount, 1);
	if (uuid)
		obd_str2uuid(&conn->c_remote_uuid, uuid->uuid);

	/*
	 * Add the newly created conn to the hash, on key collision we
	 * lost a racing addition and must destroy our newly allocated
	 * connection.	The object which exists in the hash will be
	 * returned and may be compared against out object.
	 */
	/* In the function below, .hs_keycmp resolves to
	 * conn_keycmp() */
	/* coverity[overrun-buffer-val] */
	conn2 = cfs_hash_findadd_unique(conn_hash, &peer, &conn->c_hash);
	if (conn != conn2) {
		OBD_FREE_PTR(conn);
		conn = conn2;
	}
	EXIT;
out:
	CDEBUG(D_INFO, "conn=%p refcount %d to %s\n",
	       conn, atomic_read(&conn->c_refcount),
	       libcfs_nid2str(conn->c_peer.nid));
	return conn;
}

static int auto_quota_on(struct obd_device *obd, int type)
{
        struct obd_quotactl *oqctl;
        int rc;
        ENTRY;

        LASSERT(type == USRQUOTA || type == GRPQUOTA || type == UGQUOTA);

        OBD_ALLOC_PTR(oqctl);
        if (!oqctl)
                RETURN(-ENOMEM);

        oqctl->qc_type = type;

        rc = generic_quota_on(obd, oqctl, 0);

        OBD_FREE_PTR(oqctl);
        RETURN(rc);
}

static
struct ptlrpc_cli_ctx *ctx_create_pf(struct ptlrpc_sec *sec,
                                     struct vfs_cred *vcred)
{
        struct gss_cli_ctx *gctx;
        int                 rc;

        OBD_ALLOC_PTR(gctx);
        if (gctx == NULL)
                return NULL;

        rc = gss_cli_ctx_init_common(sec, &gctx->gc_base,
                                     &gss_pipefs_ctxops, vcred);
        if (rc) {
                OBD_FREE_PTR(gctx);
                return NULL;
        }

        return &gctx->gc_base;
}

static void lprocfs_free_client_stats(struct nid_stat *client_stat)
{
	CDEBUG(D_CONFIG, "stat %p - data %p/%p\n", client_stat,
	       client_stat->nid_proc, client_stat->nid_stats);

	LASSERTF(atomic_read(&client_stat->nid_exp_ref_count) == 0,
		 "nid %s:count %d\n", libcfs_nid2str(client_stat->nid),
		 atomic_read(&client_stat->nid_exp_ref_count));

	if (client_stat->nid_proc)
		lprocfs_remove(&client_stat->nid_proc);

	if (client_stat->nid_stats)
		lprocfs_free_stats(&client_stat->nid_stats);

	if (client_stat->nid_ldlm_stats)
		lprocfs_free_stats(&client_stat->nid_ldlm_stats);

	OBD_FREE_PTR(client_stat);
	return;
}

static void osd_oi_table_put(struct osd_thread_info *info,
			     struct osd_oi **oi_table, unsigned oi_count)
{
	struct iam_container *bag;
	int		      i;

	for (i = 0; i < oi_count; i++) {
		if (oi_table[i] == NULL)
			continue;

		LASSERT(oi_table[i]->oi_inode != NULL);

		bag = &(oi_table[i]->oi_dir.od_container);
		if (bag->ic_object == oi_table[i]->oi_inode)
			iam_container_fini(bag);
		iput(oi_table[i]->oi_inode);
		oi_table[i]->oi_inode = NULL;
		OBD_FREE_PTR(oi_table[i]);
		oi_table[i] = NULL;
	}
}

/** Send a DONE_WRITING rpc. */
static void ll_done_writing(struct inode *inode)
{
        struct obd_client_handle *och = NULL;
        struct md_op_data *op_data;
        int rc;
        ENTRY;

        LASSERT(exp_connect_som(ll_i2mdexp(inode)));

        OBD_ALLOC_PTR(op_data);
        if (op_data == NULL) {
                CERROR("can't allocate op_data\n");
                EXIT;
                return;
        }

        ll_prepare_done_writing(inode, op_data, &och);
        /* If there is no @och, we do not do D_W yet. */
        if (och == NULL)
                GOTO(out, 0);

        rc = md_done_writing(ll_i2sbi(inode)->ll_md_exp, op_data, NULL);
        if (rc == -EAGAIN) {
                /* MDS has instructed us to obtain Size-on-MDS attribute from
                 * OSTs and send setattr to back to MDS. */
                rc = ll_som_update(inode, op_data);
        } else if (rc) {
                CERROR("inode %lu mdc done_writing failed: rc = %d\n",
                       inode->i_ino, rc);
        }
out:
        ll_finish_md_op_data(op_data);
        if (och) {
                md_clear_open_replay_data(ll_i2sbi(inode)->ll_md_exp, och);
                OBD_FREE_PTR(och);
        }
        EXIT;
}

void osp_precreate_fini(struct osp_device *d)
{
	struct ptlrpc_thread *thread;

	ENTRY;

	cfs_timer_disarm(&d->opd_statfs_timer);

	if (d->opd_pre == NULL)
		RETURN_EXIT;

	thread = &d->opd_pre_thread;

	thread->t_flags = SVC_STOPPING;
	wake_up(&d->opd_pre_waitq);

	wait_event(thread->t_ctl_waitq, thread->t_flags & SVC_STOPPED);

	OBD_FREE_PTR(d->opd_pre);
	d->opd_pre = NULL;

	EXIT;
}

/**
 * destroy fld cache.
 */
void fld_cache_fini(struct fld_cache *cache)
{
        __u64 pct;
        ENTRY;

        LASSERT(cache != NULL);
        fld_cache_flush(cache);

        if (cache->fci_stat.fst_count > 0) {
                pct = cache->fci_stat.fst_cache * 100;
                do_div(pct, cache->fci_stat.fst_count);
        } else {
                pct = 0;
        }

        CDEBUG(D_INFO, "FLD cache statistics (%s):\n", cache->fci_name);
        CDEBUG(D_INFO, "  Total reqs: "LPU64"\n", cache->fci_stat.fst_count);
        CDEBUG(D_INFO, "  Cache reqs: "LPU64"\n", cache->fci_stat.fst_cache);
        CDEBUG(D_INFO, "  Cache hits: "LPU64"%%\n", pct);

        OBD_FREE_PTR(cache);

        EXIT;
}

//.........这里部分代码省略.........
	spin_lock(&exp->exp_lock);
	if (exp->exp_nid_stats != NULL) {
		spin_unlock(&exp->exp_lock);
		RETURN(-EALREADY);
	}
	spin_unlock(&exp->exp_lock);

	obd = exp->exp_obd;

	CDEBUG(D_CONFIG, "using hash %p\n", obd->obd_nid_stats_hash);

	OBD_ALLOC_PTR(new_stat);
	if (new_stat == NULL)
		RETURN(-ENOMEM);

	new_stat->nid		= *nid;
	new_stat->nid_obd	= exp->exp_obd;
	/* we need set default refcount to 1 to balance obd_disconnect */
	atomic_set(&new_stat->nid_exp_ref_count, 1);

	old_stat = cfs_hash_findadd_unique(obd->obd_nid_stats_hash,
					   nid, &new_stat->nid_hash);
	CDEBUG(D_INFO, "Found stats %p for nid %s - ref %d\n",
	       old_stat, libcfs_nid2str(*nid),
	       atomic_read(&new_stat->nid_exp_ref_count));

	/* Return -EALREADY here so that we know that the /proc
	 * entry already has been created */
	if (old_stat != new_stat) {
		spin_lock(&exp->exp_lock);
		if (exp->exp_nid_stats) {
			LASSERT(exp->exp_nid_stats == old_stat);
			nidstat_putref(exp->exp_nid_stats);
		}
		exp->exp_nid_stats = old_stat;
		spin_unlock(&exp->exp_lock);
		GOTO(destroy_new, rc = -EALREADY);
	}
	/* not found - create */
	OBD_ALLOC(buffer, LNET_NIDSTR_SIZE);
	if (buffer == NULL)
		GOTO(destroy_new, rc = -ENOMEM);

	memcpy(buffer, libcfs_nid2str(*nid), LNET_NIDSTR_SIZE);
#ifndef HAVE_ONLY_PROCFS_SEQ
	new_stat->nid_proc = lprocfs_register(buffer,
					      obd->obd_proc_exports_entry,
					      NULL, NULL);
#else
	new_stat->nid_proc = lprocfs_seq_register(buffer,
						  obd->obd_proc_exports_entry,
						  NULL, NULL);
#endif
	OBD_FREE(buffer, LNET_NIDSTR_SIZE);

	if (IS_ERR(new_stat->nid_proc)) {
		rc = PTR_ERR(new_stat->nid_proc);
		new_stat->nid_proc = NULL;
		CERROR("%s: cannot create proc entry for export %s: rc = %d\n",
		       obd->obd_name, libcfs_nid2str(*nid), rc);
		GOTO(destroy_new_ns, rc);
	}

	entry = lprocfs_add_simple(new_stat->nid_proc, "uuid", new_stat,
				   &lprocfs_exp_uuid_fops);
	if (IS_ERR(entry)) {
		CWARN("Error adding the NID stats file\n");
		rc = PTR_ERR(entry);
		GOTO(destroy_new_ns, rc);
	}

	entry = lprocfs_add_simple(new_stat->nid_proc, "hash", new_stat,
				   &lprocfs_exp_hash_fops);
	if (IS_ERR(entry)) {
		CWARN("Error adding the hash file\n");
		rc = PTR_ERR(entry);
		GOTO(destroy_new_ns, rc);
	}

	spin_lock(&exp->exp_lock);
	exp->exp_nid_stats = new_stat;
	spin_unlock(&exp->exp_lock);
	*newnid = 1;
	/* protect competitive add to list, not need locking on destroy */
	spin_lock(&obd->obd_nid_lock);
	list_add(&new_stat->nid_list, &obd->obd_nid_stats);
	spin_unlock(&obd->obd_nid_lock);

	RETURN(rc);

destroy_new_ns:
	if (new_stat->nid_proc != NULL)
		lprocfs_remove(&new_stat->nid_proc);
	cfs_hash_del(obd->obd_nid_stats_hash, nid, &new_stat->nid_hash);

destroy_new:
	nidstat_putref(new_stat);
	OBD_FREE_PTR(new_stat);
	RETURN(rc);
}

static int lmv_intent_remote(struct obd_export *exp, struct lookup_intent *it,
			     const struct lu_fid *parent_fid,
			     struct ptlrpc_request **reqp,
			     ldlm_blocking_callback cb_blocking,
			     __u64 extra_lock_flags)
{
	struct obd_device	*obd = exp->exp_obd;
	struct lmv_obd		*lmv = &obd->u.lmv;
	struct ptlrpc_request	*req = NULL;
	struct lustre_handle	plock;
	struct md_op_data	*op_data;
	struct lmv_tgt_desc	*tgt;
	struct mdt_body		*body;
	int			pmode;
	int			rc = 0;
	ENTRY;

	body = req_capsule_server_get(&(*reqp)->rq_pill, &RMF_MDT_BODY);
	if (body == NULL)
		RETURN(-EPROTO);

	LASSERT((body->mbo_valid & OBD_MD_MDS));

	/*
	 * Unfortunately, we have to lie to MDC/MDS to retrieve
	 * attributes llite needs and provideproper locking.
	 */
	if (it->it_op & IT_LOOKUP)
		it->it_op = IT_GETATTR;

	/*
	 * We got LOOKUP lock, but we really need attrs.
	 */
	pmode = it->d.lustre.it_lock_mode;
	if (pmode) {
		plock.cookie = it->d.lustre.it_lock_handle;
		it->d.lustre.it_lock_mode = 0;
		it->d.lustre.it_data = NULL;
	}

	LASSERT(fid_is_sane(&body->mbo_fid1));

	tgt = lmv_find_target(lmv, &body->mbo_fid1);
	if (IS_ERR(tgt))
		GOTO(out, rc = PTR_ERR(tgt));

	OBD_ALLOC_PTR(op_data);
	if (op_data == NULL)
		GOTO(out, rc = -ENOMEM);

	op_data->op_fid1 = body->mbo_fid1;
	/* Sent the parent FID to the remote MDT */
	if (parent_fid != NULL) {
		/* The parent fid is only for remote open to
		 * check whether the open is from OBF,
		 * see mdt_cross_open */
		LASSERT(it->it_op & IT_OPEN);
		op_data->op_fid2 = *parent_fid;
	}

	op_data->op_bias = MDS_CROSS_REF;
	CDEBUG(D_INODE, "REMOTE_INTENT with fid="DFID" -> mds #%u\n",
	       PFID(&body->mbo_fid1), tgt->ltd_idx);

	rc = md_intent_lock(tgt->ltd_exp, op_data, it, &req, cb_blocking,
			    extra_lock_flags);
        if (rc)
                GOTO(out_free_op_data, rc);

	/*
	 * LLite needs LOOKUP lock to track dentry revocation in order to
	 * maintain dcache consistency. Thus drop UPDATE|PERM lock here
	 * and put LOOKUP in request.
	 */
	if (it->d.lustre.it_lock_mode != 0) {
		it->d.lustre.it_remote_lock_handle =
					it->d.lustre.it_lock_handle;
		it->d.lustre.it_remote_lock_mode = it->d.lustre.it_lock_mode;
	}

	if (pmode) {
		it->d.lustre.it_lock_handle = plock.cookie;
		it->d.lustre.it_lock_mode = pmode;
	}

	EXIT;
out_free_op_data:
	OBD_FREE_PTR(op_data);
out:
	if (rc && pmode)
		ldlm_lock_decref(&plock, pmode);

	ptlrpc_req_finished(*reqp);
	*reqp = req;
	return rc;
}

//.........这里部分代码省略.........
			while (class_parse_nid(ptr, &nid, &ptr) == 0) {
				rc = do_lcfg(mgcname, nid,
					     LCFG_ADD_UUID, niduuid, 0,0,0);
				i++;
			}
		}
	} else { /* client */
		/* Use nids from mount line: uml1,[email protected]:uml2,[email protected]:/lustre */
		ptr = lsi->lsi_lmd->lmd_dev;
		while (class_parse_nid(ptr, &nid, &ptr) == 0) {
			rc = do_lcfg(mgcname, nid,
				     LCFG_ADD_UUID, niduuid, 0,0,0);
			i++;
			/* Stop at the first failover nid */
			if (*ptr == ':')
				break;
		}
	}
	if (i == 0) {
		CERROR("No valid MGS nids found.\n");
		GOTO(out_free, rc = -EINVAL);
	}
	lsi->lsi_lmd->lmd_mgs_failnodes = 1;

	/* Random uuid for MGC allows easier reconnects */
	OBD_ALLOC_PTR(uuid);
	ll_generate_random_uuid(uuidc);
	class_uuid_unparse(uuidc, uuid);

	/* Start the MGC */
	rc = lustre_start_simple(mgcname, LUSTRE_MGC_NAME,
				 (char *)uuid->uuid, LUSTRE_MGS_OBDNAME,
				 niduuid, 0, 0);
	OBD_FREE_PTR(uuid);
	if (rc)
		GOTO(out_free, rc);

	/* Add any failover MGS nids */
	i = 1;
	while (ptr && ((*ptr == ':' ||
	       class_find_param(ptr, PARAM_MGSNODE, &ptr) == 0))) {
		/* New failover node */
		sprintf(niduuid, "%s_%x", mgcname, i);
		j = 0;
		while (class_parse_nid_quiet(ptr, &nid, &ptr) == 0) {
			j++;
			rc = do_lcfg(mgcname, nid,
				     LCFG_ADD_UUID, niduuid, 0,0,0);
			if (*ptr == ':')
				break;
		}
		if (j > 0) {
			rc = do_lcfg(mgcname, 0, LCFG_ADD_CONN,
				     niduuid, 0, 0, 0);
			i++;
		} else {
			/* at ":/fsname" */
			break;
		}
	}
	lsi->lsi_lmd->lmd_mgs_failnodes = i;

	obd = class_name2obd(mgcname);
	if (!obd) {
		CERROR("Can't find mgcobd %s\n", mgcname);
		GOTO(out_free, rc = -ENOTCONN);