static int llu_dir_do_readpage(struct inode *inode, struct page *page)
{
        struct llu_inode_info *lli = llu_i2info(inode);
        struct intnl_stat     *st = llu_i2stat(inode);
        struct llu_sb_info    *sbi = llu_i2sbi(inode);
        struct ptlrpc_request *request;
        struct lustre_handle   lockh;
        struct mdt_body       *body;
        struct lookup_intent   it = { .it_op = IT_READDIR };
        struct md_op_data      op_data = {{ 0 }};
        ldlm_policy_data_t policy = { .l_inodebits = { MDS_INODELOCK_UPDATE } };
        int rc = 0;
        ENTRY;

        llu_prep_md_op_data(&op_data, inode, NULL, NULL, 0, 0, LUSTRE_OPC_ANY);
        rc = md_lock_match(sbi->ll_md_exp, LDLM_FL_BLOCK_GRANTED,
                           &lli->lli_fid, LDLM_IBITS, &policy, LCK_CR, &lockh);
        if (!rc) {
		struct ldlm_enqueue_info einfo = {
			.ei_type	= LDLM_IBITS,
			.ei_mode	= LCK_CR,
			.ei_cb_bl	= llu_md_blocking_ast,
			.ei_cb_cp	= ldlm_completion_ast,
			.ei_cbdata	= inode,
		};

                rc = md_enqueue(sbi->ll_md_exp, &einfo, &it,
                                &op_data, &lockh, NULL, 0, NULL,
                                LDLM_FL_CANCEL_ON_BLOCK);
                request = (struct ptlrpc_request *)it.d.lustre.it_data;
                if (request)
                        ptlrpc_req_finished(request);
                if (rc < 0) {
                        CERROR("lock enqueue: err: %d\n", rc);
                        RETURN(rc);
                }
        }
        ldlm_lock_dump_handle(D_OTHER, &lockh);

        op_data.op_offset = (__u64)hash_x_index(page->index, 0);
        op_data.op_npages = 1;
        rc = md_readpage(sbi->ll_md_exp, &op_data, &page, &request);
        if (!rc) {
                body = req_capsule_server_get(&request->rq_pill, &RMF_MDT_BODY);
                LASSERT(body != NULL);         /* checked by md_readpage() */

                if (body->valid & OBD_MD_FLSIZE)
                        st->st_size = body->size;
        } else {
                CERROR("read_dir_page(%ld) error %d\n", page->index, rc);
        }
        ptlrpc_req_finished(request);
        EXIT;

        ldlm_lock_decref(&lockh, LCK_CR);
        return rc;
}

int osc_quotacheck(struct obd_device *unused, struct obd_export *exp,
                   struct obd_quotactl *oqctl)
{
        struct client_obd       *cli = &exp->exp_obd->u.cli;
        struct ptlrpc_request   *req;
        struct obd_quotactl     *body;
        int                      rc;
        ENTRY;

        req = ptlrpc_request_alloc_pack(class_exp2cliimp(exp),
                                        &RQF_OST_QUOTACHECK, LUSTRE_OST_VERSION,
                                        OST_QUOTACHECK);
        if (req == NULL)
                RETURN(-ENOMEM);

        body = req_capsule_client_get(&req->rq_pill, &RMF_OBD_QUOTACTL);
        *body = *oqctl;

        ptlrpc_request_set_replen(req);

        /* the next poll will find -ENODATA, that means quotacheck is
         * going on */
        cli->cl_qchk_stat = -ENODATA;
        rc = ptlrpc_queue_wait(req);
        if (rc)
                cli->cl_qchk_stat = rc;
        ptlrpc_req_finished(req);
        RETURN(rc);
}

/*
 *  Client's outgoing request callback
 */
void request_out_callback(lnet_event_t *ev)
{
	struct ptlrpc_cb_id   *cbid = ev->md.user_ptr;
	struct ptlrpc_request *req = cbid->cbid_arg;

	LASSERT(ev->type == LNET_EVENT_SEND ||
		ev->type == LNET_EVENT_UNLINK);
	LASSERT(ev->unlinked);

	DEBUG_REQ(D_NET, req, "type %d, status %d", ev->type, ev->status);

	sptlrpc_request_out_callback(req);
	req->rq_real_sent = cfs_time_current_sec();

	if (ev->type == LNET_EVENT_UNLINK || ev->status != 0) {

		/* Failed send: make it seem like the reply timed out, just
		 * like failing sends in client.c does currently...  */

		spin_lock(&req->rq_lock);
		req->rq_net_err = 1;
		spin_unlock(&req->rq_lock);

		ptlrpc_client_wake_req(req);
	}

	ptlrpc_req_finished(req);
}

static int llog_client_destroy(const struct lu_env *env,
			       struct llog_handle *loghandle)
{
	struct obd_import     *imp;
	struct ptlrpc_request *req = NULL;
	struct llogd_body     *body;
	int		    rc;

	LLOG_CLIENT_ENTRY(loghandle->lgh_ctxt, imp);
	req = ptlrpc_request_alloc_pack(imp, &RQF_LLOG_ORIGIN_HANDLE_DESTROY,
					LUSTRE_LOG_VERSION,
					LLOG_ORIGIN_HANDLE_DESTROY);
	if (req == NULL)
		GOTO(err_exit, rc =-ENOMEM);

	body = req_capsule_client_get(&req->rq_pill, &RMF_LLOGD_BODY);
	body->lgd_logid = loghandle->lgh_id;
	body->lgd_llh_flags = loghandle->lgh_hdr->llh_flags;

	if (!(body->lgd_llh_flags & LLOG_F_IS_PLAIN))
		CERROR("%s: wrong llog flags %x\n", imp->imp_obd->obd_name,
		       body->lgd_llh_flags);

	ptlrpc_request_set_replen(req);
	rc = ptlrpc_queue_wait(req);

	ptlrpc_req_finished(req);
err_exit:
	LLOG_CLIENT_EXIT(loghandle->lgh_ctxt, imp);
	return rc;
}

static int llog_client_next_block(const struct lu_env *env,
				  struct llog_handle *loghandle,
				  int *cur_idx, int next_idx,
				  __u64 *cur_offset, void *buf, int len)
{
	struct obd_import *imp;
	struct ptlrpc_request *req = NULL;
	struct llogd_body *body;
	void *ptr;
	int rc;

	LLOG_CLIENT_ENTRY(loghandle->lgh_ctxt, imp);
	req = ptlrpc_request_alloc_pack(imp, &RQF_LLOG_ORIGIN_HANDLE_NEXT_BLOCK,
					LUSTRE_LOG_VERSION,
					LLOG_ORIGIN_HANDLE_NEXT_BLOCK);
	if (!req) {
		rc = -ENOMEM;
		goto err_exit;
	}

	body = req_capsule_client_get(&req->rq_pill, &RMF_LLOGD_BODY);
	body->lgd_logid = loghandle->lgh_id;
	body->lgd_ctxt_idx = loghandle->lgh_ctxt->loc_idx - 1;
	body->lgd_llh_flags = loghandle->lgh_hdr->llh_flags;
	body->lgd_index = next_idx;
	body->lgd_saved_index = *cur_idx;
	body->lgd_len = len;
	body->lgd_cur_offset = *cur_offset;

	req_capsule_set_size(&req->rq_pill, &RMF_EADATA, RCL_SERVER, len);
	ptlrpc_request_set_replen(req);
	rc = ptlrpc_queue_wait(req);
	if (rc)
		goto out;

	body = req_capsule_server_get(&req->rq_pill, &RMF_LLOGD_BODY);
	if (!body) {
		rc = -EFAULT;
		goto out;
	}

	/* The log records are swabbed as they are processed */
	ptr = req_capsule_server_get(&req->rq_pill, &RMF_EADATA);
	if (!ptr) {
		rc = -EFAULT;
		goto out;
	}

	*cur_idx = body->lgd_saved_index;
	*cur_offset = body->lgd_cur_offset;

	memcpy(buf, ptr, len);
out:
	ptlrpc_req_finished(req);
err_exit:
	LLOG_CLIENT_EXIT(loghandle->lgh_ctxt, imp);
	return rc;
}

static int llog_client_read_header(const struct lu_env *env,
				   struct llog_handle *handle)
{
	struct obd_import     *imp;
	struct ptlrpc_request *req = NULL;
	struct llogd_body     *body;
	struct llog_log_hdr   *hdr;
	struct llog_rec_hdr   *llh_hdr;
	int		    rc;

	LLOG_CLIENT_ENTRY(handle->lgh_ctxt, imp);
	req = ptlrpc_request_alloc_pack(imp, &RQF_LLOG_ORIGIN_HANDLE_READ_HEADER,
					LUSTRE_LOG_VERSION,
					LLOG_ORIGIN_HANDLE_READ_HEADER);
	if (req == NULL) {
		rc = -ENOMEM;
		goto err_exit;
	}

	body = req_capsule_client_get(&req->rq_pill, &RMF_LLOGD_BODY);
	body->lgd_logid = handle->lgh_id;
	body->lgd_ctxt_idx = handle->lgh_ctxt->loc_idx - 1;
	body->lgd_llh_flags = handle->lgh_hdr->llh_flags;

	ptlrpc_request_set_replen(req);
	rc = ptlrpc_queue_wait(req);
	if (rc)
		goto out;

	hdr = req_capsule_server_get(&req->rq_pill, &RMF_LLOG_LOG_HDR);
	if (hdr == NULL) {
		rc = -EFAULT;
		goto out;
	}

	memcpy(handle->lgh_hdr, hdr, sizeof(*hdr));
	handle->lgh_last_idx = handle->lgh_hdr->llh_tail.lrt_index;

	/* sanity checks */
	llh_hdr = &handle->lgh_hdr->llh_hdr;
	if (llh_hdr->lrh_type != LLOG_HDR_MAGIC) {
		CERROR("bad log header magic: %#x (expecting %#x)\n",
		       llh_hdr->lrh_type, LLOG_HDR_MAGIC);
		rc = -EIO;
	} else if (llh_hdr->lrh_len != LLOG_CHUNK_SIZE) {
		CERROR("incorrectly sized log header: %#x "
		       "(expecting %#x)\n",
		       llh_hdr->lrh_len, LLOG_CHUNK_SIZE);
		CERROR("you may need to re-run lconf --write_conf.\n");
		rc = -EIO;
	}
out:
	ptlrpc_req_finished(req);
err_exit:
	LLOG_CLIENT_EXIT(handle->lgh_ctxt, imp);
	return rc;
}

void ll_intent_release(struct lookup_intent *it)
{
        ENTRY;

        CDEBUG(D_INFO, "intent %p released\n", it);
        ll_intent_drop_lock(it);
        /* We are still holding extra reference on a request, need to free it */
        if (it_disposition(it, DISP_ENQ_OPEN_REF))
                 ptlrpc_req_finished(it->d.lustre.it_data); /* ll_file_open */
        if (it_disposition(it, DISP_ENQ_CREATE_REF)) /* create rec */
                ptlrpc_req_finished(it->d.lustre.it_data);
        if (it_disposition(it, DISP_ENQ_COMPLETE)) /* saved req from revalidate
                                                    * to lookup */
                ptlrpc_req_finished(it->d.lustre.it_data);

        it->d.lustre.it_disposition = 0;
        it->d.lustre.it_data = NULL;
        EXIT;
}

/*
 * Send non-intent quota request to master.
 *
 * \param env    - the environment passed by the caller
 * \param exp    - is the export to use to send the acquire RPC
 * \param qbody  - quota body to be packed in request
 * \param sync   - synchronous or asynchronous
 * \param completion - completion callback
 * \param qqi    - is the qsd_qtype_info structure to pass to the completion
 *                 function
 * \param lqe    - is the qid entry to be processed
 *
 * \retval 0     - success
 * \retval -ve   - appropriate errors
 */
int qsd_send_dqacq(const struct lu_env *env, struct obd_export *exp,
		   struct quota_body *qbody, bool sync,
		   qsd_req_completion_t completion, struct qsd_qtype_info *qqi,
		   struct lustre_handle *lockh, struct lquota_entry *lqe)
{
	struct ptlrpc_request	*req;
	struct quota_body	*req_qbody;
	struct qsd_async_args	*aa;
	int			 rc;
	ENTRY;

	LASSERT(exp);

	req = ptlrpc_request_alloc(class_exp2cliimp(exp), &RQF_QUOTA_DQACQ);
	if (req == NULL)
		GOTO(out, rc = -ENOMEM);

	req->rq_no_resend = req->rq_no_delay = 1;
	req->rq_no_retry_einprogress = 1;
	rc = ptlrpc_request_pack(req, LUSTRE_MDS_VERSION, QUOTA_DQACQ);
	if (rc) {
		ptlrpc_request_free(req);
		GOTO(out, rc);
	}

	req->rq_request_portal = MDS_READPAGE_PORTAL;
	req_qbody = req_capsule_client_get(&req->rq_pill, &RMF_QUOTA_BODY);
	*req_qbody = *qbody;

	ptlrpc_request_set_replen(req);

	CLASSERT(sizeof(*aa) <= sizeof(req->rq_async_args));
	aa = ptlrpc_req_async_args(req);
	aa->aa_exp = exp;
	aa->aa_qqi = qqi;
	aa->aa_arg = (void *)lqe;
	aa->aa_completion = completion;
	lustre_handle_copy(&aa->aa_lockh, lockh);

	if (sync) {
		rc = ptlrpc_queue_wait(req);
		rc = qsd_dqacq_interpret(env, req, aa, rc);
		ptlrpc_req_finished(req);
	} else {
		req->rq_interpret_reply = qsd_dqacq_interpret;
		ptlrpcd_add_req(req);
	}

	RETURN(rc);
out:
	completion(env, qqi, qbody, NULL, lockh, NULL, lqe, rc);
	return rc;
}

static int llog_client_prev_block(const struct lu_env *env,
				  struct llog_handle *loghandle,
				  int prev_idx, void *buf, int len)
{
	struct obd_import *imp;
	struct ptlrpc_request *req = NULL;
	struct llogd_body *body;
	void *ptr;
	int rc;

	LLOG_CLIENT_ENTRY(loghandle->lgh_ctxt, imp);
	req = ptlrpc_request_alloc_pack(imp, &RQF_LLOG_ORIGIN_HANDLE_PREV_BLOCK,
					LUSTRE_LOG_VERSION,
					LLOG_ORIGIN_HANDLE_PREV_BLOCK);
	if (!req) {
		rc = -ENOMEM;
		goto err_exit;
	}

	body = req_capsule_client_get(&req->rq_pill, &RMF_LLOGD_BODY);
	body->lgd_logid = loghandle->lgh_id;
	body->lgd_ctxt_idx = loghandle->lgh_ctxt->loc_idx - 1;
	body->lgd_llh_flags = loghandle->lgh_hdr->llh_flags;
	body->lgd_index = prev_idx;
	body->lgd_len = len;

	req_capsule_set_size(&req->rq_pill, &RMF_EADATA, RCL_SERVER, len);
	ptlrpc_request_set_replen(req);

	rc = ptlrpc_queue_wait(req);
	if (rc)
		goto out;

	body = req_capsule_server_get(&req->rq_pill, &RMF_LLOGD_BODY);
	if (!body) {
		rc = -EFAULT;
		goto out;
	}

	ptr = req_capsule_server_get(&req->rq_pill, &RMF_EADATA);
	if (!ptr) {
		rc = -EFAULT;
		goto out;
	}

	memcpy(buf, ptr, len);
out:
	ptlrpc_req_finished(req);
err_exit:
	LLOG_CLIENT_EXIT(loghandle->lgh_ctxt, imp);
	return rc;
}

/**
 * Cliens updates SOM attributes on MDS (including llog cookies):
 * obd_getattr with no lock and md_setattr.
 */
int ll_som_update(struct inode *inode, struct md_op_data *op_data)
{
        struct ll_inode_info *lli = ll_i2info(inode);
        struct ptlrpc_request *request = NULL;
        __u32 old_flags;
        struct obdo *oa;
        int rc;
        ENTRY;

        LASSERT(op_data != NULL);
        if (lli->lli_flags & LLIF_MDS_SIZE_LOCK)
                CERROR("ino %lu/%u(flags %u) som valid it just after "
                       "recovery\n", inode->i_ino, inode->i_generation,
                       lli->lli_flags);

        OBDO_ALLOC(oa);
        if (!oa) {
                CERROR("can't allocate memory for Size-on-MDS update.\n");
                RETURN(-ENOMEM);
        }

        old_flags = op_data->op_flags;
        op_data->op_flags = MF_SOM_CHANGE;

        /* If inode is already in another epoch, skip getattr from OSTs. */
        if (lli->lli_ioepoch == op_data->op_ioepoch) {
                rc = ll_inode_getattr(inode, oa, op_data->op_ioepoch,
                                      old_flags & MF_GETATTR_LOCK);
                if (rc) {
                        oa->o_valid = 0;
			if (rc != -ENOENT)
                                CERROR("inode_getattr failed (%d): unable to "
                                       "send a Size-on-MDS attribute update "
                                       "for inode %lu/%u\n", rc, inode->i_ino,
                                       inode->i_generation);
                } else {
                        CDEBUG(D_INODE, "Size-on-MDS update on "DFID"\n",
                               PFID(&lli->lli_fid));
                }
                /* Install attributes into op_data. */
                md_from_obdo(op_data, oa, oa->o_valid);
        }

        rc = md_setattr(ll_i2sbi(inode)->ll_md_exp, op_data,
                        NULL, 0, NULL, 0, &request, NULL);
        ptlrpc_req_finished(request);

        OBDO_FREE(oa);
        RETURN(rc);
}

struct inode *search_inode_for_lustre(struct super_block *sb,
				      const struct lu_fid *fid)
{
	struct ll_sb_info     *sbi = ll_s2sbi(sb);
	struct ptlrpc_request *req = NULL;
	struct inode	  *inode = NULL;
	int		   eadatalen = 0;
	unsigned long	      hash = cl_fid_build_ino(fid,
						      ll_need_32bit_api(sbi));
	struct  md_op_data    *op_data;
	int		   rc;

	CDEBUG(D_INFO, "searching inode for:(%lu,"DFID")\n", hash, PFID(fid));

	inode = ilookup5(sb, hash, ll_nfs_test_inode, (void *)fid);
	if (inode)
		return inode;

	rc = ll_get_default_mdsize(sbi, &eadatalen);
	if (rc)
		return ERR_PTR(rc);

	/* Because inode is NULL, ll_prep_md_op_data can not
	 * be used here. So we allocate op_data ourselves */
	op_data = kzalloc(sizeof(*op_data), GFP_NOFS);
	if (!op_data)
		return ERR_PTR(-ENOMEM);

	op_data->op_fid1 = *fid;
	op_data->op_mode = eadatalen;
	op_data->op_valid = OBD_MD_FLEASIZE;

	/* mds_fid2dentry ignores f_type */
	rc = md_getattr(sbi->ll_md_exp, op_data, &req);
	kfree(op_data);
	if (rc) {
		CERROR("can't get object attrs, fid "DFID", rc %d\n",
		       PFID(fid), rc);
		return ERR_PTR(rc);
	}
	rc = ll_prep_inode(&inode, req, sb, NULL);
	ptlrpc_req_finished(req);
	if (rc)
		return ERR_PTR(rc);

	return inode;
}

/**
 * Implementation of dt_object_operations::do_object_lock
 *
 * Enqueue a lock (by ldlm_cli_enqueue()) of remote object on the remote MDT,
 * which will lock the object in the global namespace. And because the
 * cross-MDT locks are relatively rare compared with normal local MDT operation,
 * let's release it right away, instead of putting it into the LRU list.
 *
 * \param[in] env	execution environment
 * \param[in] dt	object to be locked
 * \param[out] lh	lock handle
 * \param[in] einfo	enqueue information
 * \param[in] policy	lock policy
 *
 * \retval		ELDLM_OK if locking the object succeeds.
 * \retval		negative errno if locking fails.
 */
static int osp_md_object_lock(const struct lu_env *env,
			      struct dt_object *dt,
			      struct lustre_handle *lh,
			      struct ldlm_enqueue_info *einfo,
			      union ldlm_policy_data *policy)
{
	struct ldlm_res_id	*res_id;
	struct dt_device	*dt_dev = lu2dt_dev(dt->do_lu.lo_dev);
	struct osp_device	*osp = dt2osp_dev(dt_dev);
	struct ptlrpc_request	*req;
	int			rc = 0;
	__u64			flags = 0;
	enum ldlm_mode		mode;

	res_id = einfo->ei_res_id;
	LASSERT(res_id != NULL);

	mode = ldlm_lock_match(osp->opd_obd->obd_namespace,
			       LDLM_FL_BLOCK_GRANTED, res_id,
			       einfo->ei_type, policy,
			       einfo->ei_mode, lh, 0);
	if (mode > 0)
		return ELDLM_OK;

	if (einfo->ei_nonblock)
		flags |= LDLM_FL_BLOCK_NOWAIT;

	req = ldlm_enqueue_pack(osp->opd_exp, 0);
	if (IS_ERR(req))
		RETURN(PTR_ERR(req));

	rc = ldlm_cli_enqueue(osp->opd_exp, &req, einfo, res_id,
			      (const union ldlm_policy_data *)policy,
			      &flags, NULL, 0, LVB_T_NONE, lh, 0);

	ptlrpc_req_finished(req);
	if (rc == ELDLM_OK) {
		struct ldlm_lock *lock;

		lock = __ldlm_handle2lock(lh, 0);
		ldlm_set_cbpending(lock);
		LDLM_LOCK_PUT(lock);
	}

	return rc == ELDLM_OK ? 0 : -EIO;
}

int ptlrpc_obd_ping(struct obd_device *obd)
{
	int rc;
	struct ptlrpc_request *req;

	req = ptlrpc_prep_ping(obd->u.cli.cl_import);
	if (!req)
		return -ENOMEM;

	req->rq_send_state = LUSTRE_IMP_FULL;

	rc = ptlrpc_queue_wait(req);

	ptlrpc_req_finished(req);

	return rc;
}

static int llu_lookup_it(struct inode *parent, struct pnode *pnode,
                         struct lookup_intent *it, int flags)
{
        struct md_op_data op_data = {{ 0 }};
        struct ptlrpc_request *req = NULL;
        struct lookup_intent lookup_it = { .it_op = IT_LOOKUP };
        __u32 opc;
        int rc;
        ENTRY;

        if (pnode->p_base->pb_name.len > EXT2_NAME_LEN)
                RETURN(-ENAMETOOLONG);

        if (!it) {
                it = &lookup_it;
                it->it_op_release = ll_intent_release;
        }

	if (it->it_op & IT_CREAT)
		opc = LUSTRE_OPC_CREATE;
	else
		opc = LUSTRE_OPC_ANY;

        llu_prep_md_op_data(&op_data, parent, NULL,
                            pnode->p_base->pb_name.name,
                            pnode->p_base->pb_name.len, flags, opc);

        rc = md_intent_lock(llu_i2mdexp(parent), &op_data, NULL, 0, it,
                            flags, &req, llu_md_blocking_ast,
                            LDLM_FL_CANCEL_ON_BLOCK);
        if (rc < 0)
                GOTO(out, rc);

	rc = lookup_it_finish(req, DLM_REPLY_REC_OFF, it, parent, pnode);
        if (rc != 0) {
                ll_intent_release(it);
                GOTO(out, rc);
        }

        llu_lookup_finish_locks(it, pnode);

 out:
        if (req)
                ptlrpc_req_finished(req);
        return rc;
}

int llu_md_close(struct obd_export *md_exp, struct inode *inode)
{
        struct llu_inode_info *lli = llu_i2info(inode);
        struct ll_file_data *fd = lli->lli_file_data;
        struct ptlrpc_request *req = NULL;
        struct obd_client_handle *och = &fd->fd_mds_och;
        struct intnl_stat *st = llu_i2stat(inode);
        struct md_op_data op_data = { { 0 } };
        int rc;
        ENTRY;

        /* clear group lock, if present */
        if (fd->fd_flags & LL_FILE_GROUP_LOCKED)
                llu_put_grouplock(inode, fd->fd_grouplock.cg_gid);

        llu_prepare_close(inode, &op_data, fd);
        rc = md_close(md_exp, &op_data, och->och_mod, &req);
        if (rc == -EAGAIN) {
                /* We are the last writer, so the MDS has instructed us to get
                 * the file size and any write cookies, then close again. */
                LASSERT(lli->lli_open_flags & FMODE_WRITE);
                rc = llu_som_update(inode, &op_data);
                if (rc) {
                        CERROR("inode %llu mdc Size-on-MDS update failed: "
                               "rc = %d\n", (long long)st->st_ino, rc);
                        rc = 0;
                }
        } else if (rc) {
                CERROR("inode %llu close failed: rc %d\n",
                       (long long)st->st_ino, rc);
        } else {
                rc = llu_objects_destroy(req, inode);
                if (rc)
                        CERROR("inode %llu ll_objects destroy: rc = %d\n",
                               (long long)st->st_ino, rc);
        }

        md_clear_open_replay_data(md_exp, och);
        ptlrpc_req_finished(req);
        och->och_fh.cookie = DEAD_HANDLE_MAGIC;
        lli->lli_file_data = NULL;
        OBD_FREE(fd, sizeof(*fd));

        RETURN(rc);
}

int ptlrpc_obd_ping(struct obd_device *obd)
{
        int rc;
        struct ptlrpc_request *req;
        ENTRY;

        req = ptlrpc_prep_ping(obd->u.cli.cl_import);
        if (req == NULL)
                RETURN(-ENOMEM);

        req->rq_send_state = LUSTRE_IMP_FULL;

        rc = ptlrpc_queue_wait(req);

        ptlrpc_req_finished(req);

        RETURN(rc);
}

/**
 * Implementation of dt_object_operations::do_object_lock
 *
 * Enqueue a lock (by ldlm_cli_enqueue()) of remote object on the remote MDT,
 * which will lock the object in the global namespace. And because the
 * cross-MDT locks are relatively rare compared with normal local MDT operation,
 * let's release it right away, instead of putting it into the LRU list.
 *
 * \param[in] env	execution environment
 * \param[in] dt	object to be locked
 * \param[out] lh	lock handle
 * \param[in] einfo	enqueue information
 * \param[in] policy	lock policy
 *
 * \retval		ELDLM_OK if locking the object succeeds.
 * \retval		negative errno if locking fails.
 */
static int osp_md_object_lock(const struct lu_env *env,
			      struct dt_object *dt,
			      struct lustre_handle *lh,
			      struct ldlm_enqueue_info *einfo,
			      union ldlm_policy_data *policy)
{
	struct ldlm_res_id	*res_id;
	struct dt_device	*dt_dev = lu2dt_dev(dt->do_lu.lo_dev);
	struct osp_device	*osp = dt2osp_dev(dt_dev);
	struct lu_device	*top_device;
	struct ptlrpc_request	*req;
	int			rc = 0;
	__u64			flags = LDLM_FL_NO_LRU;

	res_id = einfo->ei_res_id;
	LASSERT(res_id != NULL);

	if (einfo->ei_nonblock)
		flags |= LDLM_FL_BLOCK_NOWAIT;
	if (einfo->ei_mode & (LCK_EX | LCK_PW))
		flags |= LDLM_FL_COS_INCOMPAT;

	req = ldlm_enqueue_pack(osp->opd_exp, 0);
	if (IS_ERR(req))
		RETURN(PTR_ERR(req));

	/* During recovery, it needs to let OSP send enqueue
	 * without checking recoverying status, in case the
	 * other target is being recovered at the same time,
	 * and if we wait here for the import to be recovered,
	 * it might cause deadlock */
	top_device = dt_dev->dd_lu_dev.ld_site->ls_top_dev;
	if (top_device->ld_obd->obd_recovering)
		req->rq_allow_replay = 1;

	rc = ldlm_cli_enqueue(osp->opd_exp, &req, einfo, res_id,
			      (const union ldlm_policy_data *)policy,
			      &flags, NULL, 0, LVB_T_NONE, lh, 0);

	ptlrpc_req_finished(req);

	return rc == ELDLM_OK ? 0 : -EIO;
}

int client_quota_check(struct obd_device *unused, struct obd_export *exp,
                       struct obd_quotactl *oqctl)
{
        struct client_obd       *cli = &exp->exp_obd->u.cli;
        struct ptlrpc_request   *req;
        struct obd_quotactl     *body;
        const struct req_format *rf;
        int                      ver, opc, rc;
        ENTRY;

        if (!strcmp(exp->exp_obd->obd_type->typ_name, LUSTRE_MDC_NAME)) {
                rf  = &RQF_MDS_QUOTACHECK;
                ver = LUSTRE_MDS_VERSION;
                opc = MDS_QUOTACHECK;
        } else if (!strcmp(exp->exp_obd->obd_type->typ_name, LUSTRE_OSC_NAME)) {
                rf  = &RQF_OST_QUOTACHECK;
                ver = LUSTRE_OST_VERSION;
                opc = OST_QUOTACHECK;
        } else {
                RETURN(-EINVAL);
        }

        req = ptlrpc_request_alloc_pack(class_exp2cliimp(exp), rf, ver, opc);
        if (req == NULL)
                RETURN(-ENOMEM);

        body = req_capsule_client_get(&req->rq_pill, &RMF_OBD_QUOTACTL);
        *body = *oqctl;

        ptlrpc_request_set_replen(req);

        /* the next poll will find -ENODATA, that means quotacheck is
         * going on */
        cli->cl_qchk_stat = -ENODATA;
        rc = ptlrpc_queue_wait(req);
        if (rc)
                cli->cl_qchk_stat = rc;
        ptlrpc_req_finished(req);
        RETURN(rc);
}

static int target_quotacheck_callback(struct obd_export *exp,
                                      struct obd_quotactl *oqctl)
{
        struct ptlrpc_request *req;
        struct obd_quotactl   *body;
        int                    rc;
        ENTRY;

        req = ptlrpc_request_alloc_pack(exp->exp_imp_reverse, &RQF_QC_CALLBACK,
                                        LUSTRE_OBD_VERSION, OBD_QC_CALLBACK);
        if (req == NULL)
                RETURN(-ENOMEM);

        body = req_capsule_client_get(&req->rq_pill, &RMF_OBD_QUOTACTL);
        *body = *oqctl;

        ptlrpc_request_set_replen(req);

        rc = ptlrpc_queue_wait(req);
        ptlrpc_req_finished(req);

        RETURN(rc);
}

int osc_quotactl(struct obd_device *unused, struct obd_export *exp,
                 struct obd_quotactl *oqctl)
{
        struct ptlrpc_request *req;
        struct obd_quotactl   *oqc;
        int                    rc;
        ENTRY;

        req = ptlrpc_request_alloc_pack(class_exp2cliimp(exp),
                                        &RQF_OST_QUOTACTL, LUSTRE_OST_VERSION,
                                        OST_QUOTACTL);
        if (req == NULL)
                RETURN(-ENOMEM);

        oqc = req_capsule_client_get(&req->rq_pill, &RMF_OBD_QUOTACTL);
        *oqc = *oqctl;

        ptlrpc_request_set_replen(req);
        ptlrpc_at_set_req_timeout(req);
        req->rq_no_resend = 1;

        rc = ptlrpc_queue_wait(req);
        if (rc)
                CERROR("ptlrpc_queue_wait failed, rc: %d\n", rc);

        if (req->rq_repmsg &&
            (oqc = req_capsule_server_get(&req->rq_pill, &RMF_OBD_QUOTACTL))) {
                *oqctl = *oqc;
        } else if (!rc) {
                CERROR ("Can't unpack obd_quotactl\n");
                rc = -EPROTO;
        }
        ptlrpc_req_finished(req);

        RETURN(rc);
}