drivers/staging/lustre/lustre/ptlrpc/nrs.c

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