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