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) 2014, 2016, 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 <obd_support.h>
44 #include <obd_class.h>
45 #include <lustre_net.h>
46 #include <lprocfs_status.h>
47 #include <libcfs/libcfs.h>
48 #include "ptlrpc_internal.h"
53 struct nrs_core nrs_core;
55 static int nrs_policy_init(struct ptlrpc_nrs_policy *policy)
57 return policy->pol_desc->pd_ops->op_policy_init != NULL ?
58 policy->pol_desc->pd_ops->op_policy_init(policy) : 0;
61 static void nrs_policy_fini(struct ptlrpc_nrs_policy *policy)
63 LASSERT(policy->pol_ref == 0);
64 LASSERT(refcount_read(&policy->pol_start_ref) == 0);
65 LASSERT(policy->pol_req_queued == 0);
67 if (policy->pol_desc->pd_ops->op_policy_fini != NULL)
68 policy->pol_desc->pd_ops->op_policy_fini(policy);
71 static int nrs_policy_ctl_locked(struct ptlrpc_nrs_policy *policy,
72 enum ptlrpc_nrs_ctl opc, void *arg)
75 * The policy may be stopped, but the lprocfs files and
76 * ptlrpc_nrs_policy instances remain present until unregistration time.
77 * Do not perform the ctl operation if the policy is stopped, as
78 * policy->pol_private will be NULL in such a case.
80 if (policy->pol_state == NRS_POL_STATE_STOPPED)
83 RETURN(policy->pol_desc->pd_ops->op_policy_ctl != NULL ?
84 policy->pol_desc->pd_ops->op_policy_ctl(policy, opc, arg) :
88 static void nrs_policy_stop0(struct ptlrpc_nrs_policy *policy)
92 if (policy->pol_desc->pd_ops->op_policy_stop != NULL)
93 policy->pol_desc->pd_ops->op_policy_stop(policy);
95 LASSERT(list_empty(&policy->pol_list_queued));
96 LASSERT(policy->pol_req_queued == 0 &&
97 policy->pol_req_started == 0);
99 policy->pol_private = NULL;
100 policy->pol_arg[0] = '\0';
102 policy->pol_state = NRS_POL_STATE_STOPPED;
103 wake_up(&policy->pol_wq);
105 if (atomic_dec_and_test(&policy->pol_desc->pd_refs))
106 module_put(policy->pol_desc->pd_owner);
112 * Increases the policy's usage started reference count.
114 static inline void nrs_policy_started_get(struct ptlrpc_nrs_policy *policy)
116 refcount_inc(&policy->pol_start_ref);
120 * Decreases the policy's usage started reference count, and stops the policy
121 * in case it was already stopping and have no more outstanding usage
122 * references (which indicates it has no more queued or started requests, and
123 * can be safely stopped).
125 static void nrs_policy_started_put(struct ptlrpc_nrs_policy *policy)
127 if (refcount_dec_and_test(&policy->pol_start_ref))
128 nrs_policy_stop0(policy);
131 static int nrs_policy_stop_locked(struct ptlrpc_nrs_policy *policy)
133 struct ptlrpc_nrs *nrs = policy->pol_nrs;
136 if (nrs->nrs_policy_fallback == policy && !nrs->nrs_stopping)
139 if (policy->pol_state == NRS_POL_STATE_STARTING)
142 /* In progress or already stopped */
143 if (policy->pol_state != NRS_POL_STATE_STARTED)
146 policy->pol_state = NRS_POL_STATE_STOPPING;
148 /* Immediately make it invisible */
149 if (nrs->nrs_policy_primary == policy) {
150 nrs->nrs_policy_primary = NULL;
153 LASSERT(nrs->nrs_policy_fallback == policy);
154 nrs->nrs_policy_fallback = NULL;
157 /* Drop started ref and wait for requests to be drained */
158 spin_unlock(&nrs->nrs_lock);
159 nrs_policy_started_put(policy);
161 wait_event_timeout(policy->pol_wq,
162 policy->pol_state == NRS_POL_STATE_STOPPED,
163 cfs_time_seconds(30));
165 spin_lock(&nrs->nrs_lock);
167 if (policy->pol_state != NRS_POL_STATE_STOPPED)
174 * Transitions the \a nrs NRS head's primary policy to
175 * ptlrpc_nrs_pol_state::NRS_POL_STATE_STOPPING and if the policy has no
176 * pending usage references, to ptlrpc_nrs_pol_state::NRS_POL_STATE_STOPPED.
178 * \param[in] nrs the NRS head to carry out this operation on
180 static void nrs_policy_stop_primary(struct ptlrpc_nrs *nrs)
182 struct ptlrpc_nrs_policy *tmp = nrs->nrs_policy_primary;
187 * XXX: This should really be RETURN_EXIT, but the latter does
188 * not currently print anything out, and possibly should be
195 nrs->nrs_policy_primary = NULL;
197 LASSERT(tmp->pol_state == NRS_POL_STATE_STARTED);
198 tmp->pol_state = NRS_POL_STATE_STOPPING;
200 /* Drop started ref to free the policy */
201 spin_unlock(&nrs->nrs_lock);
202 nrs_policy_started_put(tmp);
203 spin_lock(&nrs->nrs_lock);
208 * Transitions a policy across the ptlrpc_nrs_pol_state range of values, in
209 * response to an lprocfs command to start a policy.
211 * If a primary policy different to the current one is specified, this function
212 * will transition the new policy to the
213 * ptlrpc_nrs_pol_state::NRS_POL_STATE_STARTING and then to
214 * ptlrpc_nrs_pol_state::NRS_POL_STATE_STARTED, and will then transition
215 * the old primary policy (if there is one) to
216 * ptlrpc_nrs_pol_state::NRS_POL_STATE_STOPPING, and if there are no outstanding
217 * references on the policy to ptlrpc_nrs_pol_stae::NRS_POL_STATE_STOPPED.
219 * If the fallback policy is specified, this is taken to indicate an instruction
220 * to stop the current primary policy, without substituting it with another
221 * primary policy, so the primary policy (if any) is transitioned to
222 * ptlrpc_nrs_pol_state::NRS_POL_STATE_STOPPING, and if there are no outstanding
223 * references on the policy to ptlrpc_nrs_pol_stae::NRS_POL_STATE_STOPPED. In
224 * this case, the fallback policy is only left active in the NRS head.
226 static int nrs_policy_start_locked(struct ptlrpc_nrs_policy *policy, char *arg)
228 struct ptlrpc_nrs *nrs = policy->pol_nrs;
229 struct ptlrpc_service_part *svcpt = nrs->nrs_svcpt;
230 char *srv_name = svcpt->scp_service->srv_name;
235 * Don't allow multiple starting which is too complex, and has no real
238 if (nrs->nrs_policy_starting)
241 LASSERT(policy->pol_state != NRS_POL_STATE_STARTING);
243 if (policy->pol_state == NRS_POL_STATE_STOPPING)
246 if (arg && strlen(arg) >= sizeof(policy->pol_arg)) {
248 CWARN("%s.%d NRS: arg '%s' is too long: rc = %d\n",
249 srv_name, svcpt->scp_cpt, arg, rc);
253 if (policy->pol_flags & PTLRPC_NRS_FL_FALLBACK) {
255 * This is for cases in which the user sets the policy to the
256 * fallback policy (currently fifo for all services); i.e. the
257 * user is resetting the policy to the default; so we stop the
258 * primary policy, if any.
260 if (policy == nrs->nrs_policy_fallback) {
261 nrs_policy_stop_primary(nrs);
266 * If we reach here, we must be setting up the fallback policy
267 * at service startup time, and only a single policy with the
268 * nrs_policy_flags::PTLRPC_NRS_FL_FALLBACK flag set can
269 * register with NRS core.
271 LASSERT(nrs->nrs_policy_fallback == NULL);
274 * Shouldn't start primary policy if w/o fallback policy.
276 if (nrs->nrs_policy_fallback == NULL)
279 if (policy->pol_state == NRS_POL_STATE_STARTED) {
281 * If the policy argument now is different from the last time,
282 * stop the policy first and start it again with the new
285 if ((arg == NULL && strlen(policy->pol_arg) == 0) ||
286 (arg != NULL && strcmp(policy->pol_arg, arg) == 0))
289 rc = nrs_policy_stop_locked(policy);
296 * Increase the module usage count for policies registering from other
299 if (atomic_inc_return(&policy->pol_desc->pd_refs) == 1 &&
300 !try_module_get(policy->pol_desc->pd_owner)) {
301 atomic_dec(&policy->pol_desc->pd_refs);
303 CERROR("%s.%d NRS: cannot get module for policy %s (is it alive?): rc = %d\n",
304 srv_name, svcpt->scp_cpt, policy->pol_desc->pd_name, rc);
309 * Serialize policy starting across the NRS head
311 nrs->nrs_policy_starting = 1;
313 policy->pol_state = NRS_POL_STATE_STARTING;
315 if (policy->pol_desc->pd_ops->op_policy_start) {
316 spin_unlock(&nrs->nrs_lock);
318 rc = policy->pol_desc->pd_ops->op_policy_start(policy, arg);
320 spin_lock(&nrs->nrs_lock);
322 if (atomic_dec_and_test(&policy->pol_desc->pd_refs))
323 module_put(policy->pol_desc->pd_owner);
325 policy->pol_state = NRS_POL_STATE_STOPPED;
331 strlcpy(policy->pol_arg, arg, sizeof(policy->pol_arg));
333 /* take the started reference */
334 refcount_set(&policy->pol_start_ref, 1);
335 policy->pol_state = NRS_POL_STATE_STARTED;
337 if (policy->pol_flags & PTLRPC_NRS_FL_FALLBACK) {
339 * This path is only used at PTLRPC service setup time.
341 nrs->nrs_policy_fallback = policy;
344 * Try to stop the current primary policy if there is one.
346 nrs_policy_stop_primary(nrs);
349 * And set the newly-started policy as the primary one.
351 nrs->nrs_policy_primary = policy;
355 nrs->nrs_policy_starting = 0;
361 * Increases the policy's usage reference count (caller count).
363 static inline void nrs_policy_get_locked(struct ptlrpc_nrs_policy *policy)
364 __must_hold(&policy->pol_nrs->nrs_lock)
370 * Decreases the policy's usage reference count.
372 static void nrs_policy_put_locked(struct ptlrpc_nrs_policy *policy)
373 __must_hold(&policy->pol_nrs->nrs_lock)
375 LASSERT(policy->pol_ref > 0);
381 * Find and return a policy by name.
383 static struct ptlrpc_nrs_policy * nrs_policy_find_locked(struct ptlrpc_nrs *nrs,
386 struct ptlrpc_nrs_policy *tmp;
388 list_for_each_entry(tmp, &nrs->nrs_policy_list, pol_list) {
389 if (strncmp(tmp->pol_desc->pd_name, name,
390 NRS_POL_NAME_MAX) == 0) {
391 nrs_policy_get_locked(tmp);
399 * Release references for the resource hierarchy moving upwards towards the
400 * policy instance resource.
402 static void nrs_resource_put(struct ptlrpc_nrs_resource *res)
404 struct ptlrpc_nrs_policy *policy = res->res_policy;
406 if (policy->pol_desc->pd_ops->op_res_put != NULL) {
407 struct ptlrpc_nrs_resource *parent;
409 for (; res != NULL; res = parent) {
410 parent = res->res_parent;
411 policy->pol_desc->pd_ops->op_res_put(policy, res);
417 * Obtains references for each resource in the resource hierarchy for request
418 * \a nrq if it is to be handled by \a policy.
420 * \param[in] policy the policy
421 * \param[in] nrq the request
422 * \param[in] moving_req denotes whether this is a call to the function by
423 * ldlm_lock_reorder_req(), in order to move \a nrq to
424 * the high-priority NRS head; we should not sleep when
427 * \retval NULL resource hierarchy references not obtained
428 * \retval valid-pointer the bottom level of the resource hierarchy
430 * \see ptlrpc_nrs_pol_ops::op_res_get()
433 struct ptlrpc_nrs_resource * nrs_resource_get(struct ptlrpc_nrs_policy *policy,
434 struct ptlrpc_nrs_request *nrq,
438 * Set to NULL to traverse the resource hierarchy from the top.
440 struct ptlrpc_nrs_resource *res = NULL;
441 struct ptlrpc_nrs_resource *tmp = NULL;
445 rc = policy->pol_desc->pd_ops->op_res_get(policy, nrq, res,
449 nrs_resource_put(res);
453 LASSERT(tmp != NULL);
454 tmp->res_parent = res;
455 tmp->res_policy = policy;
459 * Return once we have obtained a reference to the bottom level
460 * of the resource hierarchy.
468 * Obtains resources for the resource hierarchies and policy references for
469 * the fallback and current primary policy (if any), that will later be used
470 * to handle request \a nrq.
472 * \param[in] nrs the NRS head instance that will be handling request \a nrq.
473 * \param[in] nrq the request that is being handled.
474 * \param[out] resp the array where references to the resource hierarchy are
476 * \param[in] moving_req is set when obtaining resources while moving a
477 * request from a policy on the regular NRS head to a
478 * policy on the HP NRS head (via
479 * ldlm_lock_reorder_req()). It signifies that
480 * allocations to get resources should be atomic; for
481 * a full explanation, see comment in
482 * ptlrpc_nrs_pol_ops::op_res_get().
484 static void nrs_resource_get_safe(struct ptlrpc_nrs *nrs,
485 struct ptlrpc_nrs_request *nrq,
486 struct ptlrpc_nrs_resource **resp,
489 struct ptlrpc_nrs_policy *primary = NULL;
490 struct ptlrpc_nrs_policy *fallback = NULL;
492 memset(resp, 0, sizeof(resp[0]) * NRS_RES_MAX);
495 * Obtain policy references.
497 spin_lock(&nrs->nrs_lock);
499 fallback = nrs->nrs_policy_fallback;
500 nrs_policy_started_get(fallback);
502 primary = nrs->nrs_policy_primary;
504 nrs_policy_started_get(primary);
506 spin_unlock(&nrs->nrs_lock);
509 * Obtain resource hierarchy references.
511 resp[NRS_RES_FALLBACK] = nrs_resource_get(fallback, nrq, moving_req);
512 LASSERT(resp[NRS_RES_FALLBACK] != NULL);
514 if (primary != NULL) {
515 resp[NRS_RES_PRIMARY] = nrs_resource_get(primary, nrq,
518 * A primary policy may exist which may not wish to serve a
519 * particular request for different reasons; release the
520 * reference on the policy as it will not be used for this
523 if (resp[NRS_RES_PRIMARY] == NULL)
524 nrs_policy_started_put(primary);
529 * Releases references to resource hierarchies and policies, because they are no
530 * longer required; used when request handling has been completed, or the
531 * request is moving to the high priority NRS head.
533 * \param resp the resource hierarchy that is being released
535 * \see ptlrpcnrs_req_hp_move()
536 * \see ptlrpc_nrs_req_finalize()
538 static void nrs_resource_put_safe(struct ptlrpc_nrs_resource **resp)
540 struct ptlrpc_nrs_policy *pols[NRS_RES_MAX];
543 for (i = 0; i < NRS_RES_MAX; i++) {
544 if (resp[i] != NULL) {
545 pols[i] = resp[i]->res_policy;
546 nrs_resource_put(resp[i]);
553 for (i = 0; i < NRS_RES_MAX; i++) {
557 nrs_policy_started_put(pols[i]);
562 * Obtains an NRS request from \a policy for handling or examination; the
563 * request should be removed in the 'handling' case.
565 * Calling into this function implies we already know the policy has a request
566 * waiting to be handled.
568 * \param[in] policy the policy from which a request
569 * \param[in] peek when set, signifies that we just want to examine the
570 * request, and not handle it, so the request is not removed
572 * \param[in] force when set, it will force a policy to return a request if it
575 * \retval the NRS request to be handled
578 struct ptlrpc_nrs_request * nrs_request_get(struct ptlrpc_nrs_policy *policy,
579 bool peek, bool force)
581 struct ptlrpc_nrs_request *nrq;
583 LASSERT(policy->pol_req_queued > 0);
585 /* for a non-started policy, use force mode to drain requests */
586 if (unlikely(policy->pol_state != NRS_POL_STATE_STARTED))
589 nrq = policy->pol_desc->pd_ops->op_req_get(policy, peek, force);
591 LASSERT(ergo(nrq != NULL, nrs_request_policy(nrq) == policy));
597 * Enqueues request \a nrq for later handling, via one one the policies for
598 * which resources where earlier obtained via nrs_resource_get_safe(). The
599 * function attempts to enqueue the request first on the primary policy
600 * (if any), since this is the preferred choice.
602 * \param nrq the request being enqueued
604 * \see nrs_resource_get_safe()
606 static inline void nrs_request_enqueue(struct ptlrpc_nrs_request *nrq)
608 struct ptlrpc_nrs_policy *policy;
613 * Try in descending order, because the primary policy (if any) is
614 * the preferred choice.
616 for (i = NRS_RES_MAX - 1; i >= 0; i--) {
617 if (nrq->nr_res_ptrs[i] == NULL)
621 policy = nrq->nr_res_ptrs[i]->res_policy;
623 rc = policy->pol_desc->pd_ops->op_req_enqueue(policy, nrq);
625 policy->pol_nrs->nrs_req_queued++;
626 policy->pol_req_queued++;
628 * Take an extra ref to avoid stopping policy with
629 * pending request in it
631 nrs_policy_started_get(policy);
636 * Should never get here, as at least the primary policy's
637 * ptlrpc_nrs_pol_ops::op_req_enqueue() implementation should always
644 * Called when a request has been handled
646 * \param[in] nrs the request that has been handled; can be used for
647 * job/resource control.
649 * \see ptlrpc_nrs_req_stop_nolock()
651 static inline void nrs_request_stop(struct ptlrpc_nrs_request *nrq)
653 struct ptlrpc_nrs_policy *policy = nrs_request_policy(nrq);
655 if (policy->pol_desc->pd_ops->op_req_stop)
656 policy->pol_desc->pd_ops->op_req_stop(policy, nrq);
658 LASSERT(policy->pol_nrs->nrs_req_started > 0);
659 LASSERT(policy->pol_req_started > 0);
661 policy->pol_nrs->nrs_req_started--;
662 policy->pol_req_started--;
666 * Handler for operations that can be carried out on policies.
668 * Handles opcodes that are common to all policy types within NRS core, and
669 * passes any unknown opcodes to the policy-specific control function.
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.
674 * \param[in] opc the opcode of the operation being carried out.
675 * \param[in,out] arg can be used to pass information in and out between when
676 * carrying an operation; usually data that is private to
677 * the policy at some level, or generic policy status
680 * \retval -ve error condition
681 * \retval 0 operation was carried out successfully
683 static int nrs_policy_ctl(struct ptlrpc_nrs *nrs, char *name,
684 enum ptlrpc_nrs_ctl opc, void *arg)
686 struct ptlrpc_nrs_policy *policy;
690 spin_lock(&nrs->nrs_lock);
692 policy = nrs_policy_find_locked(nrs, name);
694 GOTO(out, rc = -ENOENT);
696 if (policy->pol_state != NRS_POL_STATE_STARTED &&
697 policy->pol_state != NRS_POL_STATE_STOPPED)
698 GOTO(out, rc = -EAGAIN);
702 * Unknown opcode, pass it down to the policy-specific control
703 * function for handling.
706 rc = nrs_policy_ctl_locked(policy, opc, arg);
712 case PTLRPC_NRS_CTL_START:
713 rc = nrs_policy_start_locked(policy, arg);
718 nrs_policy_put_locked(policy);
720 spin_unlock(&nrs->nrs_lock);
726 * Unregisters a policy by name.
728 * \param[in] nrs the NRS head this policy belongs to.
729 * \param[in] name the human-readable policy name; should be the same as
730 * ptlrpc_nrs_pol_desc::pd_name
735 static int nrs_policy_unregister(struct ptlrpc_nrs *nrs, char *name)
737 struct ptlrpc_nrs_policy *policy = NULL;
738 struct ptlrpc_service_part *svcpt = nrs->nrs_svcpt;
739 char *srv_name = svcpt->scp_service->srv_name;
743 spin_lock(&nrs->nrs_lock);
745 policy = nrs_policy_find_locked(nrs, name);
746 if (policy == NULL) {
748 CERROR("%s.%d NRS: cannot find policy '%s': rc = %d\n",
749 srv_name, svcpt->scp_cpt, name, rc);
750 GOTO(out_unlock, rc);
753 if (policy->pol_ref > 1) {
755 CERROR("%s.%d NRS: policy '%s' is busy with %ld references: rc = %d\n",
756 srv_name, svcpt->scp_cpt, name, policy->pol_ref, rc);
760 LASSERT(policy->pol_req_queued == 0);
761 LASSERT(policy->pol_req_started == 0);
763 if (policy->pol_state != NRS_POL_STATE_STOPPED) {
764 rc = nrs_policy_stop_locked(policy);
766 CERROR("%s.%d NRS: failed to stop policy '%s' with refcount %d: rc = %d\n",
767 srv_name, svcpt->scp_cpt, name,
768 refcount_read(&policy->pol_start_ref), rc);
773 LASSERT(policy->pol_private == NULL);
774 list_del(&policy->pol_list);
779 nrs_policy_put_locked(policy);
781 spin_unlock(&nrs->nrs_lock);
784 nrs_policy_fini(policy);
785 OBD_FREE_PTR(policy);
792 * Register a policy from \policy descriptor \a desc with NRS head \a nrs.
794 * \param[in] nrs the NRS head on which the policy will be registered.
795 * \param[in] desc the policy descriptor from which the information will be
796 * obtained to register the policy.
801 static int nrs_policy_register(struct ptlrpc_nrs *nrs,
802 struct ptlrpc_nrs_pol_desc *desc)
804 struct ptlrpc_nrs_policy *policy;
805 struct ptlrpc_nrs_policy *tmp;
806 struct ptlrpc_service_part *svcpt = nrs->nrs_svcpt;
807 char *srv_name = svcpt->scp_service->srv_name;
811 LASSERT(svcpt != NULL);
812 LASSERT(desc->pd_ops != NULL);
813 LASSERT(desc->pd_ops->op_res_get != NULL);
814 LASSERT(desc->pd_ops->op_req_get != NULL);
815 LASSERT(desc->pd_ops->op_req_enqueue != NULL);
816 LASSERT(desc->pd_ops->op_req_dequeue != NULL);
817 LASSERT(desc->pd_compat != NULL);
819 OBD_CPT_ALLOC_GFP(policy, svcpt->scp_service->srv_cptable,
820 svcpt->scp_cpt, sizeof(*policy), GFP_NOFS);
824 policy->pol_nrs = nrs;
825 policy->pol_desc = desc;
826 policy->pol_state = NRS_POL_STATE_STOPPED;
827 policy->pol_flags = desc->pd_flags;
829 INIT_LIST_HEAD(&policy->pol_list);
830 INIT_LIST_HEAD(&policy->pol_list_queued);
832 init_waitqueue_head(&policy->pol_wq);
834 rc = nrs_policy_init(policy);
836 OBD_FREE_PTR(policy);
840 spin_lock(&nrs->nrs_lock);
842 tmp = nrs_policy_find_locked(nrs, policy->pol_desc->pd_name);
845 CERROR("%s.%d NRS: policy %s has been registered, can't register it: rc = %d\n",
846 srv_name, svcpt->scp_cpt, policy->pol_desc->pd_name,
848 nrs_policy_put_locked(tmp);
850 spin_unlock(&nrs->nrs_lock);
851 nrs_policy_fini(policy);
852 OBD_FREE_PTR(policy);
857 list_add_tail(&policy->pol_list, &nrs->nrs_policy_list);
860 if (policy->pol_flags & PTLRPC_NRS_FL_REG_START)
861 rc = nrs_policy_start_locked(policy, NULL);
863 spin_unlock(&nrs->nrs_lock);
866 (void) nrs_policy_unregister(nrs, policy->pol_desc->pd_name);
872 * Enqueue request \a req using one of the policies its resources are referring
875 * \param[in] req the request to enqueue.
877 static void ptlrpc_nrs_req_add_nolock(struct ptlrpc_request *req)
879 struct ptlrpc_nrs_policy *policy;
881 LASSERT(req->rq_nrq.nr_initialized);
882 LASSERT(!req->rq_nrq.nr_enqueued);
884 nrs_request_enqueue(&req->rq_nrq);
885 req->rq_nrq.nr_enqueued = 1;
887 policy = nrs_request_policy(&req->rq_nrq);
889 * Add the policy to the NRS head's list of policies with enqueued
890 * requests, if it has not been added there.
892 if (unlikely(list_empty(&policy->pol_list_queued)))
893 list_add_tail(&policy->pol_list_queued,
894 &policy->pol_nrs->nrs_policy_queued);
898 * Enqueue a request on the high priority NRS head.
900 * \param req the request to enqueue.
902 static void ptlrpc_nrs_hpreq_add_nolock(struct ptlrpc_request *req)
904 int opc = lustre_msg_get_opc(req->rq_reqmsg);
907 spin_lock(&req->rq_lock);
909 ptlrpc_nrs_req_add_nolock(req);
911 DEBUG_REQ(D_NET, req, "high priority req");
912 spin_unlock(&req->rq_lock);
917 * Returns a boolean predicate indicating whether the policy described by
918 * \a desc is adequate for use with service \a svc.
920 * \param[in] svc the service
921 * \param[in] desc the policy descriptor
923 * \retval false the policy is not compatible with the service
924 * \retval true the policy is compatible with the service
926 static inline bool nrs_policy_compatible(const struct ptlrpc_service *svc,
927 const struct ptlrpc_nrs_pol_desc *desc)
929 return desc->pd_compat(svc, desc);
933 * Registers all compatible policies in nrs_core.nrs_policies, for NRS head
936 * \param[in] nrs the NRS head
941 * \pre mutex_is_locked(&nrs_core.nrs_mutex)
943 * \see ptlrpc_service_nrs_setup()
945 static int nrs_register_policies_locked(struct ptlrpc_nrs *nrs)
947 struct ptlrpc_nrs_pol_desc *desc;
948 /* for convenience */
949 struct ptlrpc_service_part *svcpt = nrs->nrs_svcpt;
950 struct ptlrpc_service *svc = svcpt->scp_service;
954 LASSERT(mutex_is_locked(&nrs_core.nrs_mutex));
956 list_for_each_entry(desc, &nrs_core.nrs_policies, pd_list) {
957 if (nrs_policy_compatible(svc, desc)) {
958 rc = nrs_policy_register(nrs, desc);
960 CERROR("%s.%d NRS: Failed to register policy %s: rc = %d\n",
961 svc->srv_name, svcpt->scp_cpt,
964 * Fail registration if any of the policies'
965 * registration fails.
976 * Initializes NRS head \a nrs of service partition \a svcpt, and registers all
977 * compatible policies in NRS core, with the NRS head.
979 * \param[in] nrs the NRS head
980 * \param[in] svcpt the PTLRPC service partition to setup
985 * \pre mutex_is_locked(&nrs_core.nrs_mutex)
987 static int nrs_svcpt_setup_locked0(struct ptlrpc_nrs *nrs,
988 struct ptlrpc_service_part *svcpt)
991 enum ptlrpc_nrs_queue_type queue;
993 LASSERT(mutex_is_locked(&nrs_core.nrs_mutex));
995 if (nrs == &svcpt->scp_nrs_reg)
996 queue = PTLRPC_NRS_QUEUE_REG;
997 else if (nrs == svcpt->scp_nrs_hp)
998 queue = PTLRPC_NRS_QUEUE_HP;
1002 nrs->nrs_svcpt = svcpt;
1003 nrs->nrs_queue_type = queue;
1004 spin_lock_init(&nrs->nrs_lock);
1005 INIT_LIST_HEAD(&nrs->nrs_policy_list);
1006 INIT_LIST_HEAD(&nrs->nrs_policy_queued);
1007 nrs->nrs_throttling = 0;
1009 rc = nrs_register_policies_locked(nrs);
1015 * Allocates a regular and optionally a high-priority NRS head (if the service
1016 * handles high-priority RPCs), and then registers all available compatible
1017 * policies on those NRS heads.
1019 * \param[in,out] svcpt the PTLRPC service partition to setup
1021 * \pre mutex_is_locked(&nrs_core.nrs_mutex)
1023 static int nrs_svcpt_setup_locked(struct ptlrpc_service_part *svcpt)
1025 struct ptlrpc_nrs *nrs;
1029 LASSERT(mutex_is_locked(&nrs_core.nrs_mutex));
1032 * Initialize the regular NRS head.
1034 nrs = nrs_svcpt2nrs(svcpt, false);
1035 rc = nrs_svcpt_setup_locked0(nrs, svcpt);
1040 * Optionally allocate a high-priority NRS head.
1042 if (svcpt->scp_service->srv_ops.so_hpreq_handler == NULL)
1045 OBD_CPT_ALLOC_PTR(svcpt->scp_nrs_hp,
1046 svcpt->scp_service->srv_cptable,
1048 if (svcpt->scp_nrs_hp == NULL)
1049 GOTO(out, rc = -ENOMEM);
1051 nrs = nrs_svcpt2nrs(svcpt, true);
1052 rc = nrs_svcpt_setup_locked0(nrs, svcpt);
1059 * Unregisters all policies on all available NRS heads in a service partition;
1060 * called at PTLRPC service unregistration time.
1062 * \param[in] svcpt the PTLRPC service partition
1064 * \pre mutex_is_locked(&nrs_core.nrs_mutex)
1066 static void nrs_svcpt_cleanup_locked(struct ptlrpc_service_part *svcpt)
1068 struct ptlrpc_nrs *nrs;
1069 struct ptlrpc_nrs_policy *policy;
1070 struct ptlrpc_nrs_policy *tmp;
1075 LASSERT(mutex_is_locked(&nrs_core.nrs_mutex));
1078 /* scp_nrs_hp could be NULL due to short of memory. */
1079 nrs = hp ? svcpt->scp_nrs_hp : &svcpt->scp_nrs_reg;
1080 /* check the nrs_svcpt to see if nrs is initialized. */
1081 if (!nrs || !nrs->nrs_svcpt) {
1085 nrs->nrs_stopping = 1;
1087 list_for_each_entry_safe(policy, tmp, &nrs->nrs_policy_list,
1089 rc = nrs_policy_unregister(nrs, policy->pol_desc->pd_name);
1094 * If the service partition has an HP NRS head, clean that up as well.
1096 if (!hp && nrs_svcpt_has_hp(svcpt)) {
1108 * Returns the descriptor for a policy as identified by by \a name.
1110 * \param[in] name the policy name
1112 * \retval the policy descriptor
1115 static struct ptlrpc_nrs_pol_desc *nrs_policy_find_desc_locked(const char *name)
1117 struct ptlrpc_nrs_pol_desc *tmp;
1120 list_for_each_entry(tmp, &nrs_core.nrs_policies, pd_list) {
1121 if (strncmp(tmp->pd_name, name, NRS_POL_NAME_MAX) == 0)
1128 * Removes the policy from all supported NRS heads of all partitions of all
1131 * \param[in] desc the policy descriptor to unregister
1134 * \retval 0 successfully unregistered policy on all supported NRS heads
1136 * \pre mutex_is_locked(&nrs_core.nrs_mutex)
1137 * \pre mutex_is_locked(&ptlrpc_all_services_mutex)
1139 static int nrs_policy_unregister_locked(struct ptlrpc_nrs_pol_desc *desc)
1141 struct ptlrpc_nrs *nrs;
1142 struct ptlrpc_service *svc;
1143 struct ptlrpc_service_part *svcpt;
1148 LASSERT(mutex_is_locked(&nrs_core.nrs_mutex));
1149 LASSERT(mutex_is_locked(&ptlrpc_all_services_mutex));
1151 list_for_each_entry(svc, &ptlrpc_all_services, srv_list) {
1153 if (!nrs_policy_compatible(svc, desc) ||
1154 unlikely(svc->srv_is_stopping))
1157 ptlrpc_service_for_each_part(svcpt, i, svc) {
1158 char *srv_name = svcpt->scp_service->srv_name;
1162 nrs = nrs_svcpt2nrs(svcpt, hp);
1163 rc = nrs_policy_unregister(nrs, desc->pd_name);
1165 * Ignore -ENOENT as the policy may not have registered
1166 * successfully on all service partitions.
1168 if (rc == -ENOENT) {
1170 } else if (rc != 0) {
1171 CERROR("%s.%d NRS: Failed to unregister policy %s: rc = %d\n",
1172 srv_name, svcpt->scp_cpt, desc->pd_name,
1177 if (!hp && nrs_svc_has_hp(svc)) {
1183 if (desc->pd_ops->op_lprocfs_fini != NULL)
1184 desc->pd_ops->op_lprocfs_fini(svc);
1191 * Registers a new policy with NRS core.
1193 * The function will only succeed if policy registration with all compatible
1194 * service partitions (if any) is successful.
1196 * N.B. This function should be called either at ptlrpc module initialization
1197 * time when registering a policy that ships with NRS core, or in a
1198 * module's init() function for policies registering from other modules.
1200 * \param[in] conf configuration information for the new policy to register
1205 static int ptlrpc_nrs_policy_register(struct ptlrpc_nrs_pol_conf *conf)
1207 struct ptlrpc_service *svc;
1208 struct ptlrpc_nrs_pol_desc *desc;
1212 LASSERT(conf != NULL);
1213 LASSERT(conf->nc_ops != NULL);
1214 LASSERT(conf->nc_compat != NULL);
1215 LASSERT(ergo(conf->nc_compat == nrs_policy_compat_one,
1216 conf->nc_compat_svc_name != NULL));
1217 LASSERT(ergo((conf->nc_flags & PTLRPC_NRS_FL_REG_EXTERN) != 0,
1218 conf->nc_owner != NULL));
1220 conf->nc_name[NRS_POL_NAME_MAX - 1] = '\0';
1223 * External policies are not allowed to start immediately upon
1224 * registration, as there is a relatively higher chance that their
1225 * registration might fail. In such a case, some policy instances may
1226 * already have requests queued wen unregistration needs to happen as
1227 * part o cleanup; since there is currently no way to drain requests
1228 * from a policy unless the service is unregistering, we just disallow
1231 if ((conf->nc_flags & PTLRPC_NRS_FL_REG_EXTERN) &&
1232 (conf->nc_flags & (PTLRPC_NRS_FL_FALLBACK |
1233 PTLRPC_NRS_FL_REG_START))) {
1235 CERROR("NRS: failing to register policy %s. Please check "
1236 "policy flags; external policies cannot act as fallback "
1237 "policies, or be started immediately upon registration "
1238 "without interaction with lprocfs: rc = %d\n",
1243 mutex_lock(&nrs_core.nrs_mutex);
1245 if (nrs_policy_find_desc_locked(conf->nc_name) != NULL) {
1247 CERROR("NRS: failing to register policy %s which has already been registered with NRS core: rc = %d\n",
1252 OBD_ALLOC_PTR(desc);
1254 GOTO(fail, rc = -ENOMEM);
1256 if (strlcpy(desc->pd_name, conf->nc_name, sizeof(desc->pd_name)) >=
1257 sizeof(desc->pd_name)) {
1259 GOTO(fail, rc = -E2BIG);
1261 desc->pd_ops = conf->nc_ops;
1262 desc->pd_compat = conf->nc_compat;
1263 desc->pd_compat_svc_name = conf->nc_compat_svc_name;
1264 if ((conf->nc_flags & PTLRPC_NRS_FL_REG_EXTERN) != 0)
1265 desc->pd_owner = conf->nc_owner;
1266 desc->pd_flags = conf->nc_flags;
1267 atomic_set(&desc->pd_refs, 0);
1270 * For policies that are held in the same module as NRS (currently
1271 * ptlrpc), do not register the policy with all compatible services,
1272 * as the services will not have started at this point, since we are
1273 * calling from ptlrpc module initialization code. In such cases each
1274 * service will register all compatible policies later, via
1275 * ptlrpc_service_nrs_setup().
1277 if ((conf->nc_flags & PTLRPC_NRS_FL_REG_EXTERN) == 0)
1281 * Register the new policy on all compatible services
1283 mutex_lock(&ptlrpc_all_services_mutex);
1285 list_for_each_entry(svc, &ptlrpc_all_services, srv_list) {
1286 struct ptlrpc_service_part *svcpt;
1290 if (!nrs_policy_compatible(svc, desc) ||
1291 unlikely(svc->srv_is_stopping))
1294 ptlrpc_service_for_each_part(svcpt, i, svc) {
1295 struct ptlrpc_nrs *nrs;
1296 char *srv_name = svcpt->scp_service->srv_name;
1299 nrs = nrs_svcpt2nrs(svcpt, hp);
1300 rc = nrs_policy_register(nrs, desc);
1302 CERROR("%s.%d NRS: Failed to register policy %s: rc = %d\n",
1303 srv_name, svcpt->scp_cpt,
1306 rc2 = nrs_policy_unregister_locked(desc);
1308 * Should not fail at this point
1311 mutex_unlock(&ptlrpc_all_services_mutex);
1316 if (!hp && nrs_svc_has_hp(svc)) {
1323 * No need to take a reference to other modules here, as we
1324 * will be calling from the module's init() function.
1326 if (desc->pd_ops->op_lprocfs_init != NULL) {
1327 rc = desc->pd_ops->op_lprocfs_init(svc);
1329 rc2 = nrs_policy_unregister_locked(desc);
1331 * Should not fail at this point
1334 mutex_unlock(&ptlrpc_all_services_mutex);
1341 mutex_unlock(&ptlrpc_all_services_mutex);
1343 list_add_tail(&desc->pd_list, &nrs_core.nrs_policies);
1345 mutex_unlock(&nrs_core.nrs_mutex);
1351 * Setup NRS heads on all service partitions of service \a svc, and register
1352 * all compatible policies on those NRS heads.
1354 * To be called from withing ptl
1355 * \param[in] svc the service to setup
1357 * \retval -ve error, the calling logic should eventually call
1358 * ptlrpc_service_nrs_cleanup() to undo any work performed
1361 * \see ptlrpc_register_service()
1362 * \see ptlrpc_service_nrs_cleanup()
1364 int ptlrpc_service_nrs_setup(struct ptlrpc_service *svc)
1366 struct ptlrpc_service_part *svcpt;
1367 const struct ptlrpc_nrs_pol_desc *desc;
1371 mutex_lock(&nrs_core.nrs_mutex);
1374 * Initialize NRS heads on all service CPTs.
1376 ptlrpc_service_for_each_part(svcpt, i, svc) {
1377 rc = nrs_svcpt_setup_locked(svcpt);
1383 * Set up lprocfs interfaces for all supported policies for the
1386 list_for_each_entry(desc, &nrs_core.nrs_policies, pd_list) {
1387 if (!nrs_policy_compatible(svc, desc))
1390 if (desc->pd_ops->op_lprocfs_init != NULL) {
1391 rc = desc->pd_ops->op_lprocfs_init(svc);
1399 mutex_unlock(&nrs_core.nrs_mutex);
1405 * Unregisters all policies on all service partitions of service \a svc.
1407 * \param[in] svc the PTLRPC service to unregister
1409 void ptlrpc_service_nrs_cleanup(struct ptlrpc_service *svc)
1411 struct ptlrpc_service_part *svcpt;
1412 const struct ptlrpc_nrs_pol_desc *desc;
1415 mutex_lock(&nrs_core.nrs_mutex);
1418 * Clean up NRS heads on all service partitions
1420 ptlrpc_service_for_each_part(svcpt, i, svc)
1421 nrs_svcpt_cleanup_locked(svcpt);
1424 * Clean up lprocfs interfaces for all supported policies for the
1427 list_for_each_entry(desc, &nrs_core.nrs_policies, pd_list) {
1428 if (!nrs_policy_compatible(svc, desc))
1431 if (desc->pd_ops->op_lprocfs_fini != NULL)
1432 desc->pd_ops->op_lprocfs_fini(svc);
1435 mutex_unlock(&nrs_core.nrs_mutex);
1439 * Obtains NRS head resources for request \a req.
1441 * These could be either on the regular or HP NRS head of \a svcpt; resources
1442 * taken on the regular head can later be swapped for HP head resources by
1443 * ldlm_lock_reorder_req().
1445 * \param[in] svcpt the service partition
1446 * \param[in] req the request
1447 * \param[in] hp which NRS head of \a svcpt to use
1449 void ptlrpc_nrs_req_initialize(struct ptlrpc_service_part *svcpt,
1450 struct ptlrpc_request *req, bool hp)
1452 struct ptlrpc_nrs *nrs = nrs_svcpt2nrs(svcpt, hp);
1454 memset(&req->rq_nrq, 0, sizeof(req->rq_nrq));
1455 nrs_resource_get_safe(nrs, &req->rq_nrq, req->rq_nrq.nr_res_ptrs,
1459 * It is fine to access \e nr_initialized without locking as there is
1460 * no contention at this early stage.
1462 req->rq_nrq.nr_initialized = 1;
1466 * Releases resources for a request; is called after the request has been
1469 * \param[in] req the request
1471 * \see ptlrpc_server_finish_request()
1473 void ptlrpc_nrs_req_finalize(struct ptlrpc_request *req)
1475 if (req->rq_nrq.nr_initialized) {
1476 nrs_resource_put_safe(req->rq_nrq.nr_res_ptrs);
1477 /* no protection on bit nr_initialized because no
1478 * contention at this late stage */
1479 req->rq_nrq.nr_finalized = 1;
1483 void ptlrpc_nrs_req_stop_nolock(struct ptlrpc_request *req)
1485 if (req->rq_nrq.nr_started)
1486 nrs_request_stop(&req->rq_nrq);
1490 * Enqueues request \a req on either the regular or high-priority NRS head
1491 * of service partition \a svcpt.
1493 * \param[in] svcpt the service partition
1494 * \param[in] req the request to be enqueued
1495 * \param[in] hp whether to enqueue the request on the regular or
1496 * high-priority NRS head.
1498 void ptlrpc_nrs_req_add(struct ptlrpc_service_part *svcpt,
1499 struct ptlrpc_request *req, bool hp)
1501 spin_lock(&svcpt->scp_req_lock);
1504 ptlrpc_nrs_hpreq_add_nolock(req);
1506 ptlrpc_nrs_req_add_nolock(req);
1508 spin_unlock(&svcpt->scp_req_lock);
1511 static void nrs_request_removed(struct ptlrpc_nrs_policy *policy)
1513 LASSERT(policy->pol_nrs->nrs_req_queued > 0);
1514 LASSERT(policy->pol_req_queued > 0);
1516 policy->pol_nrs->nrs_req_queued--;
1517 policy->pol_req_queued--;
1520 * If the policy has no more requests queued, remove it from
1521 * ptlrpc_nrs::nrs_policy_queued.
1523 if (unlikely(policy->pol_req_queued == 0)) {
1524 list_del_init(&policy->pol_list_queued);
1527 * If there are other policies with queued requests, move the
1528 * current policy to the end so that we can round robin over
1529 * all policies and drain the requests.
1531 } else if (policy->pol_req_queued != policy->pol_nrs->nrs_req_queued) {
1532 LASSERT(policy->pol_req_queued <
1533 policy->pol_nrs->nrs_req_queued);
1535 list_move_tail(&policy->pol_list_queued,
1536 &policy->pol_nrs->nrs_policy_queued);
1539 /* remove the extra ref for policy pending requests */
1540 nrs_policy_started_put(policy);
1544 * Obtains a request for handling from an NRS head of service partition
1547 * \param[in] svcpt the service partition
1548 * \param[in] hp whether to obtain a request from the regular or
1549 * high-priority NRS head.
1550 * \param[in] peek when set, signifies that we just want to examine the
1551 * request, and not handle it, so the request is not removed
1553 * \param[in] force when set, it will force a policy to return a request if it
1556 * \retval the request to be handled
1557 * \retval NULL the head has no requests to serve
1559 struct ptlrpc_request *
1560 ptlrpc_nrs_req_get_nolock0(struct ptlrpc_service_part *svcpt, bool hp,
1561 bool peek, bool force)
1563 struct ptlrpc_nrs *nrs = nrs_svcpt2nrs(svcpt, hp);
1564 struct ptlrpc_nrs_policy *policy;
1565 struct ptlrpc_nrs_request *nrq;
1568 * Always try to drain requests from all NRS polices even if they are
1569 * inactive, because the user can change policy status at runtime.
1571 list_for_each_entry(policy, &nrs->nrs_policy_queued,
1573 nrq = nrs_request_get(policy, peek, force);
1575 if (likely(!peek)) {
1576 nrq->nr_started = 1;
1578 policy->pol_req_started++;
1579 policy->pol_nrs->nrs_req_started++;
1581 nrs_request_removed(policy);
1584 return container_of(nrq, struct ptlrpc_request, rq_nrq);
1592 * Dequeues request \a req from the policy it has been enqueued on.
1594 * \param[in] req the request
1596 void ptlrpc_nrs_req_del_nolock(struct ptlrpc_request *req)
1598 struct ptlrpc_nrs_policy *policy = nrs_request_policy(&req->rq_nrq);
1600 policy->pol_desc->pd_ops->op_req_dequeue(policy, &req->rq_nrq);
1602 req->rq_nrq.nr_enqueued = 0;
1604 nrs_request_removed(policy);
1608 * Returns whether there are any requests currently enqueued on any of the
1609 * policies of service partition's \a svcpt NRS head specified by \a hp. Should
1610 * be called while holding ptlrpc_service_part::scp_req_lock to get a reliable
1613 * \param[in] svcpt the service partition to enquire.
1614 * \param[in] hp whether the regular or high-priority NRS head is to be
1617 * \retval false the indicated NRS head has no enqueued requests.
1618 * \retval true the indicated NRS head has some enqueued requests.
1620 bool ptlrpc_nrs_req_pending_nolock(struct ptlrpc_service_part *svcpt, bool hp)
1622 struct ptlrpc_nrs *nrs = nrs_svcpt2nrs(svcpt, hp);
1624 return nrs->nrs_req_queued > 0;
1628 * Returns whether NRS policy is throttling reqeust
1630 * \param[in] svcpt the service partition to enquire.
1631 * \param[in] hp whether the regular or high-priority NRS head is to be
1634 * \retval false the indicated NRS head has no enqueued requests.
1635 * \retval true the indicated NRS head has some enqueued requests.
1637 bool ptlrpc_nrs_req_throttling_nolock(struct ptlrpc_service_part *svcpt,
1640 struct ptlrpc_nrs *nrs = nrs_svcpt2nrs(svcpt, hp);
1642 return !!nrs->nrs_throttling;
1646 * Moves request \a req from the regular to the high-priority NRS head.
1648 * \param[in] req the request to move
1650 void ptlrpc_nrs_req_hp_move(struct ptlrpc_request *req)
1652 struct ptlrpc_service_part *svcpt = req->rq_rqbd->rqbd_svcpt;
1653 struct ptlrpc_nrs_request *nrq = &req->rq_nrq;
1654 struct ptlrpc_nrs_resource *res1[NRS_RES_MAX];
1655 struct ptlrpc_nrs_resource *res2[NRS_RES_MAX];
1659 * Obtain the high-priority NRS head resources.
1661 nrs_resource_get_safe(nrs_svcpt2nrs(svcpt, true), nrq, res1, true);
1663 spin_lock(&svcpt->scp_req_lock);
1665 if (!ptlrpc_nrs_req_can_move(req))
1668 ptlrpc_nrs_req_del_nolock(req);
1670 memcpy(res2, nrq->nr_res_ptrs, NRS_RES_MAX * sizeof(res2[0]));
1671 memcpy(nrq->nr_res_ptrs, res1, NRS_RES_MAX * sizeof(res1[0]));
1673 ptlrpc_nrs_hpreq_add_nolock(req);
1675 memcpy(res1, res2, NRS_RES_MAX * sizeof(res1[0]));
1677 spin_unlock(&svcpt->scp_req_lock);
1680 * Release either the regular NRS head resources if we moved the
1681 * request, or the high-priority NRS head resources if we took a
1682 * reference earlier in this function and ptlrpc_nrs_req_can_move()
1685 nrs_resource_put_safe(res1);
1690 * Carries out a control operation \a opc on the policy identified by the
1691 * human-readable \a name, on either all partitions, or only on the first
1692 * partition of service \a svc.
1694 * \param[in] svc the service the policy belongs to.
1695 * \param[in] queue whether to carry out the command on the policy which
1696 * belongs to the regular, high-priority, or both NRS
1697 * heads of service partitions of \a svc.
1698 * \param[in] name the policy to act upon, by human-readable name
1699 * \param[in] opc the opcode of the operation to carry out
1700 * \param[in] single when set, the operation will only be carried out on the
1701 * NRS heads of the first service partition of \a svc.
1702 * This is useful for some policies which e.g. share
1703 * identical values on the same parameters of different
1704 * service partitions; when reading these parameters via
1705 * lprocfs, these policies may just want to obtain and
1706 * print out the values from the first service partition.
1707 * Storing these values centrally elsewhere then could be
1708 * another solution for this.
1709 * \param[in,out] arg can be used as a generic in/out buffer between control
1710 * operations and the user environment.
1712 *\retval -ve error condition
1713 *\retval 0 operation was carried out successfully
1715 int ptlrpc_nrs_policy_control(const struct ptlrpc_service *svc,
1716 enum ptlrpc_nrs_queue_type queue, char *name,
1717 enum ptlrpc_nrs_ctl opc, bool single, void *arg)
1719 struct ptlrpc_service_part *svcpt;
1724 LASSERT(opc != PTLRPC_NRS_CTL_INVALID);
1726 if ((queue & PTLRPC_NRS_QUEUE_BOTH) == 0)
1729 ptlrpc_service_for_each_part(svcpt, i, svc) {
1730 if ((queue & PTLRPC_NRS_QUEUE_REG) != 0) {
1731 rc = nrs_policy_ctl(nrs_svcpt2nrs(svcpt, false), name,
1733 if (rc != 0 || (queue == PTLRPC_NRS_QUEUE_REG &&
1738 if ((queue & PTLRPC_NRS_QUEUE_HP) != 0) {
1740 * XXX: We could optionally check for
1741 * nrs_svc_has_hp(svc) here, and return an error if it
1742 * is false. Right now we rely on the policies' lprocfs
1743 * handlers that call the present function to make this
1744 * check; if they fail to do so, they might hit the
1745 * assertion inside nrs_svcpt2nrs() below.
1747 rc = nrs_policy_ctl(nrs_svcpt2nrs(svcpt, true), name,
1749 if (rc != 0 || single)
1758 * Adds all policies that ship with the ptlrpc module, to NRS core's list of
1759 * policies \e nrs_core.nrs_policies.
1761 * \retval 0 all policies have been registered successfully
1764 int ptlrpc_nrs_init(void)
1769 mutex_init(&nrs_core.nrs_mutex);
1770 INIT_LIST_HEAD(&nrs_core.nrs_policies);
1772 rc = ptlrpc_nrs_policy_register(&nrs_conf_fifo);
1776 #ifdef HAVE_SERVER_SUPPORT
1777 rc = ptlrpc_nrs_policy_register(&nrs_conf_crrn);
1781 rc = ptlrpc_nrs_policy_register(&nrs_conf_orr);
1785 rc = ptlrpc_nrs_policy_register(&nrs_conf_trr);
1788 rc = ptlrpc_nrs_policy_register(&nrs_conf_tbf);
1791 #endif /* HAVE_SERVER_SUPPORT */
1793 rc = ptlrpc_nrs_policy_register(&nrs_conf_delay);
1800 * Since no PTLRPC services have been started at this point, all we need
1801 * to do for cleanup is to free the descriptors.
1809 * Removes all policy descriptors from nrs_core::nrs_policies, and frees the
1810 * policy descriptors.
1812 * Since all PTLRPC services are stopped at this point, there are no more
1813 * instances of any policies, because each service will have stopped its policy
1814 * instances in ptlrpc_service_nrs_cleanup(), so we just need to free the
1817 void ptlrpc_nrs_fini(void)
1819 struct ptlrpc_nrs_pol_desc *desc;
1820 struct ptlrpc_nrs_pol_desc *tmp;
1822 list_for_each_entry_safe(desc, tmp, &nrs_core.nrs_policies,
1824 list_del_init(&desc->pd_list);