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