/**
 * Server side bulk abort. Idempotent. Not thread-safe (i.e. only
 * serialises with completion callback)
 */
void ptlrpc_abort_bulk(struct ptlrpc_bulk_desc *desc)
{
	struct l_wait_info       lwi;
	int                      rc;

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

	if (!ptlrpc_server_bulk_active(desc))   /* completed or */
		return;                         /* never started */

	/* We used to poison the pages with 0xab here because we did not want to
	 * send any meaningful data over the wire for evicted clients (bug 9297)
	 * However, this is no longer safe now that we use the page cache on the
	 * OSS (bug 20560) */

	/* 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 server_bulk_callback()*/
	mdunlink_iterate_helper(desc->bd_mds, desc->bd_md_max_brw);

	for (;;) {
		/* 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(desc->bd_waitq,
				  !ptlrpc_server_bulk_active(desc), &lwi);
		if (rc == 0)
			return;

		LASSERT(rc == -ETIMEDOUT);
		CWARN("Unexpectedly long timeout: desc %p\n", desc);
	}
}

/**
 * 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;
	struct l_wait_info lwi;
	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 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_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
		 */
		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;
}

/**
 * Callback handler for receiving incoming glimpse ASTs.
 *
 * This only can happen on client side.  After handling the glimpse AST
 * we also consider dropping the lock here if it is unused locally for a
 * long time.
 */
static void ldlm_handle_gl_callback(struct ptlrpc_request *req,
				    struct ldlm_namespace *ns,
				    struct ldlm_request *dlm_req,
				    struct ldlm_lock *lock)
{
	int rc = -ENOSYS;

	LDLM_DEBUG(lock, "client glimpse AST callback handler");

	if (lock->l_glimpse_ast != NULL)
		rc = lock->l_glimpse_ast(lock, req);

	if (req->rq_repmsg != NULL) {
		ptlrpc_reply(req);
	} else {
		req->rq_status = rc;
		ptlrpc_error(req);
	}

	lock_res_and_lock(lock);
	if (lock->l_granted_mode == LCK_PW &&
	    !lock->l_readers && !lock->l_writers &&
	    cfs_time_after(cfs_time_current(),
			   cfs_time_add(lock->l_last_used,
					cfs_time_seconds(10)))) {
		unlock_res_and_lock(lock);
		if (ldlm_bl_to_thread_lock(ns, NULL, lock))
			ldlm_handle_bl_callback(ns, NULL, lock);

		return;
	}
	unlock_res_and_lock(lock);
	LDLM_LOCK_RELEASE(lock);
}

static int
proc_trigger_stack_reset(struct ctl_table *table, int write,
			 void __user *buffer, size_t *lenp, loff_t *ppos)
{
	int              rc = 0;
	int              i = 1;
	kgn_device_t    *dev;
	ENTRY;

	if (!write) {
		/* read */
		rc = proc_dointvec(table, write, buffer, lenp, ppos);
		RETURN(rc);
	}

	/* only device 0 gets the handle, see kgnilnd_dev_init */
	dev = &kgnilnd_data.kgn_devices[0];

	LASSERTF(dev != NULL, "dev 0 is NULL\n");

	kgnilnd_critical_error(dev->gnd_err_handle);

	/* Wait for the reset to complete.  This prevents any races in testing
	 * where we'd immediately try to send traffic again */
	while (kgnilnd_data.kgn_needs_reset != 0) {
		i++;
		LCONSOLE((((i) & (-i)) == i) ? D_WARNING : D_NET,
				"Waiting for stack reset request to clear\n");
		cfs_pause(cfs_time_seconds(1 * i));
	}

	RETURN(rc);
}

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;
}

static int proc_console_min_delay_cs(struct ctl_table *table, int write,
				     void __user *buffer, size_t *lenp,
				     loff_t *ppos)
{
	int rc, min_delay_cs;
	struct ctl_table dummy = *table;
	long d;

	dummy.data = &min_delay_cs;
	dummy.proc_handler = &proc_dointvec;

	if (!write) { /* read */
		min_delay_cs = cfs_duration_sec(libcfs_console_min_delay * 100);
		rc = proc_dointvec(&dummy, write, buffer, lenp, ppos);
		return rc;
	}

	/* write */
	min_delay_cs = 0;
	rc = proc_dointvec(&dummy, write, buffer, lenp, ppos);
	if (rc < 0)
		return rc;
	if (min_delay_cs <= 0)
		return -EINVAL;

	d = cfs_time_seconds(min_delay_cs) / 100;
	if (d == 0 || d > libcfs_console_max_delay)
		return -EINVAL;
	libcfs_console_min_delay = d;

	return rc;
}

void ptlrpc_ni_fini(void)
{
	wait_queue_head_t	 waitq;
	struct l_wait_info  lwi;
	int		 rc;
	int		 retries;

	/* Wait for the event queue to become idle since there may still be
	 * messages in flight with pending events (i.e. the fire-and-forget
	 * messages == client requests and "non-difficult" server
	 * replies */

	for (retries = 0;; retries++) {
		rc = LNetEQFree(ptlrpc_eq_h);
		switch (rc) {
		default:
			LBUG();

		case 0:
			LNetNIFini();
			return;

		case -EBUSY:
			if (retries != 0)
				CWARN("Event queue still busy\n");

			/* Wait for a bit */
			init_waitqueue_head(&waitq);
			lwi = LWI_TIMEOUT(cfs_time_seconds(2), NULL, NULL);
			l_wait_event(waitq, 0, &lwi);
			break;
		}
	}
	/* notreached */
}

void
cfs_wi_shutdown (void)
{
        int i;

        if (cfs_wi_data.wi_scheds == NULL)
                return;

        for (i = 0; i < cfs_wi_data.wi_nsched; i++)
                cfs_wi_sched_shutdown(&cfs_wi_data.wi_scheds[i]);

#ifdef __KERNEL__
        cfs_spin_lock(&cfs_wi_data.wi_glock);
        i = 2;
        while (cfs_wi_data.wi_nthreads != 0) {
                CDEBUG(IS_PO2(++i) ? D_WARNING : D_NET,
                       "waiting for %d threads to terminate\n",
                       cfs_wi_data.wi_nthreads);
                cfs_spin_unlock(&cfs_wi_data.wi_glock);

                cfs_pause(cfs_time_seconds(1));

                cfs_spin_lock(&cfs_wi_data.wi_glock);
        }
        cfs_spin_unlock(&cfs_wi_data.wi_glock);
#endif
        LIBCFS_FREE(cfs_wi_data.wi_scheds,
                    cfs_wi_data.wi_nsched * sizeof(cfs_wi_sched_t));
        return;
}

int LL_PROC_PROTO(proc_console_min_delay_cs)
{
        int rc, min_delay_cs;
        cfs_sysctl_table_t dummy = *table;
        cfs_duration_t d;

        dummy.data = &min_delay_cs;
        dummy.proc_handler = &proc_dointvec;

        if (!write) { /* read */
                min_delay_cs = cfs_duration_sec(libcfs_console_min_delay * 100);
                rc = ll_proc_dointvec(&dummy, write, filp, buffer, lenp, ppos);
                return rc;
        }

        /* write */
        min_delay_cs = 0;
        rc = ll_proc_dointvec(&dummy, write, filp, buffer, lenp, ppos);
        if (rc < 0)
                return rc;
        if (min_delay_cs <= 0)
                return -EINVAL;

        d = cfs_time_seconds(min_delay_cs) / 100;
        if (d == 0 || d > libcfs_console_max_delay)
                return -EINVAL;
        libcfs_console_min_delay = d;

        return rc;
}

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);
	}

/* Read from wire as much data as possible.
 * Returns 0 or 1 on succsess, <0 if error or EOF.
 * 0 means partial read, 1 - complete */
int
usocklnd_read_data(usock_conn_t *conn)
{
        struct iovec *iov;
        int           nob;
        cfs_time_t    t;

        LASSERT (conn->uc_rx_nob_wanted != 0);

        do {
                usock_peer_t *peer = conn->uc_peer;

                LASSERT (conn->uc_rx_niov > 0);

                nob = libcfs_sock_readv(conn->uc_sock,
                                        conn->uc_rx_iov, conn->uc_rx_niov);
                if (nob <= 0) {/* read nothing or error */
                        if (nob < 0)
                                conn->uc_errored = 1;
                        return nob;
                }

                LASSERT (nob <= conn->uc_rx_nob_wanted);
                conn->uc_rx_nob_wanted -= nob;
                conn->uc_rx_nob_left -= nob;
                t = cfs_time_current();
                conn->uc_rx_deadline = cfs_time_add(t, cfs_time_seconds(usock_tuns.ut_timeout));

                if(peer != NULL)
                        peer->up_last_alive = t;

                /* "consume" iov */
                iov = conn->uc_rx_iov;
                do {
                        LASSERT (conn->uc_rx_niov > 0);

                        if (nob < iov->iov_len) {
                                iov->iov_base = (void *)(((unsigned long)(iov->iov_base)) + nob);
                                iov->iov_len -= nob;
                                break;
                        }

                        nob -= iov->iov_len;
                        conn->uc_rx_iov = ++iov;
                        conn->uc_rx_niov--;
                } while (nob != 0);

        } while (conn->uc_rx_nob_wanted != 0);

        return 1; /* read complete */
}

int __cfs_fail_timeout_set(__u32 id, __u32 value, int ms, int set)
{
	int ret;

	ret = __cfs_fail_check_set(id, value, set);
	if (ret && likely(ms > 0)) {
		CERROR("cfs_fail_timeout id %x sleeping for %dms\n",
		       id, ms);
		set_current_state(TASK_UNINTERRUPTIBLE);
		schedule_timeout(cfs_time_seconds(ms) / 1000);
		CERROR("cfs_fail_timeout id %x awake\n", id);
	}
	return ret;
}

/* Send as much tx data as possible.
 * Returns 0 or 1 on succsess, <0 if fatal error.
 * 0 means partial send or non-fatal error, 1 - complete.
 * Rely on libcfs_sock_writev() for differentiating fatal and
 * non-fatal errors. An error should be considered as non-fatal if:
 * 1) it still makes sense to continue reading &&
 * 2) anyway, poll() will set up POLLHUP|POLLERR flags */
int
usocklnd_send_tx(usock_conn_t *conn, usock_tx_t *tx)
{
        struct iovec *iov;
        int           nob;
        cfs_time_t    t;

        LASSERT (tx->tx_resid != 0);

        do {
                usock_peer_t *peer = conn->uc_peer;

                LASSERT (tx->tx_niov > 0);

                nob = libcfs_sock_writev(conn->uc_sock,
                                         tx->tx_iov, tx->tx_niov);
                if (nob < 0)
                        conn->uc_errored = 1;
                if (nob <= 0) /* write queue is flow-controlled or error */
                        return nob;

                LASSERT (nob <= tx->tx_resid);
                tx->tx_resid -= nob;
                t = cfs_time_current();
                conn->uc_tx_deadline = cfs_time_add(t, cfs_time_seconds(usock_tuns.ut_timeout));

                if(peer != NULL)
                        peer->up_last_alive = t;

                /* "consume" iov */
                iov = tx->tx_iov;
                do {
                        LASSERT (tx->tx_niov > 0);

                        if (nob < iov->iov_len) {
                                iov->iov_base = (void *)(((unsigned long)(iov->iov_base)) + nob);
                                iov->iov_len -= nob;
                                break;
                        }

                        nob -= iov->iov_len;
                        tx->tx_iov = ++iov;
                        tx->tx_niov--;
                } while (nob != 0);

        } while (tx->tx_resid != 0);

        return 1; /* send complete */
}

void lc_watchdog_touch(struct lc_watchdog *lcw, int timeout)
{
        ENTRY;
        LASSERT(lcw != NULL);

        lc_watchdog_del_pending(lcw);

        lcw_update_time(lcw, "resumed");
        lcw->lcw_state = LC_WATCHDOG_ENABLED;

        cfs_timer_arm(&lcw->lcw_timer, cfs_time_current() +
                      cfs_time_seconds(timeout));

        EXIT;
}

int __obd_fail_timeout_set(__u32 id, __u32 value, int ms, int set)
{
        int ret = 0;

        ret = __obd_fail_check_set(id, value, set);
        if (ret) {
                CERROR("obd_fail_timeout id %x sleeping for %dms\n",
                       id, ms);
                cfs_schedule_timeout_and_set_state(CFS_TASK_UNINT,
                                                   cfs_time_seconds(ms) / 1000);
                cfs_set_current_state(CFS_TASK_RUNNING);
                CERROR("obd_fail_timeout id %x awake\n", id);
        }
        return ret;
}

static int param_set_delay_minmax(const char *val,
				  const struct kernel_param *kp,
				  long min, long max)
{
	long d;
	int sec;
	int rc;

	rc = kstrtoint(val, 0, &sec);
	if (rc)
		return -EINVAL;

	d = cfs_time_seconds(sec) / 100;
	if (d < min || d > max)
		return -EINVAL;

	*((unsigned int *)kp->arg) = d;

	return 0;
}

struct lc_watchdog *lc_watchdog_add(int timeout,
                                    void (*callback)(pid_t, void *),
                                    void *data)
{
        struct lc_watchdog *lcw = NULL;
        ENTRY;

        LIBCFS_ALLOC(lcw, sizeof(*lcw));
        if (lcw == NULL) {
                CDEBUG(D_INFO, "Could not allocate new lc_watchdog\n");
                RETURN(ERR_PTR(-ENOMEM));
        }

        cfs_spin_lock_init(&lcw->lcw_lock);
        lcw->lcw_refcount = 1; /* refcount for owner */
        lcw->lcw_task     = cfs_current();
        lcw->lcw_pid      = cfs_curproc_pid();
        lcw->lcw_callback = (callback != NULL) ? callback : lc_watchdog_dumplog;
        lcw->lcw_data     = data;
        lcw->lcw_state    = LC_WATCHDOG_DISABLED;

        CFS_INIT_LIST_HEAD(&lcw->lcw_list);
        cfs_timer_init(&lcw->lcw_timer, lcw_cb, lcw);

        cfs_down(&lcw_refcount_sem);
        if (++lcw_refcount == 1)
                lcw_dispatch_start();
        cfs_up(&lcw_refcount_sem);

        /* Keep this working in case we enable them by default */
        if (lcw->lcw_state == LC_WATCHDOG_ENABLED) {
                lcw->lcw_last_touched = cfs_time_current();
                cfs_timer_arm(&lcw->lcw_timer, cfs_time_seconds(timeout) +
                              cfs_time_current());
        }

        RETURN(lcw);
}

/* Request sequence-controller node to allocate new meta-sequence. */
static int seq_client_alloc_meta(const struct lu_env *env,
                                 struct lu_client_seq *seq)
{
        int rc;
        ENTRY;

        if (seq->lcs_srv) {
#ifdef HAVE_SEQ_SERVER
                LASSERT(env != NULL);
                rc = seq_server_alloc_meta(seq->lcs_srv, &seq->lcs_space, env);
#else
		rc = 0;
#endif
	} else {
		do {
			/* If meta server return -EINPROGRESS or EAGAIN,
			 * it means meta server might not be ready to
			 * allocate super sequence from sequence controller
			 * (MDT0)yet */
			rc = seq_client_rpc(seq, &seq->lcs_space,
					    SEQ_ALLOC_META, "meta");
			if (rc == -EINPROGRESS || rc == -EAGAIN) {
				wait_queue_head_t waitq;
				struct l_wait_info  lwi;

				/* MDT0 is not ready, let's wait for 2
				 * seconds and retry. */
				init_waitqueue_head(&waitq);
				lwi = LWI_TIMEOUT(cfs_time_seconds(2), NULL,
						  NULL);
				l_wait_event(waitq, 0, &lwi);
			}
		} while (rc == -EINPROGRESS || rc == -EAGAIN);
        }

        RETURN(rc);
}

/**
 * Callback handler for receiving incoming completion ASTs.
 *
 * This only can happen on client side.
 */
static void ldlm_handle_cp_callback(struct ptlrpc_request *req,
				    struct ldlm_namespace *ns,
				    struct ldlm_request *dlm_req,
				    struct ldlm_lock *lock)
{
	int lvb_len;
	LIST_HEAD(ast_list);
	int rc = 0;

	LDLM_DEBUG(lock, "client completion callback handler START");

	if (OBD_FAIL_CHECK(OBD_FAIL_LDLM_CANCEL_BL_CB_RACE)) {
		int to = cfs_time_seconds(1);
		while (to > 0) {
			schedule_timeout_and_set_state(
				TASK_INTERRUPTIBLE, to);
			if (lock->l_granted_mode == lock->l_req_mode ||
			    lock->l_flags & LDLM_FL_DESTROYED)
				break;
		}
	}

	lvb_len = req_capsule_get_size(&req->rq_pill, &RMF_DLM_LVB, RCL_CLIENT);
	if (lvb_len < 0) {
		LDLM_ERROR(lock, "Fail to get lvb_len, rc = %d", lvb_len);
		GOTO(out, rc = lvb_len);
	} else if (lvb_len > 0) {
		if (lock->l_lvb_len > 0) {
			/* for extent lock, lvb contains ost_lvb{}. */
			LASSERT(lock->l_lvb_data != NULL);

			if (unlikely(lock->l_lvb_len < lvb_len)) {
				LDLM_ERROR(lock, "Replied LVB is larger than "
					   "expectation, expected = %d, "
					   "replied = %d",
					   lock->l_lvb_len, lvb_len);
				GOTO(out, rc = -EINVAL);
			}
		} else if (ldlm_has_layout(lock)) { /* for layout lock, lvb has
						     * variable length */
			void *lvb_data;

			OBD_ALLOC(lvb_data, lvb_len);
			if (lvb_data == NULL) {
				LDLM_ERROR(lock, "No memory: %d.\n", lvb_len);
				GOTO(out, rc = -ENOMEM);
			}

			lock_res_and_lock(lock);
			LASSERT(lock->l_lvb_data == NULL);
			lock->l_lvb_data = lvb_data;
			lock->l_lvb_len = lvb_len;
			unlock_res_and_lock(lock);
		}
	}

	lock_res_and_lock(lock);
	if ((lock->l_flags & LDLM_FL_DESTROYED) ||
	    lock->l_granted_mode == lock->l_req_mode) {
		/* bug 11300: the lock has already been granted */
		unlock_res_and_lock(lock);
		LDLM_DEBUG(lock, "Double grant race happened");
		GOTO(out, rc = 0);
	}

	/* If we receive the completion AST before the actual enqueue returned,
	 * then we might need to switch lock modes, resources, or extents. */
	if (dlm_req->lock_desc.l_granted_mode != lock->l_req_mode) {
		lock->l_req_mode = dlm_req->lock_desc.l_granted_mode;
		LDLM_DEBUG(lock, "completion AST, new lock mode");
	}

	if (lock->l_resource->lr_type != LDLM_PLAIN) {
		ldlm_convert_policy_to_local(req->rq_export,
					  dlm_req->lock_desc.l_resource.lr_type,
					  &dlm_req->lock_desc.l_policy_data,
					  &lock->l_policy_data);
		LDLM_DEBUG(lock, "completion AST, new policy data");
	}

	ldlm_resource_unlink_lock(lock);
	if (memcmp(&dlm_req->lock_desc.l_resource.lr_name,
		   &lock->l_resource->lr_name,
		   sizeof(lock->l_resource->lr_name)) != 0) {
		unlock_res_and_lock(lock);
		rc = ldlm_lock_change_resource(ns, lock,
				&dlm_req->lock_desc.l_resource.lr_name);
		if (rc < 0) {
			LDLM_ERROR(lock, "Failed to allocate resource");
			GOTO(out, rc);
		}
		LDLM_DEBUG(lock, "completion AST, new resource");
		CERROR("change resource!\n");
		lock_res_and_lock(lock);
	}
//.........这里部分代码省略.........

/*
 * Release qsd_qtype_info structure which contains data associated with a
 * given quota type. This releases the accounting objects.
 * It's called on OSD cleanup when the qsd instance is released.
 *
 * \param env - is the environment passed by the caller
 * \param qsd - is the qsd instance managing the qsd_qtype_info structure
 *              to be released
 * \param qtype - is the quota type to be shutdown
 */
static void qsd_qtype_fini(const struct lu_env *env, struct qsd_instance *qsd,
			   int qtype)
{
	struct qsd_qtype_info	*qqi;
	int repeat = 0;
	ENTRY;

	if (qsd->qsd_type_array[qtype] == NULL)
		RETURN_EXIT;
	qqi = qsd->qsd_type_array[qtype];
	qsd->qsd_type_array[qtype] = NULL;

	/* all deferred work lists should be empty */
	LASSERT(cfs_list_empty(&qqi->qqi_deferred_glb));
	LASSERT(cfs_list_empty(&qqi->qqi_deferred_slv));

	/* shutdown lquota site */
	if (qqi->qqi_site != NULL && !IS_ERR(qqi->qqi_site)) {
		lquota_site_free(env, qqi->qqi_site);
		qqi->qqi_site = NULL;
	}

	/* The qqi may still be holding by global locks which are being
	 * canceled asynchronously (LU-4365), see the following steps:
	 *
	 * - On server umount, we try to clear all quota locks first by
	 *   disconnecting LWP (which will invalidate import and cleanup
	 *   all locks on it), however, if quota reint process is holding
	 *   the global lock for reintegration at that time, global lock
	 *   will fail to be cleared on LWP disconnection.
	 *
	 * - Umount process goes on and stops reint process, the global
	 *   lock will be dropped on reint process exit, however, the lock
	 *   cancel in done in asynchronous way, so the
	 *   qsd_glb_blocking_ast() might haven't been called yet when we
	 *   get here.
	 */
	while (atomic_read(&qqi->qqi_ref) > 1) {
		CDEBUG(D_QUOTA, "qqi reference count %u, repeat: %d\n",
		       atomic_read(&qqi->qqi_ref), repeat);
		repeat++;
		schedule_timeout_and_set_state(TASK_INTERRUPTIBLE,
						cfs_time_seconds(1));
	}

	/* by now, all qqi users should have gone away */
	LASSERT(atomic_read(&qqi->qqi_ref) == 1);
	lu_ref_fini(&qqi->qqi_reference);

	/* release accounting object */
	if (qqi->qqi_acct_obj != NULL && !IS_ERR(qqi->qqi_acct_obj)) {
		lu_object_put(env, &qqi->qqi_acct_obj->do_lu);
		qqi->qqi_acct_obj = NULL;
	}

	/* release slv index */
	if (qqi->qqi_slv_obj != NULL && !IS_ERR(qqi->qqi_slv_obj)) {
		lu_object_put(env, &qqi->qqi_slv_obj->do_lu);
		qqi->qqi_slv_obj = NULL;
		qqi->qqi_slv_ver = 0;
	}

	/* release global index */
	if (qqi->qqi_glb_obj != NULL && !IS_ERR(qqi->qqi_glb_obj)) {
		lu_object_put(env, &qqi->qqi_glb_obj->do_lu);
		qqi->qqi_glb_obj = NULL;
		qqi->qqi_glb_ver = 0;
	}

	OBD_FREE_PTR(qqi);
	EXIT;
}