4 * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
6 * This program is free software; you can redistribute it and/or modify
7 * it under the terms of the GNU General Public License version 2 only,
8 * as published by the Free Software Foundation.
10 * This program is distributed in the hope that it will be useful,
11 * but WITHOUT ANY WARRANTY; without even the implied warranty of
12 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
13 * GNU General Public License version 2 for more details. A copy is
14 * included in the COPYING file that accompanied this code.
16 * You should have received a copy of the GNU General Public License
17 * along with this program; if not, write to the Free Software
18 * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
23 * Copyright (c) 2011 Intel Corporation
25 * Copyright 2012 Xyratex Technology Limited
30 * Network Request Scheduler (NRS)
32 * Allows to reorder the handling of RPCs at servers.
34 * Author: Liang Zhen <liang@whamcloud.com>
35 * Author: Nikitas Angelinas <nikitas_angelinas@xyratex.com>
42 #define DEBUG_SUBSYSTEM S_RPC
43 #include "../include/obd_support.h"
44 #include "../include/obd_class.h"
45 #include "../include/lustre_net.h"
46 #include "../include/lprocfs_status.h"
47 #include "../../include/linux/libcfs/libcfs.h"
48 #include "ptlrpc_internal.h"
50 /* XXX: This is just for liblustre. Remove the #if defined directive when the
51 * "cfs_" prefix is dropped from cfs_list_head. */
52 extern struct list_head ptlrpc_all_services;
57 struct nrs_core nrs_core;
59 static int nrs_policy_init(struct ptlrpc_nrs_policy *policy)
61 return policy->pol_desc->pd_ops->op_policy_init != NULL ?
62 policy->pol_desc->pd_ops->op_policy_init(policy) : 0;
65 static void nrs_policy_fini(struct ptlrpc_nrs_policy *policy)
67 LASSERT(policy->pol_ref == 0);
68 LASSERT(policy->pol_req_queued == 0);
70 if (policy->pol_desc->pd_ops->op_policy_fini != NULL)
71 policy->pol_desc->pd_ops->op_policy_fini(policy);
74 static int nrs_policy_ctl_locked(struct ptlrpc_nrs_policy *policy,
75 enum ptlrpc_nrs_ctl opc, void *arg)
78 * The policy may be stopped, but the lprocfs files and
79 * ptlrpc_nrs_policy instances remain present until unregistration time.
80 * Do not perform the ctl operation if the policy is stopped, as
81 * policy->pol_private will be NULL in such a case.
83 if (policy->pol_state == NRS_POL_STATE_STOPPED)
86 return policy->pol_desc->pd_ops->op_policy_ctl != NULL ?
87 policy->pol_desc->pd_ops->op_policy_ctl(policy, opc, arg) :
91 static void nrs_policy_stop0(struct ptlrpc_nrs_policy *policy)
93 struct ptlrpc_nrs *nrs = policy->pol_nrs;
95 if (policy->pol_desc->pd_ops->op_policy_stop != NULL) {
96 spin_unlock(&nrs->nrs_lock);
98 policy->pol_desc->pd_ops->op_policy_stop(policy);
100 spin_lock(&nrs->nrs_lock);
103 LASSERT(list_empty(&policy->pol_list_queued));
104 LASSERT(policy->pol_req_queued == 0 &&
105 policy->pol_req_started == 0);
107 policy->pol_private = NULL;
109 policy->pol_state = NRS_POL_STATE_STOPPED;
111 if (atomic_dec_and_test(&policy->pol_desc->pd_refs))
112 module_put(policy->pol_desc->pd_owner);
115 static int nrs_policy_stop_locked(struct ptlrpc_nrs_policy *policy)
117 struct ptlrpc_nrs *nrs = policy->pol_nrs;
119 if (nrs->nrs_policy_fallback == policy && !nrs->nrs_stopping)
122 if (policy->pol_state == NRS_POL_STATE_STARTING)
125 /* In progress or already stopped */
126 if (policy->pol_state != NRS_POL_STATE_STARTED)
129 policy->pol_state = NRS_POL_STATE_STOPPING;
131 /* Immediately make it invisible */
132 if (nrs->nrs_policy_primary == policy) {
133 nrs->nrs_policy_primary = NULL;
136 LASSERT(nrs->nrs_policy_fallback == policy);
137 nrs->nrs_policy_fallback = NULL;
140 /* I have the only refcount */
141 if (policy->pol_ref == 1)
142 nrs_policy_stop0(policy);
148 * Transitions the \a nrs NRS head's primary policy to
149 * ptlrpc_nrs_pol_state::NRS_POL_STATE_STOPPING and if the policy has no
150 * pending usage references, to ptlrpc_nrs_pol_state::NRS_POL_STATE_STOPPED.
152 * \param[in] nrs the NRS head to carry out this operation on
154 static void nrs_policy_stop_primary(struct ptlrpc_nrs *nrs)
156 struct ptlrpc_nrs_policy *tmp = nrs->nrs_policy_primary;
162 nrs->nrs_policy_primary = NULL;
164 LASSERT(tmp->pol_state == NRS_POL_STATE_STARTED);
165 tmp->pol_state = NRS_POL_STATE_STOPPING;
167 if (tmp->pol_ref == 0)
168 nrs_policy_stop0(tmp);
172 * Transitions a policy across the ptlrpc_nrs_pol_state range of values, in
173 * response to an lprocfs command to start a policy.
175 * If a primary policy different to the current one is specified, this function
176 * will transition the new policy to the
177 * ptlrpc_nrs_pol_state::NRS_POL_STATE_STARTING and then to
178 * ptlrpc_nrs_pol_state::NRS_POL_STATE_STARTED, and will then transition
179 * the old primary policy (if there is one) to
180 * ptlrpc_nrs_pol_state::NRS_POL_STATE_STOPPING, and if there are no outstanding
181 * references on the policy to ptlrpc_nrs_pol_stae::NRS_POL_STATE_STOPPED.
183 * If the fallback policy is specified, this is taken to indicate an instruction
184 * to stop the current primary policy, without substituting it with another
185 * primary policy, so the primary policy (if any) is transitioned to
186 * ptlrpc_nrs_pol_state::NRS_POL_STATE_STOPPING, and if there are no outstanding
187 * references on the policy to ptlrpc_nrs_pol_stae::NRS_POL_STATE_STOPPED. In
188 * this case, the fallback policy is only left active in the NRS head.
190 static int nrs_policy_start_locked(struct ptlrpc_nrs_policy *policy)
192 struct ptlrpc_nrs *nrs = policy->pol_nrs;
196 * Don't allow multiple starting which is too complex, and has no real
199 if (nrs->nrs_policy_starting)
202 LASSERT(policy->pol_state != NRS_POL_STATE_STARTING);
204 if (policy->pol_state == NRS_POL_STATE_STOPPING)
207 if (policy->pol_flags & PTLRPC_NRS_FL_FALLBACK) {
209 * This is for cases in which the user sets the policy to the
210 * fallback policy (currently fifo for all services); i.e. the
211 * user is resetting the policy to the default; so we stop the
212 * primary policy, if any.
214 if (policy == nrs->nrs_policy_fallback) {
215 nrs_policy_stop_primary(nrs);
220 * If we reach here, we must be setting up the fallback policy
221 * at service startup time, and only a single policy with the
222 * nrs_policy_flags::PTLRPC_NRS_FL_FALLBACK flag set can
223 * register with NRS core.
225 LASSERT(nrs->nrs_policy_fallback == NULL);
228 * Shouldn't start primary policy if w/o fallback policy.
230 if (nrs->nrs_policy_fallback == NULL)
233 if (policy->pol_state == NRS_POL_STATE_STARTED)
238 * Increase the module usage count for policies registering from other
241 if (atomic_inc_return(&policy->pol_desc->pd_refs) == 1 &&
242 !try_module_get(policy->pol_desc->pd_owner)) {
243 atomic_dec(&policy->pol_desc->pd_refs);
244 CERROR("NRS: cannot get module for policy %s; is it alive?\n",
245 policy->pol_desc->pd_name);
250 * Serialize policy starting across the NRS head
252 nrs->nrs_policy_starting = 1;
254 policy->pol_state = NRS_POL_STATE_STARTING;
256 if (policy->pol_desc->pd_ops->op_policy_start) {
257 spin_unlock(&nrs->nrs_lock);
259 rc = policy->pol_desc->pd_ops->op_policy_start(policy);
261 spin_lock(&nrs->nrs_lock);
263 if (atomic_dec_and_test(&policy->pol_desc->pd_refs))
264 module_put(policy->pol_desc->pd_owner);
266 policy->pol_state = NRS_POL_STATE_STOPPED;
271 policy->pol_state = NRS_POL_STATE_STARTED;
273 if (policy->pol_flags & PTLRPC_NRS_FL_FALLBACK) {
275 * This path is only used at PTLRPC service setup time.
277 nrs->nrs_policy_fallback = policy;
280 * Try to stop the current primary policy if there is one.
282 nrs_policy_stop_primary(nrs);
285 * And set the newly-started policy as the primary one.
287 nrs->nrs_policy_primary = policy;
291 nrs->nrs_policy_starting = 0;
297 * Increases the policy's usage reference count.
299 static inline void nrs_policy_get_locked(struct ptlrpc_nrs_policy *policy)
305 * Decreases the policy's usage reference count, and stops the policy in case it
306 * was already stopping and have no more outstanding usage references (which
307 * indicates it has no more queued or started requests, and can be safely
310 static void nrs_policy_put_locked(struct ptlrpc_nrs_policy *policy)
312 LASSERT(policy->pol_ref > 0);
315 if (unlikely(policy->pol_ref == 0 &&
316 policy->pol_state == NRS_POL_STATE_STOPPING))
317 nrs_policy_stop0(policy);
320 static void nrs_policy_put(struct ptlrpc_nrs_policy *policy)
322 spin_lock(&policy->pol_nrs->nrs_lock);
323 nrs_policy_put_locked(policy);
324 spin_unlock(&policy->pol_nrs->nrs_lock);
328 * Find and return a policy by name.
330 static struct ptlrpc_nrs_policy *nrs_policy_find_locked(struct ptlrpc_nrs *nrs,
333 struct ptlrpc_nrs_policy *tmp;
335 list_for_each_entry(tmp, &nrs->nrs_policy_list, pol_list) {
336 if (strncmp(tmp->pol_desc->pd_name, name,
337 NRS_POL_NAME_MAX) == 0) {
338 nrs_policy_get_locked(tmp);
346 * Release references for the resource hierarchy moving upwards towards the
347 * policy instance resource.
349 static void nrs_resource_put(struct ptlrpc_nrs_resource *res)
351 struct ptlrpc_nrs_policy *policy = res->res_policy;
353 if (policy->pol_desc->pd_ops->op_res_put != NULL) {
354 struct ptlrpc_nrs_resource *parent;
356 for (; res != NULL; res = parent) {
357 parent = res->res_parent;
358 policy->pol_desc->pd_ops->op_res_put(policy, res);
364 * Obtains references for each resource in the resource hierarchy for request
365 * \a nrq if it is to be handled by \a policy.
367 * \param[in] policy the policy
368 * \param[in] nrq the request
369 * \param[in] moving_req denotes whether this is a call to the function by
370 * ldlm_lock_reorder_req(), in order to move \a nrq to
371 * the high-priority NRS head; we should not sleep when
374 * \retval NULL resource hierarchy references not obtained
375 * \retval valid-pointer the bottom level of the resource hierarchy
377 * \see ptlrpc_nrs_pol_ops::op_res_get()
380 struct ptlrpc_nrs_resource *nrs_resource_get(struct ptlrpc_nrs_policy *policy,
381 struct ptlrpc_nrs_request *nrq,
385 * Set to NULL to traverse the resource hierarchy from the top.
387 struct ptlrpc_nrs_resource *res = NULL;
388 struct ptlrpc_nrs_resource *tmp = NULL;
392 rc = policy->pol_desc->pd_ops->op_res_get(policy, nrq, res,
396 nrs_resource_put(res);
400 LASSERT(tmp != NULL);
401 tmp->res_parent = res;
402 tmp->res_policy = policy;
406 * Return once we have obtained a reference to the bottom level
407 * of the resource hierarchy.
415 * Obtains resources for the resource hierarchies and policy references for
416 * the fallback and current primary policy (if any), that will later be used
417 * to handle request \a nrq.
419 * \param[in] nrs the NRS head instance that will be handling request \a nrq.
420 * \param[in] nrq the request that is being handled.
421 * \param[out] resp the array where references to the resource hierarchy are
423 * \param[in] moving_req is set when obtaining resources while moving a
424 * request from a policy on the regular NRS head to a
425 * policy on the HP NRS head (via
426 * ldlm_lock_reorder_req()). It signifies that
427 * allocations to get resources should be atomic; for
428 * a full explanation, see comment in
429 * ptlrpc_nrs_pol_ops::op_res_get().
431 static void nrs_resource_get_safe(struct ptlrpc_nrs *nrs,
432 struct ptlrpc_nrs_request *nrq,
433 struct ptlrpc_nrs_resource **resp,
436 struct ptlrpc_nrs_policy *primary = NULL;
437 struct ptlrpc_nrs_policy *fallback = NULL;
439 memset(resp, 0, sizeof(resp[0]) * NRS_RES_MAX);
442 * Obtain policy references.
444 spin_lock(&nrs->nrs_lock);
446 fallback = nrs->nrs_policy_fallback;
447 nrs_policy_get_locked(fallback);
449 primary = nrs->nrs_policy_primary;
451 nrs_policy_get_locked(primary);
453 spin_unlock(&nrs->nrs_lock);
456 * Obtain resource hierarchy references.
458 resp[NRS_RES_FALLBACK] = nrs_resource_get(fallback, nrq, moving_req);
459 LASSERT(resp[NRS_RES_FALLBACK] != NULL);
461 if (primary != NULL) {
462 resp[NRS_RES_PRIMARY] = nrs_resource_get(primary, nrq,
465 * A primary policy may exist which may not wish to serve a
466 * particular request for different reasons; release the
467 * reference on the policy as it will not be used for this
470 if (resp[NRS_RES_PRIMARY] == NULL)
471 nrs_policy_put(primary);
476 * Releases references to resource hierarchies and policies, because they are no
477 * longer required; used when request handling has been completed, or the
478 * request is moving to the high priority NRS head.
480 * \param resp the resource hierarchy that is being released
482 * \see ptlrpcnrs_req_hp_move()
483 * \see ptlrpc_nrs_req_finalize()
485 static void nrs_resource_put_safe(struct ptlrpc_nrs_resource **resp)
487 struct ptlrpc_nrs_policy *pols[NRS_RES_MAX];
488 struct ptlrpc_nrs *nrs = NULL;
491 for (i = 0; i < NRS_RES_MAX; i++) {
492 if (resp[i] != NULL) {
493 pols[i] = resp[i]->res_policy;
494 nrs_resource_put(resp[i]);
501 for (i = 0; i < NRS_RES_MAX; i++) {
506 nrs = pols[i]->pol_nrs;
507 spin_lock(&nrs->nrs_lock);
509 nrs_policy_put_locked(pols[i]);
513 spin_unlock(&nrs->nrs_lock);
517 * Obtains an NRS request from \a policy for handling or examination; the
518 * request should be removed in the 'handling' case.
520 * Calling into this function implies we already know the policy has a request
521 * waiting to be handled.
523 * \param[in] policy the policy from which a request
524 * \param[in] peek when set, signifies that we just want to examine the
525 * request, and not handle it, so the request is not removed
527 * \param[in] force when set, it will force a policy to return a request if it
530 * \retval the NRS request to be handled
533 struct ptlrpc_nrs_request *nrs_request_get(struct ptlrpc_nrs_policy *policy,
534 bool peek, bool force)
536 struct ptlrpc_nrs_request *nrq;
538 LASSERT(policy->pol_req_queued > 0);
540 nrq = policy->pol_desc->pd_ops->op_req_get(policy, peek, force);
542 LASSERT(ergo(nrq != NULL, nrs_request_policy(nrq) == policy));
548 * Enqueues request \a nrq for later handling, via one one the policies for
549 * which resources where earlier obtained via nrs_resource_get_safe(). The
550 * function attempts to enqueue the request first on the primary policy
551 * (if any), since this is the preferred choice.
553 * \param nrq the request being enqueued
555 * \see nrs_resource_get_safe()
557 static inline void nrs_request_enqueue(struct ptlrpc_nrs_request *nrq)
559 struct ptlrpc_nrs_policy *policy;
564 * Try in descending order, because the primary policy (if any) is
565 * the preferred choice.
567 for (i = NRS_RES_MAX - 1; i >= 0; i--) {
568 if (nrq->nr_res_ptrs[i] == NULL)
572 policy = nrq->nr_res_ptrs[i]->res_policy;
574 rc = policy->pol_desc->pd_ops->op_req_enqueue(policy, nrq);
576 policy->pol_nrs->nrs_req_queued++;
577 policy->pol_req_queued++;
582 * Should never get here, as at least the primary policy's
583 * ptlrpc_nrs_pol_ops::op_req_enqueue() implementation should always
590 * Called when a request has been handled
592 * \param[in] nrs the request that has been handled; can be used for
593 * job/resource control.
595 * \see ptlrpc_nrs_req_stop_nolock()
597 static inline void nrs_request_stop(struct ptlrpc_nrs_request *nrq)
599 struct ptlrpc_nrs_policy *policy = nrs_request_policy(nrq);
601 if (policy->pol_desc->pd_ops->op_req_stop)
602 policy->pol_desc->pd_ops->op_req_stop(policy, nrq);
604 LASSERT(policy->pol_nrs->nrs_req_started > 0);
605 LASSERT(policy->pol_req_started > 0);
607 policy->pol_nrs->nrs_req_started--;
608 policy->pol_req_started--;
612 * Handler for operations that can be carried out on policies.
614 * Handles opcodes that are common to all policy types within NRS core, and
615 * passes any unknown opcodes to the policy-specific control function.
617 * \param[in] nrs the NRS head this policy belongs to.
618 * \param[in] name the human-readable policy name; should be the same as
619 * ptlrpc_nrs_pol_desc::pd_name.
620 * \param[in] opc the opcode of the operation being carried out.
621 * \param[in,out] arg can be used to pass information in and out between when
622 * carrying an operation; usually data that is private to
623 * the policy at some level, or generic policy status
626 * \retval -ve error condition
627 * \retval 0 operation was carried out successfully
629 static int nrs_policy_ctl(struct ptlrpc_nrs *nrs, char *name,
630 enum ptlrpc_nrs_ctl opc, void *arg)
632 struct ptlrpc_nrs_policy *policy;
635 spin_lock(&nrs->nrs_lock);
637 policy = nrs_policy_find_locked(nrs, name);
638 if (policy == NULL) {
645 * Unknown opcode, pass it down to the policy-specific control
646 * function for handling.
649 rc = nrs_policy_ctl_locked(policy, opc, arg);
655 case PTLRPC_NRS_CTL_START:
656 rc = nrs_policy_start_locked(policy);
661 nrs_policy_put_locked(policy);
663 spin_unlock(&nrs->nrs_lock);
669 * Unregisters a policy by name.
671 * \param[in] nrs the NRS head this policy belongs to.
672 * \param[in] name the human-readable policy name; should be the same as
673 * ptlrpc_nrs_pol_desc::pd_name
678 static int nrs_policy_unregister(struct ptlrpc_nrs *nrs, char *name)
680 struct ptlrpc_nrs_policy *policy = NULL;
682 spin_lock(&nrs->nrs_lock);
684 policy = nrs_policy_find_locked(nrs, name);
685 if (policy == NULL) {
686 spin_unlock(&nrs->nrs_lock);
688 CERROR("Can't find NRS policy %s\n", name);
692 if (policy->pol_ref > 1) {
693 CERROR("Policy %s is busy with %d references\n", name,
694 (int)policy->pol_ref);
695 nrs_policy_put_locked(policy);
697 spin_unlock(&nrs->nrs_lock);
701 LASSERT(policy->pol_req_queued == 0);
702 LASSERT(policy->pol_req_started == 0);
704 if (policy->pol_state != NRS_POL_STATE_STOPPED) {
705 nrs_policy_stop_locked(policy);
706 LASSERT(policy->pol_state == NRS_POL_STATE_STOPPED);
709 list_del(&policy->pol_list);
712 nrs_policy_put_locked(policy);
714 spin_unlock(&nrs->nrs_lock);
716 nrs_policy_fini(policy);
718 LASSERT(policy->pol_private == NULL);
719 OBD_FREE_PTR(policy);
725 * Register a policy from \policy descriptor \a desc with NRS head \a nrs.
727 * \param[in] nrs the NRS head on which the policy will be registered.
728 * \param[in] desc the policy descriptor from which the information will be
729 * obtained to register the policy.
734 static int nrs_policy_register(struct ptlrpc_nrs *nrs,
735 struct ptlrpc_nrs_pol_desc *desc)
737 struct ptlrpc_nrs_policy *policy;
738 struct ptlrpc_nrs_policy *tmp;
739 struct ptlrpc_service_part *svcpt = nrs->nrs_svcpt;
742 LASSERT(svcpt != NULL);
743 LASSERT(desc->pd_ops != NULL);
744 LASSERT(desc->pd_ops->op_res_get != NULL);
745 LASSERT(desc->pd_ops->op_req_get != NULL);
746 LASSERT(desc->pd_ops->op_req_enqueue != NULL);
747 LASSERT(desc->pd_ops->op_req_dequeue != NULL);
748 LASSERT(desc->pd_compat != NULL);
750 OBD_CPT_ALLOC_GFP(policy, svcpt->scp_service->srv_cptable,
751 svcpt->scp_cpt, sizeof(*policy), GFP_NOFS);
755 policy->pol_nrs = nrs;
756 policy->pol_desc = desc;
757 policy->pol_state = NRS_POL_STATE_STOPPED;
758 policy->pol_flags = desc->pd_flags;
760 INIT_LIST_HEAD(&policy->pol_list);
761 INIT_LIST_HEAD(&policy->pol_list_queued);
763 rc = nrs_policy_init(policy);
765 OBD_FREE_PTR(policy);
769 spin_lock(&nrs->nrs_lock);
771 tmp = nrs_policy_find_locked(nrs, policy->pol_desc->pd_name);
773 CERROR("NRS policy %s has been registered, can't register it for %s\n",
774 policy->pol_desc->pd_name,
775 svcpt->scp_service->srv_name);
776 nrs_policy_put_locked(tmp);
778 spin_unlock(&nrs->nrs_lock);
779 nrs_policy_fini(policy);
780 OBD_FREE_PTR(policy);
785 list_add_tail(&policy->pol_list, &nrs->nrs_policy_list);
788 if (policy->pol_flags & PTLRPC_NRS_FL_REG_START)
789 rc = nrs_policy_start_locked(policy);
791 spin_unlock(&nrs->nrs_lock);
794 (void) nrs_policy_unregister(nrs, policy->pol_desc->pd_name);
800 * Enqueue request \a req using one of the policies its resources are referring
803 * \param[in] req the request to enqueue.
805 static void ptlrpc_nrs_req_add_nolock(struct ptlrpc_request *req)
807 struct ptlrpc_nrs_policy *policy;
809 LASSERT(req->rq_nrq.nr_initialized);
810 LASSERT(!req->rq_nrq.nr_enqueued);
812 nrs_request_enqueue(&req->rq_nrq);
813 req->rq_nrq.nr_enqueued = 1;
815 policy = nrs_request_policy(&req->rq_nrq);
817 * Add the policy to the NRS head's list of policies with enqueued
818 * requests, if it has not been added there.
820 if (unlikely(list_empty(&policy->pol_list_queued)))
821 list_add_tail(&policy->pol_list_queued,
822 &policy->pol_nrs->nrs_policy_queued);
826 * Enqueue a request on the high priority NRS head.
828 * \param req the request to enqueue.
830 static void ptlrpc_nrs_hpreq_add_nolock(struct ptlrpc_request *req)
832 int opc = lustre_msg_get_opc(req->rq_reqmsg);
834 spin_lock(&req->rq_lock);
836 ptlrpc_nrs_req_add_nolock(req);
838 DEBUG_REQ(D_NET, req, "high priority req");
839 spin_unlock(&req->rq_lock);
843 * Returns a boolean predicate indicating whether the policy described by
844 * \a desc is adequate for use with service \a svc.
846 * \param[in] svc the service
847 * \param[in] desc the policy descriptor
849 * \retval false the policy is not compatible with the service
850 * \retval true the policy is compatible with the service
852 static inline bool nrs_policy_compatible(const struct ptlrpc_service *svc,
853 const struct ptlrpc_nrs_pol_desc *desc)
855 return desc->pd_compat(svc, desc);
859 * Registers all compatible policies in nrs_core.nrs_policies, for NRS head
862 * \param[in] nrs the NRS head
867 * \pre mutex_is_locked(&nrs_core.nrs_mutex)
869 * \see ptlrpc_service_nrs_setup()
871 static int nrs_register_policies_locked(struct ptlrpc_nrs *nrs)
873 struct ptlrpc_nrs_pol_desc *desc;
874 /* for convenience */
875 struct ptlrpc_service_part *svcpt = nrs->nrs_svcpt;
876 struct ptlrpc_service *svc = svcpt->scp_service;
879 LASSERT(mutex_is_locked(&nrs_core.nrs_mutex));
881 list_for_each_entry(desc, &nrs_core.nrs_policies, pd_list) {
882 if (nrs_policy_compatible(svc, desc)) {
883 rc = nrs_policy_register(nrs, desc);
885 CERROR("Failed to register NRS policy %s for partition %d of service %s: %d\n",
886 desc->pd_name, svcpt->scp_cpt,
889 * Fail registration if any of the policies'
890 * registration fails.
901 * Initializes NRS head \a nrs of service partition \a svcpt, and registers all
902 * compatible policies in NRS core, with the NRS head.
904 * \param[in] nrs the NRS head
905 * \param[in] svcpt the PTLRPC service partition to setup
910 * \pre mutex_is_locked(&nrs_core.nrs_mutex)
912 static int nrs_svcpt_setup_locked0(struct ptlrpc_nrs *nrs,
913 struct ptlrpc_service_part *svcpt)
916 enum ptlrpc_nrs_queue_type queue;
918 LASSERT(mutex_is_locked(&nrs_core.nrs_mutex));
920 if (nrs == &svcpt->scp_nrs_reg)
921 queue = PTLRPC_NRS_QUEUE_REG;
922 else if (nrs == svcpt->scp_nrs_hp)
923 queue = PTLRPC_NRS_QUEUE_HP;
927 nrs->nrs_svcpt = svcpt;
928 nrs->nrs_queue_type = queue;
929 spin_lock_init(&nrs->nrs_lock);
930 INIT_LIST_HEAD(&nrs->nrs_policy_list);
931 INIT_LIST_HEAD(&nrs->nrs_policy_queued);
933 rc = nrs_register_policies_locked(nrs);
939 * Allocates a regular and optionally a high-priority NRS head (if the service
940 * handles high-priority RPCs), and then registers all available compatible
941 * policies on those NRS heads.
943 * \param[in,out] svcpt the PTLRPC service partition to setup
945 * \pre mutex_is_locked(&nrs_core.nrs_mutex)
947 static int nrs_svcpt_setup_locked(struct ptlrpc_service_part *svcpt)
949 struct ptlrpc_nrs *nrs;
952 LASSERT(mutex_is_locked(&nrs_core.nrs_mutex));
955 * Initialize the regular NRS head.
957 nrs = nrs_svcpt2nrs(svcpt, false);
958 rc = nrs_svcpt_setup_locked0(nrs, svcpt);
963 * Optionally allocate a high-priority NRS head.
965 if (svcpt->scp_service->srv_ops.so_hpreq_handler == NULL)
968 OBD_CPT_ALLOC_PTR(svcpt->scp_nrs_hp,
969 svcpt->scp_service->srv_cptable,
971 if (svcpt->scp_nrs_hp == NULL) {
976 nrs = nrs_svcpt2nrs(svcpt, true);
977 rc = nrs_svcpt_setup_locked0(nrs, svcpt);
984 * Unregisters all policies on all available NRS heads in a service partition;
985 * called at PTLRPC service unregistration time.
987 * \param[in] svcpt the PTLRPC service partition
989 * \pre mutex_is_locked(&nrs_core.nrs_mutex)
991 static void nrs_svcpt_cleanup_locked(struct ptlrpc_service_part *svcpt)
993 struct ptlrpc_nrs *nrs;
994 struct ptlrpc_nrs_policy *policy;
995 struct ptlrpc_nrs_policy *tmp;
999 LASSERT(mutex_is_locked(&nrs_core.nrs_mutex));
1002 nrs = nrs_svcpt2nrs(svcpt, hp);
1003 nrs->nrs_stopping = 1;
1005 list_for_each_entry_safe(policy, tmp, &nrs->nrs_policy_list,
1007 rc = nrs_policy_unregister(nrs, policy->pol_desc->pd_name);
1012 * If the service partition has an HP NRS head, clean that up as well.
1014 if (!hp && nrs_svcpt_has_hp(svcpt)) {
1024 * Returns the descriptor for a policy as identified by by \a name.
1026 * \param[in] name the policy name
1028 * \retval the policy descriptor
1031 static struct ptlrpc_nrs_pol_desc *nrs_policy_find_desc_locked(const char *name)
1033 struct ptlrpc_nrs_pol_desc *tmp;
1035 list_for_each_entry(tmp, &nrs_core.nrs_policies, pd_list) {
1036 if (strncmp(tmp->pd_name, name, NRS_POL_NAME_MAX) == 0)
1043 * Removes the policy from all supported NRS heads of all partitions of all
1046 * \param[in] desc the policy descriptor to unregister
1049 * \retval 0 successfully unregistered policy on all supported NRS heads
1051 * \pre mutex_is_locked(&nrs_core.nrs_mutex)
1052 * \pre mutex_is_locked(&ptlrpc_all_services_mutex)
1054 static int nrs_policy_unregister_locked(struct ptlrpc_nrs_pol_desc *desc)
1056 struct ptlrpc_nrs *nrs;
1057 struct ptlrpc_service *svc;
1058 struct ptlrpc_service_part *svcpt;
1062 LASSERT(mutex_is_locked(&nrs_core.nrs_mutex));
1063 LASSERT(mutex_is_locked(&ptlrpc_all_services_mutex));
1065 list_for_each_entry(svc, &ptlrpc_all_services, srv_list) {
1067 if (!nrs_policy_compatible(svc, desc) ||
1068 unlikely(svc->srv_is_stopping))
1071 ptlrpc_service_for_each_part(svcpt, i, svc) {
1075 nrs = nrs_svcpt2nrs(svcpt, hp);
1076 rc = nrs_policy_unregister(nrs, desc->pd_name);
1078 * Ignore -ENOENT as the policy may not have registered
1079 * successfully on all service partitions.
1081 if (rc == -ENOENT) {
1083 } else if (rc != 0) {
1084 CERROR("Failed to unregister NRS policy %s for partition %d of service %s: %d\n",
1085 desc->pd_name, svcpt->scp_cpt,
1086 svcpt->scp_service->srv_name, rc);
1090 if (!hp && nrs_svc_has_hp(svc)) {
1096 if (desc->pd_ops->op_lprocfs_fini != NULL)
1097 desc->pd_ops->op_lprocfs_fini(svc);
1104 * Registers a new policy with NRS core.
1106 * The function will only succeed if policy registration with all compatible
1107 * service partitions (if any) is successful.
1109 * N.B. This function should be called either at ptlrpc module initialization
1110 * time when registering a policy that ships with NRS core, or in a
1111 * module's init() function for policies registering from other modules.
1113 * \param[in] conf configuration information for the new policy to register
1118 int ptlrpc_nrs_policy_register(struct ptlrpc_nrs_pol_conf *conf)
1120 struct ptlrpc_service *svc;
1121 struct ptlrpc_nrs_pol_desc *desc;
1124 LASSERT(conf != NULL);
1125 LASSERT(conf->nc_ops != NULL);
1126 LASSERT(conf->nc_compat != NULL);
1127 LASSERT(ergo(conf->nc_compat == nrs_policy_compat_one,
1128 conf->nc_compat_svc_name != NULL));
1129 LASSERT(ergo((conf->nc_flags & PTLRPC_NRS_FL_REG_EXTERN) != 0,
1130 conf->nc_owner != NULL));
1132 conf->nc_name[NRS_POL_NAME_MAX - 1] = '\0';
1135 * External policies are not allowed to start immediately upon
1136 * registration, as there is a relatively higher chance that their
1137 * registration might fail. In such a case, some policy instances may
1138 * already have requests queued wen unregistration needs to happen as
1139 * part o cleanup; since there is currently no way to drain requests
1140 * from a policy unless the service is unregistering, we just disallow
1143 if ((conf->nc_flags & PTLRPC_NRS_FL_REG_EXTERN) &&
1144 (conf->nc_flags & (PTLRPC_NRS_FL_FALLBACK |
1145 PTLRPC_NRS_FL_REG_START))) {
1146 CERROR("NRS: failing to register policy %s. Please check policy flags; external policies cannot act as fallback policies, or be started immediately upon registration without interaction with lprocfs\n",
1151 mutex_lock(&nrs_core.nrs_mutex);
1153 if (nrs_policy_find_desc_locked(conf->nc_name) != NULL) {
1154 CERROR("NRS: failing to register policy %s which has already been registered with NRS core!\n",
1160 OBD_ALLOC_PTR(desc);
1166 strncpy(desc->pd_name, conf->nc_name, NRS_POL_NAME_MAX);
1167 desc->pd_ops = conf->nc_ops;
1168 desc->pd_compat = conf->nc_compat;
1169 desc->pd_compat_svc_name = conf->nc_compat_svc_name;
1170 if ((conf->nc_flags & PTLRPC_NRS_FL_REG_EXTERN) != 0)
1171 desc->pd_owner = conf->nc_owner;
1172 desc->pd_flags = conf->nc_flags;
1173 atomic_set(&desc->pd_refs, 0);
1176 * For policies that are held in the same module as NRS (currently
1177 * ptlrpc), do not register the policy with all compatible services,
1178 * as the services will not have started at this point, since we are
1179 * calling from ptlrpc module initialization code. In such cases each
1180 * service will register all compatible policies later, via
1181 * ptlrpc_service_nrs_setup().
1183 if ((conf->nc_flags & PTLRPC_NRS_FL_REG_EXTERN) == 0)
1187 * Register the new policy on all compatible services
1189 mutex_lock(&ptlrpc_all_services_mutex);
1191 list_for_each_entry(svc, &ptlrpc_all_services, srv_list) {
1192 struct ptlrpc_service_part *svcpt;
1196 if (!nrs_policy_compatible(svc, desc) ||
1197 unlikely(svc->srv_is_stopping))
1200 ptlrpc_service_for_each_part(svcpt, i, svc) {
1201 struct ptlrpc_nrs *nrs;
1204 nrs = nrs_svcpt2nrs(svcpt, hp);
1205 rc = nrs_policy_register(nrs, desc);
1207 CERROR("Failed to register NRS policy %s for partition %d of service %s: %d\n",
1208 desc->pd_name, svcpt->scp_cpt,
1209 svcpt->scp_service->srv_name, rc);
1211 rc2 = nrs_policy_unregister_locked(desc);
1213 * Should not fail at this point
1216 mutex_unlock(&ptlrpc_all_services_mutex);
1221 if (!hp && nrs_svc_has_hp(svc)) {
1228 * No need to take a reference to other modules here, as we
1229 * will be calling from the module's init() function.
1231 if (desc->pd_ops->op_lprocfs_init != NULL) {
1232 rc = desc->pd_ops->op_lprocfs_init(svc);
1234 rc2 = nrs_policy_unregister_locked(desc);
1236 * Should not fail at this point
1239 mutex_unlock(&ptlrpc_all_services_mutex);
1246 mutex_unlock(&ptlrpc_all_services_mutex);
1248 list_add_tail(&desc->pd_list, &nrs_core.nrs_policies);
1250 mutex_unlock(&nrs_core.nrs_mutex);
1254 EXPORT_SYMBOL(ptlrpc_nrs_policy_register);
1257 * Unregisters a previously registered policy with NRS core. All instances of
1258 * the policy on all NRS heads of all supported services are removed.
1260 * N.B. This function should only be called from a module's exit() function.
1261 * Although it can be used for policies that ship alongside NRS core, the
1262 * function is primarily intended for policies that register externally,
1263 * from other modules.
1265 * \param[in] conf configuration information for the policy to unregister
1270 int ptlrpc_nrs_policy_unregister(struct ptlrpc_nrs_pol_conf *conf)
1272 struct ptlrpc_nrs_pol_desc *desc;
1275 LASSERT(conf != NULL);
1277 if (conf->nc_flags & PTLRPC_NRS_FL_FALLBACK) {
1278 CERROR("Unable to unregister a fallback policy, unless the PTLRPC service is stopping.\n");
1282 conf->nc_name[NRS_POL_NAME_MAX - 1] = '\0';
1284 mutex_lock(&nrs_core.nrs_mutex);
1286 desc = nrs_policy_find_desc_locked(conf->nc_name);
1288 CERROR("Failing to unregister NRS policy %s which has not been registered with NRS core!\n",
1294 mutex_lock(&ptlrpc_all_services_mutex);
1296 rc = nrs_policy_unregister_locked(desc);
1299 CERROR("Please first stop policy %s on all service partitions and then retry to unregister the policy.\n",
1304 CDEBUG(D_INFO, "Unregistering policy %s from NRS core.\n",
1307 list_del(&desc->pd_list);
1311 mutex_unlock(&ptlrpc_all_services_mutex);
1314 mutex_unlock(&nrs_core.nrs_mutex);
1318 EXPORT_SYMBOL(ptlrpc_nrs_policy_unregister);
1321 * Setup NRS heads on all service partitions of service \a svc, and register
1322 * all compatible policies on those NRS heads.
1324 * To be called from within ptl
1325 * \param[in] svc the service to setup
1327 * \retval -ve error, the calling logic should eventually call
1328 * ptlrpc_service_nrs_cleanup() to undo any work performed
1331 * \see ptlrpc_register_service()
1332 * \see ptlrpc_service_nrs_cleanup()
1334 int ptlrpc_service_nrs_setup(struct ptlrpc_service *svc)
1336 struct ptlrpc_service_part *svcpt;
1337 const struct ptlrpc_nrs_pol_desc *desc;
1341 mutex_lock(&nrs_core.nrs_mutex);
1344 * Initialize NRS heads on all service CPTs.
1346 ptlrpc_service_for_each_part(svcpt, i, svc) {
1347 rc = nrs_svcpt_setup_locked(svcpt);
1353 * Set up lprocfs interfaces for all supported policies for the
1356 list_for_each_entry(desc, &nrs_core.nrs_policies, pd_list) {
1357 if (!nrs_policy_compatible(svc, desc))
1360 if (desc->pd_ops->op_lprocfs_init != NULL) {
1361 rc = desc->pd_ops->op_lprocfs_init(svc);
1369 mutex_unlock(&nrs_core.nrs_mutex);
1375 * Unregisters all policies on all service partitions of service \a svc.
1377 * \param[in] svc the PTLRPC service to unregister
1379 void ptlrpc_service_nrs_cleanup(struct ptlrpc_service *svc)
1381 struct ptlrpc_service_part *svcpt;
1382 const struct ptlrpc_nrs_pol_desc *desc;
1385 mutex_lock(&nrs_core.nrs_mutex);
1388 * Clean up NRS heads on all service partitions
1390 ptlrpc_service_for_each_part(svcpt, i, svc)
1391 nrs_svcpt_cleanup_locked(svcpt);
1394 * Clean up lprocfs interfaces for all supported policies for the
1397 list_for_each_entry(desc, &nrs_core.nrs_policies, pd_list) {
1398 if (!nrs_policy_compatible(svc, desc))
1401 if (desc->pd_ops->op_lprocfs_fini != NULL)
1402 desc->pd_ops->op_lprocfs_fini(svc);
1405 mutex_unlock(&nrs_core.nrs_mutex);
1409 * Obtains NRS head resources for request \a req.
1411 * These could be either on the regular or HP NRS head of \a svcpt; resources
1412 * taken on the regular head can later be swapped for HP head resources by
1413 * ldlm_lock_reorder_req().
1415 * \param[in] svcpt the service partition
1416 * \param[in] req the request
1417 * \param[in] hp which NRS head of \a svcpt to use
1419 void ptlrpc_nrs_req_initialize(struct ptlrpc_service_part *svcpt,
1420 struct ptlrpc_request *req, bool hp)
1422 struct ptlrpc_nrs *nrs = nrs_svcpt2nrs(svcpt, hp);
1424 memset(&req->rq_nrq, 0, sizeof(req->rq_nrq));
1425 nrs_resource_get_safe(nrs, &req->rq_nrq, req->rq_nrq.nr_res_ptrs,
1429 * It is fine to access \e nr_initialized without locking as there is
1430 * no contention at this early stage.
1432 req->rq_nrq.nr_initialized = 1;
1436 * Releases resources for a request; is called after the request has been
1439 * \param[in] req the request
1441 * \see ptlrpc_server_finish_request()
1443 void ptlrpc_nrs_req_finalize(struct ptlrpc_request *req)
1445 if (req->rq_nrq.nr_initialized) {
1446 nrs_resource_put_safe(req->rq_nrq.nr_res_ptrs);
1447 /* no protection on bit nr_initialized because no
1448 * contention at this late stage */
1449 req->rq_nrq.nr_finalized = 1;
1453 void ptlrpc_nrs_req_stop_nolock(struct ptlrpc_request *req)
1455 if (req->rq_nrq.nr_started)
1456 nrs_request_stop(&req->rq_nrq);
1460 * Enqueues request \a req on either the regular or high-priority NRS head
1461 * of service partition \a svcpt.
1463 * \param[in] svcpt the service partition
1464 * \param[in] req the request to be enqueued
1465 * \param[in] hp whether to enqueue the request on the regular or
1466 * high-priority NRS head.
1468 void ptlrpc_nrs_req_add(struct ptlrpc_service_part *svcpt,
1469 struct ptlrpc_request *req, bool hp)
1471 spin_lock(&svcpt->scp_req_lock);
1474 ptlrpc_nrs_hpreq_add_nolock(req);
1476 ptlrpc_nrs_req_add_nolock(req);
1478 spin_unlock(&svcpt->scp_req_lock);
1481 static void nrs_request_removed(struct ptlrpc_nrs_policy *policy)
1483 LASSERT(policy->pol_nrs->nrs_req_queued > 0);
1484 LASSERT(policy->pol_req_queued > 0);
1486 policy->pol_nrs->nrs_req_queued--;
1487 policy->pol_req_queued--;
1490 * If the policy has no more requests queued, remove it from
1491 * ptlrpc_nrs::nrs_policy_queued.
1493 if (unlikely(policy->pol_req_queued == 0)) {
1494 list_del_init(&policy->pol_list_queued);
1497 * If there are other policies with queued requests, move the
1498 * current policy to the end so that we can round robin over
1499 * all policies and drain the requests.
1501 } else if (policy->pol_req_queued != policy->pol_nrs->nrs_req_queued) {
1502 LASSERT(policy->pol_req_queued <
1503 policy->pol_nrs->nrs_req_queued);
1505 list_move_tail(&policy->pol_list_queued,
1506 &policy->pol_nrs->nrs_policy_queued);
1511 * Obtains a request for handling from an NRS head of service partition
1514 * \param[in] svcpt the service partition
1515 * \param[in] hp whether to obtain a request from the regular or
1516 * high-priority NRS head.
1517 * \param[in] peek when set, signifies that we just want to examine the
1518 * request, and not handle it, so the request is not removed
1520 * \param[in] force when set, it will force a policy to return a request if it
1523 * \retval the request to be handled
1524 * \retval NULL the head has no requests to serve
1526 struct ptlrpc_request *
1527 ptlrpc_nrs_req_get_nolock0(struct ptlrpc_service_part *svcpt, bool hp,
1528 bool peek, bool force)
1530 struct ptlrpc_nrs *nrs = nrs_svcpt2nrs(svcpt, hp);
1531 struct ptlrpc_nrs_policy *policy;
1532 struct ptlrpc_nrs_request *nrq;
1535 * Always try to drain requests from all NRS polices even if they are
1536 * inactive, because the user can change policy status at runtime.
1538 list_for_each_entry(policy, &nrs->nrs_policy_queued,
1540 nrq = nrs_request_get(policy, peek, force);
1542 if (likely(!peek)) {
1543 nrq->nr_started = 1;
1545 policy->pol_req_started++;
1546 policy->pol_nrs->nrs_req_started++;
1548 nrs_request_removed(policy);
1551 return container_of(nrq, struct ptlrpc_request, rq_nrq);
1559 * Dequeues request \a req from the policy it has been enqueued on.
1561 * \param[in] req the request
1563 void ptlrpc_nrs_req_del_nolock(struct ptlrpc_request *req)
1565 struct ptlrpc_nrs_policy *policy = nrs_request_policy(&req->rq_nrq);
1567 policy->pol_desc->pd_ops->op_req_dequeue(policy, &req->rq_nrq);
1569 req->rq_nrq.nr_enqueued = 0;
1571 nrs_request_removed(policy);
1575 * Returns whether there are any requests currently enqueued on any of the
1576 * policies of service partition's \a svcpt NRS head specified by \a hp. Should
1577 * be called while holding ptlrpc_service_part::scp_req_lock to get a reliable
1580 * \param[in] svcpt the service partition to enquire.
1581 * \param[in] hp whether the regular or high-priority NRS head is to be
1584 * \retval false the indicated NRS head has no enqueued requests.
1585 * \retval true the indicated NRS head has some enqueued requests.
1587 bool ptlrpc_nrs_req_pending_nolock(struct ptlrpc_service_part *svcpt, bool hp)
1589 struct ptlrpc_nrs *nrs = nrs_svcpt2nrs(svcpt, hp);
1591 return nrs->nrs_req_queued > 0;
1595 * Moves request \a req from the regular to the high-priority NRS head.
1597 * \param[in] req the request to move
1599 void ptlrpc_nrs_req_hp_move(struct ptlrpc_request *req)
1601 struct ptlrpc_service_part *svcpt = req->rq_rqbd->rqbd_svcpt;
1602 struct ptlrpc_nrs_request *nrq = &req->rq_nrq;
1603 struct ptlrpc_nrs_resource *res1[NRS_RES_MAX];
1604 struct ptlrpc_nrs_resource *res2[NRS_RES_MAX];
1607 * Obtain the high-priority NRS head resources.
1609 nrs_resource_get_safe(nrs_svcpt2nrs(svcpt, true), nrq, res1, true);
1611 spin_lock(&svcpt->scp_req_lock);
1613 if (!ptlrpc_nrs_req_can_move(req))
1616 ptlrpc_nrs_req_del_nolock(req);
1618 memcpy(res2, nrq->nr_res_ptrs, NRS_RES_MAX * sizeof(res2[0]));
1619 memcpy(nrq->nr_res_ptrs, res1, NRS_RES_MAX * sizeof(res1[0]));
1621 ptlrpc_nrs_hpreq_add_nolock(req);
1623 memcpy(res1, res2, NRS_RES_MAX * sizeof(res1[0]));
1625 spin_unlock(&svcpt->scp_req_lock);
1628 * Release either the regular NRS head resources if we moved the
1629 * request, or the high-priority NRS head resources if we took a
1630 * reference earlier in this function and ptlrpc_nrs_req_can_move()
1633 nrs_resource_put_safe(res1);
1637 * Carries out a control operation \a opc on the policy identified by the
1638 * human-readable \a name, on either all partitions, or only on the first
1639 * partition of service \a svc.
1641 * \param[in] svc the service the policy belongs to.
1642 * \param[in] queue whether to carry out the command on the policy which
1643 * belongs to the regular, high-priority, or both NRS
1644 * heads of service partitions of \a svc.
1645 * \param[in] name the policy to act upon, by human-readable name
1646 * \param[in] opc the opcode of the operation to carry out
1647 * \param[in] single when set, the operation will only be carried out on the
1648 * NRS heads of the first service partition of \a svc.
1649 * This is useful for some policies which e.g. share
1650 * identical values on the same parameters of different
1651 * service partitions; when reading these parameters via
1652 * lprocfs, these policies may just want to obtain and
1653 * print out the values from the first service partition.
1654 * Storing these values centrally elsewhere then could be
1655 * another solution for this.
1656 * \param[in,out] arg can be used as a generic in/out buffer between control
1657 * operations and the user environment.
1659 *\retval -ve error condition
1660 *\retval 0 operation was carried out successfully
1662 int ptlrpc_nrs_policy_control(const struct ptlrpc_service *svc,
1663 enum ptlrpc_nrs_queue_type queue, char *name,
1664 enum ptlrpc_nrs_ctl opc, bool single, void *arg)
1666 struct ptlrpc_service_part *svcpt;
1670 LASSERT(opc != PTLRPC_NRS_CTL_INVALID);
1672 if ((queue & PTLRPC_NRS_QUEUE_BOTH) == 0)
1675 ptlrpc_service_for_each_part(svcpt, i, svc) {
1676 if ((queue & PTLRPC_NRS_QUEUE_REG) != 0) {
1677 rc = nrs_policy_ctl(nrs_svcpt2nrs(svcpt, false), name,
1679 if (rc != 0 || (queue == PTLRPC_NRS_QUEUE_REG &&
1684 if ((queue & PTLRPC_NRS_QUEUE_HP) != 0) {
1686 * XXX: We could optionally check for
1687 * nrs_svc_has_hp(svc) here, and return an error if it
1688 * is false. Right now we rely on the policies' lprocfs
1689 * handlers that call the present function to make this
1690 * check; if they fail to do so, they might hit the
1691 * assertion inside nrs_svcpt2nrs() below.
1693 rc = nrs_policy_ctl(nrs_svcpt2nrs(svcpt, true), name,
1695 if (rc != 0 || single)
1704 /* ptlrpc/nrs_fifo.c */
1705 extern struct ptlrpc_nrs_pol_conf nrs_conf_fifo;
1708 * Adds all policies that ship with the ptlrpc module, to NRS core's list of
1709 * policies \e nrs_core.nrs_policies.
1711 * \retval 0 all policies have been registered successfully
1714 int ptlrpc_nrs_init(void)
1718 mutex_init(&nrs_core.nrs_mutex);
1719 INIT_LIST_HEAD(&nrs_core.nrs_policies);
1721 rc = ptlrpc_nrs_policy_register(&nrs_conf_fifo);
1729 * Since no PTLRPC services have been started at this point, all we need
1730 * to do for cleanup is to free the descriptors.
1738 * Removes all policy descriptors from nrs_core::nrs_policies, and frees the
1739 * policy descriptors.
1741 * Since all PTLRPC services are stopped at this point, there are no more
1742 * instances of any policies, because each service will have stopped its policy
1743 * instances in ptlrpc_service_nrs_cleanup(), so we just need to free the
1746 void ptlrpc_nrs_fini(void)
1748 struct ptlrpc_nrs_pol_desc *desc;
1749 struct ptlrpc_nrs_pol_desc *tmp;
1751 list_for_each_entry_safe(desc, tmp, &nrs_core.nrs_policies,
1753 list_del_init(&desc->pd_list);