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