Whamcloud - gitweb
f4700f815c7948eca5955f17d6f076048b345b49
[fs/lustre-release.git] / lustre / ptlrpc / nrs.c
1 /*
2  * GPL HEADER START
3  *
4  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
5  *
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.
9
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.
15
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
19  *
20  * GPL HEADER END
21  */
22 /*
23  * Copyright (c) 2011 Intel Corporation
24  *
25  * Copyright 2012 Xyratex Technology Limited
26  */
27 /*
28  * lustre/ptlrpc/nrs.c
29  *
30  * Network Request Scheduler (NRS)
31  *
32  * Allows to reorder the handling of RPCs at servers.
33  *
34  * Author: Liang Zhen <liang@whamcloud.com>
35  * Author: Nikitas Angelinas <nikitas_angelinas@xyratex.com>
36  */
37 /**
38  * \addtogoup nrs
39  * @{
40  */
41
42 #define DEBUG_SUBSYSTEM S_RPC
43 #ifndef __KERNEL__
44 #include <liblustre.h>
45 #endif
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"
52
53 /* XXX: This is just for liblustre. Remove the #if defined directive when the
54  * "cfs_" prefix is dropped from cfs_list_head. */
55 #if defined (__linux__) && defined(__KERNEL__)
56 extern struct list_head ptlrpc_all_services;
57 #else
58 extern struct cfs_list_head ptlrpc_all_services;
59 #endif
60
61 /**
62  * NRS core object.
63  */
64 struct nrs_core nrs_core;
65
66 static int nrs_policy_init(struct ptlrpc_nrs_policy *policy)
67 {
68         return policy->pol_desc->pd_ops->op_policy_init != NULL ?
69                policy->pol_desc->pd_ops->op_policy_init(policy) : 0;
70 }
71
72 static void nrs_policy_fini(struct ptlrpc_nrs_policy *policy)
73 {
74         LASSERT(policy->pol_ref == 0);
75         LASSERT(policy->pol_req_queued == 0);
76
77         if (policy->pol_desc->pd_ops->op_policy_fini != NULL)
78                 policy->pol_desc->pd_ops->op_policy_fini(policy);
79 }
80
81 static int nrs_policy_ctl_locked(struct ptlrpc_nrs_policy *policy,
82                                  enum ptlrpc_nrs_ctl opc, void *arg)
83 {
84         /**
85          * The policy may be stopped, but the lprocfs files and
86          * ptlrpc_nrs_policy instances remain present until unregistration time.
87          * Do not perform the ctl operation if the policy is stopped, as
88          * policy->pol_private will be NULL in such a case.
89          */
90         if (policy->pol_state == NRS_POL_STATE_STOPPED)
91                 RETURN(-ENODEV);
92
93         RETURN(policy->pol_desc->pd_ops->op_policy_ctl != NULL ?
94                policy->pol_desc->pd_ops->op_policy_ctl(policy, opc, arg) :
95                -ENOSYS);
96 }
97
98 static void nrs_policy_stop0(struct ptlrpc_nrs_policy *policy)
99 {
100         struct ptlrpc_nrs *nrs = policy->pol_nrs;
101         ENTRY;
102
103         if (policy->pol_desc->pd_ops->op_policy_stop != NULL) {
104                 spin_unlock(&nrs->nrs_lock);
105
106                 policy->pol_desc->pd_ops->op_policy_stop(policy);
107
108                 spin_lock(&nrs->nrs_lock);
109         }
110
111         LASSERT(cfs_list_empty(&policy->pol_list_queued));
112         LASSERT(policy->pol_req_queued == 0 &&
113                 policy->pol_req_started == 0);
114
115         policy->pol_private = NULL;
116
117         policy->pol_state = NRS_POL_STATE_STOPPED;
118
119         if (cfs_atomic_dec_and_test(&policy->pol_desc->pd_refs))
120                 cfs_module_put(policy->pol_desc->pd_owner);
121
122         EXIT;
123 }
124
125 static int nrs_policy_stop_locked(struct ptlrpc_nrs_policy *policy)
126 {
127         struct ptlrpc_nrs *nrs = policy->pol_nrs;
128         ENTRY;
129
130         if (nrs->nrs_policy_fallback == policy && !nrs->nrs_stopping)
131                 RETURN(-EPERM);
132
133         if (policy->pol_state == NRS_POL_STATE_STARTING)
134                 RETURN(-EAGAIN);
135
136         /* In progress or already stopped */
137         if (policy->pol_state != NRS_POL_STATE_STARTED)
138                 RETURN(0);
139
140         policy->pol_state = NRS_POL_STATE_STOPPING;
141
142         /* Immediately make it invisible */
143         if (nrs->nrs_policy_primary == policy) {
144                 nrs->nrs_policy_primary = NULL;
145
146         } else {
147                 LASSERT(nrs->nrs_policy_fallback == policy);
148                 nrs->nrs_policy_fallback = NULL;
149         }
150
151         /* I have the only refcount */
152         if (policy->pol_ref == 1)
153                 nrs_policy_stop0(policy);
154
155         RETURN(0);
156 }
157
158 /**
159  * Transitions the \a nrs NRS head's primary policy to
160  * ptlrpc_nrs_pol_state::NRS_POL_STATE_STOPPING and if the policy has no
161  * pending usage references, to ptlrpc_nrs_pol_state::NRS_POL_STATE_STOPPED.
162  *
163  * \param[in] nrs the NRS head to carry out this operation on
164  */
165 static void nrs_policy_stop_primary(struct ptlrpc_nrs *nrs)
166 {
167         struct ptlrpc_nrs_policy *tmp = nrs->nrs_policy_primary;
168         ENTRY;
169
170         if (tmp == NULL) {
171                 /**
172                  * XXX: This should really be RETURN_EXIT, but the latter does
173                  * not currently print anything out, and possibly should be
174                  * fixed to do so.
175                  */
176                 EXIT;
177                 return;
178         }
179
180         nrs->nrs_policy_primary = NULL;
181
182         LASSERT(tmp->pol_state == NRS_POL_STATE_STARTED);
183         tmp->pol_state = NRS_POL_STATE_STOPPING;
184
185         if (tmp->pol_ref == 0)
186                 nrs_policy_stop0(tmp);
187         EXIT;
188 }
189
190 /**
191  * Transitions a policy across the ptlrpc_nrs_pol_state range of values, in
192  * response to an lprocfs command to start a policy.
193  *
194  * If a primary policy different to the current one is specified, this function
195  * will transition the new policy to the
196  * ptlrpc_nrs_pol_state::NRS_POL_STATE_STARTING and then to
197  * ptlrpc_nrs_pol_state::NRS_POL_STATE_STARTED, and will then transition
198  * the old primary policy (if there is one) 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.
201  *
202  * If the fallback policy is specified, this is taken to indicate an instruction
203  * to stop the current primary policy, without substituting it with another
204  * primary policy, so the primary policy (if any) is transitioned to
205  * ptlrpc_nrs_pol_state::NRS_POL_STATE_STOPPING, and if there are no outstanding
206  * references on the policy to ptlrpc_nrs_pol_stae::NRS_POL_STATE_STOPPED. In
207  * this case, the fallback policy is only left active in the NRS head.
208  */
209 static int nrs_policy_start_locked(struct ptlrpc_nrs_policy *policy)
210 {
211         struct ptlrpc_nrs      *nrs = policy->pol_nrs;
212         int                     rc = 0;
213         ENTRY;
214
215         /**
216          * Don't allow multiple starting which is too complex, and has no real
217          * benefit.
218          */
219         if (nrs->nrs_policy_starting)
220                 RETURN(-EAGAIN);
221
222         LASSERT(policy->pol_state != NRS_POL_STATE_STARTING);
223
224         if (policy->pol_state == NRS_POL_STATE_STOPPING)
225                 RETURN(-EAGAIN);
226
227         if (policy->pol_flags & PTLRPC_NRS_FL_FALLBACK) {
228                 /**
229                  * This is for cases in which the user sets the policy to the
230                  * fallback policy (currently fifo for all services); i.e. the
231                  * user is resetting the policy to the default; so we stop the
232                  * primary policy, if any.
233                  */
234                 if (policy == nrs->nrs_policy_fallback) {
235                         nrs_policy_stop_primary(nrs);
236                         RETURN(0);
237                 }
238
239                 /**
240                  * If we reach here, we must be setting up the fallback policy
241                  * at service startup time, and only a single policy with the
242                  * nrs_policy_flags::PTLRPC_NRS_FL_FALLBACK flag set can
243                  * register with NRS core.
244                  */
245                 LASSERT(nrs->nrs_policy_fallback == NULL);
246         } else {
247                 /**
248                  * Shouldn't start primary policy if w/o fallback policy.
249                  */
250                 if (nrs->nrs_policy_fallback == NULL)
251                         RETURN(-EPERM);
252
253                 if (policy->pol_state == NRS_POL_STATE_STARTED)
254                         RETURN(0);
255         }
256
257         /**
258          * Increase the module usage count for policies registering from other
259          * modules.
260          */
261         if (cfs_atomic_inc_return(&policy->pol_desc->pd_refs) == 1 &&
262             !cfs_try_module_get(policy->pol_desc->pd_owner)) {
263                 cfs_atomic_dec(&policy->pol_desc->pd_refs);
264                 CERROR("NRS: cannot get module for policy %s; is it alive?\n",
265                        policy->pol_desc->pd_name);
266                 RETURN(-ENODEV);
267         }
268
269         /**
270          * Serialize policy starting across the NRS head
271          */
272         nrs->nrs_policy_starting = 1;
273
274         policy->pol_state = NRS_POL_STATE_STARTING;
275
276         if (policy->pol_desc->pd_ops->op_policy_start) {
277                 spin_unlock(&nrs->nrs_lock);
278
279                 rc = policy->pol_desc->pd_ops->op_policy_start(policy);
280
281                 spin_lock(&nrs->nrs_lock);
282                 if (rc != 0) {
283                         if (cfs_atomic_dec_and_test(&policy->pol_desc->pd_refs))
284                                 cfs_module_put(policy->pol_desc->pd_owner);
285
286                         policy->pol_state = NRS_POL_STATE_STOPPED;
287                         GOTO(out, rc);
288                 }
289         }
290
291         policy->pol_state = NRS_POL_STATE_STARTED;
292
293         if (policy->pol_flags & PTLRPC_NRS_FL_FALLBACK) {
294                 /**
295                  * This path is only used at PTLRPC service setup time.
296                  */
297                 nrs->nrs_policy_fallback = policy;
298         } else {
299                 /*
300                  * Try to stop the current primary policy if there is one.
301                  */
302                 nrs_policy_stop_primary(nrs);
303
304                 /**
305                  * And set the newly-started policy as the primary one.
306                  */
307                 nrs->nrs_policy_primary = policy;
308         }
309
310 out:
311         nrs->nrs_policy_starting = 0;
312
313         RETURN(rc);
314 }
315
316 /**
317  * Increases the policy's usage reference count.
318  */
319 static inline void nrs_policy_get_locked(struct ptlrpc_nrs_policy *policy)
320 {
321         policy->pol_ref++;
322 }
323
324 /**
325  * Decreases the policy's usage reference count, and stops the policy in case it
326  * was already stopping and have no more outstanding usage references (which
327  * indicates it has no more queued or started requests, and can be safely
328  * stopped).
329  */
330 static void nrs_policy_put_locked(struct ptlrpc_nrs_policy *policy)
331 {
332         LASSERT(policy->pol_ref > 0);
333
334         policy->pol_ref--;
335         if (unlikely(policy->pol_ref == 0 &&
336             policy->pol_state == NRS_POL_STATE_STOPPING))
337                 nrs_policy_stop0(policy);
338 }
339
340 static void nrs_policy_put(struct ptlrpc_nrs_policy *policy)
341 {
342         spin_lock(&policy->pol_nrs->nrs_lock);
343         nrs_policy_put_locked(policy);
344         spin_unlock(&policy->pol_nrs->nrs_lock);
345 }
346
347 /**
348  * Find and return a policy by name.
349  */
350 static struct ptlrpc_nrs_policy * nrs_policy_find_locked(struct ptlrpc_nrs *nrs,
351                                                          char *name)
352 {
353         struct ptlrpc_nrs_policy *tmp;
354
355         cfs_list_for_each_entry(tmp, &nrs->nrs_policy_list, pol_list) {
356                 if (strncmp(tmp->pol_desc->pd_name, name,
357                             NRS_POL_NAME_MAX) == 0) {
358                         nrs_policy_get_locked(tmp);
359                         return tmp;
360                 }
361         }
362         return NULL;
363 }
364
365 /**
366  * Release references for the resource hierarchy moving upwards towards the
367  * policy instance resource.
368  */
369 static void nrs_resource_put(struct ptlrpc_nrs_resource *res)
370 {
371         struct ptlrpc_nrs_policy *policy = res->res_policy;
372
373         if (policy->pol_desc->pd_ops->op_res_put != NULL) {
374                 struct ptlrpc_nrs_resource *parent;
375
376                 for (; res != NULL; res = parent) {
377                         parent = res->res_parent;
378                         policy->pol_desc->pd_ops->op_res_put(policy, res);
379                 }
380         }
381 }
382
383 /**
384  * Obtains references for each resource in the resource hierarchy for request
385  * \a nrq if it is to be handled by \a policy.
386  *
387  * \param[in] policy      the policy
388  * \param[in] nrq         the request
389  * \param[in] moving_req  denotes whether this is a call to the function by
390  *                        ldlm_lock_reorder_req(), in order to move \a nrq to
391  *                        the high-priority NRS head; we should not sleep when
392  *                        set.
393  *
394  * \retval NULL           resource hierarchy references not obtained
395  * \retval valid-pointer  the bottom level of the resource hierarchy
396  *
397  * \see ptlrpc_nrs_pol_ops::op_res_get()
398  */
399 static
400 struct ptlrpc_nrs_resource * nrs_resource_get(struct ptlrpc_nrs_policy *policy,
401                                               struct ptlrpc_nrs_request *nrq,
402                                               bool moving_req)
403 {
404         /**
405          * Set to NULL to traverse the resource hierarchy from the top.
406          */
407         struct ptlrpc_nrs_resource *res = NULL;
408         struct ptlrpc_nrs_resource *tmp = NULL;
409         int                         rc;
410
411         while (1) {
412                 rc = policy->pol_desc->pd_ops->op_res_get(policy, nrq, res,
413                                                           &tmp, moving_req);
414                 if (rc < 0) {
415                         if (res != NULL)
416                                 nrs_resource_put(res);
417                         return NULL;
418                 }
419
420                 LASSERT(tmp != NULL);
421                 tmp->res_parent = res;
422                 tmp->res_policy = policy;
423                 res = tmp;
424                 tmp = NULL;
425                 /**
426                  * Return once we have obtained a reference to the bottom level
427                  * of the resource hierarchy.
428                  */
429                 if (rc > 0)
430                         return res;
431         }
432 }
433
434 /**
435  * Obtains resources for the resource hierarchies and policy references for
436  * the fallback and current primary policy (if any), that will later be used
437  * to handle request \a nrq.
438  *
439  * \param[in]  nrs  the NRS head instance that will be handling request \a nrq.
440  * \param[in]  nrq  the request that is being handled.
441  * \param[out] resp the array where references to the resource hierarchy are
442  *                  stored.
443  * \param[in]  moving_req  is set when obtaining resources while moving a
444  *                         request from a policy on the regular NRS head to a
445  *                         policy on the HP NRS head (via
446  *                         ldlm_lock_reorder_req()). It signifies that
447  *                         allocations to get resources should be atomic; for
448  *                         a full explanation, see comment in
449  *                         ptlrpc_nrs_pol_ops::op_res_get().
450  */
451 static void nrs_resource_get_safe(struct ptlrpc_nrs *nrs,
452                                   struct ptlrpc_nrs_request *nrq,
453                                   struct ptlrpc_nrs_resource **resp,
454                                   bool moving_req)
455 {
456         struct ptlrpc_nrs_policy   *primary = NULL;
457         struct ptlrpc_nrs_policy   *fallback = NULL;
458
459         memset(resp, 0, sizeof(resp[0]) * NRS_RES_MAX);
460
461         /**
462          * Obtain policy references.
463          */
464         spin_lock(&nrs->nrs_lock);
465
466         fallback = nrs->nrs_policy_fallback;
467         nrs_policy_get_locked(fallback);
468
469         primary = nrs->nrs_policy_primary;
470         if (primary != NULL)
471                 nrs_policy_get_locked(primary);
472
473         spin_unlock(&nrs->nrs_lock);
474
475         /**
476          * Obtain resource hierarchy references.
477          */
478         resp[NRS_RES_FALLBACK] = nrs_resource_get(fallback, nrq, moving_req);
479         LASSERT(resp[NRS_RES_FALLBACK] != NULL);
480
481         if (primary != NULL) {
482                 resp[NRS_RES_PRIMARY] = nrs_resource_get(primary, nrq,
483                                                          moving_req);
484                 /**
485                  * A primary policy may exist which may not wish to serve a
486                  * particular request for different reasons; release the
487                  * reference on the policy as it will not be used for this
488                  * request.
489                  */
490                 if (resp[NRS_RES_PRIMARY] == NULL)
491                         nrs_policy_put(primary);
492         }
493 }
494
495 /**
496  * Releases references to resource hierarchies and policies, because they are no
497  * longer required; used when request handling has been completed, or the
498  * request is moving to the high priority NRS head.
499  *
500  * \param resp  the resource hierarchy that is being released
501  *
502  * \see ptlrpcnrs_req_hp_move()
503  * \see ptlrpc_nrs_req_finalize()
504  */
505 static void nrs_resource_put_safe(struct ptlrpc_nrs_resource **resp)
506 {
507         struct ptlrpc_nrs_policy *pols[NRS_RES_MAX];
508         struct ptlrpc_nrs        *nrs = NULL;
509         int                       i;
510
511         for (i = 0; i < NRS_RES_MAX; i++) {
512                 if (resp[i] != NULL) {
513                         pols[i] = resp[i]->res_policy;
514                         nrs_resource_put(resp[i]);
515                         resp[i] = NULL;
516                 } else {
517                         pols[i] = NULL;
518                 }
519         }
520
521         for (i = 0; i < NRS_RES_MAX; i++) {
522                 if (pols[i] == NULL)
523                         continue;
524
525                 if (nrs == NULL) {
526                         nrs = pols[i]->pol_nrs;
527                         spin_lock(&nrs->nrs_lock);
528                 }
529                 nrs_policy_put_locked(pols[i]);
530         }
531
532         if (nrs != NULL)
533                 spin_unlock(&nrs->nrs_lock);
534 }
535
536 /**
537  * Obtains an NRS request from \a policy for handling or examination; the
538  * request should be removed in the 'handling' case.
539  *
540  * Calling into this function implies we already know the policy has a request
541  * waiting to be handled.
542  *
543  * \param[in] policy the policy from which a request
544  * \param[in] peek   when set, signifies that we just want to examine the
545  *                   request, and not handle it, so the request is not removed
546  *                   from the policy.
547  * \param[in] force  when set, it will force a policy to return a request if it
548  *                   has one pending
549  *
550  * \retval the NRS request to be handled
551  */
552 static inline
553 struct ptlrpc_nrs_request * nrs_request_get(struct ptlrpc_nrs_policy *policy,
554                                             bool peek, bool force)
555 {
556         struct ptlrpc_nrs_request *nrq;
557
558         LASSERT(policy->pol_req_queued > 0);
559
560         nrq = policy->pol_desc->pd_ops->op_req_get(policy, peek, force);
561
562         LASSERT(ergo(nrq != NULL, nrs_request_policy(nrq) == policy));
563
564         return nrq;
565 }
566
567 /**
568  * Enqueues request \a nrq for later handling, via one one the policies for
569  * which resources where earlier obtained via nrs_resource_get_safe(). The
570  * function attempts to enqueue the request first on the primary policy
571  * (if any), since this is the preferred choice.
572  *
573  * \param nrq the request being enqueued
574  *
575  * \see nrs_resource_get_safe()
576  */
577 static inline void nrs_request_enqueue(struct ptlrpc_nrs_request *nrq)
578 {
579         struct ptlrpc_nrs_policy *policy;
580         int                       rc;
581         int                       i;
582
583         /**
584          * Try in descending order, because the primary policy (if any) is
585          * the preferred choice.
586          */
587         for (i = NRS_RES_MAX - 1; i >= 0; i--) {
588                 if (nrq->nr_res_ptrs[i] == NULL)
589                         continue;
590
591                 nrq->nr_res_idx = i;
592                 policy = nrq->nr_res_ptrs[i]->res_policy;
593
594                 rc = policy->pol_desc->pd_ops->op_req_enqueue(policy, nrq);
595                 if (rc == 0) {
596                         policy->pol_nrs->nrs_req_queued++;
597                         policy->pol_req_queued++;
598                         return;
599                 }
600         }
601         /**
602          * Should never get here, as at least the primary policy's
603          * ptlrpc_nrs_pol_ops::op_req_enqueue() implementation should always
604          * succeed.
605          */
606         LBUG();
607 }
608
609 /**
610  * Called when a request has been handled
611  *
612  * \param[in] nrs the request that has been handled; can be used for
613  *                job/resource control.
614  *
615  * \see ptlrpc_nrs_req_stop_nolock()
616  */
617 static inline void nrs_request_stop(struct ptlrpc_nrs_request *nrq)
618 {
619         struct ptlrpc_nrs_policy *policy = nrs_request_policy(nrq);
620
621         if (policy->pol_desc->pd_ops->op_req_stop)
622                 policy->pol_desc->pd_ops->op_req_stop(policy, nrq);
623
624         LASSERT(policy->pol_nrs->nrs_req_started > 0);
625         LASSERT(policy->pol_req_started > 0);
626
627         policy->pol_nrs->nrs_req_started--;
628         policy->pol_req_started--;
629 }
630
631 /**
632  * Handler for operations that can be carried out on policies.
633  *
634  * Handles opcodes that are common to all policy types within NRS core, and
635  * passes any unknown opcodes to the policy-specific control function.
636  *
637  * \param[in]     nrs  the NRS head this policy belongs to.
638  * \param[in]     name the human-readable policy name; should be the same as
639  *                     ptlrpc_nrs_pol_desc::pd_name.
640  * \param[in]     opc  the opcode of the operation being carried out.
641  * \param[in,out] arg  can be used to pass information in and out between when
642  *                     carrying an operation; usually data that is private to
643  *                     the policy at some level, or generic policy status
644  *                     information.
645  *
646  * \retval -ve error condition
647  * \retval   0 operation was carried out successfully
648  */
649 static int nrs_policy_ctl(struct ptlrpc_nrs *nrs, char *name,
650                           enum ptlrpc_nrs_ctl opc, void *arg)
651 {
652         struct ptlrpc_nrs_policy       *policy;
653         int                             rc = 0;
654         ENTRY;
655
656         spin_lock(&nrs->nrs_lock);
657
658         policy = nrs_policy_find_locked(nrs, name);
659         if (policy == NULL)
660                 GOTO(out, rc = -ENOENT);
661
662         switch (opc) {
663                 /**
664                  * Unknown opcode, pass it down to the policy-specific control
665                  * function for handling.
666                  */
667         default:
668                 rc = nrs_policy_ctl_locked(policy, opc, arg);
669                 break;
670
671                 /**
672                  * Start \e policy
673                  */
674         case PTLRPC_NRS_CTL_START:
675                 rc = nrs_policy_start_locked(policy);
676                 break;
677         }
678 out:
679         if (policy != NULL)
680                 nrs_policy_put_locked(policy);
681
682         spin_unlock(&nrs->nrs_lock);
683
684         RETURN(rc);
685 }
686
687 /**
688  * Unregisters a policy by name.
689  *
690  * \param[in] nrs  the NRS head this policy belongs to.
691  * \param[in] name the human-readable policy name; should be the same as
692  *                 ptlrpc_nrs_pol_desc::pd_name
693  *
694  * \retval -ve error
695  * \retval   0 success
696  */
697 static int nrs_policy_unregister(struct ptlrpc_nrs *nrs, char *name)
698 {
699         struct ptlrpc_nrs_policy *policy = NULL;
700         ENTRY;
701
702         spin_lock(&nrs->nrs_lock);
703
704         policy = nrs_policy_find_locked(nrs, name);
705         if (policy == NULL) {
706                 spin_unlock(&nrs->nrs_lock);
707
708                 CERROR("Can't find NRS policy %s\n", name);
709                 RETURN(-ENOENT);
710         }
711
712         if (policy->pol_ref > 1) {
713                 CERROR("Policy %s is busy with %d references\n", name,
714                        (int)policy->pol_ref);
715                 nrs_policy_put_locked(policy);
716
717                 spin_unlock(&nrs->nrs_lock);
718                 RETURN(-EBUSY);
719         }
720
721         LASSERT(policy->pol_req_queued == 0);
722         LASSERT(policy->pol_req_started == 0);
723
724         if (policy->pol_state != NRS_POL_STATE_STOPPED) {
725                 nrs_policy_stop_locked(policy);
726                 LASSERT(policy->pol_state == NRS_POL_STATE_STOPPED);
727         }
728
729         cfs_list_del(&policy->pol_list);
730         nrs->nrs_num_pols--;
731
732         nrs_policy_put_locked(policy);
733
734         spin_unlock(&nrs->nrs_lock);
735
736         nrs_policy_fini(policy);
737
738         LASSERT(policy->pol_private == NULL);
739         OBD_FREE_PTR(policy);
740
741         RETURN(0);
742 }
743
744 /**
745  * Register a policy from \policy descriptor \a desc with NRS head \a nrs.
746  *
747  * \param[in] nrs   the NRS head on which the policy will be registered.
748  * \param[in] desc  the policy descriptor from which the information will be
749  *                  obtained to register the policy.
750  *
751  * \retval -ve error
752  * \retval   0 success
753  */
754 static int nrs_policy_register(struct ptlrpc_nrs *nrs,
755                                struct ptlrpc_nrs_pol_desc *desc)
756 {
757         struct ptlrpc_nrs_policy       *policy;
758         struct ptlrpc_nrs_policy       *tmp;
759         struct ptlrpc_service_part     *svcpt = nrs->nrs_svcpt;
760         int                             rc;
761         ENTRY;
762
763         LASSERT(svcpt != NULL);
764         LASSERT(desc->pd_ops != NULL);
765         LASSERT(desc->pd_ops->op_res_get != NULL);
766         LASSERT(desc->pd_ops->op_req_get != NULL);
767         LASSERT(desc->pd_ops->op_req_enqueue != NULL);
768         LASSERT(desc->pd_ops->op_req_dequeue != NULL);
769         LASSERT(desc->pd_compat != NULL);
770
771         OBD_CPT_ALLOC_GFP(policy, svcpt->scp_service->srv_cptable,
772                           svcpt->scp_cpt, sizeof(*policy), CFS_ALLOC_IO);
773         if (policy == NULL)
774                 RETURN(-ENOMEM);
775
776         policy->pol_nrs     = nrs;
777         policy->pol_desc    = desc;
778         policy->pol_state   = NRS_POL_STATE_STOPPED;
779         policy->pol_flags   = desc->pd_flags;
780
781         CFS_INIT_LIST_HEAD(&policy->pol_list);
782         CFS_INIT_LIST_HEAD(&policy->pol_list_queued);
783
784         rc = nrs_policy_init(policy);
785         if (rc != 0) {
786                 OBD_FREE_PTR(policy);
787                 RETURN(rc);
788         }
789
790         spin_lock(&nrs->nrs_lock);
791
792         tmp = nrs_policy_find_locked(nrs, policy->pol_desc->pd_name);
793         if (tmp != NULL) {
794                 CERROR("NRS policy %s has been registered, can't register it "
795                        "for %s\n", policy->pol_desc->pd_name,
796                        svcpt->scp_service->srv_name);
797                 nrs_policy_put_locked(tmp);
798
799                 spin_unlock(&nrs->nrs_lock);
800                 nrs_policy_fini(policy);
801                 OBD_FREE_PTR(policy);
802
803                 RETURN(-EEXIST);
804         }
805
806         cfs_list_add_tail(&policy->pol_list, &nrs->nrs_policy_list);
807         nrs->nrs_num_pols++;
808
809         if (policy->pol_flags & PTLRPC_NRS_FL_REG_START)
810                 rc = nrs_policy_start_locked(policy);
811
812         spin_unlock(&nrs->nrs_lock);
813
814         if (rc != 0)
815                 (void) nrs_policy_unregister(nrs, policy->pol_desc->pd_name);
816
817         RETURN(rc);
818 }
819
820 /**
821  * Enqueue request \a req using one of the policies its resources are referring
822  * to.
823  *
824  * \param[in] req the request to enqueue.
825  */
826 static void ptlrpc_nrs_req_add_nolock(struct ptlrpc_request *req)
827 {
828         struct ptlrpc_nrs_policy       *policy;
829
830         LASSERT(req->rq_nrq.nr_initialized);
831         LASSERT(!req->rq_nrq.nr_enqueued);
832
833         nrs_request_enqueue(&req->rq_nrq);
834         req->rq_nrq.nr_enqueued = 1;
835
836         policy = nrs_request_policy(&req->rq_nrq);
837         /**
838          * Add the policy to the NRS head's list of policies with enqueued
839          * requests, if it has not been added there.
840          */
841         if (unlikely(cfs_list_empty(&policy->pol_list_queued)))
842                 cfs_list_add_tail(&policy->pol_list_queued,
843                                   &policy->pol_nrs->nrs_policy_queued);
844 }
845
846 /**
847  * Enqueue a request on the high priority NRS head.
848  *
849  * \param req the request to enqueue.
850  */
851 static void ptlrpc_nrs_hpreq_add_nolock(struct ptlrpc_request *req)
852 {
853         int     opc = lustre_msg_get_opc(req->rq_reqmsg);
854         ENTRY;
855
856         spin_lock(&req->rq_lock);
857         req->rq_hp = 1;
858         ptlrpc_nrs_req_add_nolock(req);
859         if (opc != OBD_PING)
860                 DEBUG_REQ(D_NET, req, "high priority req");
861         spin_unlock(&req->rq_lock);
862         EXIT;
863 }
864
865 /**
866  * Returns a boolean predicate indicating whether the policy described by
867  * \a desc is adequate for use with service \a svc.
868  *
869  * \param[in] svc  the service
870  * \param[in] desc the policy descriptor
871  *
872  * \retval false the policy is not compatible with the service
873  * \retval true  the policy is compatible with the service
874  */
875 static inline bool nrs_policy_compatible(const struct ptlrpc_service *svc,
876                                          const struct ptlrpc_nrs_pol_desc *desc)
877 {
878         return desc->pd_compat(svc, desc);
879 }
880
881 /**
882  * Registers all compatible policies in nrs_core.nrs_policies, for NRS head
883  * \a nrs.
884  *
885  * \param[in] nrs the NRS head
886  *
887  * \retval -ve error
888  * \retval   0 success
889  *
890  * \pre mutex_is_locked(&nrs_core.nrs_mutex)
891  *
892  * \see ptlrpc_service_nrs_setup()
893  */
894 static int nrs_register_policies_locked(struct ptlrpc_nrs *nrs)
895 {
896         struct ptlrpc_nrs_pol_desc *desc;
897         /* for convenience */
898         struct ptlrpc_service_part       *svcpt = nrs->nrs_svcpt;
899         struct ptlrpc_service            *svc = svcpt->scp_service;
900         int                               rc = -EINVAL;
901         ENTRY;
902
903         LASSERT(mutex_is_locked(&nrs_core.nrs_mutex));
904
905         cfs_list_for_each_entry(desc, &nrs_core.nrs_policies, pd_list) {
906                 if (nrs_policy_compatible(svc, desc)) {
907                         rc = nrs_policy_register(nrs, desc);
908                         if (rc != 0) {
909                                 CERROR("Failed to register NRS policy %s for "
910                                        "partition %d of service %s: %d\n",
911                                        desc->pd_name, svcpt->scp_cpt,
912                                        svc->srv_name, rc);
913                                 /**
914                                  * Fail registration if any of the policies'
915                                  * registration fails.
916                                  */
917                                 break;
918                         }
919                 }
920         }
921
922         RETURN(rc);
923 }
924
925 /**
926  * Initializes NRS head \a nrs of service partition \a svcpt, and registers all
927  * compatible policies in NRS core, with the NRS head.
928  *
929  * \param[in] nrs   the NRS head
930  * \param[in] svcpt the PTLRPC service partition to setup
931  *
932  * \retval -ve error
933  * \retval   0 success
934  *
935  * \pre mutex_is_locked(&nrs_core.nrs_mutex)
936  */
937 static int nrs_svcpt_setup_locked0(struct ptlrpc_nrs *nrs,
938                                    struct ptlrpc_service_part *svcpt)
939 {
940         int                             rc;
941         enum ptlrpc_nrs_queue_type      queue;
942
943         LASSERT(mutex_is_locked(&nrs_core.nrs_mutex));
944
945         if (nrs == &svcpt->scp_nrs_reg)
946                 queue = PTLRPC_NRS_QUEUE_REG;
947         else if (nrs == svcpt->scp_nrs_hp)
948                 queue = PTLRPC_NRS_QUEUE_HP;
949         else
950                 LBUG();
951
952         nrs->nrs_svcpt = svcpt;
953         nrs->nrs_queue_type = queue;
954         spin_lock_init(&nrs->nrs_lock);
955         CFS_INIT_LIST_HEAD(&nrs->nrs_policy_list);
956         CFS_INIT_LIST_HEAD(&nrs->nrs_policy_queued);
957
958         rc = nrs_register_policies_locked(nrs);
959
960         RETURN(rc);
961 }
962
963 /**
964  * Allocates a regular and optionally a high-priority NRS head (if the service
965  * handles high-priority RPCs), and then registers all available compatible
966  * policies on those NRS heads.
967  *
968  * \param[in,out] svcpt the PTLRPC service partition to setup
969  *
970  * \pre mutex_is_locked(&nrs_core.nrs_mutex)
971  */
972 static int nrs_svcpt_setup_locked(struct ptlrpc_service_part *svcpt)
973 {
974         struct ptlrpc_nrs              *nrs;
975         int                             rc;
976         ENTRY;
977
978         LASSERT(mutex_is_locked(&nrs_core.nrs_mutex));
979
980         /**
981          * Initialize the regular NRS head.
982          */
983         nrs = nrs_svcpt2nrs(svcpt, false);
984         rc = nrs_svcpt_setup_locked0(nrs, svcpt);
985         if (rc < 0)
986                 GOTO(out, rc);
987
988         /**
989          * Optionally allocate a high-priority NRS head.
990          */
991         if (svcpt->scp_service->srv_ops.so_hpreq_handler == NULL)
992                 GOTO(out, rc);
993
994         OBD_CPT_ALLOC_PTR(svcpt->scp_nrs_hp,
995                           svcpt->scp_service->srv_cptable,
996                           svcpt->scp_cpt);
997         if (svcpt->scp_nrs_hp == NULL)
998                 GOTO(out, rc = -ENOMEM);
999
1000         nrs = nrs_svcpt2nrs(svcpt, true);
1001         rc = nrs_svcpt_setup_locked0(nrs, svcpt);
1002
1003 out:
1004         RETURN(rc);
1005 }
1006
1007 /**
1008  * Unregisters all policies on all available NRS heads in a service partition;
1009  * called at PTLRPC service unregistration time.
1010  *
1011  * \param[in] svcpt the PTLRPC service partition
1012  *
1013  * \pre mutex_is_locked(&nrs_core.nrs_mutex)
1014  */
1015 static void nrs_svcpt_cleanup_locked(struct ptlrpc_service_part *svcpt)
1016 {
1017         struct ptlrpc_nrs              *nrs;
1018         struct ptlrpc_nrs_policy       *policy;
1019         struct ptlrpc_nrs_policy       *tmp;
1020         int                             rc;
1021         bool                            hp = false;
1022         ENTRY;
1023
1024         LASSERT(mutex_is_locked(&nrs_core.nrs_mutex));
1025
1026 again:
1027         nrs = nrs_svcpt2nrs(svcpt, hp);
1028         nrs->nrs_stopping = 1;
1029
1030         cfs_list_for_each_entry_safe(policy, tmp, &nrs->nrs_policy_list,
1031                                      pol_list) {
1032                 rc = nrs_policy_unregister(nrs, policy->pol_desc->pd_name);
1033                 LASSERT(rc == 0);
1034         }
1035
1036         /**
1037          * If the service partition has an HP NRS head, clean that up as well.
1038          */
1039         if (!hp && nrs_svcpt_has_hp(svcpt)) {
1040                 hp = true;
1041                 goto again;
1042         }
1043
1044         if (hp)
1045                 OBD_FREE_PTR(nrs);
1046
1047         EXIT;
1048 }
1049
1050 /**
1051  * Returns the descriptor for a policy as identified by by \a name.
1052  *
1053  * \param[in] name the policy name
1054  *
1055  * \retval the policy descriptor
1056  * \retval NULL
1057  */
1058 static struct ptlrpc_nrs_pol_desc *nrs_policy_find_desc_locked(const char *name)
1059 {
1060         struct ptlrpc_nrs_pol_desc     *tmp;
1061         ENTRY;
1062
1063         cfs_list_for_each_entry(tmp, &nrs_core.nrs_policies, pd_list) {
1064                 if (strncmp(tmp->pd_name, name, NRS_POL_NAME_MAX) == 0)
1065                         RETURN(tmp);
1066         }
1067         RETURN(NULL);
1068 }
1069
1070 /**
1071  * Removes the policy from all supported NRS heads of all partitions of all
1072  * PTLRPC services.
1073  *
1074  * \param[in] desc the policy descriptor to unregister
1075  *
1076  * \retval -ve error
1077  * \retval  0  successfully unregistered policy on all supported NRS heads
1078  *
1079  * \pre mutex_is_locked(&nrs_core.nrs_mutex)
1080  * \pre mutex_is_locked(&ptlrpc_all_services_mutex)
1081  */
1082 static int nrs_policy_unregister_locked(struct ptlrpc_nrs_pol_desc *desc)
1083 {
1084         struct ptlrpc_nrs              *nrs;
1085         struct ptlrpc_service          *svc;
1086         struct ptlrpc_service_part     *svcpt;
1087         int                             i;
1088         int                             rc = 0;
1089         ENTRY;
1090
1091         LASSERT(mutex_is_locked(&nrs_core.nrs_mutex));
1092         LASSERT(mutex_is_locked(&ptlrpc_all_services_mutex));
1093
1094         cfs_list_for_each_entry(svc, &ptlrpc_all_services, srv_list) {
1095
1096                 if (!nrs_policy_compatible(svc, desc) ||
1097                     unlikely(svc->srv_is_stopping))
1098                         continue;
1099
1100                 ptlrpc_service_for_each_part(svcpt, i, svc) {
1101                         bool hp = false;
1102
1103 again:
1104                         nrs = nrs_svcpt2nrs(svcpt, hp);
1105                         rc = nrs_policy_unregister(nrs, desc->pd_name);
1106                         /**
1107                          * Ignore -ENOENT as the policy may not have registered
1108                          * successfully on all service partitions.
1109                          */
1110                         if (rc == -ENOENT) {
1111                                 rc = 0;
1112                         } else if (rc != 0) {
1113                                 CERROR("Failed to unregister NRS policy %s for "
1114                                        "partition %d of service %s: %d\n",
1115                                        desc->pd_name, svcpt->scp_cpt,
1116                                        svcpt->scp_service->srv_name, rc);
1117                                 RETURN(rc);
1118                         }
1119
1120                         if (!hp && nrs_svc_has_hp(svc)) {
1121                                 hp = true;
1122                                 goto again;
1123                         }
1124                 }
1125
1126                 if (desc->pd_ops->op_lprocfs_fini != NULL)
1127                         desc->pd_ops->op_lprocfs_fini(svc);
1128         }
1129
1130         RETURN(rc);
1131 }
1132
1133 /**
1134  * Registers a new policy with NRS core.
1135  *
1136  * The function will only succeed if policy registration with all compatible
1137  * service partitions (if any) is successful.
1138  *
1139  * N.B. This function should be called either at ptlrpc module initialization
1140  *      time when registering a policy that ships with NRS core, or in a
1141  *      module's init() function for policies registering from other modules.
1142  *
1143  * \param[in] conf configuration information for the new policy to register
1144  *
1145  * \retval -ve error
1146  * \retval   0 success
1147  */
1148 int ptlrpc_nrs_policy_register(struct ptlrpc_nrs_pol_conf *conf)
1149 {
1150         struct ptlrpc_service          *svc;
1151         struct ptlrpc_nrs_pol_desc     *desc;
1152         int                             rc = 0;
1153         ENTRY;
1154
1155         LASSERT(conf != NULL);
1156         LASSERT(conf->nc_ops != NULL);
1157         LASSERT(conf->nc_compat != NULL);
1158         LASSERT(ergo(conf->nc_compat == nrs_policy_compat_one,
1159                 conf->nc_compat_svc_name != NULL));
1160         LASSERT(ergo((conf->nc_flags & PTLRPC_NRS_FL_REG_EXTERN) != 0,
1161                      conf->nc_owner != NULL));
1162
1163         conf->nc_name[NRS_POL_NAME_MAX - 1] = '\0';
1164
1165         /**
1166          * External policies are not allowed to start immediately upon
1167          * registration, as there is a relatively higher chance that their
1168          * registration might fail. In such a case, some policy instances may
1169          * already have requests queued wen unregistration needs to happen as
1170          * part o cleanup; since there is currently no way to drain requests
1171          * from a policy unless the service is unregistering, we just disallow
1172          * this.
1173          */
1174         if ((conf->nc_flags & PTLRPC_NRS_FL_REG_EXTERN) &&
1175             (conf->nc_flags & (PTLRPC_NRS_FL_FALLBACK |
1176                                PTLRPC_NRS_FL_REG_START))) {
1177                 CERROR("NRS: failing to register policy %s. Please check "
1178                        "policy flags; external policies cannot act as fallback "
1179                        "policies, or be started immediately upon registration "
1180                        "without interaction with lprocfs\n", conf->nc_name);
1181                 RETURN(-EINVAL);
1182         }
1183
1184         mutex_lock(&nrs_core.nrs_mutex);
1185
1186         if (nrs_policy_find_desc_locked(conf->nc_name) != NULL) {
1187                 CERROR("NRS: failing to register policy %s which has already "
1188                        "been registered with NRS core!\n",
1189                        conf->nc_name);
1190                 GOTO(fail, rc = -EEXIST);
1191         }
1192
1193         OBD_ALLOC_PTR(desc);
1194         if (desc == NULL)
1195                 GOTO(fail, rc = -ENOMEM);
1196
1197         strncpy(desc->pd_name, conf->nc_name, NRS_POL_NAME_MAX);
1198         desc->pd_ops             = conf->nc_ops;
1199         desc->pd_compat          = conf->nc_compat;
1200         desc->pd_compat_svc_name = conf->nc_compat_svc_name;
1201         if ((conf->nc_flags & PTLRPC_NRS_FL_REG_EXTERN) != 0)
1202                 desc->pd_owner   = conf->nc_owner;
1203         desc->pd_flags           = conf->nc_flags;
1204         cfs_atomic_set(&desc->pd_refs, 0);
1205
1206         /**
1207          * For policies that are held in the same module as NRS (currently
1208          * ptlrpc), do not register the policy with all compatible services,
1209          * as the services will not have started at this point, since we are
1210          * calling from ptlrpc module initialization code. In such cases each
1211          * service will register all compatible policies later, via
1212          * ptlrpc_service_nrs_setup().
1213          */
1214         if ((conf->nc_flags & PTLRPC_NRS_FL_REG_EXTERN) == 0)
1215                 goto internal;
1216
1217         /**
1218          * Register the new policy on all compatible services
1219          */
1220         mutex_lock(&ptlrpc_all_services_mutex);
1221
1222         cfs_list_for_each_entry(svc, &ptlrpc_all_services, srv_list) {
1223                 struct ptlrpc_service_part     *svcpt;
1224                 int                             i;
1225                 int                             rc2;
1226
1227                 if (!nrs_policy_compatible(svc, desc) ||
1228                     unlikely(svc->srv_is_stopping))
1229                         continue;
1230
1231                 ptlrpc_service_for_each_part(svcpt, i, svc) {
1232                         struct ptlrpc_nrs      *nrs;
1233                         bool                    hp = false;
1234 again:
1235                         nrs = nrs_svcpt2nrs(svcpt, hp);
1236                         rc = nrs_policy_register(nrs, desc);
1237                         if (rc != 0) {
1238                                 CERROR("Failed to register NRS policy %s for "
1239                                        "partition %d of service %s: %d\n",
1240                                        desc->pd_name, svcpt->scp_cpt,
1241                                        svcpt->scp_service->srv_name, rc);
1242
1243                                 rc2 = nrs_policy_unregister_locked(desc);
1244                                 /**
1245                                  * Should not fail at this point
1246                                  */
1247                                 LASSERT(rc2 == 0);
1248                                 mutex_unlock(&ptlrpc_all_services_mutex);
1249                                 OBD_FREE_PTR(desc);
1250                                 GOTO(fail, rc);
1251                         }
1252
1253                         if (!hp && nrs_svc_has_hp(svc)) {
1254                                 hp = true;
1255                                 goto again;
1256                         }
1257                 }
1258
1259                 /**
1260                  * No need to take a reference to other modules here, as we
1261                  * will be calling from the module's init() function.
1262                  */
1263                 if (desc->pd_ops->op_lprocfs_init != NULL) {
1264                         rc = desc->pd_ops->op_lprocfs_init(svc);
1265                         if (rc != 0) {
1266                                 rc2 = nrs_policy_unregister_locked(desc);
1267                                 /**
1268                                  * Should not fail at this point
1269                                  */
1270                                 LASSERT(rc2 == 0);
1271                                 mutex_unlock(&ptlrpc_all_services_mutex);
1272                                 OBD_FREE_PTR(desc);
1273                                 GOTO(fail, rc);
1274                         }
1275                 }
1276         }
1277
1278         mutex_unlock(&ptlrpc_all_services_mutex);
1279 internal:
1280         cfs_list_add_tail(&desc->pd_list, &nrs_core.nrs_policies);
1281 fail:
1282         mutex_unlock(&nrs_core.nrs_mutex);
1283
1284         RETURN(rc);
1285 }
1286 EXPORT_SYMBOL(ptlrpc_nrs_policy_register);
1287
1288 /**
1289  * Unregisters a previously registered policy with NRS core. All instances of
1290  * the policy on all NRS heads of all supported services are removed.
1291  *
1292  * N.B. This function should only be called from a module's exit() function.
1293  *      Although it can be used for policies that ship alongside NRS core, the
1294  *      function is primarily intended for policies that register externally,
1295  *      from other modules.
1296  *
1297  * \param[in] conf configuration information for the policy to unregister
1298  *
1299  * \retval -ve error
1300  * \retval   0 success
1301  */
1302 int ptlrpc_nrs_policy_unregister(struct ptlrpc_nrs_pol_conf *conf)
1303 {
1304         struct ptlrpc_nrs_pol_desc      *desc;
1305         int                              rc;
1306         ENTRY;
1307
1308         LASSERT(conf != NULL);
1309
1310         if (conf->nc_flags & PTLRPC_NRS_FL_FALLBACK) {
1311                 CERROR("Unable to unregister a fallback policy, unless the "
1312                        "PTLRPC service is stopping.\n");
1313                 RETURN(-EPERM);
1314         }
1315
1316         conf->nc_name[NRS_POL_NAME_MAX - 1] = '\0';
1317
1318         mutex_lock(&nrs_core.nrs_mutex);
1319
1320         desc = nrs_policy_find_desc_locked(conf->nc_name);
1321         if (desc == NULL) {
1322                 CERROR("Failing to unregister NRS policy %s which has "
1323                        "not been registered with NRS core!\n",
1324                        conf->nc_name);
1325                 GOTO(not_exist, rc = -ENOENT);
1326         }
1327
1328         mutex_lock(&ptlrpc_all_services_mutex);
1329
1330         rc = nrs_policy_unregister_locked(desc);
1331         if (rc < 0) {
1332                 if (rc == -EBUSY)
1333                         CERROR("Please first stop policy %s on all service "
1334                                "partitions and then retry to unregister the "
1335                                "policy.\n", conf->nc_name);
1336                 GOTO(fail, rc);
1337         }
1338
1339         CDEBUG(D_INFO, "Unregistering policy %s from NRS core.\n",
1340                conf->nc_name);
1341
1342         cfs_list_del(&desc->pd_list);
1343         OBD_FREE_PTR(desc);
1344
1345 fail:
1346         mutex_unlock(&ptlrpc_all_services_mutex);
1347
1348 not_exist:
1349         mutex_unlock(&nrs_core.nrs_mutex);
1350
1351         RETURN(rc);
1352 }
1353 EXPORT_SYMBOL(ptlrpc_nrs_policy_unregister);
1354
1355 /**
1356  * Setup NRS heads on all service partitions of service \a svc, and register
1357  * all compatible policies on those NRS heads.
1358  *
1359  * To be called from withing ptl
1360  * \param[in] svc the service to setup
1361  *
1362  * \retval -ve error, the calling logic should eventually call
1363  *                    ptlrpc_service_nrs_cleanup() to undo any work performed
1364  *                    by this function.
1365  *
1366  * \see ptlrpc_register_service()
1367  * \see ptlrpc_service_nrs_cleanup()
1368  */
1369 int ptlrpc_service_nrs_setup(struct ptlrpc_service *svc)
1370 {
1371         struct ptlrpc_service_part             *svcpt;
1372         const struct ptlrpc_nrs_pol_desc       *desc;
1373         int                                     i;
1374         int                                     rc = 0;
1375
1376         mutex_lock(&nrs_core.nrs_mutex);
1377
1378         /**
1379          * Initialize NRS heads on all service CPTs.
1380          */
1381         ptlrpc_service_for_each_part(svcpt, i, svc) {
1382                 rc = nrs_svcpt_setup_locked(svcpt);
1383                 if (rc != 0)
1384                         GOTO(failed, rc);
1385         }
1386
1387         /**
1388          * Set up lprocfs interfaces for all supported policies for the
1389          * service.
1390          */
1391         cfs_list_for_each_entry(desc, &nrs_core.nrs_policies, pd_list) {
1392                 if (!nrs_policy_compatible(svc, desc))
1393                         continue;
1394
1395                 if (desc->pd_ops->op_lprocfs_init != NULL) {
1396                         rc = desc->pd_ops->op_lprocfs_init(svc);
1397                         if (rc != 0)
1398                                 GOTO(failed, rc);
1399                 }
1400         }
1401
1402 failed:
1403
1404         mutex_unlock(&nrs_core.nrs_mutex);
1405
1406         RETURN(rc);
1407 }
1408
1409 /**
1410  * Unregisters all policies on all service partitions of service \a svc.
1411  *
1412  * \param[in] svc the PTLRPC service to unregister
1413  */
1414 void ptlrpc_service_nrs_cleanup(struct ptlrpc_service *svc)
1415 {
1416         struct ptlrpc_service_part           *svcpt;
1417         const struct ptlrpc_nrs_pol_desc     *desc;
1418         int                                   i;
1419
1420         mutex_lock(&nrs_core.nrs_mutex);
1421
1422         /**
1423          * Clean up NRS heads on all service partitions
1424          */
1425         ptlrpc_service_for_each_part(svcpt, i, svc)
1426                 nrs_svcpt_cleanup_locked(svcpt);
1427
1428         /**
1429          * Clean up lprocfs interfaces for all supported policies for the
1430          * service.
1431          */
1432         cfs_list_for_each_entry(desc, &nrs_core.nrs_policies, pd_list) {
1433                 if (!nrs_policy_compatible(svc, desc))
1434                         continue;
1435
1436                 if (desc->pd_ops->op_lprocfs_fini != NULL)
1437                         desc->pd_ops->op_lprocfs_fini(svc);
1438         }
1439
1440         mutex_unlock(&nrs_core.nrs_mutex);
1441 }
1442
1443 /**
1444  * Obtains NRS head resources for request \a req.
1445  *
1446  * These could be either on the regular or HP NRS head of \a svcpt; resources
1447  * taken on the regular head can later be swapped for HP head resources by
1448  * ldlm_lock_reorder_req().
1449  *
1450  * \param[in] svcpt the service partition
1451  * \param[in] req   the request
1452  * \param[in] hp    which NRS head of \a svcpt to use
1453  */
1454 void ptlrpc_nrs_req_initialize(struct ptlrpc_service_part *svcpt,
1455                                struct ptlrpc_request *req, bool hp)
1456 {
1457         struct ptlrpc_nrs       *nrs = nrs_svcpt2nrs(svcpt, hp);
1458
1459         memset(&req->rq_nrq, 0, sizeof(req->rq_nrq));
1460         nrs_resource_get_safe(nrs, &req->rq_nrq, req->rq_nrq.nr_res_ptrs,
1461                               false);
1462
1463         /**
1464          * It is fine to access \e nr_initialized without locking as there is
1465          * no contention at this early stage.
1466          */
1467         req->rq_nrq.nr_initialized = 1;
1468 }
1469
1470 /**
1471  * Releases resources for a request; is called after the request has been
1472  * handled.
1473  *
1474  * \param[in] req the request
1475  *
1476  * \see ptlrpc_server_finish_request()
1477  */
1478 void ptlrpc_nrs_req_finalize(struct ptlrpc_request *req)
1479 {
1480         if (req->rq_nrq.nr_initialized) {
1481                 nrs_resource_put_safe(req->rq_nrq.nr_res_ptrs);
1482                 /* no protection on bit nr_initialized because no
1483                  * contention at this late stage */
1484                 req->rq_nrq.nr_finalized = 1;
1485         }
1486 }
1487
1488 void ptlrpc_nrs_req_stop_nolock(struct ptlrpc_request *req)
1489 {
1490         if (req->rq_nrq.nr_started)
1491                 nrs_request_stop(&req->rq_nrq);
1492 }
1493
1494 /**
1495  * Enqueues request \a req on either the regular or high-priority NRS head
1496  * of service partition \a svcpt.
1497  *
1498  * \param[in] svcpt the service partition
1499  * \param[in] req   the request to be enqueued
1500  * \param[in] hp    whether to enqueue the request on the regular or
1501  *                  high-priority NRS head.
1502  */
1503 void ptlrpc_nrs_req_add(struct ptlrpc_service_part *svcpt,
1504                         struct ptlrpc_request *req, bool hp)
1505 {
1506         spin_lock(&svcpt->scp_req_lock);
1507
1508         if (hp)
1509                 ptlrpc_nrs_hpreq_add_nolock(req);
1510         else
1511                 ptlrpc_nrs_req_add_nolock(req);
1512
1513         spin_unlock(&svcpt->scp_req_lock);
1514 }
1515
1516 static void nrs_request_removed(struct ptlrpc_nrs_policy *policy)
1517 {
1518         LASSERT(policy->pol_nrs->nrs_req_queued > 0);
1519         LASSERT(policy->pol_req_queued > 0);
1520
1521         policy->pol_nrs->nrs_req_queued--;
1522         policy->pol_req_queued--;
1523
1524         /**
1525          * If the policy has no more requests queued, remove it from
1526          * ptlrpc_nrs::nrs_policy_queued.
1527          */
1528         if (unlikely(policy->pol_req_queued == 0)) {
1529                 cfs_list_del_init(&policy->pol_list_queued);
1530
1531                 /**
1532                  * If there are other policies with queued requests, move the
1533                  * current policy to the end so that we can round robin over
1534                  * all policies and drain the requests.
1535                  */
1536         } else if (policy->pol_req_queued != policy->pol_nrs->nrs_req_queued) {
1537                 LASSERT(policy->pol_req_queued <
1538                         policy->pol_nrs->nrs_req_queued);
1539
1540                 cfs_list_move_tail(&policy->pol_list_queued,
1541                                    &policy->pol_nrs->nrs_policy_queued);
1542         }
1543 }
1544
1545 /**
1546  * Obtains a request for handling from an NRS head of service partition
1547  * \a svcpt.
1548  *
1549  * \param[in] svcpt the service partition
1550  * \param[in] hp    whether to obtain a request from the regular or
1551  *                  high-priority NRS head.
1552  * \param[in] peek  when set, signifies that we just want to examine the
1553  *                  request, and not handle it, so the request is not removed
1554  *                  from the policy.
1555  * \param[in] force when set, it will force a policy to return a request if it
1556  *                  has one pending
1557  *
1558  * \retval the  request to be handled
1559  * \retval NULL the head has no requests to serve
1560  */
1561 struct ptlrpc_request *
1562 ptlrpc_nrs_req_get_nolock0(struct ptlrpc_service_part *svcpt, bool hp,
1563                            bool peek, bool force)
1564 {
1565         struct ptlrpc_nrs         *nrs = nrs_svcpt2nrs(svcpt, hp);
1566         struct ptlrpc_nrs_policy  *policy;
1567         struct ptlrpc_nrs_request *nrq;
1568
1569         /**
1570          * Always try to drain requests from all NRS polices even if they are
1571          * inactive, because the user can change policy status at runtime.
1572          */
1573         cfs_list_for_each_entry(policy, &nrs->nrs_policy_queued,
1574                                 pol_list_queued) {
1575                 nrq = nrs_request_get(policy, peek, force);
1576                 if (nrq != NULL) {
1577                         if (likely(!peek)) {
1578                                 nrq->nr_started = 1;
1579
1580                                 policy->pol_req_started++;
1581                                 policy->pol_nrs->nrs_req_started++;
1582
1583                                 nrs_request_removed(policy);
1584                         }
1585
1586                         return container_of(nrq, struct ptlrpc_request, rq_nrq);
1587                 }
1588         }
1589
1590         return NULL;
1591 }
1592
1593 /**
1594  * Dequeues request \a req from the policy it has been enqueued on.
1595  *
1596  * \param[in] req the request
1597  */
1598 void ptlrpc_nrs_req_del_nolock(struct ptlrpc_request *req)
1599 {
1600         struct ptlrpc_nrs_policy *policy = nrs_request_policy(&req->rq_nrq);
1601
1602         policy->pol_desc->pd_ops->op_req_dequeue(policy, &req->rq_nrq);
1603
1604         req->rq_nrq.nr_enqueued = 0;
1605
1606         nrs_request_removed(policy);
1607 }
1608
1609 /**
1610  * Returns whether there are any requests currently enqueued on any of the
1611  * policies of service partition's \a svcpt NRS head specified by \a hp. Should
1612  * be called while holding ptlrpc_service_part::scp_req_lock to get a reliable
1613  * result.
1614  *
1615  * \param[in] svcpt the service partition to enquire.
1616  * \param[in] hp    whether the regular or high-priority NRS head is to be
1617  *                  enquired.
1618  *
1619  * \retval false the indicated NRS head has no enqueued requests.
1620  * \retval true  the indicated NRS head has some enqueued requests.
1621  */
1622 bool ptlrpc_nrs_req_pending_nolock(struct ptlrpc_service_part *svcpt, bool hp)
1623 {
1624         struct ptlrpc_nrs *nrs = nrs_svcpt2nrs(svcpt, hp);
1625
1626         return nrs->nrs_req_queued > 0;
1627 };
1628
1629 /**
1630  * Moves request \a req from the regular to the high-priority NRS head.
1631  *
1632  * \param[in] req the request to move
1633  */
1634 void ptlrpc_nrs_req_hp_move(struct ptlrpc_request *req)
1635 {
1636         struct ptlrpc_service_part      *svcpt = req->rq_rqbd->rqbd_svcpt;
1637         struct ptlrpc_nrs_request       *nrq = &req->rq_nrq;
1638         struct ptlrpc_nrs_resource      *res1[NRS_RES_MAX];
1639         struct ptlrpc_nrs_resource      *res2[NRS_RES_MAX];
1640         ENTRY;
1641
1642         /**
1643          * Obtain the high-priority NRS head resources.
1644          */
1645         nrs_resource_get_safe(nrs_svcpt2nrs(svcpt, true), nrq, res1, true);
1646
1647         spin_lock(&svcpt->scp_req_lock);
1648
1649         if (!ptlrpc_nrs_req_can_move(req))
1650                 goto out;
1651
1652         ptlrpc_nrs_req_del_nolock(req);
1653
1654         memcpy(res2, nrq->nr_res_ptrs, NRS_RES_MAX * sizeof(res2[0]));
1655         memcpy(nrq->nr_res_ptrs, res1, NRS_RES_MAX * sizeof(res1[0]));
1656
1657         ptlrpc_nrs_hpreq_add_nolock(req);
1658
1659         memcpy(res1, res2, NRS_RES_MAX * sizeof(res1[0]));
1660 out:
1661         spin_unlock(&svcpt->scp_req_lock);
1662
1663         /**
1664          * Release either the regular NRS head resources if we moved the
1665          * request, or the high-priority NRS head resources if we took a
1666          * reference earlier in this function and ptlrpc_nrs_req_can_move()
1667          * returned false.
1668          */
1669         nrs_resource_put_safe(res1);
1670         EXIT;
1671 }
1672
1673 /**
1674  * Carries out a control operation \a opc on the policy identified by the
1675  * human-readable \a name, on either all partitions, or only on the first
1676  * partition of service \a svc.
1677  *
1678  * \param[in]     svc    the service the policy belongs to.
1679  * \param[in]     queue  whether to carry out the command on the policy which
1680  *                       belongs to the regular, high-priority, or both NRS
1681  *                       heads of service partitions of \a svc.
1682  * \param[in]     name   the policy to act upon, by human-readable name
1683  * \param[in]     opc    the opcode of the operation to carry out
1684  * \param[in]     single when set, the operation will only be carried out on the
1685  *                       NRS heads of the first service partition of \a svc.
1686  *                       This is useful for some policies which e.g. share
1687  *                       identical values on the same parameters of different
1688  *                       service partitions; when reading these parameters via
1689  *                       lprocfs, these policies may just want to obtain and
1690  *                       print out the values from the first service partition.
1691  *                       Storing these values centrally elsewhere then could be
1692  *                       another solution for this.
1693  * \param[in,out] arg    can be used as a generic in/out buffer between control
1694  *                       operations and the user environment.
1695  *
1696  *\retval -ve error condition
1697  *\retval   0 operation was carried out successfully
1698  */
1699 int ptlrpc_nrs_policy_control(const struct ptlrpc_service *svc,
1700                               enum ptlrpc_nrs_queue_type queue, char *name,
1701                               enum ptlrpc_nrs_ctl opc, bool single, void *arg)
1702 {
1703         struct ptlrpc_service_part     *svcpt;
1704         int                             i;
1705         int                             rc = 0;
1706         ENTRY;
1707
1708         LASSERT(opc != PTLRPC_NRS_CTL_INVALID);
1709
1710         if ((queue & PTLRPC_NRS_QUEUE_BOTH) == 0)
1711                 return -EINVAL;
1712
1713         ptlrpc_service_for_each_part(svcpt, i, svc) {
1714                 if ((queue & PTLRPC_NRS_QUEUE_REG) != 0) {
1715                         rc = nrs_policy_ctl(nrs_svcpt2nrs(svcpt, false), name,
1716                                             opc, arg);
1717                         if (rc != 0 || (queue == PTLRPC_NRS_QUEUE_REG &&
1718                                         single))
1719                                 GOTO(out, rc);
1720                 }
1721
1722                 if ((queue & PTLRPC_NRS_QUEUE_HP) != 0) {
1723                         /**
1724                          * XXX: We could optionally check for
1725                          * nrs_svc_has_hp(svc) here, and return an error if it
1726                          * is false. Right now we rely on the policies' lprocfs
1727                          * handlers that call the present function to make this
1728                          * check; if they fail to do so, they might hit the
1729                          * assertion inside nrs_svcpt2nrs() below.
1730                          */
1731                         rc = nrs_policy_ctl(nrs_svcpt2nrs(svcpt, true), name,
1732                                             opc, arg);
1733                         if (rc != 0 || single)
1734                                 GOTO(out, rc);
1735                 }
1736         }
1737 out:
1738         RETURN(rc);
1739 }
1740
1741
1742 /* ptlrpc/nrs_fifo.c */
1743 extern struct ptlrpc_nrs_pol_conf nrs_conf_fifo;
1744 #if defined HAVE_SERVER_SUPPORT && defined(__KERNEL__)
1745 /* ptlrpc/nrs_crr.c */
1746 extern struct ptlrpc_nrs_pol_conf nrs_conf_crrn;
1747 #endif
1748
1749 /**
1750  * Adds all policies that ship with the ptlrpc module, to NRS core's list of
1751  * policies \e nrs_core.nrs_policies.
1752  *
1753  * \retval 0 all policies have been registered successfully
1754  * \retval -ve error
1755  */
1756 int ptlrpc_nrs_init(void)
1757 {
1758         int     rc;
1759         ENTRY;
1760
1761         mutex_init(&nrs_core.nrs_mutex);
1762         CFS_INIT_LIST_HEAD(&nrs_core.nrs_policies);
1763
1764         rc = ptlrpc_nrs_policy_register(&nrs_conf_fifo);
1765         if (rc != 0)
1766                 GOTO(fail, rc);
1767
1768 #if defined HAVE_SERVER_SUPPORT && defined(__KERNEL__)
1769         rc = ptlrpc_nrs_policy_register(&nrs_conf_crrn);
1770         if (rc != 0)
1771                 GOTO(fail, rc);
1772 #endif
1773
1774         RETURN(rc);
1775 fail:
1776         /**
1777          * Since no PTLRPC services have been started at this point, all we need
1778          * to do for cleanup is to free the descriptors.
1779          */
1780         ptlrpc_nrs_fini();
1781
1782         RETURN(rc);
1783 }
1784
1785 /**
1786  * Removes all policy desciptors from nrs_core::nrs_policies, and frees the
1787  * policy descriptors.
1788  *
1789  * Since all PTLRPC services are stopped at this point, there are no more
1790  * instances of any policies, because each service will have stopped its policy
1791  * instances in ptlrpc_service_nrs_cleanup(), so we just need to free the
1792  * descriptors here.
1793  */
1794 void ptlrpc_nrs_fini(void)
1795 {
1796         struct ptlrpc_nrs_pol_desc *desc;
1797         struct ptlrpc_nrs_pol_desc *tmp;
1798
1799         cfs_list_for_each_entry_safe(desc, tmp, &nrs_core.nrs_policies,
1800                                      pd_list) {
1801                 cfs_list_del_init(&desc->pd_list);
1802                 OBD_FREE_PTR(desc);
1803         }
1804 }
1805
1806 /** @} nrs */