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