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