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