lustre/ptlrpc/nrs.c

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