static int osd_scrub_next(const struct lu_env *env, struct osd_device *dev,
			  struct lu_fid *fid, uint64_t *oid)
{
	struct l_wait_info lwi = { 0 };
	struct lustre_scrub *scrub = &dev->od_scrub;
	struct ptlrpc_thread *thread = &scrub->os_thread;
	struct osd_otable_it *it = dev->od_otable_it;
	struct lustre_mdt_attrs *lma = NULL;
	nvlist_t *nvbuf = NULL;
	int size = 0;
	int rc = 0;
	ENTRY;

	if (OBD_FAIL_CHECK(OBD_FAIL_OSD_SCRUB_DELAY) && cfs_fail_val > 0) {
		lwi = LWI_TIMEOUT(cfs_time_seconds(cfs_fail_val), NULL, NULL);
		if (likely(lwi.lwi_timeout > 0)) {
			l_wait_event(thread->t_ctl_waitq,
				!list_empty(&scrub->os_inconsistent_items) ||
				!thread_is_running(thread),
				&lwi);
			if (unlikely(!thread_is_running(thread)))
				RETURN(SCRUB_NEXT_EXIT);
		}
	}

	if (OBD_FAIL_CHECK(OBD_FAIL_OSD_SCRUB_CRASH)) {
		spin_lock(&scrub->os_lock);
		thread_set_flags(thread, SVC_STOPPING);
		spin_unlock(&scrub->os_lock);
		RETURN(SCRUB_NEXT_CRASH);
	}

	if (OBD_FAIL_CHECK(OBD_FAIL_OSD_SCRUB_FATAL))
		RETURN(SCRUB_NEXT_FATAL);

again:
	if (nvbuf) {
		nvlist_free(nvbuf);
		nvbuf = NULL;
		lma = NULL;
	}

	if (!list_empty(&scrub->os_inconsistent_items)) {
		spin_lock(&scrub->os_lock);
		if (likely(!list_empty(&scrub->os_inconsistent_items))) {
			struct osd_inconsistent_item *oii;

			oii = list_entry(scrub->os_inconsistent_items.next,
				struct osd_inconsistent_item, oii_list);
			*fid = oii->oii_cache.oic_fid;
			*oid = oii->oii_cache.oic_dnode;
			scrub->os_in_prior = 1;
			spin_unlock(&scrub->os_lock);

			GOTO(out, rc = 0);
		}
		spin_unlock(&scrub->os_lock);
	}

int osc_object_is_contended(struct osc_object *obj)
{
	struct osc_device *dev  = lu2osc_dev(obj->oo_cl.co_lu.lo_dev);
	int osc_contention_time = dev->od_contention_time;
	unsigned long cur_time     = cfs_time_current();
	unsigned long retry_time;

	if (OBD_FAIL_CHECK(OBD_FAIL_OSC_OBJECT_CONTENTION))
		return 1;

	if (!obj->oo_contended)
		return 0;

	/*
	 * I like copy-paste. the code is copied from
	 * ll_file_is_contended.
	 */
	retry_time = cfs_time_add(obj->oo_contention_time,
				  cfs_time_seconds(osc_contention_time));
	if (cfs_time_after(cur_time, retry_time)) {
		osc_object_clear_contended(obj);
		return 0;
	}
	return 1;
}

/**
 * Implementation of struct cl_req_operations::cro_attr_set() for VVP
 * layer. VVP is responsible for
 *
 *    - o_[mac]time
 *
 *    - o_mode
 *
 *    - o_parent_seq
 *
 *    - o_[ug]id
 *
 *    - o_parent_oid
 *
 *    - o_parent_ver
 *
 *    - o_ioepoch,
 *
 */
static void vvp_req_attr_set(const struct lu_env *env,
			     const struct cl_req_slice *slice,
			     const struct cl_object *obj,
			     struct cl_req_attr *attr, u64 flags)
{
	struct inode *inode;
	struct obdo  *oa;
	u32	      valid_flags;

	oa = attr->cra_oa;
	inode = vvp_object_inode(obj);
	valid_flags = OBD_MD_FLTYPE;

	if (slice->crs_req->crq_type == CRT_WRITE) {
		if (flags & OBD_MD_FLEPOCH) {
			oa->o_valid |= OBD_MD_FLEPOCH;
			oa->o_ioepoch = ll_i2info(inode)->lli_ioepoch;
			valid_flags |= OBD_MD_FLMTIME | OBD_MD_FLCTIME |
				       OBD_MD_FLUID | OBD_MD_FLGID;
		}
	}
	obdo_from_inode(oa, inode, valid_flags & flags);
	obdo_set_parent_fid(oa, &ll_i2info(inode)->lli_fid);
	if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_INVALID_PFID))
		oa->o_parent_oid++;
	memcpy(attr->cra_jobid, ll_i2info(inode)->lli_jobid,
	       LUSTRE_JOBID_SIZE);
}

/**
 * Disconnect a bulk desc from the network. Idempotent. Not
 * thread-safe (i.e. only interlocks with completion callback).
 * Returns 1 on success or 0 if network unregistration failed for whatever
 * reason.
 */
int ptlrpc_unregister_bulk(struct ptlrpc_request *req, int async)
{
	struct ptlrpc_bulk_desc *desc = req->rq_bulk;
	wait_queue_head_t *wq;
	int rc;

	LASSERT(!in_interrupt());     /* might sleep */

	/* Let's setup deadline for reply unlink. */
	if (OBD_FAIL_CHECK(OBD_FAIL_PTLRPC_LONG_BULK_UNLINK) &&
	    async && req->rq_bulk_deadline == 0 && cfs_fail_val == 0)
		req->rq_bulk_deadline = ktime_get_real_seconds() + LONG_UNLINK;

	if (ptlrpc_client_bulk_active(req) == 0)	/* completed or */
		return 1;				/* never registered */

	LASSERT(desc->bd_req == req);  /* bd_req NULL until registered */

	/* the unlink ensures the callback happens ASAP and is the last
	 * one.  If it fails, it must be because completion just happened,
	 * but we must still wait_event() in this case to give liblustre
	 * a chance to run client_bulk_callback()
	 */
	mdunlink_iterate_helper(desc->bd_mds, desc->bd_md_max_brw);

	if (ptlrpc_client_bulk_active(req) == 0)	/* completed or */
		return 1;				/* never registered */

	/* Move to "Unregistering" phase as bulk was not unlinked yet. */
	ptlrpc_rqphase_move(req, RQ_PHASE_UNREG_BULK);

	/* Do not wait for unlink to finish. */
	if (async)
		return 0;

	if (req->rq_set)
		wq = &req->rq_set->set_waitq;
	else
		wq = &req->rq_reply_waitq;

	for (;;) {
		/* Network access will complete in finite time but the HUGE
		 * timeout lets us CWARN for visibility of sluggish LNDs
		 */
		int cnt = 0;
		while (cnt < LONG_UNLINK &&
		       (rc = wait_event_idle_timeout(*wq,
						     !ptlrpc_client_bulk_active(req),
						     HZ)) == 0)
			cnt += 1;
		if (rc > 0) {
			ptlrpc_rqphase_move(req, req->rq_next_phase);
			return 1;
		}

		DEBUG_REQ(D_WARNING, req, "Unexpectedly long timeout: desc %p",
			  desc);
	}
	return 0;
}

int osd_xattr_set(const struct lu_env *env, struct dt_object *dt,
		  const struct lu_buf *buf, const char *name, int fl,
		  struct thandle *handle)
{
	struct osd_object  *obj = osd_dt_obj(dt);
	struct osd_thandle *oh;
	int rc = 0;
	ENTRY;

	LASSERT(handle != NULL);
	LASSERT(osd_invariant(obj));

	if (!osd_obj2dev(obj)->od_posix_acl &&
	    (strcmp(name, POSIX_ACL_XATTR_ACCESS) == 0 ||
	     strcmp(name, POSIX_ACL_XATTR_DEFAULT) == 0))
		RETURN(-EOPNOTSUPP);

	if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_LINKEA_OVERFLOW) &&
	    strcmp(name, XATTR_NAME_LINK) == 0)
		RETURN(-ENOSPC);

	oh = container_of0(handle, struct osd_thandle, ot_super);

	down_write(&obj->oo_guard);
	CDEBUG(D_INODE, "Setting xattr %s with size %d\n",
		name, (int)buf->lb_len);
	rc = osd_xattr_set_internal(env, obj, buf, name, fl, oh);
	up_write(&obj->oo_guard);

	RETURN(rc);
}

/**
 * Disconnect a bulk desc from the network. Idempotent. Not
 * thread-safe (i.e. only interlocks with completion callback).
 * Returns 1 on success or 0 if network unregistration failed for whatever
 * reason.
 */
int ptlrpc_unregister_bulk(struct ptlrpc_request *req, int async)
{
	struct ptlrpc_bulk_desc *desc = req->rq_bulk;
	struct l_wait_info       lwi;
	int                      rc;
	ENTRY;

	LASSERT(!in_interrupt());     /* might sleep */

	/* Let's setup deadline for reply unlink. */
	if (OBD_FAIL_CHECK(OBD_FAIL_PTLRPC_LONG_BULK_UNLINK) &&
	    async && req->rq_bulk_deadline == 0)
		req->rq_bulk_deadline = cfs_time_current_sec() + LONG_UNLINK;

	if (ptlrpc_client_bulk_active(req) == 0)	/* completed or */
		RETURN(1);				/* never registered */

	LASSERT(desc->bd_req == req);  /* bd_req NULL until registered */

	/* the unlink ensures the callback happens ASAP and is the last
	 * one.  If it fails, it must be because completion just happened,
	 * but we must still l_wait_event() in this case to give liblustre
	 * a chance to run client_bulk_callback() */
	mdunlink_iterate_helper(desc->bd_mds, desc->bd_md_max_brw);

	if (ptlrpc_client_bulk_active(req) == 0)	/* completed or */
		RETURN(1);				/* never registered */

        /* Move to "Unregistering" phase as bulk was not unlinked yet. */
        ptlrpc_rqphase_move(req, RQ_PHASE_UNREGISTERING);

        /* Do not wait for unlink to finish. */
        if (async)
                RETURN(0);

        for (;;) {
#ifdef __KERNEL__
		/* The wq argument is ignored by user-space wait_event macros */
		wait_queue_head_t *wq = (req->rq_set != NULL) ?
					&req->rq_set->set_waitq :
					&req->rq_reply_waitq;
#endif
                /* Network access will complete in finite time but the HUGE
                 * timeout lets us CWARN for visibility of sluggish NALs */
                lwi = LWI_TIMEOUT_INTERVAL(cfs_time_seconds(LONG_UNLINK),
                                           cfs_time_seconds(1), NULL, NULL);
                rc = l_wait_event(*wq, !ptlrpc_client_bulk_active(req), &lwi);
                if (rc == 0) {
                        ptlrpc_rqphase_move(req, req->rq_next_phase);
                        RETURN(1);
                }

                LASSERT(rc == -ETIMEDOUT);
                DEBUG_REQ(D_WARNING, req, "Unexpectedly long timeout: desc %p",
                          desc);
        }
        RETURN(0);
}

/* Add log records for each OSC that this object is striped over, and return
 * cookies for each one.  We _would_ have nice abstraction here, except that
 * we need to keep cookies in stripe order, even if some are NULL, so that
 * the right cookies are passed back to the right OSTs at the client side.
 * Unset cookies should be all-zero (which will never occur naturally). */
static int lov_llog_origin_add(const struct lu_env *env,
			       struct llog_ctxt *ctxt,
			       struct llog_rec_hdr *rec,
			       struct lov_stripe_md *lsm,
			       struct llog_cookie *logcookies, int numcookies)
{
        struct obd_device *obd = ctxt->loc_obd;
        struct lov_obd *lov = &obd->u.lov;
        int i, rc = 0, cookies = 0;
        ENTRY;

        LASSERTF(logcookies && numcookies >= lsm->lsm_stripe_count,
                 "logcookies %p, numcookies %d lsm->lsm_stripe_count %d \n",
                 logcookies, numcookies, lsm->lsm_stripe_count);

        for (i = 0; i < lsm->lsm_stripe_count; i++) {
                struct lov_oinfo *loi = lsm->lsm_oinfo[i];
                struct obd_device *child =
                        lov->lov_tgts[loi->loi_ost_idx]->ltd_exp->exp_obd;
                struct llog_ctxt *cctxt = llog_get_context(child, ctxt->loc_idx);

                /* fill mds unlink/setattr log record */
                switch (rec->lrh_type) {
                case MDS_UNLINK_REC: {
                        struct llog_unlink_rec *lur = (struct llog_unlink_rec *)rec;
                        lur->lur_oid = ostid_id(&loi->loi_oi);
                        lur->lur_oseq = (__u32)ostid_seq(&loi->loi_oi);
                        break;
                }
                case MDS_SETATTR64_REC: {
                        struct llog_setattr64_rec *lsr = (struct llog_setattr64_rec *)rec;
                        lsr->lsr_oi = loi->loi_oi;
                        break;
                }
                default:
                        break;
                }

		/* inject error in llog_obd_add() below */
		if (OBD_FAIL_CHECK(OBD_FAIL_MDS_FAIL_LOV_LOG_ADD)) {
			llog_ctxt_put(cctxt);
			cctxt = NULL;
		}
		rc = llog_obd_add(env, cctxt, rec, NULL, logcookies + cookies,
				  numcookies - cookies);
                llog_ctxt_put(cctxt);
                if (rc < 0) {
                        CERROR("Can't add llog (rc = %d) for stripe %d\n",
                               rc, cookies);
                        memset(logcookies + cookies, 0,
                               sizeof(struct llog_cookie));
                        rc = 1; /* skip this cookie */
                }
		/* Note that rc is always 1 if llog_obd_add was successful */
                cookies += rc;
        }
        RETURN(cookies);
}

/* return EADATA length to the caller. negative value means error */
static int mdt_getxattr_pack_reply(struct mdt_thread_info * info)
{
        struct req_capsule     *pill = info->mti_pill ;
        struct ptlrpc_request  *req = mdt_info_req(info);
        char                   *xattr_name;
        __u64                   valid = info->mti_body->valid;
        static const char       user_string[] = "user.";
        int                     size, rc;
        ENTRY;

        if (OBD_FAIL_CHECK(OBD_FAIL_MDS_GETXATTR_PACK))
                RETURN(-ENOMEM);

        /* Determine how many bytes we need */
        if (valid & OBD_MD_FLXATTR) {
                xattr_name = req_capsule_client_get(pill, &RMF_NAME);
                if (!xattr_name)
                        RETURN(-EFAULT);

                if (!(req->rq_export->exp_connect_flags & OBD_CONNECT_XATTR) &&
                    !strncmp(xattr_name, user_string, sizeof(user_string) - 1))
                        RETURN(-EOPNOTSUPP);

                size = mo_xattr_get(info->mti_env,
                                    mdt_object_child(info->mti_object),
                                    &LU_BUF_NULL, xattr_name);
        } else if (valid & OBD_MD_FLXATTRLS) {
                size = mo_xattr_list(info->mti_env,
                                     mdt_object_child(info->mti_object),
                                     &LU_BUF_NULL);
        } else {
                CDEBUG(D_INFO, "Valid bits: "LPX64"\n", info->mti_body->valid);
                RETURN(-EINVAL);
        }

        if (size == -ENODATA) {
                size = 0;
        } else if (size < 0) {
                CERROR("Error geting EA size: %d\n", size);
                RETURN(size);
        }

        if (info->mti_body->eadatasize != 0 &&
            info->mti_body->eadatasize < size)
                RETURN(-ERANGE);

        req_capsule_set_size(pill, &RMF_EADATA, RCL_SERVER,
                             info->mti_body->eadatasize == 0 ? 0 : size);
        rc = req_capsule_server_pack(pill);
        if (rc) {
                LASSERT(rc < 0);
                RETURN(rc);
        }

        RETURN(size);
}

/**
 * Allocate new fid on passed client @seq and save it to @fid.
 *
 * \param[in] env	pointer to the thread context
 * \param[in,out] seq	pointer to the client sequence manager
 * \param[out] fid	to hold the new allocated fid
 *
 * \retval		1 for notify the caller that sequence switch
 *			is performed to allow it to setup FLD for it.
 * \retval		0 for new FID allocated in current sequence.
 * \retval		Negative error number on failure.
 */
int seq_client_alloc_fid(const struct lu_env *env,
			 struct lu_client_seq *seq, struct lu_fid *fid)
{
	wait_queue_t link;
	int rc;
	ENTRY;

	LASSERT(seq != NULL);
	LASSERT(fid != NULL);

	init_waitqueue_entry(&link, current);
	mutex_lock(&seq->lcs_mutex);

	if (OBD_FAIL_CHECK(OBD_FAIL_SEQ_EXHAUST))
		seq->lcs_fid.f_oid = seq->lcs_width;

	while (1) {
		u64 seqnr;

		if (unlikely(!fid_is_zero(&seq->lcs_fid) &&
			     fid_oid(&seq->lcs_fid) < seq->lcs_width)) {
			/* Just bump last allocated fid and return to caller. */
			seq->lcs_fid.f_oid++;
			rc = 0;
			break;
		}

		/* Release seq::lcs_mutex via seq_fid_alloc_prep() to avoid
		 * deadlock during seq_client_alloc_seq(). */
		rc = seq_fid_alloc_prep(seq, &link);
		if (rc)
			continue;

		rc = seq_client_alloc_seq(env, seq, &seqnr);
		/* Re-take seq::lcs_mutex via seq_fid_alloc_fini(). */
		seq_fid_alloc_fini(seq, rc ? 0 : seqnr, false);
		if (rc) {
			CERROR("%s: Can't allocate new sequence: rc = %d\n",
			       seq->lcs_name, rc);
			mutex_unlock(&seq->lcs_mutex);

			RETURN(rc);
		}

		rc = 1;
		break;
	}

	*fid = seq->lcs_fid;
	mutex_unlock(&seq->lcs_mutex);

	CDEBUG(D_INFO, "%s: Allocated FID "DFID"\n", seq->lcs_name,  PFID(fid));

	RETURN(rc);
}

/**
 * Pack a link_ea_entry.
 * All elements are stored as chars to avoid alignment issues.
 * Numbers are always big-endian
 * \retval record length
 */
int linkea_entry_pack(struct link_ea_entry *lee, const struct lu_name *lname,
		      const struct lu_fid *pfid)
{
	struct lu_fid   tmpfid;
	int             reclen;

	tmpfid = *pfid;
	if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_MUL_REF))
		tmpfid.f_oid--;
	if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_LINKEA_CRASH))
		tmpfid.f_ver = ~0;
	fid_cpu_to_be(&tmpfid, &tmpfid);
	memcpy(&lee->lee_parent_fid, &tmpfid, sizeof(tmpfid));
	memcpy(lee->lee_name, lname->ln_name, lname->ln_namelen);
	reclen = sizeof(struct link_ea_entry) + lname->ln_namelen;

	lee->lee_reclen[0] = (reclen >> 8) & 0xff;
	lee->lee_reclen[1] = reclen & 0xff;
	return reclen;
}

/**
 * Register request buffer descriptor for request receiving.
 */
int ptlrpc_register_rqbd(struct ptlrpc_request_buffer_desc *rqbd)
{
	struct ptlrpc_service *service = rqbd->rqbd_svcpt->scp_service;
	static lnet_process_id_t match_id = {LNET_NID_ANY, LNET_PID_ANY};
	int rc;
	lnet_md_t md;
	lnet_handle_me_t me_h;

	CDEBUG(D_NET, "LNetMEAttach: portal %d\n",
	       service->srv_req_portal);

	if (OBD_FAIL_CHECK(OBD_FAIL_PTLRPC_RQBD))
		return -ENOMEM;

	/* NB: CPT affinity service should use new LNet flag LNET_INS_LOCAL,
	 * which means buffer can only be attached on local CPT, and LND
	 * threads can find it by grabbing a local lock
	 */
	rc = LNetMEAttach(service->srv_req_portal,
			  match_id, 0, ~0, LNET_UNLINK,
			  rqbd->rqbd_svcpt->scp_cpt >= 0 ?
			  LNET_INS_LOCAL : LNET_INS_AFTER, &me_h);
	if (rc != 0) {
		CERROR("LNetMEAttach failed: %d\n", rc);
		return -ENOMEM;
	}

	LASSERT(rqbd->rqbd_refcount == 0);
	rqbd->rqbd_refcount = 1;

	md.start = rqbd->rqbd_buffer;
	md.length = service->srv_buf_size;
	md.max_size = service->srv_max_req_size;
	md.threshold = LNET_MD_THRESH_INF;
	md.options = PTLRPC_MD_OPTIONS | LNET_MD_OP_PUT | LNET_MD_MAX_SIZE;
	md.user_ptr = &rqbd->rqbd_cbid;
	md.eq_handle = ptlrpc_eq_h;

	rc = LNetMDAttach(me_h, md, LNET_UNLINK, &rqbd->rqbd_md_h);
	if (rc == 0)
		return 0;

	CERROR("LNetMDAttach failed: %d;\n", rc);
	LASSERT(rc == -ENOMEM);
	rc = LNetMEUnlink(me_h);
	LASSERT(rc == 0);
	rqbd->rqbd_refcount = 0;

	return -ENOMEM;
}

/**
 *  delete an orphan \a obj from orphan index.
 *  \param obj file or directory.
 *  \param th  transaction for index deletion and object destruction.
 *
 *  \pre obj->mod_count == 0 && ORPHAN_OBJ is set for obj.
 *
 *  \retval 0  success
 *  \retval  -ve index operation error.
 */
int mdd_orphan_delete(const struct lu_env *env, struct mdd_object *obj,
		      struct thandle *th)
{
	struct mdd_device *mdd = mdo2mdd(&obj->mod_obj);
	struct dt_object *dor = mdd->mdd_orphans;
	struct dt_key *key;
	int rc = 0;

	ENTRY;

	LASSERT(mdd_write_locked(env, obj) != 0);
	LASSERT(obj->mod_flags & ORPHAN_OBJ);
	LASSERT(obj->mod_count == 0);

	LASSERT(dor);

	key = mdd_orphan_key_fill(env, mdo2fid(obj));
	dt_write_lock(env, mdd->mdd_orphans, MOR_TGT_ORPHAN);

	if (OBD_FAIL_CHECK(OBD_FAIL_MDS_ORPHAN_DELETE))
		goto ref_del;

	rc = dt_delete(env, mdd->mdd_orphans, key, th);
	if (rc == -ENOENT) {
		key = mdd_orphan_key_fill_20(env, mdo2fid(obj));
		rc = dt_delete(env, mdd->mdd_orphans, key, th);
	}

ref_del:
	if (!rc) {
		/* lov objects will be destroyed by caller */
		mdo_ref_del(env, obj, th);
		if (S_ISDIR(mdd_object_type(obj))) {
			mdo_ref_del(env, obj, th);
			dt_ref_del(env, mdd->mdd_orphans, th);
		}
		obj->mod_flags &= ~ORPHAN_OBJ;
	} else {
		CERROR("%s: could not delete orphan object "DFID": rc = %d\n",
		       mdd2obd_dev(mdd)->obd_name, PFID(mdo2fid(obj)), rc);
	}

	dt_write_unlock(env, mdd->mdd_orphans);
	RETURN(rc);
}

/*
 * This function implements new seq allocation algorithm using async
 * updates to seq file on disk. ref bug 18857 for details.
 * there are four variable to keep track of this process
 *
 * lss_space; - available lss_space
 * lss_lowater_set; - lu_seq_range for all seqs before barrier, i.e. safe to use
 * lss_hiwater_set; - lu_seq_range after barrier, i.e. allocated but may be
 *                    not yet committed
 *
 * when lss_lowater_set reaches the end it is replaced with hiwater one and
 * a write operation is initiated to allocate new hiwater range.
 * if last seq write opearion is still not commited, current operation is
 * flaged as sync write op.
 */
static int range_alloc_set(const struct lu_env *env,
			   struct lu_seq_range *out,
			   struct lu_server_seq *seq)
{
	struct lu_seq_range *space = &seq->lss_space;
	struct lu_seq_range *loset = &seq->lss_lowater_set;
	struct lu_seq_range *hiset = &seq->lss_hiwater_set;
	int rc = 0;

	if (lu_seq_range_is_zero(loset))
		__seq_set_init(env, seq);

	if (OBD_FAIL_CHECK(OBD_FAIL_SEQ_ALLOC)) /* exhaust set */
		loset->lsr_start = loset->lsr_end;

	if (lu_seq_range_is_exhausted(loset)) {
		/* reached high water mark. */
		struct lu_device *dev = seq->lss_site->ss_lu->ls_top_dev;
		int obd_num_clients = dev->ld_obd->obd_num_exports;
		__u64 set_sz;

		/* calculate new seq width based on number of clients */
		set_sz = max(seq->lss_set_width,
			     obd_num_clients * seq->lss_width);
		set_sz = min(lu_seq_range_space(space), set_sz);

		/* Switch to hiwater range now */
		*loset = *hiset;
		/* allocate new hiwater range */
		range_alloc(hiset, space, set_sz);

		/* update ondisk seq with new *space */
		rc = seq_store_update(env, seq, NULL, seq->lss_need_sync);
	}

	LASSERTF(!lu_seq_range_is_exhausted(loset) ||
		 lu_seq_range_is_sane(loset),
		 DRANGE"\n", PRANGE(loset));

	if (rc == 0)
		range_alloc(out, loset, seq->lss_width);

	RETURN(rc);
}

/**
 * FLR: verify the layout version of object.
 *
 * \param[in] env	execution environment
 * \param[in] fo	OFD object
 * \param[in] oa	OBDO structure with layout version
 *
 * \retval		0 on successful verification
 * \retval		-EINPROGRESS layout version is in transfer
 * \retval		-ESTALE the layout version on client is stale
 */
int ofd_verify_layout_version(const struct lu_env *env,
			      struct ofd_object *fo, const struct obdo *oa)
{
	__u32 layout_version;
	int rc;
	ENTRY;

	if (unlikely(OBD_FAIL_CHECK(OBD_FAIL_OST_SKIP_LV_CHECK)))
		GOTO(out, rc = 0);

	rc = ofd_object_ff_load(env, fo);
	if (rc < 0) {
		if (rc == -ENODATA)
			rc = -EINPROGRESS;
		GOTO(out, rc);
	}

	layout_version = fo->ofo_ff.ff_layout_version;
	if (oa->o_layout_version >= layout_version &&
	    oa->o_layout_version <= layout_version + fo->ofo_ff.ff_range)
		GOTO(out, rc = 0);

	/* normal traffic, decide if to return ESTALE or EINPROGRESS */
	layout_version &= ~LU_LAYOUT_RESYNC;

	/* this update is not legitimate */
	if ((oa->o_layout_version & ~LU_LAYOUT_RESYNC) <= layout_version)
		GOTO(out, rc = -ESTALE);

	/* layout version may not be transmitted yet */
	if ((oa->o_layout_version & ~LU_LAYOUT_RESYNC) > layout_version)
		GOTO(out, rc = -EINPROGRESS);

	EXIT;

out:
	CDEBUG(D_INODE, DFID " verify layout version: %u vs. %u/%u, rc: %d\n",
	       PFID(lu_object_fid(&fo->ofo_obj.do_lu)),
	       oa->o_layout_version, fo->ofo_ff.ff_layout_version,
	       fo->ofo_ff.ff_range, rc);
	return rc;

}

static int mdt_reint_create(struct mdt_thread_info *info,
                            struct mdt_lock_handle *lhc)
{
        struct ptlrpc_request   *req = mdt_info_req(info);
        int                     rc;
        ENTRY;

        if (OBD_FAIL_CHECK(OBD_FAIL_MDS_REINT_CREATE))
                RETURN(err_serious(-ESTALE));

        if (info->mti_dlm_req)
                ldlm_request_cancel(mdt_info_req(info), info->mti_dlm_req, 0);

        switch (info->mti_attr.ma_attr.la_mode & S_IFMT) {
        case S_IFDIR:{
                /* Cross-ref case. */
                /* TODO: we can add LPROC_MDT_CROSS for cross-ref stats */
                if (info->mti_cross_ref) {
                        rc = mdt_md_mkobj(info);
                } else {
                        LASSERT(info->mti_rr.rr_namelen > 0);
                        mdt_counter_incr(req->rq_export, LPROC_MDT_MKDIR);
                        rc = mdt_md_create(info);
                }
                break;
        }
        case S_IFREG:
        case S_IFLNK:
        case S_IFCHR:
        case S_IFBLK:
        case S_IFIFO:
        case S_IFSOCK:{
                /* Special file should stay on the same node as parent. */
                LASSERT(info->mti_rr.rr_namelen > 0);
                mdt_counter_incr(req->rq_export, LPROC_MDT_MKNOD);
                rc = mdt_md_create(info);
                break;
        }
        default:
                rc = err_serious(-EOPNOTSUPP);
        }
        RETURN(rc);
}

static void vvp_req_attr_set(const struct lu_env *env, struct cl_object *obj,
			     struct cl_req_attr *attr)
{
	struct inode *inode;
	struct obdo  *oa;
	u64 valid_flags = OBD_MD_FLTYPE;

	oa = attr->cra_oa;
	inode = vvp_object_inode(obj);

	if (attr->cra_type == CRT_WRITE)
		valid_flags |= OBD_MD_FLMTIME | OBD_MD_FLCTIME |
			       OBD_MD_FLUID | OBD_MD_FLGID;
	obdo_from_inode(oa, inode, valid_flags & attr->cra_flags);
	obdo_set_parent_fid(oa, &ll_i2info(inode)->lli_fid);
	if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_INVALID_PFID))
		oa->o_parent_oid++;
	memcpy(attr->cra_jobid, ll_i2info(inode)->lli_jobid, LUSTRE_JOBID_SIZE);
}

/**
 * Match client and OST server connection feature flags.
 *
 * Compute the compatibility flags for a connection request based on
 * features mutually supported by client and server.
 *
 * The obd_export::exp_connect_data.ocd_connect_flags field in \a exp
 * must not be updated here, otherwise a partially initialized value may
 * be exposed. After the connection request is successfully processed,
 * the top-level tgt_connect() request handler atomically updates the export
 * connect flags from the obd_connect_data::ocd_connect_flags field of the
 * reply. \see tgt_connect().
 *
 * \param[in] env		execution environment
 * \param[in] exp		the obd_export associated with this
 *				client/target pair
 * \param[in] data		stores data for this connect request
 * \param[in] new_connection	is this connection new or not
 *
 * \retval		0 if success
 * \retval		-EPROTO client and server feature requirements are
 *			incompatible
 * \retval		-EBADF  OST index in connect request doesn't match
 *			real OST index
 */
static int ofd_parse_connect_data(const struct lu_env *env,
				  struct obd_export *exp,
				  struct obd_connect_data *data,
				  bool new_connection)
{
	struct ofd_device		 *ofd = ofd_exp(exp);
	struct filter_export_data	 *fed = &exp->exp_filter_data;

	if (!data)
		RETURN(0);

	CDEBUG(D_RPCTRACE, "%s: cli %s/%p ocd_connect_flags: "LPX64
	       " ocd_version: %x ocd_grant: %d ocd_index: %u"
	       " ocd_group %u\n",
	       exp->exp_obd->obd_name, exp->exp_client_uuid.uuid, exp,
	       data->ocd_connect_flags, data->ocd_version,
	       data->ocd_grant, data->ocd_index, data->ocd_group);

	if (fed->fed_group != 0 && fed->fed_group != data->ocd_group) {
		CWARN("!!! This export (nid %s) used object group %d "
		      "earlier; now it's trying to use group %d!  This could "
		      "be a bug in the MDS. Please report to "
		      "https://jira.hpdd.intel.com/\n",
		      obd_export_nid2str(exp), fed->fed_group,
		      data->ocd_group);
		RETURN(-EPROTO);
	}
	fed->fed_group = data->ocd_group;

	data->ocd_connect_flags &= OST_CONNECT_SUPPORTED;
	data->ocd_version = LUSTRE_VERSION_CODE;

	/* Kindly make sure the SKIP_ORPHAN flag is from MDS. */
	if (data->ocd_connect_flags & OBD_CONNECT_MDS)
		CDEBUG(D_HA, "%s: Received MDS connection for group %u\n",
		       exp->exp_obd->obd_name, data->ocd_group);
	else if (data->ocd_connect_flags & OBD_CONNECT_SKIP_ORPHAN)
		RETURN(-EPROTO);

	if (ofd_grant_param_supp(exp)) {
		exp->exp_filter_data.fed_pagesize = data->ocd_blocksize;
		/* ocd_{blocksize,inodespace} are log2 values */
		data->ocd_blocksize  = ofd->ofd_blockbits;
		data->ocd_inodespace = ofd->ofd_dt_conf.ddp_inodespace;
		/* ocd_grant_extent is in 1K blocks */
		data->ocd_grant_extent = ofd->ofd_dt_conf.ddp_grant_frag >> 10;
	}

	if (data->ocd_connect_flags & OBD_CONNECT_GRANT)
		data->ocd_grant = ofd_grant_connect(env, exp, data->ocd_grant,
						    new_connection);

	if (data->ocd_connect_flags & OBD_CONNECT_INDEX) {
		struct lr_server_data *lsd = &ofd->ofd_lut.lut_lsd;
		int		       index = lsd->lsd_osd_index;

		if (index != data->ocd_index) {
			LCONSOLE_ERROR_MSG(0x136, "Connection from %s to index"
					   " %u doesn't match actual OST index"
					   " %u in last_rcvd file, bad "
					   "configuration?\n",
					   obd_export_nid2str(exp), index,
					   data->ocd_index);
			RETURN(-EBADF);
		}
		if (!(lsd->lsd_feature_compat & OBD_COMPAT_OST)) {
			/* this will only happen on the first connect */
			lsd->lsd_feature_compat |= OBD_COMPAT_OST;
			/* sync is not needed here as lut_client_add will
			 * set exp_need_sync flag */
			tgt_server_data_update(env, &ofd->ofd_lut, 0);
		}
	}
	if (OBD_FAIL_CHECK(OBD_FAIL_OST_BRW_SIZE)) {
		data->ocd_brw_size = 65536;
//.........这里部分代码省略.........

int mdt_reint_setxattr(struct mdt_thread_info *info,
                       struct mdt_lock_handle *unused)
{
        struct ptlrpc_request   *req = mdt_info_req(info);
        struct md_ucred         *uc  = mdt_ucred(info);
        struct mdt_lock_handle  *lh;
        const struct lu_env     *env  = info->mti_env;
        struct lu_buf           *buf  = &info->mti_buf;
        struct mdt_reint_record *rr   = &info->mti_rr;
        struct md_attr          *ma = &info->mti_attr;
        struct lu_attr          *attr = &info->mti_attr.ma_attr;
        struct mdt_object       *obj;
        struct md_object        *child;
        __u64                    valid = attr->la_valid;
        const char              *xattr_name = rr->rr_name;
        int                      xattr_len = rr->rr_eadatalen;
        __u64                    lockpart;
        int                      rc;
        posix_acl_xattr_header  *new_xattr = NULL;
        __u32                    remote = exp_connect_rmtclient(info->mti_exp);
        __u32                    perm;
        ENTRY;

        CDEBUG(D_INODE, "setxattr for "DFID"\n", PFID(rr->rr_fid1));

        if (OBD_FAIL_CHECK(OBD_FAIL_MDS_SETXATTR))
                RETURN(err_serious(-ENOMEM));

        CDEBUG(D_INODE, "%s xattr %s\n",
               valid & OBD_MD_FLXATTR ? "set" : "remove", xattr_name);

        rc = mdt_init_ucred_reint(info);
        if (rc != 0)
                RETURN(rc);

        if (valid & OBD_MD_FLRMTRSETFACL) {
                if (unlikely(!remote))
                        GOTO(out, rc = err_serious(-EINVAL));

                perm = mdt_identity_get_perm(uc->mu_identity, remote,
                                             req->rq_peer.nid);
                if (!(perm & CFS_RMTACL_PERM))
                        GOTO(out, rc = err_serious(-EPERM));
        }

        if (strncmp(xattr_name, XATTR_USER_PREFIX,
                    sizeof(XATTR_USER_PREFIX) - 1) == 0) {
                if (!(req->rq_export->exp_connect_flags & OBD_CONNECT_XATTR))
                        GOTO(out, rc = -EOPNOTSUPP);
                if (strcmp(xattr_name, XATTR_NAME_LOV) == 0)
                        GOTO(out, rc = -EACCES);
                if (strcmp(xattr_name, XATTR_NAME_LMA) == 0)
                        GOTO(out, rc = 0);
                if (strcmp(xattr_name, XATTR_NAME_LINK) == 0)
                        GOTO(out, rc = 0);
        } else if ((valid & OBD_MD_FLXATTR) &&
                   (strncmp(xattr_name, XATTR_NAME_ACL_ACCESS,
                            sizeof(XATTR_NAME_ACL_ACCESS) - 1) == 0 ||
                    strncmp(xattr_name, XATTR_NAME_ACL_DEFAULT,
                            sizeof(XATTR_NAME_ACL_DEFAULT) - 1) == 0)) {
                /* currently lustre limit acl access size */
                if (xattr_len > LUSTRE_POSIX_ACL_MAX_SIZE)
                        GOTO(out, -ERANGE);
        }

        lockpart = MDS_INODELOCK_UPDATE;
        /* Revoke all clients' lookup lock, since the access
         * permissions for this inode is changed when ACL_ACCESS is
         * set. This isn't needed for ACL_DEFAULT, since that does
         * not change the access permissions of this inode, nor any
         * other existing inodes. It is setting the ACLs inherited
         * by new directories/files at create time. */
        if (!strcmp(xattr_name, XATTR_NAME_ACL_ACCESS))
                lockpart |= MDS_INODELOCK_LOOKUP;

        lh = &info->mti_lh[MDT_LH_PARENT];
        /* ACLs were sent to clients under LCK_CR locks, so taking LCK_EX
         * to cancel them. */
        mdt_lock_reg_init(lh, LCK_EX);
        obj = mdt_object_find_lock(info, rr->rr_fid1, lh, lockpart);
        if (IS_ERR(obj))
                GOTO(out, rc =  PTR_ERR(obj));

        info->mti_mos = obj;
        rc = mdt_version_get_check_save(info, obj, 0);
        if (rc)
                GOTO(out_unlock, rc);

        if (unlikely(!(valid & OBD_MD_FLCTIME))) {
                /* This isn't strictly an error, but all current clients
                 * should set OBD_MD_FLCTIME when setting attributes. */
                CWARN("%s: client miss to set OBD_MD_FLCTIME when "
                      "setxattr %s: [object "DFID"] [valid "LPU64"]\n",
                      info->mti_exp->exp_obd->obd_name, xattr_name,
                      PFID(rr->rr_fid1), valid);
                attr->la_ctime = cfs_time_current_sec();
        }
        attr->la_valid = LA_CTIME;
        child = mdt_object_child(obj);
        if (valid & OBD_MD_FLXATTR) {
//.........这里部分代码省略.........

/* return EADATA length to the caller. negative value means error */
static int mdt_getxattr_pack_reply(struct mdt_thread_info * info)
{
        struct req_capsule     *pill = info->mti_pill ;
        struct ptlrpc_request  *req = mdt_info_req(info);
        char                   *xattr_name;
        __u64                   valid;
        static const char       user_string[] = "user.";
        int                     size, rc;
        ENTRY;

        if (OBD_FAIL_CHECK(OBD_FAIL_MDS_GETXATTR_PACK))
                RETURN(-ENOMEM);

	valid = info->mti_body->valid & (OBD_MD_FLXATTR | OBD_MD_FLXATTRLS);

        /* Determine how many bytes we need */
        if (valid == OBD_MD_FLXATTR) {
                xattr_name = req_capsule_client_get(pill, &RMF_NAME);
                if (!xattr_name)
                        RETURN(-EFAULT);

		if (!(exp_connect_flags(req->rq_export) & OBD_CONNECT_XATTR) &&
		    !strncmp(xattr_name, user_string, sizeof(user_string) - 1))
			RETURN(-EOPNOTSUPP);

                size = mo_xattr_get(info->mti_env,
                                    mdt_object_child(info->mti_object),
                                    &LU_BUF_NULL, xattr_name);
        } else if (valid == OBD_MD_FLXATTRLS) {
                size = mo_xattr_list(info->mti_env,
                                     mdt_object_child(info->mti_object),
                                     &LU_BUF_NULL);
	} else if (valid == OBD_MD_FLXATTRALL) {
		/* N.B. eadatasize = 0 is not valid for FLXATTRALL */
		/* We could calculate accurate sizes, but this would
		 * introduce a lot of overhead, let's do it later... */
		size = info->mti_body->eadatasize;
		req_capsule_set_size(pill, &RMF_EAVALS, RCL_SERVER, size);
		req_capsule_set_size(pill, &RMF_EAVALS_LENS, RCL_SERVER, size);
        } else {
                CDEBUG(D_INFO, "Valid bits: "LPX64"\n", info->mti_body->valid);
                RETURN(-EINVAL);
        }

        if (size == -ENODATA) {
                size = 0;
        } else if (size < 0) {
                CERROR("Error geting EA size: %d\n", size);
                RETURN(size);
        }

        req_capsule_set_size(pill, &RMF_EADATA, RCL_SERVER,
                             info->mti_body->eadatasize == 0 ? 0 : size);
        rc = req_capsule_server_pack(pill);
        if (rc) {
                LASSERT(rc < 0);
                RETURN(rc);
        }

        RETURN(size);
}