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