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
44 #include <liblustre.h>
46 #include <obd_support.h>
47 #include <obd_class.h>
48 #include <lustre_net.h>
49 #include <lprocfs_status.h>
50 #include <libcfs/libcfs.h>
51 #include "ptlrpc_internal.h"
53 extern struct list_head ptlrpc_all_services;
58 struct nrs_core nrs_core;
60 static int nrs_policy_init(struct ptlrpc_nrs_policy *policy)
62 return policy->pol_desc->pd_ops->op_policy_init != NULL ?
63 policy->pol_desc->pd_ops->op_policy_init(policy) : 0;
66 static void nrs_policy_fini(struct ptlrpc_nrs_policy *policy)
68 LASSERT(policy->pol_ref == 0);
69 LASSERT(policy->pol_req_queued == 0);
71 if (policy->pol_desc->pd_ops->op_policy_fini != NULL)
72 policy->pol_desc->pd_ops->op_policy_fini(policy);
75 static int nrs_policy_ctl_locked(struct ptlrpc_nrs_policy *policy,
76 enum ptlrpc_nrs_ctl opc, void *arg)
79 * The policy may be stopped, but the lprocfs files and
80 * ptlrpc_nrs_policy instances remain present until unregistration time.
81 * Do not perform the ctl operation if the policy is stopped, as
82 * policy->pol_private will be NULL in such a case.
84 if (policy->pol_state == NRS_POL_STATE_STOPPED)
87 RETURN(policy->pol_desc->pd_ops->op_policy_ctl != NULL ?
88 policy->pol_desc->pd_ops->op_policy_ctl(policy, opc, arg) :
92 static void nrs_policy_stop0(struct ptlrpc_nrs_policy *policy)
94 struct ptlrpc_nrs *nrs = policy->pol_nrs;
97 if (policy->pol_desc->pd_ops->op_policy_stop != NULL) {
98 spin_unlock(&nrs->nrs_lock);
100 policy->pol_desc->pd_ops->op_policy_stop(policy);
102 spin_lock(&nrs->nrs_lock);
105 LASSERT(cfs_list_empty(&policy->pol_list_queued));
106 LASSERT(policy->pol_req_queued == 0 &&
107 policy->pol_req_started == 0);
109 policy->pol_private = NULL;
111 policy->pol_state = NRS_POL_STATE_STOPPED;
113 if (cfs_atomic_dec_and_test(&policy->pol_desc->pd_refs))
114 module_put(policy->pol_desc->pd_owner);
119 static int nrs_policy_stop_locked(struct ptlrpc_nrs_policy *policy)
121 struct ptlrpc_nrs *nrs = policy->pol_nrs;
124 if (nrs->nrs_policy_fallback == policy && !nrs->nrs_stopping)
127 if (policy->pol_state == NRS_POL_STATE_STARTING)
130 /* In progress or already stopped */
131 if (policy->pol_state != NRS_POL_STATE_STARTED)
134 policy->pol_state = NRS_POL_STATE_STOPPING;
136 /* Immediately make it invisible */
137 if (nrs->nrs_policy_primary == policy) {
138 nrs->nrs_policy_primary = NULL;
141 LASSERT(nrs->nrs_policy_fallback == policy);
142 nrs->nrs_policy_fallback = NULL;
145 /* I have the only refcount */
146 if (policy->pol_ref == 1)
147 nrs_policy_stop0(policy);
153 * Transitions the \a nrs NRS head's primary policy to
154 * ptlrpc_nrs_pol_state::NRS_POL_STATE_STOPPING and if the policy has no
155 * pending usage references, to ptlrpc_nrs_pol_state::NRS_POL_STATE_STOPPED.
157 * \param[in] nrs the NRS head to carry out this operation on
159 static void nrs_policy_stop_primary(struct ptlrpc_nrs *nrs)
161 struct ptlrpc_nrs_policy *tmp = nrs->nrs_policy_primary;
166 * XXX: This should really be RETURN_EXIT, but the latter does
167 * not currently print anything out, and possibly should be
174 nrs->nrs_policy_primary = NULL;
176 LASSERT(tmp->pol_state == NRS_POL_STATE_STARTED);
177 tmp->pol_state = NRS_POL_STATE_STOPPING;
179 if (tmp->pol_ref == 0)
180 nrs_policy_stop0(tmp);
185 * Transitions a policy across the ptlrpc_nrs_pol_state range of values, in
186 * response to an lprocfs command to start a policy.
188 * If a primary policy different to the current one is specified, this function
189 * will transition the new policy to the
190 * ptlrpc_nrs_pol_state::NRS_POL_STATE_STARTING and then to
191 * ptlrpc_nrs_pol_state::NRS_POL_STATE_STARTED, and will then transition
192 * the old primary policy (if there is one) to
193 * ptlrpc_nrs_pol_state::NRS_POL_STATE_STOPPING, and if there are no outstanding
194 * references on the policy to ptlrpc_nrs_pol_stae::NRS_POL_STATE_STOPPED.
196 * If the fallback policy is specified, this is taken to indicate an instruction
197 * to stop the current primary policy, without substituting it with another
198 * primary policy, so the primary policy (if any) is transitioned to
199 * ptlrpc_nrs_pol_state::NRS_POL_STATE_STOPPING, and if there are no outstanding
200 * references on the policy to ptlrpc_nrs_pol_stae::NRS_POL_STATE_STOPPED. In
201 * this case, the fallback policy is only left active in the NRS head.
203 static int nrs_policy_start_locked(struct ptlrpc_nrs_policy *policy, char *arg)
205 struct ptlrpc_nrs *nrs = policy->pol_nrs;
210 * Don't allow multiple starting which is too complex, and has no real
213 if (nrs->nrs_policy_starting)
216 LASSERT(policy->pol_state != NRS_POL_STATE_STARTING);
218 if (policy->pol_state == NRS_POL_STATE_STOPPING)
221 if (policy->pol_flags & PTLRPC_NRS_FL_FALLBACK) {
223 * This is for cases in which the user sets the policy to the
224 * fallback policy (currently fifo for all services); i.e. the
225 * user is resetting the policy to the default; so we stop the
226 * primary policy, if any.
228 if (policy == nrs->nrs_policy_fallback) {
229 nrs_policy_stop_primary(nrs);
234 * If we reach here, we must be setting up the fallback policy
235 * at service startup time, and only a single policy with the
236 * nrs_policy_flags::PTLRPC_NRS_FL_FALLBACK flag set can
237 * register with NRS core.
239 LASSERT(nrs->nrs_policy_fallback == NULL);
242 * Shouldn't start primary policy if w/o fallback policy.
244 if (nrs->nrs_policy_fallback == NULL)
247 if (policy->pol_state == NRS_POL_STATE_STARTED)
252 * Increase the module usage count for policies registering from other
255 if (cfs_atomic_inc_return(&policy->pol_desc->pd_refs) == 1 &&
256 !try_module_get(policy->pol_desc->pd_owner)) {
257 cfs_atomic_dec(&policy->pol_desc->pd_refs);
258 CERROR("NRS: cannot get module for policy %s; is it alive?\n",
259 policy->pol_desc->pd_name);
264 * Serialize policy starting across the NRS head
266 nrs->nrs_policy_starting = 1;
268 policy->pol_state = NRS_POL_STATE_STARTING;
270 if (policy->pol_desc->pd_ops->op_policy_start) {
271 spin_unlock(&nrs->nrs_lock);
273 rc = policy->pol_desc->pd_ops->op_policy_start(policy, arg);
275 spin_lock(&nrs->nrs_lock);
277 if (cfs_atomic_dec_and_test(&policy->pol_desc->pd_refs))
278 module_put(policy->pol_desc->pd_owner);
280 policy->pol_state = NRS_POL_STATE_STOPPED;
285 policy->pol_state = NRS_POL_STATE_STARTED;
287 if (policy->pol_flags & PTLRPC_NRS_FL_FALLBACK) {
289 * This path is only used at PTLRPC service setup time.
291 nrs->nrs_policy_fallback = policy;
294 * Try to stop the current primary policy if there is one.
296 nrs_policy_stop_primary(nrs);
299 * And set the newly-started policy as the primary one.
301 nrs->nrs_policy_primary = policy;
305 nrs->nrs_policy_starting = 0;
311 * Increases the policy's usage reference count.
313 static inline void nrs_policy_get_locked(struct ptlrpc_nrs_policy *policy)
319 * Decreases the policy's usage reference count, and stops the policy in case it
320 * was already stopping and have no more outstanding usage references (which
321 * indicates it has no more queued or started requests, and can be safely
324 static void nrs_policy_put_locked(struct ptlrpc_nrs_policy *policy)
326 LASSERT(policy->pol_ref > 0);
329 if (unlikely(policy->pol_ref == 0 &&
330 policy->pol_state == NRS_POL_STATE_STOPPING))
331 nrs_policy_stop0(policy);
334 static void nrs_policy_put(struct ptlrpc_nrs_policy *policy)
336 spin_lock(&policy->pol_nrs->nrs_lock);
337 nrs_policy_put_locked(policy);
338 spin_unlock(&policy->pol_nrs->nrs_lock);
342 * Find and return a policy by name.
344 static struct ptlrpc_nrs_policy * nrs_policy_find_locked(struct ptlrpc_nrs *nrs,
347 struct ptlrpc_nrs_policy *tmp;
349 cfs_list_for_each_entry(tmp, &nrs->nrs_policy_list, pol_list) {
350 if (strncmp(tmp->pol_desc->pd_name, name,
351 NRS_POL_NAME_MAX) == 0) {
352 nrs_policy_get_locked(tmp);
360 * Release references for the resource hierarchy moving upwards towards the
361 * policy instance resource.
363 static void nrs_resource_put(struct ptlrpc_nrs_resource *res)
365 struct ptlrpc_nrs_policy *policy = res->res_policy;
367 if (policy->pol_desc->pd_ops->op_res_put != NULL) {
368 struct ptlrpc_nrs_resource *parent;
370 for (; res != NULL; res = parent) {
371 parent = res->res_parent;
372 policy->pol_desc->pd_ops->op_res_put(policy, res);
378 * Obtains references for each resource in the resource hierarchy for request
379 * \a nrq if it is to be handled by \a policy.
381 * \param[in] policy the policy
382 * \param[in] nrq the request
383 * \param[in] moving_req denotes whether this is a call to the function by
384 * ldlm_lock_reorder_req(), in order to move \a nrq to
385 * the high-priority NRS head; we should not sleep when
388 * \retval NULL resource hierarchy references not obtained
389 * \retval valid-pointer the bottom level of the resource hierarchy
391 * \see ptlrpc_nrs_pol_ops::op_res_get()
394 struct ptlrpc_nrs_resource * nrs_resource_get(struct ptlrpc_nrs_policy *policy,
395 struct ptlrpc_nrs_request *nrq,
399 * Set to NULL to traverse the resource hierarchy from the top.
401 struct ptlrpc_nrs_resource *res = NULL;
402 struct ptlrpc_nrs_resource *tmp = NULL;
406 rc = policy->pol_desc->pd_ops->op_res_get(policy, nrq, res,
410 nrs_resource_put(res);
414 LASSERT(tmp != NULL);
415 tmp->res_parent = res;
416 tmp->res_policy = policy;
420 * Return once we have obtained a reference to the bottom level
421 * of the resource hierarchy.
429 * Obtains resources for the resource hierarchies and policy references for
430 * the fallback and current primary policy (if any), that will later be used
431 * to handle request \a nrq.
433 * \param[in] nrs the NRS head instance that will be handling request \a nrq.
434 * \param[in] nrq the request that is being handled.
435 * \param[out] resp the array where references to the resource hierarchy are
437 * \param[in] moving_req is set when obtaining resources while moving a
438 * request from a policy on the regular NRS head to a
439 * policy on the HP NRS head (via
440 * ldlm_lock_reorder_req()). It signifies that
441 * allocations to get resources should be atomic; for
442 * a full explanation, see comment in
443 * ptlrpc_nrs_pol_ops::op_res_get().
445 static void nrs_resource_get_safe(struct ptlrpc_nrs *nrs,
446 struct ptlrpc_nrs_request *nrq,
447 struct ptlrpc_nrs_resource **resp,
450 struct ptlrpc_nrs_policy *primary = NULL;
451 struct ptlrpc_nrs_policy *fallback = NULL;
453 memset(resp, 0, sizeof(resp[0]) * NRS_RES_MAX);
456 * Obtain policy references.
458 spin_lock(&nrs->nrs_lock);
460 fallback = nrs->nrs_policy_fallback;
461 nrs_policy_get_locked(fallback);
463 primary = nrs->nrs_policy_primary;
465 nrs_policy_get_locked(primary);
467 spin_unlock(&nrs->nrs_lock);
470 * Obtain resource hierarchy references.
472 resp[NRS_RES_FALLBACK] = nrs_resource_get(fallback, nrq, moving_req);
473 LASSERT(resp[NRS_RES_FALLBACK] != NULL);
475 if (primary != NULL) {
476 resp[NRS_RES_PRIMARY] = nrs_resource_get(primary, nrq,
479 * A primary policy may exist which may not wish to serve a
480 * particular request for different reasons; release the
481 * reference on the policy as it will not be used for this
484 if (resp[NRS_RES_PRIMARY] == NULL)
485 nrs_policy_put(primary);
490 * Releases references to resource hierarchies and policies, because they are no
491 * longer required; used when request handling has been completed, or the
492 * request is moving to the high priority NRS head.
494 * \param resp the resource hierarchy that is being released
496 * \see ptlrpcnrs_req_hp_move()
497 * \see ptlrpc_nrs_req_finalize()
499 static void nrs_resource_put_safe(struct ptlrpc_nrs_resource **resp)
501 struct ptlrpc_nrs_policy *pols[NRS_RES_MAX];
502 struct ptlrpc_nrs *nrs = NULL;
505 for (i = 0; i < NRS_RES_MAX; i++) {
506 if (resp[i] != NULL) {
507 pols[i] = resp[i]->res_policy;
508 nrs_resource_put(resp[i]);
515 for (i = 0; i < NRS_RES_MAX; i++) {
520 nrs = pols[i]->pol_nrs;
521 spin_lock(&nrs->nrs_lock);
523 nrs_policy_put_locked(pols[i]);
527 spin_unlock(&nrs->nrs_lock);
531 * Obtains an NRS request from \a policy for handling or examination; the
532 * request should be removed in the 'handling' case.
534 * Calling into this function implies we already know the policy has a request
535 * waiting to be handled.
537 * \param[in] policy the policy from which a request
538 * \param[in] peek when set, signifies that we just want to examine the
539 * request, and not handle it, so the request is not removed
541 * \param[in] force when set, it will force a policy to return a request if it
544 * \retval the NRS request to be handled
547 struct ptlrpc_nrs_request * nrs_request_get(struct ptlrpc_nrs_policy *policy,
548 bool peek, bool force)
550 struct ptlrpc_nrs_request *nrq;
552 LASSERT(policy->pol_req_queued > 0);
554 nrq = policy->pol_desc->pd_ops->op_req_get(policy, peek, force);
556 LASSERT(ergo(nrq != NULL, nrs_request_policy(nrq) == policy));
562 * Enqueues request \a nrq for later handling, via one one the policies for
563 * which resources where earlier obtained via nrs_resource_get_safe(). The
564 * function attempts to enqueue the request first on the primary policy
565 * (if any), since this is the preferred choice.
567 * \param nrq the request being enqueued
569 * \see nrs_resource_get_safe()
571 static inline void nrs_request_enqueue(struct ptlrpc_nrs_request *nrq)
573 struct ptlrpc_nrs_policy *policy;
578 * Try in descending order, because the primary policy (if any) is
579 * the preferred choice.
581 for (i = NRS_RES_MAX - 1; i >= 0; i--) {
582 if (nrq->nr_res_ptrs[i] == NULL)
586 policy = nrq->nr_res_ptrs[i]->res_policy;
588 rc = policy->pol_desc->pd_ops->op_req_enqueue(policy, nrq);
590 policy->pol_nrs->nrs_req_queued++;
591 policy->pol_req_queued++;
596 * Should never get here, as at least the primary policy's
597 * ptlrpc_nrs_pol_ops::op_req_enqueue() implementation should always
604 * Called when a request has been handled
606 * \param[in] nrs the request that has been handled; can be used for
607 * job/resource control.
609 * \see ptlrpc_nrs_req_stop_nolock()
611 static inline void nrs_request_stop(struct ptlrpc_nrs_request *nrq)
613 struct ptlrpc_nrs_policy *policy = nrs_request_policy(nrq);
615 if (policy->pol_desc->pd_ops->op_req_stop)
616 policy->pol_desc->pd_ops->op_req_stop(policy, nrq);
618 LASSERT(policy->pol_nrs->nrs_req_started > 0);
619 LASSERT(policy->pol_req_started > 0);
621 policy->pol_nrs->nrs_req_started--;
622 policy->pol_req_started--;
626 * Handler for operations that can be carried out on policies.
628 * Handles opcodes that are common to all policy types within NRS core, and
629 * passes any unknown opcodes to the policy-specific control function.
631 * \param[in] nrs the NRS head this policy belongs to.
632 * \param[in] name the human-readable policy name; should be the same as
633 * ptlrpc_nrs_pol_desc::pd_name.
634 * \param[in] opc the opcode of the operation being carried out.
635 * \param[in,out] arg can be used to pass information in and out between when
636 * carrying an operation; usually data that is private to
637 * the policy at some level, or generic policy status
640 * \retval -ve error condition
641 * \retval 0 operation was carried out successfully
643 static int nrs_policy_ctl(struct ptlrpc_nrs *nrs, char *name,
644 enum ptlrpc_nrs_ctl opc, void *arg)
646 struct ptlrpc_nrs_policy *policy;
650 spin_lock(&nrs->nrs_lock);
652 policy = nrs_policy_find_locked(nrs, name);
654 GOTO(out, rc = -ENOENT);
658 * Unknown opcode, pass it down to the policy-specific control
659 * function for handling.
662 rc = nrs_policy_ctl_locked(policy, opc, arg);
668 case PTLRPC_NRS_CTL_START:
669 rc = nrs_policy_start_locked(policy, arg);
674 nrs_policy_put_locked(policy);
676 spin_unlock(&nrs->nrs_lock);
682 * Unregisters a policy by name.
684 * \param[in] nrs the NRS head this policy belongs to.
685 * \param[in] name the human-readable policy name; should be the same as
686 * ptlrpc_nrs_pol_desc::pd_name
691 static int nrs_policy_unregister(struct ptlrpc_nrs *nrs, char *name)
693 struct ptlrpc_nrs_policy *policy = NULL;
696 spin_lock(&nrs->nrs_lock);
698 policy = nrs_policy_find_locked(nrs, name);
699 if (policy == NULL) {
700 spin_unlock(&nrs->nrs_lock);
702 CERROR("Can't find NRS policy %s\n", name);
706 if (policy->pol_ref > 1) {
707 CERROR("Policy %s is busy with %d references\n", name,
708 (int)policy->pol_ref);
709 nrs_policy_put_locked(policy);
711 spin_unlock(&nrs->nrs_lock);
715 LASSERT(policy->pol_req_queued == 0);
716 LASSERT(policy->pol_req_started == 0);
718 if (policy->pol_state != NRS_POL_STATE_STOPPED) {
719 nrs_policy_stop_locked(policy);
720 LASSERT(policy->pol_state == NRS_POL_STATE_STOPPED);
723 cfs_list_del(&policy->pol_list);
726 nrs_policy_put_locked(policy);
728 spin_unlock(&nrs->nrs_lock);
730 nrs_policy_fini(policy);
732 LASSERT(policy->pol_private == NULL);
733 OBD_FREE_PTR(policy);
739 * Register a policy from \policy descriptor \a desc with NRS head \a nrs.
741 * \param[in] nrs the NRS head on which the policy will be registered.
742 * \param[in] desc the policy descriptor from which the information will be
743 * obtained to register the policy.
748 static int nrs_policy_register(struct ptlrpc_nrs *nrs,
749 struct ptlrpc_nrs_pol_desc *desc)
751 struct ptlrpc_nrs_policy *policy;
752 struct ptlrpc_nrs_policy *tmp;
753 struct ptlrpc_service_part *svcpt = nrs->nrs_svcpt;
757 LASSERT(svcpt != NULL);
758 LASSERT(desc->pd_ops != NULL);
759 LASSERT(desc->pd_ops->op_res_get != NULL);
760 LASSERT(desc->pd_ops->op_req_get != NULL);
761 LASSERT(desc->pd_ops->op_req_enqueue != NULL);
762 LASSERT(desc->pd_ops->op_req_dequeue != NULL);
763 LASSERT(desc->pd_compat != NULL);
765 OBD_CPT_ALLOC_GFP(policy, svcpt->scp_service->srv_cptable,
766 svcpt->scp_cpt, sizeof(*policy), __GFP_IO);
770 policy->pol_nrs = nrs;
771 policy->pol_desc = desc;
772 policy->pol_state = NRS_POL_STATE_STOPPED;
773 policy->pol_flags = desc->pd_flags;
775 CFS_INIT_LIST_HEAD(&policy->pol_list);
776 CFS_INIT_LIST_HEAD(&policy->pol_list_queued);
778 rc = nrs_policy_init(policy);
780 OBD_FREE_PTR(policy);
784 spin_lock(&nrs->nrs_lock);
786 tmp = nrs_policy_find_locked(nrs, policy->pol_desc->pd_name);
788 CERROR("NRS policy %s has been registered, can't register it "
789 "for %s\n", policy->pol_desc->pd_name,
790 svcpt->scp_service->srv_name);
791 nrs_policy_put_locked(tmp);
793 spin_unlock(&nrs->nrs_lock);
794 nrs_policy_fini(policy);
795 OBD_FREE_PTR(policy);
800 cfs_list_add_tail(&policy->pol_list, &nrs->nrs_policy_list);
803 if (policy->pol_flags & PTLRPC_NRS_FL_REG_START)
804 rc = nrs_policy_start_locked(policy, NULL);
806 spin_unlock(&nrs->nrs_lock);
809 (void) nrs_policy_unregister(nrs, policy->pol_desc->pd_name);
815 * Enqueue request \a req using one of the policies its resources are referring
818 * \param[in] req the request to enqueue.
820 static void ptlrpc_nrs_req_add_nolock(struct ptlrpc_request *req)
822 struct ptlrpc_nrs_policy *policy;
824 LASSERT(req->rq_nrq.nr_initialized);
825 LASSERT(!req->rq_nrq.nr_enqueued);
827 nrs_request_enqueue(&req->rq_nrq);
828 req->rq_nrq.nr_enqueued = 1;
830 policy = nrs_request_policy(&req->rq_nrq);
832 * Add the policy to the NRS head's list of policies with enqueued
833 * requests, if it has not been added there.
835 if (unlikely(cfs_list_empty(&policy->pol_list_queued)))
836 cfs_list_add_tail(&policy->pol_list_queued,
837 &policy->pol_nrs->nrs_policy_queued);
841 * Enqueue a request on the high priority NRS head.
843 * \param req the request to enqueue.
845 static void ptlrpc_nrs_hpreq_add_nolock(struct ptlrpc_request *req)
847 int opc = lustre_msg_get_opc(req->rq_reqmsg);
850 spin_lock(&req->rq_lock);
852 ptlrpc_nrs_req_add_nolock(req);
854 DEBUG_REQ(D_NET, req, "high priority req");
855 spin_unlock(&req->rq_lock);
860 * Returns a boolean predicate indicating whether the policy described by
861 * \a desc is adequate for use with service \a svc.
863 * \param[in] svc the service
864 * \param[in] desc the policy descriptor
866 * \retval false the policy is not compatible with the service
867 * \retval true the policy is compatible with the service
869 static inline bool nrs_policy_compatible(const struct ptlrpc_service *svc,
870 const struct ptlrpc_nrs_pol_desc *desc)
872 return desc->pd_compat(svc, desc);
876 * Registers all compatible policies in nrs_core.nrs_policies, for NRS head
879 * \param[in] nrs the NRS head
884 * \pre mutex_is_locked(&nrs_core.nrs_mutex)
886 * \see ptlrpc_service_nrs_setup()
888 static int nrs_register_policies_locked(struct ptlrpc_nrs *nrs)
890 struct ptlrpc_nrs_pol_desc *desc;
891 /* for convenience */
892 struct ptlrpc_service_part *svcpt = nrs->nrs_svcpt;
893 struct ptlrpc_service *svc = svcpt->scp_service;
897 LASSERT(mutex_is_locked(&nrs_core.nrs_mutex));
899 cfs_list_for_each_entry(desc, &nrs_core.nrs_policies, pd_list) {
900 if (nrs_policy_compatible(svc, desc)) {
901 rc = nrs_policy_register(nrs, desc);
903 CERROR("Failed to register NRS policy %s for "
904 "partition %d of service %s: %d\n",
905 desc->pd_name, svcpt->scp_cpt,
908 * Fail registration if any of the policies'
909 * registration fails.
920 * Initializes NRS head \a nrs of service partition \a svcpt, and registers all
921 * compatible policies in NRS core, with the NRS head.
923 * \param[in] nrs the NRS head
924 * \param[in] svcpt the PTLRPC service partition to setup
929 * \pre mutex_is_locked(&nrs_core.nrs_mutex)
931 static int nrs_svcpt_setup_locked0(struct ptlrpc_nrs *nrs,
932 struct ptlrpc_service_part *svcpt)
935 enum ptlrpc_nrs_queue_type queue;
937 LASSERT(mutex_is_locked(&nrs_core.nrs_mutex));
939 if (nrs == &svcpt->scp_nrs_reg)
940 queue = PTLRPC_NRS_QUEUE_REG;
941 else if (nrs == svcpt->scp_nrs_hp)
942 queue = PTLRPC_NRS_QUEUE_HP;
946 nrs->nrs_svcpt = svcpt;
947 nrs->nrs_queue_type = queue;
948 spin_lock_init(&nrs->nrs_lock);
949 CFS_INIT_LIST_HEAD(&nrs->nrs_policy_list);
950 CFS_INIT_LIST_HEAD(&nrs->nrs_policy_queued);
951 nrs->nrs_throttling = 0;
953 rc = nrs_register_policies_locked(nrs);
959 * Allocates a regular and optionally a high-priority NRS head (if the service
960 * handles high-priority RPCs), and then registers all available compatible
961 * policies on those NRS heads.
963 * \param[in,out] svcpt the PTLRPC service partition to setup
965 * \pre mutex_is_locked(&nrs_core.nrs_mutex)
967 static int nrs_svcpt_setup_locked(struct ptlrpc_service_part *svcpt)
969 struct ptlrpc_nrs *nrs;
973 LASSERT(mutex_is_locked(&nrs_core.nrs_mutex));
976 * Initialize the regular NRS head.
978 nrs = nrs_svcpt2nrs(svcpt, false);
979 rc = nrs_svcpt_setup_locked0(nrs, svcpt);
984 * Optionally allocate a high-priority NRS head.
986 if (svcpt->scp_service->srv_ops.so_hpreq_handler == NULL)
989 OBD_CPT_ALLOC_PTR(svcpt->scp_nrs_hp,
990 svcpt->scp_service->srv_cptable,
992 if (svcpt->scp_nrs_hp == NULL)
993 GOTO(out, rc = -ENOMEM);
995 nrs = nrs_svcpt2nrs(svcpt, true);
996 rc = nrs_svcpt_setup_locked0(nrs, svcpt);
1003 * Unregisters all policies on all available NRS heads in a service partition;
1004 * called at PTLRPC service unregistration time.
1006 * \param[in] svcpt the PTLRPC service partition
1008 * \pre mutex_is_locked(&nrs_core.nrs_mutex)
1010 static void nrs_svcpt_cleanup_locked(struct ptlrpc_service_part *svcpt)
1012 struct ptlrpc_nrs *nrs;
1013 struct ptlrpc_nrs_policy *policy;
1014 struct ptlrpc_nrs_policy *tmp;
1019 LASSERT(mutex_is_locked(&nrs_core.nrs_mutex));
1022 nrs = nrs_svcpt2nrs(svcpt, hp);
1023 nrs->nrs_stopping = 1;
1025 cfs_list_for_each_entry_safe(policy, tmp, &nrs->nrs_policy_list,
1027 rc = nrs_policy_unregister(nrs, policy->pol_desc->pd_name);
1032 * If the service partition has an HP NRS head, clean that up as well.
1034 if (!hp && nrs_svcpt_has_hp(svcpt)) {
1046 * Returns the descriptor for a policy as identified by by \a name.
1048 * \param[in] name the policy name
1050 * \retval the policy descriptor
1053 static struct ptlrpc_nrs_pol_desc *nrs_policy_find_desc_locked(const char *name)
1055 struct ptlrpc_nrs_pol_desc *tmp;
1058 cfs_list_for_each_entry(tmp, &nrs_core.nrs_policies, pd_list) {
1059 if (strncmp(tmp->pd_name, name, NRS_POL_NAME_MAX) == 0)
1066 * Removes the policy from all supported NRS heads of all partitions of all
1069 * \param[in] desc the policy descriptor to unregister
1072 * \retval 0 successfully unregistered policy on all supported NRS heads
1074 * \pre mutex_is_locked(&nrs_core.nrs_mutex)
1075 * \pre mutex_is_locked(&ptlrpc_all_services_mutex)
1077 static int nrs_policy_unregister_locked(struct ptlrpc_nrs_pol_desc *desc)
1079 struct ptlrpc_nrs *nrs;
1080 struct ptlrpc_service *svc;
1081 struct ptlrpc_service_part *svcpt;
1086 LASSERT(mutex_is_locked(&nrs_core.nrs_mutex));
1087 LASSERT(mutex_is_locked(&ptlrpc_all_services_mutex));
1089 cfs_list_for_each_entry(svc, &ptlrpc_all_services, srv_list) {
1091 if (!nrs_policy_compatible(svc, desc) ||
1092 unlikely(svc->srv_is_stopping))
1095 ptlrpc_service_for_each_part(svcpt, i, svc) {
1099 nrs = nrs_svcpt2nrs(svcpt, hp);
1100 rc = nrs_policy_unregister(nrs, desc->pd_name);
1102 * Ignore -ENOENT as the policy may not have registered
1103 * successfully on all service partitions.
1105 if (rc == -ENOENT) {
1107 } else if (rc != 0) {
1108 CERROR("Failed to unregister NRS policy %s for "
1109 "partition %d of service %s: %d\n",
1110 desc->pd_name, svcpt->scp_cpt,
1111 svcpt->scp_service->srv_name, rc);
1115 if (!hp && nrs_svc_has_hp(svc)) {
1121 if (desc->pd_ops->op_lprocfs_fini != NULL)
1122 desc->pd_ops->op_lprocfs_fini(svc);
1129 * Registers a new policy with NRS core.
1131 * The function will only succeed if policy registration with all compatible
1132 * service partitions (if any) is successful.
1134 * N.B. This function should be called either at ptlrpc module initialization
1135 * time when registering a policy that ships with NRS core, or in a
1136 * module's init() function for policies registering from other modules.
1138 * \param[in] conf configuration information for the new policy to register
1143 int ptlrpc_nrs_policy_register(struct ptlrpc_nrs_pol_conf *conf)
1145 struct ptlrpc_service *svc;
1146 struct ptlrpc_nrs_pol_desc *desc;
1150 LASSERT(conf != NULL);
1151 LASSERT(conf->nc_ops != NULL);
1152 LASSERT(conf->nc_compat != NULL);
1153 LASSERT(ergo(conf->nc_compat == nrs_policy_compat_one,
1154 conf->nc_compat_svc_name != NULL));
1155 LASSERT(ergo((conf->nc_flags & PTLRPC_NRS_FL_REG_EXTERN) != 0,
1156 conf->nc_owner != NULL));
1158 conf->nc_name[NRS_POL_NAME_MAX - 1] = '\0';
1161 * External policies are not allowed to start immediately upon
1162 * registration, as there is a relatively higher chance that their
1163 * registration might fail. In such a case, some policy instances may
1164 * already have requests queued wen unregistration needs to happen as
1165 * part o cleanup; since there is currently no way to drain requests
1166 * from a policy unless the service is unregistering, we just disallow
1169 if ((conf->nc_flags & PTLRPC_NRS_FL_REG_EXTERN) &&
1170 (conf->nc_flags & (PTLRPC_NRS_FL_FALLBACK |
1171 PTLRPC_NRS_FL_REG_START))) {
1172 CERROR("NRS: failing to register policy %s. Please check "
1173 "policy flags; external policies cannot act as fallback "
1174 "policies, or be started immediately upon registration "
1175 "without interaction with lprocfs\n", conf->nc_name);
1179 mutex_lock(&nrs_core.nrs_mutex);
1181 if (nrs_policy_find_desc_locked(conf->nc_name) != NULL) {
1182 CERROR("NRS: failing to register policy %s which has already "
1183 "been registered with NRS core!\n",
1185 GOTO(fail, rc = -EEXIST);
1188 OBD_ALLOC_PTR(desc);
1190 GOTO(fail, rc = -ENOMEM);
1192 if (strlcpy(desc->pd_name, conf->nc_name, sizeof(desc->pd_name)) >=
1193 sizeof(desc->pd_name)) {
1195 GOTO(fail, rc = -E2BIG);
1197 desc->pd_ops = conf->nc_ops;
1198 desc->pd_compat = conf->nc_compat;
1199 desc->pd_compat_svc_name = conf->nc_compat_svc_name;
1200 if ((conf->nc_flags & PTLRPC_NRS_FL_REG_EXTERN) != 0)
1201 desc->pd_owner = conf->nc_owner;
1202 desc->pd_flags = conf->nc_flags;
1203 cfs_atomic_set(&desc->pd_refs, 0);
1206 * For policies that are held in the same module as NRS (currently
1207 * ptlrpc), do not register the policy with all compatible services,
1208 * as the services will not have started at this point, since we are
1209 * calling from ptlrpc module initialization code. In such cases each
1210 * service will register all compatible policies later, via
1211 * ptlrpc_service_nrs_setup().
1213 if ((conf->nc_flags & PTLRPC_NRS_FL_REG_EXTERN) == 0)
1217 * Register the new policy on all compatible services
1219 mutex_lock(&ptlrpc_all_services_mutex);
1221 cfs_list_for_each_entry(svc, &ptlrpc_all_services, srv_list) {
1222 struct ptlrpc_service_part *svcpt;
1226 if (!nrs_policy_compatible(svc, desc) ||
1227 unlikely(svc->srv_is_stopping))
1230 ptlrpc_service_for_each_part(svcpt, i, svc) {
1231 struct ptlrpc_nrs *nrs;
1234 nrs = nrs_svcpt2nrs(svcpt, hp);
1235 rc = nrs_policy_register(nrs, desc);
1237 CERROR("Failed to register NRS policy %s for "
1238 "partition %d of service %s: %d\n",
1239 desc->pd_name, svcpt->scp_cpt,
1240 svcpt->scp_service->srv_name, rc);
1242 rc2 = nrs_policy_unregister_locked(desc);
1244 * Should not fail at this point
1247 mutex_unlock(&ptlrpc_all_services_mutex);
1252 if (!hp && nrs_svc_has_hp(svc)) {
1259 * No need to take a reference to other modules here, as we
1260 * will be calling from the module's init() function.
1262 if (desc->pd_ops->op_lprocfs_init != NULL) {
1263 rc = desc->pd_ops->op_lprocfs_init(svc);
1265 rc2 = nrs_policy_unregister_locked(desc);
1267 * Should not fail at this point
1270 mutex_unlock(&ptlrpc_all_services_mutex);
1277 mutex_unlock(&ptlrpc_all_services_mutex);
1279 cfs_list_add_tail(&desc->pd_list, &nrs_core.nrs_policies);
1281 mutex_unlock(&nrs_core.nrs_mutex);
1285 EXPORT_SYMBOL(ptlrpc_nrs_policy_register);
1288 * Unregisters a previously registered policy with NRS core. All instances of
1289 * the policy on all NRS heads of all supported services are removed.
1291 * N.B. This function should only be called from a module's exit() function.
1292 * Although it can be used for policies that ship alongside NRS core, the
1293 * function is primarily intended for policies that register externally,
1294 * from other modules.
1296 * \param[in] conf configuration information for the policy to unregister
1301 int ptlrpc_nrs_policy_unregister(struct ptlrpc_nrs_pol_conf *conf)
1303 struct ptlrpc_nrs_pol_desc *desc;
1307 LASSERT(conf != NULL);
1309 if (conf->nc_flags & PTLRPC_NRS_FL_FALLBACK) {
1310 CERROR("Unable to unregister a fallback policy, unless the "
1311 "PTLRPC service is stopping.\n");
1315 conf->nc_name[NRS_POL_NAME_MAX - 1] = '\0';
1317 mutex_lock(&nrs_core.nrs_mutex);
1319 desc = nrs_policy_find_desc_locked(conf->nc_name);
1321 CERROR("Failing to unregister NRS policy %s which has "
1322 "not been registered with NRS core!\n",
1324 GOTO(not_exist, rc = -ENOENT);
1327 mutex_lock(&ptlrpc_all_services_mutex);
1329 rc = nrs_policy_unregister_locked(desc);
1332 CERROR("Please first stop policy %s on all service "
1333 "partitions and then retry to unregister the "
1334 "policy.\n", conf->nc_name);
1338 CDEBUG(D_INFO, "Unregistering policy %s from NRS core.\n",
1341 cfs_list_del(&desc->pd_list);
1345 mutex_unlock(&ptlrpc_all_services_mutex);
1348 mutex_unlock(&nrs_core.nrs_mutex);
1352 EXPORT_SYMBOL(ptlrpc_nrs_policy_unregister);
1355 * Setup NRS heads on all service partitions of service \a svc, and register
1356 * all compatible policies on those NRS heads.
1358 * To be called from withing ptl
1359 * \param[in] svc the service to setup
1361 * \retval -ve error, the calling logic should eventually call
1362 * ptlrpc_service_nrs_cleanup() to undo any work performed
1365 * \see ptlrpc_register_service()
1366 * \see ptlrpc_service_nrs_cleanup()
1368 int ptlrpc_service_nrs_setup(struct ptlrpc_service *svc)
1370 struct ptlrpc_service_part *svcpt;
1371 const struct ptlrpc_nrs_pol_desc *desc;
1375 mutex_lock(&nrs_core.nrs_mutex);
1378 * Initialize NRS heads on all service CPTs.
1380 ptlrpc_service_for_each_part(svcpt, i, svc) {
1381 rc = nrs_svcpt_setup_locked(svcpt);
1387 * Set up lprocfs interfaces for all supported policies for the
1390 cfs_list_for_each_entry(desc, &nrs_core.nrs_policies, pd_list) {
1391 if (!nrs_policy_compatible(svc, desc))
1394 if (desc->pd_ops->op_lprocfs_init != NULL) {
1395 rc = desc->pd_ops->op_lprocfs_init(svc);
1403 mutex_unlock(&nrs_core.nrs_mutex);
1409 * Unregisters all policies on all service partitions of service \a svc.
1411 * \param[in] svc the PTLRPC service to unregister
1413 void ptlrpc_service_nrs_cleanup(struct ptlrpc_service *svc)
1415 struct ptlrpc_service_part *svcpt;
1416 const struct ptlrpc_nrs_pol_desc *desc;
1419 mutex_lock(&nrs_core.nrs_mutex);
1422 * Clean up NRS heads on all service partitions
1424 ptlrpc_service_for_each_part(svcpt, i, svc)
1425 nrs_svcpt_cleanup_locked(svcpt);
1428 * Clean up lprocfs interfaces for all supported policies for the
1431 cfs_list_for_each_entry(desc, &nrs_core.nrs_policies, pd_list) {
1432 if (!nrs_policy_compatible(svc, desc))
1435 if (desc->pd_ops->op_lprocfs_fini != NULL)
1436 desc->pd_ops->op_lprocfs_fini(svc);
1439 mutex_unlock(&nrs_core.nrs_mutex);
1443 * Obtains NRS head resources for request \a req.
1445 * These could be either on the regular or HP NRS head of \a svcpt; resources
1446 * taken on the regular head can later be swapped for HP head resources by
1447 * ldlm_lock_reorder_req().
1449 * \param[in] svcpt the service partition
1450 * \param[in] req the request
1451 * \param[in] hp which NRS head of \a svcpt to use
1453 void ptlrpc_nrs_req_initialize(struct ptlrpc_service_part *svcpt,
1454 struct ptlrpc_request *req, bool hp)
1456 struct ptlrpc_nrs *nrs = nrs_svcpt2nrs(svcpt, hp);
1458 memset(&req->rq_nrq, 0, sizeof(req->rq_nrq));
1459 nrs_resource_get_safe(nrs, &req->rq_nrq, req->rq_nrq.nr_res_ptrs,
1463 * It is fine to access \e nr_initialized without locking as there is
1464 * no contention at this early stage.
1466 req->rq_nrq.nr_initialized = 1;
1470 * Releases resources for a request; is called after the request has been
1473 * \param[in] req the request
1475 * \see ptlrpc_server_finish_request()
1477 void ptlrpc_nrs_req_finalize(struct ptlrpc_request *req)
1479 if (req->rq_nrq.nr_initialized) {
1480 nrs_resource_put_safe(req->rq_nrq.nr_res_ptrs);
1481 /* no protection on bit nr_initialized because no
1482 * contention at this late stage */
1483 req->rq_nrq.nr_finalized = 1;
1487 void ptlrpc_nrs_req_stop_nolock(struct ptlrpc_request *req)
1489 if (req->rq_nrq.nr_started)
1490 nrs_request_stop(&req->rq_nrq);
1494 * Enqueues request \a req on either the regular or high-priority NRS head
1495 * of service partition \a svcpt.
1497 * \param[in] svcpt the service partition
1498 * \param[in] req the request to be enqueued
1499 * \param[in] hp whether to enqueue the request on the regular or
1500 * high-priority NRS head.
1502 void ptlrpc_nrs_req_add(struct ptlrpc_service_part *svcpt,
1503 struct ptlrpc_request *req, bool hp)
1505 spin_lock(&svcpt->scp_req_lock);
1508 ptlrpc_nrs_hpreq_add_nolock(req);
1510 ptlrpc_nrs_req_add_nolock(req);
1512 spin_unlock(&svcpt->scp_req_lock);
1515 static void nrs_request_removed(struct ptlrpc_nrs_policy *policy)
1517 LASSERT(policy->pol_nrs->nrs_req_queued > 0);
1518 LASSERT(policy->pol_req_queued > 0);
1520 policy->pol_nrs->nrs_req_queued--;
1521 policy->pol_req_queued--;
1524 * If the policy has no more requests queued, remove it from
1525 * ptlrpc_nrs::nrs_policy_queued.
1527 if (unlikely(policy->pol_req_queued == 0)) {
1528 cfs_list_del_init(&policy->pol_list_queued);
1531 * If there are other policies with queued requests, move the
1532 * current policy to the end so that we can round robin over
1533 * all policies and drain the requests.
1535 } else if (policy->pol_req_queued != policy->pol_nrs->nrs_req_queued) {
1536 LASSERT(policy->pol_req_queued <
1537 policy->pol_nrs->nrs_req_queued);
1539 cfs_list_move_tail(&policy->pol_list_queued,
1540 &policy->pol_nrs->nrs_policy_queued);
1545 * Obtains a request for handling from an NRS head of service partition
1548 * \param[in] svcpt the service partition
1549 * \param[in] hp whether to obtain a request from the regular or
1550 * high-priority NRS head.
1551 * \param[in] peek when set, signifies that we just want to examine the
1552 * request, and not handle it, so the request is not removed
1554 * \param[in] force when set, it will force a policy to return a request if it
1557 * \retval the request to be handled
1558 * \retval NULL the head has no requests to serve
1560 struct ptlrpc_request *
1561 ptlrpc_nrs_req_get_nolock0(struct ptlrpc_service_part *svcpt, bool hp,
1562 bool peek, bool force)
1564 struct ptlrpc_nrs *nrs = nrs_svcpt2nrs(svcpt, hp);
1565 struct ptlrpc_nrs_policy *policy;
1566 struct ptlrpc_nrs_request *nrq;
1569 * Always try to drain requests from all NRS polices even if they are
1570 * inactive, because the user can change policy status at runtime.
1572 cfs_list_for_each_entry(policy, &nrs->nrs_policy_queued,
1574 nrq = nrs_request_get(policy, peek, force);
1576 if (likely(!peek)) {
1577 nrq->nr_started = 1;
1579 policy->pol_req_started++;
1580 policy->pol_nrs->nrs_req_started++;
1582 nrs_request_removed(policy);
1585 return container_of(nrq, struct ptlrpc_request, rq_nrq);
1593 * Dequeues request \a req from the policy it has been enqueued on.
1595 * \param[in] req the request
1597 void ptlrpc_nrs_req_del_nolock(struct ptlrpc_request *req)
1599 struct ptlrpc_nrs_policy *policy = nrs_request_policy(&req->rq_nrq);
1601 policy->pol_desc->pd_ops->op_req_dequeue(policy, &req->rq_nrq);
1603 req->rq_nrq.nr_enqueued = 0;
1605 nrs_request_removed(policy);
1609 * Returns whether there are any requests currently enqueued on any of the
1610 * policies of service partition's \a svcpt NRS head specified by \a hp. Should
1611 * be called while holding ptlrpc_service_part::scp_req_lock to get a reliable
1614 * \param[in] svcpt the service partition to enquire.
1615 * \param[in] hp whether the regular or high-priority NRS head is to be
1618 * \retval false the indicated NRS head has no enqueued requests.
1619 * \retval true the indicated NRS head has some enqueued requests.
1621 bool ptlrpc_nrs_req_pending_nolock(struct ptlrpc_service_part *svcpt, bool hp)
1623 struct ptlrpc_nrs *nrs = nrs_svcpt2nrs(svcpt, hp);
1625 return nrs->nrs_req_queued > 0;
1629 * Returns whether NRS policy is throttling reqeust
1631 * \param[in] svcpt the service partition to enquire.
1632 * \param[in] hp whether the regular or high-priority NRS head is to be
1635 * \retval false the indicated NRS head has no enqueued requests.
1636 * \retval true the indicated NRS head has some enqueued requests.
1638 bool ptlrpc_nrs_req_throttling_nolock(struct ptlrpc_service_part *svcpt,
1641 struct ptlrpc_nrs *nrs = nrs_svcpt2nrs(svcpt, hp);
1643 return !!nrs->nrs_throttling;
1647 * Moves request \a req from the regular to the high-priority NRS head.
1649 * \param[in] req the request to move
1651 void ptlrpc_nrs_req_hp_move(struct ptlrpc_request *req)
1653 struct ptlrpc_service_part *svcpt = req->rq_rqbd->rqbd_svcpt;
1654 struct ptlrpc_nrs_request *nrq = &req->rq_nrq;
1655 struct ptlrpc_nrs_resource *res1[NRS_RES_MAX];
1656 struct ptlrpc_nrs_resource *res2[NRS_RES_MAX];
1660 * Obtain the high-priority NRS head resources.
1662 nrs_resource_get_safe(nrs_svcpt2nrs(svcpt, true), nrq, res1, true);
1664 spin_lock(&svcpt->scp_req_lock);
1666 if (!ptlrpc_nrs_req_can_move(req))
1669 ptlrpc_nrs_req_del_nolock(req);
1671 memcpy(res2, nrq->nr_res_ptrs, NRS_RES_MAX * sizeof(res2[0]));
1672 memcpy(nrq->nr_res_ptrs, res1, NRS_RES_MAX * sizeof(res1[0]));
1674 ptlrpc_nrs_hpreq_add_nolock(req);
1676 memcpy(res1, res2, NRS_RES_MAX * sizeof(res1[0]));
1678 spin_unlock(&svcpt->scp_req_lock);
1681 * Release either the regular NRS head resources if we moved the
1682 * request, or the high-priority NRS head resources if we took a
1683 * reference earlier in this function and ptlrpc_nrs_req_can_move()
1686 nrs_resource_put_safe(res1);
1691 * Carries out a control operation \a opc on the policy identified by the
1692 * human-readable \a name, on either all partitions, or only on the first
1693 * partition of service \a svc.
1695 * \param[in] svc the service the policy belongs to.
1696 * \param[in] queue whether to carry out the command on the policy which
1697 * belongs to the regular, high-priority, or both NRS
1698 * heads of service partitions of \a svc.
1699 * \param[in] name the policy to act upon, by human-readable name
1700 * \param[in] opc the opcode of the operation to carry out
1701 * \param[in] single when set, the operation will only be carried out on the
1702 * NRS heads of the first service partition of \a svc.
1703 * This is useful for some policies which e.g. share
1704 * identical values on the same parameters of different
1705 * service partitions; when reading these parameters via
1706 * lprocfs, these policies may just want to obtain and
1707 * print out the values from the first service partition.
1708 * Storing these values centrally elsewhere then could be
1709 * another solution for this.
1710 * \param[in,out] arg can be used as a generic in/out buffer between control
1711 * operations and the user environment.
1713 *\retval -ve error condition
1714 *\retval 0 operation was carried out successfully
1716 int ptlrpc_nrs_policy_control(const struct ptlrpc_service *svc,
1717 enum ptlrpc_nrs_queue_type queue, char *name,
1718 enum ptlrpc_nrs_ctl opc, bool single, void *arg)
1720 struct ptlrpc_service_part *svcpt;
1725 LASSERT(opc != PTLRPC_NRS_CTL_INVALID);
1727 if ((queue & PTLRPC_NRS_QUEUE_BOTH) == 0)
1730 ptlrpc_service_for_each_part(svcpt, i, svc) {
1731 if ((queue & PTLRPC_NRS_QUEUE_REG) != 0) {
1732 rc = nrs_policy_ctl(nrs_svcpt2nrs(svcpt, false), name,
1734 if (rc != 0 || (queue == PTLRPC_NRS_QUEUE_REG &&
1739 if ((queue & PTLRPC_NRS_QUEUE_HP) != 0) {
1741 * XXX: We could optionally check for
1742 * nrs_svc_has_hp(svc) here, and return an error if it
1743 * is false. Right now we rely on the policies' lprocfs
1744 * handlers that call the present function to make this
1745 * check; if they fail to do so, they might hit the
1746 * assertion inside nrs_svcpt2nrs() below.
1748 rc = nrs_policy_ctl(nrs_svcpt2nrs(svcpt, true), name,
1750 if (rc != 0 || single)
1759 /* ptlrpc/nrs_fifo.c */
1760 extern struct ptlrpc_nrs_pol_conf nrs_conf_fifo;
1761 #if defined HAVE_SERVER_SUPPORT && defined(__KERNEL__)
1762 /* ptlrpc/nrs_crr.c */
1763 extern struct ptlrpc_nrs_pol_conf nrs_conf_crrn;
1764 /* ptlrpc/nrs_orr.c */
1765 extern struct ptlrpc_nrs_pol_conf nrs_conf_orr;
1766 extern struct ptlrpc_nrs_pol_conf nrs_conf_trr;
1767 extern struct ptlrpc_nrs_pol_conf nrs_conf_tbf;
1771 * Adds all policies that ship with the ptlrpc module, to NRS core's list of
1772 * policies \e nrs_core.nrs_policies.
1774 * \retval 0 all policies have been registered successfully
1777 int ptlrpc_nrs_init(void)
1782 mutex_init(&nrs_core.nrs_mutex);
1783 CFS_INIT_LIST_HEAD(&nrs_core.nrs_policies);
1785 rc = ptlrpc_nrs_policy_register(&nrs_conf_fifo);
1789 #if defined HAVE_SERVER_SUPPORT && defined(__KERNEL__)
1790 rc = ptlrpc_nrs_policy_register(&nrs_conf_crrn);
1794 rc = ptlrpc_nrs_policy_register(&nrs_conf_orr);
1798 rc = ptlrpc_nrs_policy_register(&nrs_conf_trr);
1801 rc = ptlrpc_nrs_policy_register(&nrs_conf_tbf);
1809 * Since no PTLRPC services have been started at this point, all we need
1810 * to do for cleanup is to free the descriptors.
1818 * Removes all policy desciptors from nrs_core::nrs_policies, and frees the
1819 * policy descriptors.
1821 * Since all PTLRPC services are stopped at this point, there are no more
1822 * instances of any policies, because each service will have stopped its policy
1823 * instances in ptlrpc_service_nrs_cleanup(), so we just need to free the
1826 void ptlrpc_nrs_fini(void)
1828 struct ptlrpc_nrs_pol_desc *desc;
1829 struct ptlrpc_nrs_pol_desc *tmp;
1831 cfs_list_for_each_entry_safe(desc, tmp, &nrs_core.nrs_policies,
1833 cfs_list_del_init(&desc->pd_list);