lustre/ptlrpc/nrs_orr.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) 2013, Intel Corporation.
  24  *
  25  * Copyright 2012 Xyratex Technology Limited
  26  */
  27 /*
  28  * lustre/ptlrpc/nrs_orr.c
  29  *
  30  * Network Request Scheduler (NRS) ORR and TRR policies
  31  *
  32  * Request scheduling in a Round-Robin manner over backend-fs objects and OSTs
  33  * respectively
  34  *
  35  * Author: Liang Zhen <liang@whamcloud.com>
  36  * Author: Nikitas Angelinas <nikitas_angelinas@xyratex.com>
  37  */
  38 #ifdef HAVE_SERVER_SUPPORT
  39
  40 /**
  41  * \addtogoup nrs
  42  * @{
  43  */
  44 #define DEBUG_SUBSYSTEM S_RPC
  45 #include <obd_support.h>
  46 #include <obd_class.h>
  47 #include <lustre_net.h>
  48 #include <lustre/lustre_idl.h>
  49 #include <lustre_req_layout.h>
  50 #include "ptlrpc_internal.h"
  51
  52 /**
  53  * \name ORR/TRR policy
  54  *
  55  * ORR/TRR (Object-based Round Robin/Target-based Round Robin) NRS policies
  56  *
  57  * ORR performs batched Round Robin shceduling of brw RPCs, based on the FID of
  58  * the backend-fs object that the brw RPC pertains to; the TRR policy performs
  59  * batched Round Robin scheduling of brw RPCs, based on the OST index that the
  60  * RPC pertains to. Both policies also order RPCs in each batch in ascending
  61  * offset order, which is lprocfs-tunable between logical file offsets, and
  62  * physical disk offsets, as reported by fiemap.
  63  *
  64  * The TRR policy reuses much of the functionality of ORR. These two scheduling
  65  * algorithms could alternatively be implemented under a single NRS policy, that
  66  * uses an lprocfs tunable in order to switch between the two types of
  67  * scheduling behaviour. The two algorithms have been implemented as separate
  68  * policies for reasons of clarity to the user, and to avoid issues that would
  69  * otherwise arise at the point of switching between behaviours in the case of
  70  * having a single policy, such as resource cleanup for nrs_orr_object
  71  * instances. It is possible that this may need to be re-examined in the future,
  72  * along with potentially coalescing other policies that perform batched request
  73  * scheduling in a Round-Robin manner, all into one policy.
  74  *
  75  * @{
  76  */
  77
  78 #define NRS_POL_NAME_ORR        "orr"
  79 #define NRS_POL_NAME_TRR        "trr"
  80
  81 /**
  82  * Checks if the RPC type of \a nrq is currently handled by an ORR/TRR policy
  83  *
  84  * \param[in]  orrd   the ORR/TRR policy scheduler instance
  85  * \param[in]  nrq    the request
  86  * \param[out] opcode the opcode is saved here, just in order to avoid calling
  87  *                    lustre_msg_get_opc() again later
  88  *
  89  * \retval true  request type is supported by the policy instance
  90  * \retval false request type is not supported by the policy instance
  91  */
  92 static bool nrs_orr_req_supported(struct nrs_orr_data *orrd,
  93                                   struct ptlrpc_nrs_request *nrq, __u32 *opcode)
  94 {
  95         struct ptlrpc_request  *req = container_of(nrq, struct ptlrpc_request,
  96                                                    rq_nrq);
  97         __u32                   opc = lustre_msg_get_opc(req->rq_reqmsg);
  98         bool                    rc = false;
  99
 100         /**
 101          * XXX: nrs_orr_data::od_supp accessed unlocked.
 102          */
 103         switch (opc) {
 104         case OST_READ:
 105                 rc = orrd->od_supp & NOS_OST_READ;
 106                 break;
 107         case OST_WRITE:
 108                 rc = orrd->od_supp & NOS_OST_WRITE;
 109                 break;
 110         }
 111
 112         if (rc)
 113                 *opcode = opc;
 114
 115         return rc;
 116 }
 117
 118 /**
 119  * Returns the ORR/TRR key fields for the request \a nrq in \a key.
 120  *
 121  * \param[in]  orrd the ORR/TRR policy scheduler instance
 122  * \param[in]  nrq  the request
 123  * \param[in]  opc  the request's opcode
 124  * \param[in]  name the policy name
 125  * \param[out] key  fields of the key are returned here.
 126  *
 127  * \retval 0   key filled successfully
 128  * \retval < 0 error
 129  */
 130 static int nrs_orr_key_fill(struct nrs_orr_data *orrd,
 131                             struct ptlrpc_nrs_request *nrq, __u32 opc,
 132                             char *name, struct nrs_orr_key *key)
 133 {
 134         struct ptlrpc_request  *req = container_of(nrq, struct ptlrpc_request,
 135                                                    rq_nrq);
 136         struct ost_body        *body;
 137         __u32                   ost_idx;
 138         bool                    is_orr = strncmp(name, NRS_POL_NAME_ORR,
 139                                                  NRS_POL_NAME_MAX) == 0;
 140
 141         LASSERT(req != NULL);
 142
 143         /**
 144          * This is an attempt to fill in the request key fields while
 145          * moving a request from the regular to the high-priority NRS
 146          * head (via ldlm_lock_reorder_req()), but the request key has
 147          * been adequately filled when nrs_orr_res_get() was called through
 148          * ptlrpc_nrs_req_initialize() for the regular NRS head's ORR/TRR
 149          * policy, so there is nothing to do.
 150          */
 151         if ((is_orr && nrq->nr_u.orr.or_orr_set) ||
 152             (!is_orr && nrq->nr_u.orr.or_trr_set)) {
 153                 *key = nrq->nr_u.orr.or_key;
 154                 return 0;
 155         }
 156
 157         if (nrq->nr_u.orr.or_orr_set || nrq->nr_u.orr.or_trr_set)
 158                 memset(&nrq->nr_u.orr.or_key, 0, sizeof(nrq->nr_u.orr.or_key));
 159
 160         ost_idx = class_server_data(req->rq_export->exp_obd)->lsd_osd_index;
 161
 162         if (is_orr) {
 163                 int     rc;
 164                 /**
 165                  * The request pill for OST_READ and OST_WRITE requests is
 166                  * initialized in the ost_io service's
 167                  * ptlrpc_service_ops::so_hpreq_handler, ost_io_hpreq_handler(),
 168                  * so no need to redo it here.
 169                  */
 170                 body = req_capsule_client_get(&req->rq_pill, &RMF_OST_BODY);
 171                 if (body == NULL)
 172                         RETURN(-EFAULT);
 173
 174                 rc = ostid_to_fid(&key->ok_fid, &body->oa.o_oi, ost_idx);
 175                 if (rc < 0)
 176                         return rc;
 177
 178                 nrq->nr_u.orr.or_orr_set = 1;
 179         } else {
 180                 key->ok_idx = ost_idx;
 181                 nrq->nr_u.orr.or_trr_set = 1;
 182         }
 183
 184         return 0;
 185 }
 186
 187 /**
 188  * Populates the range values in \a range with logical offsets obtained via
 189  * \a nb.
 190  *
 191  * \param[in]  nb       niobuf_remote struct array for this request
 192  * \param[in]  niocount count of niobuf_remote structs for this request
 193  * \param[out] range    the offset range is returned here
 194  */
 195 static void nrs_orr_range_fill_logical(struct niobuf_remote *nb, int niocount,
 196                                        struct nrs_orr_req_range *range)
 197 {
 198         /* Should we do this at page boundaries ? */
 199         range->or_start = nb[0].offset & CFS_PAGE_MASK;
 200         range->or_end = (nb[niocount - 1].offset +
 201                          nb[niocount - 1].len - 1) | ~CFS_PAGE_MASK;
 202 }
 203
 204 /**
 205  * We obtain information just for a single extent, as the request can only be in
 206  * a single place in the binary heap anyway.
 207  */
 208 #define ORR_NUM_EXTENTS 1
 209
 210 /**
 211  * Converts the logical file offset range in \a range, to a physical disk offset
 212  * range in \a range, for a request. Uses obd_get_info() in order to carry out a
 213  * fiemap call and obtain backend-fs extent information. The returned range is
 214  * in physical block numbers.
 215  *
 216  * \param[in]     nrq   the request
 217  * \param[in]     oa    obdo struct for this request
 218  * \param[in,out] range the offset range in bytes; logical range in, physical
 219  *                      range out
 220  *
 221  * \retval 0    physical offsets obtained successfully
 222  * \retvall < 0 error
 223  */
 224 static int nrs_orr_range_fill_physical(struct ptlrpc_nrs_request *nrq,
 225                                        struct obdo *oa,
 226                                        struct nrs_orr_req_range *range)
 227 {
 228         struct ptlrpc_request     *req = container_of(nrq,
 229                                                       struct ptlrpc_request,
 230                                                       rq_nrq);
 231         char                       fiemap_buf[offsetof(struct ll_user_fiemap,
 232                                                   fm_extents[ORR_NUM_EXTENTS])];
 233         struct ll_user_fiemap     *fiemap = (struct ll_user_fiemap *)fiemap_buf;
 234         struct ll_fiemap_info_key  key;
 235         loff_t                     start;
 236         loff_t                     end;
 237         int                        rc;
 238
 239         key = (typeof(key)) {
 240                 .name = KEY_FIEMAP,
 241                 .oa = *oa,
 242                 .fiemap = {
 243                         .fm_start = range->or_start,
 244                         .fm_length = range->or_end - range->or_start,
 245                         .fm_extent_count = ORR_NUM_EXTENTS
 246                 }
 247         };
 248
 249         rc = obd_get_info(req->rq_svc_thread->t_env, req->rq_export,
 250                           sizeof(key), &key, NULL, fiemap, NULL);
 251         if (rc < 0)
 252                 GOTO(out, rc);
 253
 254         if (fiemap->fm_mapped_extents == 0 ||
 255             fiemap->fm_mapped_extents > ORR_NUM_EXTENTS)
 256                 GOTO(out, rc = -EFAULT);
 257
 258         /**
 259          * Calculate the physical offset ranges for the request from the extent
 260          * information and the logical request offsets.
 261          */
 262         start = fiemap->fm_extents[0].fe_physical + range->or_start -
 263                 fiemap->fm_extents[0].fe_logical;
 264         end = start + range->or_end - range->or_start;
 265
 266         range->or_start = start;
 267         range->or_end = end;
 268
 269         nrq->nr_u.orr.or_physical_set = 1;
 270 out:
 271         return rc;
 272 }
 273
 274 /**
 275  * Sets the offset range the request covers; either in logical file
 276  * offsets or in physical disk offsets.
 277  *
 278  * \param[in] nrq        the request
 279  * \param[in] orrd       the ORR/TRR policy scheduler instance
 280  * \param[in] opc        the request's opcode
 281  * \param[in] moving_req is the request in the process of moving onto the
 282  *                       high-priority NRS head?
 283  *
 284  * \retval 0    range filled successfully
 285  * \retval != 0 error
 286  */
 287 static int nrs_orr_range_fill(struct ptlrpc_nrs_request *nrq,
 288                               struct nrs_orr_data *orrd, __u32 opc,
 289                               bool moving_req)
 290 {
 291         struct ptlrpc_request       *req = container_of(nrq,
 292                                                         struct ptlrpc_request,
 293                                                         rq_nrq);
 294         struct obd_ioobj            *ioo;
 295         struct niobuf_remote        *nb;
 296         struct ost_body             *body;
 297         struct nrs_orr_req_range     range;
 298         int                          niocount;
 299         int                          rc = 0;
 300
 301         /**
 302          * If we are scheduling using physical disk offsets, but we have filled
 303          * the offset information in the request previously
 304          * (i.e. ldlm_lock_reorder_req() is moving the request to the
 305          * high-priority NRS head), there is no need to do anything, and we can
 306          * exit. Moreover than the lack of need, we would be unable to perform
 307          * the obd_get_info() call required in nrs_orr_range_fill_physical(),
 308          * because ldlm_lock_reorder_lock() calls into here while holding a
 309          * spinlock, and retrieving fiemap information via obd_get_info() is a
 310          * potentially sleeping operation.
 311          */
 312         if (orrd->od_physical && nrq->nr_u.orr.or_physical_set)
 313                 return 0;
 314
 315         ioo = req_capsule_client_get(&req->rq_pill, &RMF_OBD_IOOBJ);
 316         if (ioo == NULL)
 317                 GOTO(out, rc = -EFAULT);
 318
 319         niocount = ioo->ioo_bufcnt;
 320
 321         nb = req_capsule_client_get(&req->rq_pill, &RMF_NIOBUF_REMOTE);
 322         if (nb == NULL)
 323                 GOTO(out, rc = -EFAULT);
 324
 325         /**
 326          * Use logical information from niobuf_remote structures.
 327          */
 328         nrs_orr_range_fill_logical(nb, niocount, &range);
 329
 330         /**
 331          * Obtain physical offsets if selected, and this is an OST_READ RPC
 332          * RPC. We do not enter this block if moving_req is set which indicates
 333          * that the request is being moved to the high-priority NRS head by
 334          * ldlm_lock_reorder_req(), as that function calls in here while holding
 335          * a spinlock, and nrs_orr_range_physical() can sleep, so we just use
 336          * logical file offsets for the range values for such requests.
 337          */
 338         if (orrd->od_physical && opc == OST_READ && !moving_req) {
 339                 body = req_capsule_client_get(&req->rq_pill, &RMF_OST_BODY);
 340                 if (body == NULL)
 341                         GOTO(out, rc = -EFAULT);
 342
 343                 /**
 344                  * Translate to physical block offsets from backend filesystem
 345                  * extents.
 346                  * Ignore return values; if obtaining the physical offsets
 347                  * fails, use the logical offsets.
 348                  */
 349                 nrs_orr_range_fill_physical(nrq, &body->oa, &range);
 350         }
 351
 352         nrq->nr_u.orr.or_range = range;
 353 out:
 354         return rc;
 355 }
 356
 357 /**
 358  * Generates a character string that can be used in order to register uniquely
 359  * named libcfs_hash and slab objects for ORR/TRR policy instances. The
 360  * character string is unique per policy instance, as it includes the policy's
 361  * name, the CPT number, and a {reg|hp} token, and there is one policy instance
 362  * per NRS head on each CPT, and the policy is only compatible with the ost_io
 363  * service.
 364  *
 365  * \param[in] policy the policy instance
 366  * \param[out] name  the character array that will hold the generated name
 367  */
 368 static void nrs_orr_genobjname(struct ptlrpc_nrs_policy *policy, char *name)
 369 {
 370         snprintf(name, NRS_ORR_OBJ_NAME_MAX, "%s%s%s%d",
 371                  "nrs_", policy->pol_desc->pd_name,
 372                  policy->pol_nrs->nrs_queue_type == PTLRPC_NRS_QUEUE_REG ?
 373                  "_reg_" : "_hp_", nrs_pol2cptid(policy));
 374 }
 375
 376 /**
 377  * ORR/TRR hash operations
 378  */
 379 #define NRS_ORR_BITS            24
 380 #define NRS_ORR_BKT_BITS        12
 381 #define NRS_ORR_HASH_FLAGS      (CFS_HASH_SPIN_BKTLOCK | CFS_HASH_ASSERT_EMPTY)
 382
 383 #define NRS_TRR_BITS            4
 384 #define NRS_TRR_BKT_BITS        2
 385 #define NRS_TRR_HASH_FLAGS      CFS_HASH_SPIN_BKTLOCK
 386
 387 static unsigned nrs_orr_hop_hash(cfs_hash_t *hs, const void *key, unsigned mask)
 388 {
 389         return cfs_hash_djb2_hash(key, sizeof(struct nrs_orr_key), mask);
 390 }
 391
 392 static void *nrs_orr_hop_key(cfs_hlist_node_t *hnode)
 393 {
 394         struct nrs_orr_object *orro = cfs_hlist_entry(hnode,
 395                                                       struct nrs_orr_object,
 396                                                       oo_hnode);
 397         return &orro->oo_key;
 398 }
 399
 400 static int nrs_orr_hop_keycmp(const void *key, cfs_hlist_node_t *hnode)
 401 {
 402         struct nrs_orr_object *orro = cfs_hlist_entry(hnode,
 403                                                       struct nrs_orr_object,
 404                                                       oo_hnode);
 405
 406         return lu_fid_eq(&orro->oo_key.ok_fid,
 407                          &((struct nrs_orr_key *)key)->ok_fid);
 408 }
 409
 410 static void *nrs_orr_hop_object(cfs_hlist_node_t *hnode)
 411 {
 412         return cfs_hlist_entry(hnode, struct nrs_orr_object, oo_hnode);
 413 }
 414
 415 static void nrs_orr_hop_get(cfs_hash_t *hs, cfs_hlist_node_t *hnode)
 416 {
 417         struct nrs_orr_object *orro = cfs_hlist_entry(hnode,
 418                                                       struct nrs_orr_object,
 419                                                       oo_hnode);
 420         orro->oo_ref++;
 421 }
 422
 423 /**
 424  * Removes an nrs_orr_object the hash and frees its memory, if the object has
 425  * no active users.
 426  */
 427 static void nrs_orr_hop_put_free(cfs_hash_t *hs, cfs_hlist_node_t *hnode)
 428 {
 429         struct nrs_orr_object *orro = cfs_hlist_entry(hnode,
 430                                                       struct nrs_orr_object,
 431                                                       oo_hnode);
 432         struct nrs_orr_data   *orrd = container_of(orro->oo_res.res_parent,
 433                                                    struct nrs_orr_data, od_res);
 434         cfs_hash_bd_t          bd;
 435
 436         cfs_hash_bd_get_and_lock(hs, &orro->oo_key, &bd, 1);
 437
 438         if (--orro->oo_ref > 1) {
 439                 cfs_hash_bd_unlock(hs, &bd, 1);
 440
 441                 return;
 442         }
 443         LASSERT(orro->oo_ref == 1);
 444
 445         cfs_hash_bd_del_locked(hs, &bd, hnode);
 446         cfs_hash_bd_unlock(hs, &bd, 1);
 447
 448         OBD_SLAB_FREE_PTR(orro, orrd->od_cache);
 449 }
 450
 451 static void nrs_orr_hop_put(cfs_hash_t *hs, cfs_hlist_node_t *hnode)
 452 {
 453         struct nrs_orr_object *orro = cfs_hlist_entry(hnode,
 454                                                       struct nrs_orr_object,
 455                                                       oo_hnode);
 456         orro->oo_ref--;
 457 }
 458
 459 static int nrs_trr_hop_keycmp(const void *key, cfs_hlist_node_t *hnode)
 460 {
 461         struct nrs_orr_object *orro = cfs_hlist_entry(hnode,
 462                                                       struct nrs_orr_object,
 463                                                       oo_hnode);
 464
 465         return orro->oo_key.ok_idx == ((struct nrs_orr_key *)key)->ok_idx;
 466 }
 467
 468 static void nrs_trr_hop_exit(cfs_hash_t *hs, cfs_hlist_node_t *hnode)
 469 {
 470         struct nrs_orr_object *orro = cfs_hlist_entry(hnode,
 471                                                       struct nrs_orr_object,
 472                                                       oo_hnode);
 473         struct nrs_orr_data   *orrd = container_of(orro->oo_res.res_parent,
 474                                                    struct nrs_orr_data, od_res);
 475
 476         LASSERTF(orro->oo_ref == 0,
 477                  "Busy NRS TRR policy object for OST with index %u, with %ld "
 478                  "refs\n", orro->oo_key.ok_idx, orro->oo_ref);
 479
 480         OBD_SLAB_FREE_PTR(orro, orrd->od_cache);
 481 }
 482
 483 static cfs_hash_ops_t nrs_orr_hash_ops = {
 484         .hs_hash        = nrs_orr_hop_hash,
 485         .hs_key         = nrs_orr_hop_key,
 486         .hs_keycmp      = nrs_orr_hop_keycmp,
 487         .hs_object      = nrs_orr_hop_object,
 488         .hs_get         = nrs_orr_hop_get,
 489         .hs_put         = nrs_orr_hop_put_free,
 490         .hs_put_locked  = nrs_orr_hop_put,
 491 };
 492
 493 static cfs_hash_ops_t nrs_trr_hash_ops = {
 494         .hs_hash        = nrs_orr_hop_hash,
 495         .hs_key         = nrs_orr_hop_key,
 496         .hs_keycmp      = nrs_trr_hop_keycmp,
 497         .hs_object      = nrs_orr_hop_object,
 498         .hs_get         = nrs_orr_hop_get,
 499         .hs_put         = nrs_orr_hop_put,
 500         .hs_put_locked  = nrs_orr_hop_put,
 501         .hs_exit        = nrs_trr_hop_exit,
 502 };
 503
 504 #define NRS_ORR_QUANTUM_DFLT    256
 505
 506 /**
 507  * Binary heap predicate.
 508  *
 509  * Uses
 510  * ptlrpc_nrs_request::nr_u::orr::or_round,
 511  * ptlrpc_nrs_request::nr_u::orr::or_sequence, and
 512  * ptlrpc_nrs_request::nr_u::orr::or_range to compare two binheap nodes and
 513  * produce a binary predicate that indicates their relative priority, so that
 514  * the binary heap can perform the necessary sorting operations.
 515  *
 516  * \param[in] e1 the first binheap node to compare
 517  * \param[in] e2 the second binheap node to compare
 518  *
 519  * \retval 0 e1 > e2
 520  * \retval 1 e1 < e2
 521  */
 522 static int orr_req_compare(cfs_binheap_node_t *e1, cfs_binheap_node_t *e2)
 523 {
 524         struct ptlrpc_nrs_request *nrq1;
 525         struct ptlrpc_nrs_request *nrq2;
 526
 527         nrq1 = container_of(e1, struct ptlrpc_nrs_request, nr_node);
 528         nrq2 = container_of(e2, struct ptlrpc_nrs_request, nr_node);
 529
 530         /**
 531          * Requests have been scheduled against a different scheduling round.
 532          */
 533         if (nrq1->nr_u.orr.or_round < nrq2->nr_u.orr.or_round)
 534                 return 1;
 535         else if (nrq1->nr_u.orr.or_round > nrq2->nr_u.orr.or_round)
 536                 return 0;
 537
 538         /**
 539          * Requests have been scheduled against the same scheduling round, but
 540          * belong to a different batch, i.e. they pertain to a different
 541          * backend-fs object (for ORR policy instances) or OST (for TRR policy
 542          * instances).
 543          */
 544         if (nrq1->nr_u.orr.or_sequence < nrq2->nr_u.orr.or_sequence)
 545                 return 1;
 546         else if (nrq1->nr_u.orr.or_sequence > nrq2->nr_u.orr.or_sequence)
 547                 return 0;
 548
 549         /**
 550          * If round numbers and sequence numbers are equal, the two requests
 551          * have been scheduled on the same round, and belong to the same batch,
 552          * which means they pertain to the same backend-fs object (if this is an
 553          * ORR policy instance), or to the same OST (if this is a TRR policy
 554          * instance), so these requests should be sorted by ascending offset
 555          * order.
 556          */
 557         if (nrq1->nr_u.orr.or_range.or_start <
 558             nrq2->nr_u.orr.or_range.or_start) {
 559                 return 1;
 560         } else if (nrq1->nr_u.orr.or_range.or_start >
 561                  nrq2->nr_u.orr.or_range.or_start) {
 562                 return 0;
 563         } else {
 564                 /**
 565                  * Requests start from the same offset; Dispatch the shorter one
 566                  * first; perhaps slightly more chances of hitting caches like
 567                  * this.
 568                  */
 569                 return nrq1->nr_u.orr.or_range.or_end <
 570                        nrq2->nr_u.orr.or_range.or_end;
 571         }
 572 }
 573
 574 /**
 575  * ORR binary heap operations
 576  */
 577 static cfs_binheap_ops_t nrs_orr_heap_ops = {
 578         .hop_enter      = NULL,
 579         .hop_exit       = NULL,
 580         .hop_compare    = orr_req_compare,
 581 };
 582
 583 /**
 584  * Prints a warning message if an ORR/TRR policy is started on a service with
 585  * more than one CPT.  Not printed on the console for now, since we don't
 586  * have any performance metrics in the first place, and it is annoying.
 587  *
 588  * \param[in] policy the policy instance
 589  *
 590  * \retval 0 success
 591  */
 592 static int nrs_orr_init(struct ptlrpc_nrs_policy *policy)
 593 {
 594         if (policy->pol_nrs->nrs_svcpt->scp_service->srv_ncpts > 1)
 595                 CDEBUG(D_CONFIG, "%s: The %s NRS policy was registered on a "
 596                       "service with multiple service partitions. This policy "
 597                       "may perform better with a single partition.\n",
 598                       policy->pol_nrs->nrs_svcpt->scp_service->srv_name,
 599                       policy->pol_desc->pd_name);
 600
 601         return 0;
 602 }
 603
 604 /**
 605  * Called when an ORR policy instance is started.
 606  *
 607  * \param[in] policy the policy
 608  *
 609  * \retval -ENOMEM OOM error
 610  * \retval 0       success
 611  */
 612 static int nrs_orr_start(struct ptlrpc_nrs_policy *policy)
 613 {
 614         struct nrs_orr_data    *orrd;
 615         cfs_hash_ops_t         *ops;
 616         unsigned                cur_bits;
 617         unsigned                max_bits;
 618         unsigned                bkt_bits;
 619         unsigned                flags;
 620         int                     rc = 0;
 621         ENTRY;
 622
 623         OBD_CPT_ALLOC_PTR(orrd, nrs_pol2cptab(policy), nrs_pol2cptid(policy));
 624         if (orrd == NULL)
 625                 RETURN(-ENOMEM);
 626
 627         /*
 628          * Binary heap instance for sorted incoming requests.
 629          */
 630         orrd->od_binheap = cfs_binheap_create(&nrs_orr_heap_ops,
 631                                               CBH_FLAG_ATOMIC_GROW, 4096, NULL,
 632                                               nrs_pol2cptab(policy),
 633                                               nrs_pol2cptid(policy));
 634         if (orrd->od_binheap == NULL)
 635                 GOTO(failed, rc = -ENOMEM);
 636
 637         nrs_orr_genobjname(policy, orrd->od_objname);
 638
 639         /**
 640          * Slab cache for NRS ORR/TRR objects.
 641          */
 642         orrd->od_cache = kmem_cache_create(orrd->od_objname,
 643                                            sizeof(struct nrs_orr_object),
 644                                            0, 0, NULL);
 645         if (orrd->od_cache == NULL)
 646                 GOTO(failed, rc = -ENOMEM);
 647
 648         if (strncmp(policy->pol_desc->pd_name, NRS_POL_NAME_ORR,
 649                     NRS_POL_NAME_MAX) == 0) {
 650                 ops = &nrs_orr_hash_ops;
 651                 cur_bits = NRS_ORR_BITS;
 652                 max_bits = NRS_ORR_BITS;
 653                 bkt_bits = NRS_ORR_BKT_BITS;
 654                 flags = NRS_ORR_HASH_FLAGS;
 655         } else {
 656                 ops = &nrs_trr_hash_ops;
 657                 cur_bits = NRS_TRR_BITS;
 658                 max_bits = NRS_TRR_BITS;
 659                 bkt_bits = NRS_TRR_BKT_BITS;
 660                 flags = NRS_TRR_HASH_FLAGS;
 661         }
 662
 663         /**
 664          * Hash for finding objects by struct nrs_orr_key.
 665          * XXX: For TRR, it might be better to avoid using libcfs_hash?
 666          * All that needs to be resolved are OST indices, and they
 667          * will stay relatively stable during an OSS node's lifetime.
 668          */
 669         orrd->od_obj_hash = cfs_hash_create(orrd->od_objname, cur_bits,
 670                                             max_bits, bkt_bits, 0,
 671                                             CFS_HASH_MIN_THETA,
 672                                             CFS_HASH_MAX_THETA, ops, flags);
 673         if (orrd->od_obj_hash == NULL)
 674                 GOTO(failed, rc = -ENOMEM);
 675
 676         /* XXX: Fields accessed unlocked */
 677         orrd->od_quantum = NRS_ORR_QUANTUM_DFLT;
 678         orrd->od_supp = NOS_DFLT;
 679         orrd->od_physical = true;
 680         /**
 681          * Set to 1 so that the test inside nrs_orr_req_add() can evaluate to
 682          * true.
 683          */
 684         orrd->od_sequence = 1;
 685
 686         policy->pol_private = orrd;
 687
 688         RETURN(rc);
 689
 690 failed:
 691         if (orrd->od_cache) {
 692                 kmem_cache_destroy(orrd->od_cache);
 693                 LASSERTF(rc == 0, "Could not destroy od_cache slab\n");
 694         }
 695         if (orrd->od_binheap != NULL)
 696                 cfs_binheap_destroy(orrd->od_binheap);
 697
 698         OBD_FREE_PTR(orrd);
 699
 700         RETURN(rc);
 701 }
 702
 703 /**
 704  * Called when an ORR/TRR policy instance is stopped.
 705  *
 706  * Called when the policy has been instructed to transition to the
 707  * ptlrpc_nrs_pol_state::NRS_POL_STATE_STOPPED state and has no more
 708  * pending requests to serve.
 709  *
 710  * \param[in] policy the policy
 711  */
 712 static void nrs_orr_stop(struct ptlrpc_nrs_policy *policy)
 713 {
 714         struct nrs_orr_data *orrd = policy->pol_private;
 715         ENTRY;
 716
 717         LASSERT(orrd != NULL);
 718         LASSERT(orrd->od_binheap != NULL);
 719         LASSERT(orrd->od_obj_hash != NULL);
 720         LASSERT(orrd->od_cache != NULL);
 721         LASSERT(cfs_binheap_is_empty(orrd->od_binheap));
 722
 723         cfs_binheap_destroy(orrd->od_binheap);
 724         cfs_hash_putref(orrd->od_obj_hash);
 725         kmem_cache_destroy(orrd->od_cache);
 726
 727         OBD_FREE_PTR(orrd);
 728 }
 729
 730 /**
 731  * Performs a policy-specific ctl function on ORR/TRR policy instances; similar
 732  * to ioctl.
 733  *
 734  * \param[in]     policy the policy instance
 735  * \param[in]     opc    the opcode
 736  * \param[in,out] arg    used for passing parameters and information
 737  *
 738  * \pre spin_is_locked(&policy->pol_nrs->->nrs_lock)
 739  * \post spin_is_locked(&policy->pol_nrs->->nrs_lock)
 740  *
 741  * \retval 0   operation carried successfully
 742  * \retval -ve error
 743  */
 744 int nrs_orr_ctl(struct ptlrpc_nrs_policy *policy, enum ptlrpc_nrs_ctl opc,
 745                 void *arg)
 746 {
 747         LASSERT(spin_is_locked(&policy->pol_nrs->nrs_lock));
 748
 749         switch((enum nrs_ctl_orr)opc) {
 750         default:
 751                 RETURN(-EINVAL);
 752
 753         case NRS_CTL_ORR_RD_QUANTUM: {
 754                 struct nrs_orr_data     *orrd = policy->pol_private;
 755
 756                 *(__u16 *)arg = orrd->od_quantum;
 757                 }
 758                 break;
 759
 760         case NRS_CTL_ORR_WR_QUANTUM: {
 761                 struct nrs_orr_data     *orrd = policy->pol_private;
 762
 763                 orrd->od_quantum = *(__u16 *)arg;
 764                 LASSERT(orrd->od_quantum != 0);
 765                 }
 766                 break;
 767
 768         case NRS_CTL_ORR_RD_OFF_TYPE: {
 769                 struct nrs_orr_data     *orrd = policy->pol_private;
 770
 771                 *(bool *)arg = orrd->od_physical;
 772                 }
 773                 break;
 774
 775         case NRS_CTL_ORR_WR_OFF_TYPE: {
 776                 struct nrs_orr_data     *orrd = policy->pol_private;
 777
 778                 orrd->od_physical = *(bool *)arg;
 779                 }
 780                 break;
 781
 782         case NRS_CTL_ORR_RD_SUPP_REQ: {
 783                 struct nrs_orr_data     *orrd = policy->pol_private;
 784
 785                 *(enum nrs_orr_supp *)arg = orrd->od_supp;
 786                 }
 787                 break;
 788
 789         case NRS_CTL_ORR_WR_SUPP_REQ: {
 790                 struct nrs_orr_data     *orrd = policy->pol_private;
 791
 792                 orrd->od_supp = *(enum nrs_orr_supp *)arg;
 793                 LASSERT((orrd->od_supp & NOS_OST_RW) != 0);
 794                 }
 795                 break;
 796         }
 797         RETURN(0);
 798 }
 799
 800 /**
 801  * Obtains resources for ORR/TRR policy instances. The top-level resource lives
 802  * inside \e nrs_orr_data and the second-level resource inside
 803  * \e nrs_orr_object instances.
 804  *
 805  * \param[in]  policy     the policy for which resources are being taken for
 806  *                        request \a nrq
 807  * \param[in]  nrq        the request for which resources are being taken
 808  * \param[in]  parent     parent resource, embedded in nrs_orr_data for the
 809  *                        ORR/TRR policies
 810  * \param[out] resp       used to return resource references
 811  * \param[in]  moving_req signifies limited caller context; used to perform
 812  *                        memory allocations in an atomic context in this
 813  *                        policy
 814  *
 815  * \retval 0   we are returning a top-level, parent resource, one that is
 816  *             embedded in an nrs_orr_data object
 817  * \retval 1   we are returning a bottom-level resource, one that is embedded
 818  *             in an nrs_orr_object object
 819  *
 820  * \see nrs_resource_get_safe()
 821  */
 822 int nrs_orr_res_get(struct ptlrpc_nrs_policy *policy,
 823                     struct ptlrpc_nrs_request *nrq,
 824                     const struct ptlrpc_nrs_resource *parent,
 825                     struct ptlrpc_nrs_resource **resp, bool moving_req)
 826 {
 827         struct nrs_orr_data            *orrd;
 828         struct nrs_orr_object          *orro;
 829         struct nrs_orr_object          *tmp;
 830         struct nrs_orr_key              key = { { { 0 } } };
 831         __u32                           opc;
 832         int                             rc = 0;
 833
 834         /**
 835          * struct nrs_orr_data is requested.
 836          */
 837         if (parent == NULL) {
 838                 *resp = &((struct nrs_orr_data *)policy->pol_private)->od_res;
 839                 return 0;
 840         }
 841
 842         orrd = container_of(parent, struct nrs_orr_data, od_res);
 843
 844         /**
 845          * If the request type is not supported, fail the enqueuing; the RPC
 846          * will be handled by the fallback NRS policy.
 847          */
 848         if (!nrs_orr_req_supported(orrd, nrq, &opc))
 849                 return -1;
 850
 851         /**
 852          * Fill in the key for the request; OST FID for ORR policy instances,
 853          * and OST index for TRR policy instances.
 854          */
 855         rc = nrs_orr_key_fill(orrd, nrq, opc, policy->pol_desc->pd_name, &key);
 856         if (rc < 0)
 857                 RETURN(rc);
 858
 859         /**
 860          * Set the offset range the request covers
 861          */
 862         rc = nrs_orr_range_fill(nrq, orrd, opc, moving_req);
 863         if (rc < 0)
 864                 RETURN(rc);
 865
 866         orro = cfs_hash_lookup(orrd->od_obj_hash, &key);
 867         if (orro != NULL)
 868                 goto out;
 869
 870         OBD_SLAB_CPT_ALLOC_PTR_GFP(orro, orrd->od_cache,
 871                                    nrs_pol2cptab(policy), nrs_pol2cptid(policy),
 872                                    moving_req ? GFP_ATOMIC : __GFP_IO);
 873         if (orro == NULL)
 874                 RETURN(-ENOMEM);
 875
 876         orro->oo_key = key;
 877         orro->oo_ref = 1;
 878
 879         tmp = cfs_hash_findadd_unique(orrd->od_obj_hash, &orro->oo_key,
 880                                       &orro->oo_hnode);
 881         if (tmp != orro) {
 882                 OBD_SLAB_FREE_PTR(orro, orrd->od_cache);
 883                 orro = tmp;
 884         }
 885 out:
 886         /**
 887          * For debugging purposes
 888          */
 889         nrq->nr_u.orr.or_key = orro->oo_key;
 890
 891         *resp = &orro->oo_res;
 892
 893         return 1;
 894 }
 895
 896 /**
 897  * Called when releasing references to the resource hierachy obtained for a
 898  * request for scheduling using ORR/TRR policy instances
 899  *
 900  * \param[in] policy   the policy the resource belongs to
 901  * \param[in] res      the resource to be released
 902  */
 903 static void nrs_orr_res_put(struct ptlrpc_nrs_policy *policy,
 904                             const struct ptlrpc_nrs_resource *res)
 905 {
 906         struct nrs_orr_data     *orrd;
 907         struct nrs_orr_object   *orro;
 908
 909         /**
 910          * Do nothing for freeing parent, nrs_orr_data resources.
 911          */
 912         if (res->res_parent == NULL)
 913                 return;
 914
 915         orro = container_of(res, struct nrs_orr_object, oo_res);
 916         orrd = container_of(res->res_parent, struct nrs_orr_data, od_res);
 917
 918         cfs_hash_put(orrd->od_obj_hash, &orro->oo_hnode);
 919 }
 920
 921 /**
 922  * Called when polling an ORR/TRR policy instance for a request so that it can
 923  * be served. Returns the request that is at the root of the binary heap, as
 924  * that is the lowest priority one (i.e. libcfs_heap is an implementation of a
 925  * min-heap)
 926  *
 927  * \param[in] policy the policy instance being polled
 928  * \param[in] peek   when set, signifies that we just want to examine the
 929  *                   request, and not handle it, so the request is not removed
 930  *                   from the policy.
 931  * \param[in] force  force the policy to return a request; unused in this policy
 932  *
 933  * \retval the request to be handled
 934  * \retval NULL no request available
 935  *
 936  * \see ptlrpc_nrs_req_get_nolock()
 937  * \see nrs_request_get()
 938  */
 939 static
 940 struct ptlrpc_nrs_request *nrs_orr_req_get(struct ptlrpc_nrs_policy *policy,
 941                                            bool peek, bool force)
 942 {
 943         struct nrs_orr_data       *orrd = policy->pol_private;
 944         cfs_binheap_node_t        *node = cfs_binheap_root(orrd->od_binheap);
 945         struct ptlrpc_nrs_request *nrq;
 946
 947         nrq = unlikely(node == NULL) ? NULL :
 948               container_of(node, struct ptlrpc_nrs_request, nr_node);
 949
 950         if (likely(!peek && nrq != NULL)) {
 951                 struct nrs_orr_object *orro;
 952
 953                 orro = container_of(nrs_request_resource(nrq),
 954                                     struct nrs_orr_object, oo_res);
 955
 956                 LASSERT(nrq->nr_u.orr.or_round <= orro->oo_round);
 957
 958                 cfs_binheap_remove(orrd->od_binheap, &nrq->nr_node);
 959                 orro->oo_active--;
 960
 961                 if (strncmp(policy->pol_desc->pd_name, NRS_POL_NAME_ORR,
 962                                  NRS_POL_NAME_MAX) == 0)
 963                         CDEBUG(D_RPCTRACE,
 964                                "NRS: starting to handle %s request for object "
 965                                "with FID "DFID", from OST with index %u, with "
 966                                "round "LPU64"\n", NRS_POL_NAME_ORR,
 967                                PFID(&orro->oo_key.ok_fid),
 968                                nrq->nr_u.orr.or_key.ok_idx,
 969                                nrq->nr_u.orr.or_round);
 970                 else
 971                         CDEBUG(D_RPCTRACE,
 972                                "NRS: starting to handle %s request from OST "
 973                                "with index %u, with round "LPU64"\n",
 974                                NRS_POL_NAME_TRR, nrq->nr_u.orr.or_key.ok_idx,
 975                                nrq->nr_u.orr.or_round);
 976
 977                 /** Peek at the next request to be served */
 978                 node = cfs_binheap_root(orrd->od_binheap);
 979
 980                 /** No more requests */
 981                 if (unlikely(node == NULL)) {
 982                         orrd->od_round++;
 983                 } else {
 984                         struct ptlrpc_nrs_request *next;
 985
 986                         next = container_of(node, struct ptlrpc_nrs_request,
 987                                             nr_node);
 988
 989                         if (orrd->od_round < next->nr_u.orr.or_round)
 990                                 orrd->od_round = next->nr_u.orr.or_round;
 991                 }
 992         }
 993
 994         return nrq;
 995 }
 996
 997 /**
 998  * Sort-adds request \a nrq to an ORR/TRR \a policy instance's set of queued
 999  * requests in the policy's binary heap.
1000  *
1001  * A scheduling round is a stream of requests that have been sorted in batches
1002  * according to the backend-fs object (for ORR policy instances) or OST (for TRR
1003  * policy instances) that they pertain to (as identified by its IDIF FID or OST
1004  * index respectively); there can be only one batch for each object or OST in
1005  * each round. The batches are of maximum size nrs_orr_data:od_quantum. When a
1006  * new request arrives for scheduling for an object or OST that has exhausted
1007  * its quantum in its current round, the request will be scheduled on the next
1008  * scheduling round. Requests are allowed to be scheduled against a round until
1009  * all requests for the round are serviced, so an object or OST might miss a
1010  * round if requests are not scheduled for it for a long enough period of time.
1011  * Objects or OSTs that miss a round will continue with having their next
1012  * request scheduled, starting at the round that requests are being dispatched
1013  * for, at the time of arrival of this request.
1014  *
1015  * Requests are tagged with the round number and a sequence number; the sequence
1016  * number indicates the relative ordering amongst the batches of requests in a
1017  * round, and is identical for all requests in a batch, as is the round number.
1018  * The round and sequence numbers are used by orr_req_compare() in order to use
1019  * nrs_orr_data::od_binheap in order to maintain an ordered set of rounds, with
1020  * each round consisting of an ordered set of batches of requests, and each
1021  * batch consisting of an ordered set of requests according to their logical
1022  * file or physical disk offsets.
1023  *
1024  * \param[in] policy the policy
1025  * \param[in] nrq    the request to add
1026  *
1027  * \retval 0    request successfully added
1028  * \retval != 0 error
1029  */
1030 static int nrs_orr_req_add(struct ptlrpc_nrs_policy *policy,
1031                            struct ptlrpc_nrs_request *nrq)
1032 {
1033         struct nrs_orr_data     *orrd;
1034         struct nrs_orr_object   *orro;
1035         int                      rc;
1036
1037         orro = container_of(nrs_request_resource(nrq),
1038                             struct nrs_orr_object, oo_res);
1039         orrd = container_of(nrs_request_resource(nrq)->res_parent,
1040                             struct nrs_orr_data, od_res);
1041
1042         if (orro->oo_quantum == 0 || orro->oo_round < orrd->od_round ||
1043             (orro->oo_active == 0 && orro->oo_quantum > 0)) {
1044
1045                 /**
1046                  * If there are no pending requests for the object/OST, but some
1047                  * of its quantum still remains unused, which implies we did not
1048                  * get a chance to schedule up to its maximum allowed batch size
1049                  * of requests in the previous round this object/OST
1050                  * participated in, schedule this next request on a new round;
1051                  * this avoids fragmentation of request batches caused by
1052                  * intermittent inactivity on the object/OST, at the expense of
1053                  * potentially slightly increased service time for the request
1054                  * batch this request will be a part of.
1055                  */
1056                 if (orro->oo_active == 0 && orro->oo_quantum > 0)
1057                         orro->oo_round++;
1058
1059                 /** A new scheduling round has commenced */
1060                 if (orro->oo_round < orrd->od_round)
1061                         orro->oo_round = orrd->od_round;
1062
1063                 /** I was not the last object/OST that scheduled a request */
1064                 if (orro->oo_sequence < orrd->od_sequence)
1065                         orro->oo_sequence = ++orrd->od_sequence;
1066                 /**
1067                  * Reset the quantum if we have reached the maximum quantum
1068                  * size for this batch, or even if we have not managed to
1069                  * complete a batch size up to its maximum allowed size.
1070                  * XXX: Accessed unlocked
1071                  */
1072                 orro->oo_quantum = orrd->od_quantum;
1073         }
1074
1075         nrq->nr_u.orr.or_round = orro->oo_round;
1076         nrq->nr_u.orr.or_sequence = orro->oo_sequence;
1077
1078         rc = cfs_binheap_insert(orrd->od_binheap, &nrq->nr_node);
1079         if (rc == 0) {
1080                 orro->oo_active++;
1081                 if (--orro->oo_quantum == 0)
1082                         orro->oo_round++;
1083         }
1084         return rc;
1085 }
1086
1087 /**
1088  * Removes request \a nrq from an ORR/TRR \a policy instance's set of queued
1089  * requests.
1090  *
1091  * \param[in] policy the policy
1092  * \param[in] nrq    the request to remove
1093  */
1094 static void nrs_orr_req_del(struct ptlrpc_nrs_policy *policy,
1095                             struct ptlrpc_nrs_request *nrq)
1096 {
1097         struct nrs_orr_data     *orrd;
1098         struct nrs_orr_object   *orro;
1099         bool                     is_root;
1100
1101         orro = container_of(nrs_request_resource(nrq),
1102                             struct nrs_orr_object, oo_res);
1103         orrd = container_of(nrs_request_resource(nrq)->res_parent,
1104                             struct nrs_orr_data, od_res);
1105
1106         LASSERT(nrq->nr_u.orr.or_round <= orro->oo_round);
1107
1108         is_root = &nrq->nr_node == cfs_binheap_root(orrd->od_binheap);
1109
1110         cfs_binheap_remove(orrd->od_binheap, &nrq->nr_node);
1111         orro->oo_active--;
1112
1113         /**
1114          * If we just deleted the node at the root of the binheap, we may have
1115          * to adjust round numbers.
1116          */
1117         if (unlikely(is_root)) {
1118                 /** Peek at the next request to be served */
1119                 cfs_binheap_node_t *node = cfs_binheap_root(orrd->od_binheap);
1120
1121                 /** No more requests */
1122                 if (unlikely(node == NULL)) {
1123                         orrd->od_round++;
1124                 } else {
1125                         nrq = container_of(node, struct ptlrpc_nrs_request,
1126                                            nr_node);
1127
1128                         if (orrd->od_round < nrq->nr_u.orr.or_round)
1129                                 orrd->od_round = nrq->nr_u.orr.or_round;
1130                 }
1131         }
1132 }
1133
1134 /**
1135  * Called right after the request \a nrq finishes being handled by ORR policy
1136  * instance \a policy.
1137  *
1138  * \param[in] policy the policy that handled the request
1139  * \param[in] nrq    the request that was handled
1140  */
1141 static void nrs_orr_req_stop(struct ptlrpc_nrs_policy *policy,
1142                              struct ptlrpc_nrs_request *nrq)
1143 {
1144         /** NB: resource control, credits etc can be added here */
1145         if (strncmp(policy->pol_desc->pd_name, NRS_POL_NAME_ORR,
1146                     NRS_POL_NAME_MAX) == 0)
1147                 CDEBUG(D_RPCTRACE,
1148                        "NRS: finished handling %s request for object with FID "
1149                        DFID", from OST with index %u, with round "LPU64"\n",
1150                        NRS_POL_NAME_ORR, PFID(&nrq->nr_u.orr.or_key.ok_fid),
1151                        nrq->nr_u.orr.or_key.ok_idx, nrq->nr_u.orr.or_round);
1152         else
1153                 CDEBUG(D_RPCTRACE,
1154                        "NRS: finished handling %s request from OST with index %u,"
1155                        " with round "LPU64"\n",
1156                        NRS_POL_NAME_TRR, nrq->nr_u.orr.or_key.ok_idx,
1157                        nrq->nr_u.orr.or_round);
1158 }
1159
1160 /**
1161  * lprocfs interface
1162  */
1163
1164 #ifdef LPROCFS
1165
1166 /**
1167  * This allows to bundle the policy name into the lprocfs_vars::data pointer
1168  * so that lprocfs read/write functions can be used by both the ORR and TRR
1169  * policies.
1170  */
1171 struct nrs_lprocfs_orr_data {
1172         struct ptlrpc_service   *svc;
1173         char                    *name;
1174 } lprocfs_orr_data = {
1175         .name = NRS_POL_NAME_ORR
1176 }, lprocfs_trr_data = {
1177         .name = NRS_POL_NAME_TRR
1178 };
1179
1180 /**
1181  * Retrieves the value of the Round Robin quantum (i.e. the maximum batch size)
1182  * for ORR/TRR policy instances on both the regular and high-priority NRS head
1183  * of a service, as long as a policy instance is not in the
1184  * ptlrpc_nrs_pol_state::NRS_POL_STATE_STOPPED state; policy instances in this
1185  * state are skipped later by nrs_orr_ctl().
1186  *
1187  * Quantum values are in # of RPCs, and the output is in YAML format.
1188  *
1189  * For example:
1190  *
1191  *      reg_quantum:256
1192  *      hp_quantum:8
1193  *
1194  * XXX: the CRR-N version of this, ptlrpc_lprocfs_rd_nrs_crrn_quantum() is
1195  * almost identical; it can be reworked and then reused for ORR/TRR.
1196  */
1197 static int ptlrpc_lprocfs_rd_nrs_orr_quantum(char *page, char **start,
1198                                              off_t off, int count, int *eof,
1199                                              void *data)
1200 {
1201         struct nrs_lprocfs_orr_data *orr_data = data;
1202         struct ptlrpc_service       *svc = orr_data->svc;
1203         __u16                        quantum;
1204         int                          rc;
1205         int                          rc2 = 0;
1206
1207         /**
1208          * Perform two separate calls to this as only one of the NRS heads'
1209          * policies may be in the ptlrpc_nrs_pol_state::NRS_POL_STATE_STARTED or
1210          * ptlrpc_nrs_pol_state::NRS_POL_STATE_STOPPING state.
1211          */
1212         rc = ptlrpc_nrs_policy_control(svc, PTLRPC_NRS_QUEUE_REG,
1213                                        orr_data->name,
1214                                        NRS_CTL_ORR_RD_QUANTUM,
1215                                        true, &quantum);
1216         if (rc == 0) {
1217                 *eof = 1;
1218                 rc2 = snprintf(page, count, NRS_LPROCFS_QUANTUM_NAME_REG
1219                                "%-5d\n", quantum);
1220                 /**
1221                  * Ignore -ENODEV as the regular NRS head's policy may be in the
1222                  * ptlrpc_nrs_pol_state::NRS_POL_STATE_STOPPED state.
1223                  */
1224         } else if (rc != -ENODEV) {
1225                 return rc;
1226         }
1227
1228         /**
1229          * We know the ost_io service which is the only one ORR/TRR policies are
1230          * compatible with, do have an HP NRS head, but it may be best to guard
1231          * against a possible change of this in the future.
1232          */
1233         if (!nrs_svc_has_hp(svc))
1234                 goto no_hp;
1235
1236         rc = ptlrpc_nrs_policy_control(svc, PTLRPC_NRS_QUEUE_HP,
1237                                        orr_data->name, NRS_CTL_ORR_RD_QUANTUM,
1238                                        true, &quantum);
1239         if (rc == 0) {
1240                 *eof = 1;
1241                 rc2 += snprintf(page + rc2, count - rc2,
1242                                 NRS_LPROCFS_QUANTUM_NAME_HP"%-5d\n", quantum);
1243                 /**
1244                  * Ignore -ENODEV as the high priority NRS head's policy may be
1245                  * in the ptlrpc_nrs_pol_state::NRS_POL_STATE_STOPPED state.
1246                  */
1247         } else if (rc != -ENODEV) {
1248                 return rc;
1249         }
1250
1251 no_hp:
1252
1253         return rc2 ? : rc;
1254 }
1255
1256 /**
1257  * Sets the value of the Round Robin quantum (i.e. the maximum batch size)
1258  * for ORR/TRR policy instances of a service. The user can set the quantum size
1259  * for the regular and high priority NRS head separately by specifying each
1260  * value, or both together in a single invocation.
1261  *
1262  * For example:
1263  *
1264  * lctl set_param ost.OSS.ost_io.nrs_orr_quantum=req_quantum:64, to set the
1265  * request quantum size of the ORR policy instance on the regular NRS head of
1266  * the ost_io service to 64
1267  *
1268  * lctl set_param ost.OSS.ost_io.nrs_trr_quantum=hp_quantum:8 to set the request
1269  * quantum size of the TRR policy instance on the high priority NRS head of the
1270  * ost_io service to 8
1271  *
1272  * lctl set_param ost.OSS.ost_io.nrs_orr_quantum=32, to set both the request
1273  * quantum size of the ORR policy instance on both the regular and the high
1274  * priority NRS head of the ost_io service to 32
1275  *
1276  * policy instances in the ptlrpc_nrs_pol_state::NRS_POL_STATE_STOPPED state
1277  * are skipped later by nrs_orr_ctl().
1278  *
1279  * XXX: the CRR-N version of this, ptlrpc_lprocfs_wr_nrs_crrn_quantum() is
1280  * almost identical; it can be reworked and then reused for ORR/TRR.
1281  */
1282 static int ptlrpc_lprocfs_wr_nrs_orr_quantum(struct file *file,
1283                                              const char *buffer,
1284                                              unsigned long count, void *data)
1285 {
1286         struct nrs_lprocfs_orr_data *orr_data = data;
1287         struct ptlrpc_service       *svc = orr_data->svc;
1288         enum ptlrpc_nrs_queue_type   queue = 0;
1289         char                         kernbuf[LPROCFS_NRS_WR_QUANTUM_MAX_CMD];
1290         char                        *val;
1291         long                         quantum_reg;
1292         long                         quantum_hp;
1293         /** lprocfs_find_named_value() modifies its argument, so keep a copy */
1294         size_t                       count_copy;
1295         int                          rc = 0;
1296         int                          rc2 = 0;
1297
1298         if (count > (sizeof(kernbuf) - 1))
1299                 return -EINVAL;
1300
1301         if (copy_from_user(kernbuf, buffer, count))
1302                 return -EFAULT;
1303
1304         kernbuf[count] = '\0';
1305
1306         count_copy = count;
1307
1308         /**
1309          * Check if the regular quantum value has been specified
1310          */
1311         val = lprocfs_find_named_value(kernbuf, NRS_LPROCFS_QUANTUM_NAME_REG,
1312                                        &count_copy);
1313         if (val != kernbuf) {
1314                 quantum_reg = simple_strtol(val, NULL, 10);
1315
1316                 queue |= PTLRPC_NRS_QUEUE_REG;
1317         }
1318
1319         count_copy = count;
1320
1321         /**
1322          * Check if the high priority quantum value has been specified
1323          */
1324         val = lprocfs_find_named_value(kernbuf, NRS_LPROCFS_QUANTUM_NAME_HP,
1325                                        &count_copy);
1326         if (val != kernbuf) {
1327                 if (!nrs_svc_has_hp(svc))
1328                         return -ENODEV;
1329
1330                 quantum_hp = simple_strtol(val, NULL, 10);
1331
1332                 queue |= PTLRPC_NRS_QUEUE_HP;
1333         }
1334
1335         /**
1336          * If none of the queues has been specified, look for a valid numerical
1337          * value
1338          */
1339         if (queue == 0) {
1340                 if (!isdigit(kernbuf[0]))
1341                         return -EINVAL;
1342
1343                 quantum_reg = simple_strtol(kernbuf, NULL, 10);
1344
1345                 queue = PTLRPC_NRS_QUEUE_REG;
1346
1347                 if (nrs_svc_has_hp(svc)) {
1348                         queue |= PTLRPC_NRS_QUEUE_HP;
1349                         quantum_hp = quantum_reg;
1350                 }
1351         }
1352
1353         if ((((queue & PTLRPC_NRS_QUEUE_REG) != 0) &&
1354             ((quantum_reg > LPROCFS_NRS_QUANTUM_MAX || quantum_reg <= 0))) ||
1355             (((queue & PTLRPC_NRS_QUEUE_HP) != 0) &&
1356             ((quantum_hp > LPROCFS_NRS_QUANTUM_MAX || quantum_hp <= 0))))
1357                 return -EINVAL;
1358
1359         /**
1360          * We change the values on regular and HP NRS heads separately, so that
1361          * we do not exit early from ptlrpc_nrs_policy_control() with an error
1362          * returned by nrs_policy_ctl_locked(), in cases where the user has not
1363          * started the policy on either the regular or HP NRS head; i.e. we are
1364          * ignoring -ENODEV within nrs_policy_ctl_locked(). -ENODEV is returned
1365          * only if the operation fails with -ENODEV on all heads that have been
1366          * specified by the command; if at least one operation succeeds,
1367          * success is returned.
1368          */
1369         if ((queue & PTLRPC_NRS_QUEUE_REG) != 0) {
1370                 rc = ptlrpc_nrs_policy_control(svc, PTLRPC_NRS_QUEUE_REG,
1371                                                orr_data->name,
1372                                                NRS_CTL_ORR_WR_QUANTUM, false,
1373                                                &quantum_reg);
1374                 if ((rc < 0 && rc != -ENODEV) ||
1375                     (rc == -ENODEV && queue == PTLRPC_NRS_QUEUE_REG))
1376                         return rc;
1377         }
1378
1379         if ((queue & PTLRPC_NRS_QUEUE_HP) != 0) {
1380                 rc2 = ptlrpc_nrs_policy_control(svc, PTLRPC_NRS_QUEUE_HP,
1381                                                 orr_data->name,
1382                                                 NRS_CTL_ORR_WR_QUANTUM, false,
1383                                                 &quantum_hp);
1384                 if ((rc2 < 0 && rc2 != -ENODEV) ||
1385                     (rc2 == -ENODEV && queue == PTLRPC_NRS_QUEUE_HP))
1386                         return rc2;
1387         }
1388
1389         return rc == -ENODEV && rc2 == -ENODEV ? -ENODEV : count;
1390 }
1391
1392 #define LPROCFS_NRS_OFF_NAME_REG                "reg_offset_type:"
1393 #define LPROCFS_NRS_OFF_NAME_HP                 "hp_offset_type:"
1394
1395 #define LPROCFS_NRS_OFF_NAME_PHYSICAL           "physical"
1396 #define LPROCFS_NRS_OFF_NAME_LOGICAL            "logical"
1397
1398 /**
1399  * Retrieves the offset type used by ORR/TRR policy instances on both the
1400  * regular and high-priority NRS head of a service, as long as a policy
1401  * instance is not in the ptlrpc_nrs_pol_state::NRS_POL_STATE_STOPPED state;
1402  * policy instances in this state are skipped later by nrs_orr_ctl().
1403  *
1404  * Offset type information is a (physical|logical) string, and output is
1405  * in YAML format.
1406  *
1407  * For example:
1408  *
1409  *      reg_offset_type:physical
1410  *      hp_offset_type:logical
1411  */
1412 static int ptlrpc_lprocfs_rd_nrs_orr_offset_type(char *page, char **start,
1413                                                  off_t off, int count, int *eof,
1414                                                  void *data)
1415 {
1416         struct nrs_lprocfs_orr_data *orr_data = data;
1417         struct ptlrpc_service       *svc = orr_data->svc;
1418         bool                         physical;
1419         int                          rc;
1420         int                          rc2 = 0;
1421
1422         /**
1423          * Perform two separate calls to this as only one of the NRS heads'
1424          * policies may be in the ptlrpc_nrs_pol_state::NRS_POL_STATE_STARTED
1425          * or ptlrpc_nrs_pol_state::NRS_POL_STATE_STOPPING state.
1426          */
1427         rc = ptlrpc_nrs_policy_control(svc, PTLRPC_NRS_QUEUE_REG,
1428                                        orr_data->name, NRS_CTL_ORR_RD_OFF_TYPE,
1429                                        true, &physical);
1430         if (rc == 0) {
1431                 *eof = 1;
1432                 rc2 = snprintf(page, count,
1433                                LPROCFS_NRS_OFF_NAME_REG"%s\n",
1434                                physical ? LPROCFS_NRS_OFF_NAME_PHYSICAL :
1435                                LPROCFS_NRS_OFF_NAME_LOGICAL);
1436                 /**
1437                  * Ignore -ENODEV as the regular NRS head's policy may be in the
1438                  * ptlrpc_nrs_pol_state::NRS_POL_STATE_STOPPED state.
1439                  */
1440         } else if (rc != -ENODEV) {
1441                 return rc;
1442         }
1443
1444         /**
1445          * We know the ost_io service which is the only one ORR/TRR policies are
1446          * compatible with, do have an HP NRS head, but it may be best to guard
1447          * against a possible change of this in the future.
1448          */
1449         if (!nrs_svc_has_hp(svc))
1450                 goto no_hp;
1451
1452         rc = ptlrpc_nrs_policy_control(svc, PTLRPC_NRS_QUEUE_HP,
1453                                        orr_data->name, NRS_CTL_ORR_RD_OFF_TYPE,
1454                                        true, &physical);
1455         if (rc == 0) {
1456                 *eof = 1;
1457                 rc2 += snprintf(page + rc2, count - rc2,
1458                                 LPROCFS_NRS_OFF_NAME_HP"%s\n",
1459                                 physical ? LPROCFS_NRS_OFF_NAME_PHYSICAL :
1460                                 LPROCFS_NRS_OFF_NAME_LOGICAL);
1461                 /**
1462                  * Ignore -ENODEV as the high priority NRS head's policy may be
1463                  * in the ptlrpc_nrs_pol_state::NRS_POL_STATE_STOPPED state.
1464                  */
1465         } else if (rc != -ENODEV) {
1466                 return rc;
1467         }
1468
1469 no_hp:
1470
1471         return rc2 ? : rc;
1472 }
1473
1474 /**
1475  * Max valid command string is the size of the labels, plus "physical" twice.
1476  * plus a separating ' '
1477  */
1478 #define LPROCFS_NRS_WR_OFF_TYPE_MAX_CMD                                        \
1479         sizeof(LPROCFS_NRS_OFF_NAME_REG LPROCFS_NRS_OFF_NAME_PHYSICAL " "      \
1480                LPROCFS_NRS_OFF_NAME_HP LPROCFS_NRS_OFF_NAME_PHYSICAL)
1481
1482 /**
1483  * Sets the type of offsets used to order RPCs in ORR/TRR policy instances. The
1484  * user can set offset type for the regular or high priority NRS head
1485  * separately by specifying each value, or both together in a single invocation.
1486  *
1487  * For example:
1488  *
1489  * lctl set_param ost.OSS.ost_io.nrs_orr_offset_type=
1490  * reg_offset_type:physical, to enable the ORR policy instance on the regular
1491  * NRS head of the ost_io service to use physical disk offset ordering.
1492  *
1493  * lctl set_param ost.OSS.ost_io.nrs_trr_offset_type=logical, to enable the TRR
1494  * policy instances on both the regular ang high priority NRS heads of the
1495  * ost_io service to use logical file offset ordering.
1496  *
1497  * policy instances in the ptlrpc_nrs_pol_state::NRS_POL_STATE_STOPPED state are
1498  * are skipped later by nrs_orr_ctl().
1499  */
1500 static int ptlrpc_lprocfs_wr_nrs_orr_offset_type(struct file *file,
1501                                                  const char *buffer,
1502                                                  unsigned long count,
1503                                                  void *data)
1504 {
1505         struct nrs_lprocfs_orr_data *orr_data = data;
1506         struct ptlrpc_service       *svc = orr_data->svc;
1507         enum ptlrpc_nrs_queue_type   queue = 0;
1508         char                         kernbuf[LPROCFS_NRS_WR_OFF_TYPE_MAX_CMD];
1509         char                        *val_reg;
1510         char                        *val_hp;
1511         bool                         physical_reg;
1512         bool                         physical_hp;
1513         size_t                       count_copy;
1514         int                          rc = 0;
1515         int                          rc2 = 0;
1516
1517         if (count > (sizeof(kernbuf) - 1))
1518                 return -EINVAL;
1519
1520         if (copy_from_user(kernbuf, buffer, count))
1521                 return -EFAULT;
1522
1523         kernbuf[count] = '\0';
1524
1525         count_copy = count;
1526
1527         /**
1528          * Check if the regular offset type has been specified
1529          */
1530         val_reg = lprocfs_find_named_value(kernbuf,
1531                                            LPROCFS_NRS_OFF_NAME_REG,
1532                                            &count_copy);
1533         if (val_reg != kernbuf)
1534                 queue |= PTLRPC_NRS_QUEUE_REG;
1535
1536         count_copy = count;
1537
1538         /**
1539          * Check if the high priority offset type has been specified
1540          */
1541         val_hp = lprocfs_find_named_value(kernbuf, LPROCFS_NRS_OFF_NAME_HP,
1542                                           &count_copy);
1543         if (val_hp != kernbuf) {
1544                 if (!nrs_svc_has_hp(svc))
1545                         return -ENODEV;
1546
1547                 queue |= PTLRPC_NRS_QUEUE_HP;
1548         }
1549
1550         /**
1551          * If none of the queues has been specified, there may be a valid
1552          * command string at the start of the buffer.
1553          */
1554         if (queue == 0) {
1555                 queue = PTLRPC_NRS_QUEUE_REG;
1556
1557                 if (nrs_svc_has_hp(svc))
1558                         queue |= PTLRPC_NRS_QUEUE_HP;
1559         }
1560
1561         if ((queue & PTLRPC_NRS_QUEUE_REG) != 0) {
1562                 if (strncmp(val_reg, LPROCFS_NRS_OFF_NAME_PHYSICAL,
1563                             sizeof(LPROCFS_NRS_OFF_NAME_PHYSICAL) - 1) == 0)
1564                         physical_reg = true;
1565                 else if (strncmp(val_reg, LPROCFS_NRS_OFF_NAME_LOGICAL,
1566                          sizeof(LPROCFS_NRS_OFF_NAME_LOGICAL) - 1) == 0)
1567                         physical_reg = false;
1568                 else
1569                         return -EINVAL;
1570         }
1571
1572         if ((queue & PTLRPC_NRS_QUEUE_HP) != 0) {
1573                 if (strncmp(val_hp, LPROCFS_NRS_OFF_NAME_PHYSICAL,
1574                             sizeof(LPROCFS_NRS_OFF_NAME_PHYSICAL) - 1) == 0)
1575                         physical_hp = true;
1576                 else if (strncmp(val_hp, LPROCFS_NRS_OFF_NAME_LOGICAL,
1577                                  sizeof(LPROCFS_NRS_OFF_NAME_LOGICAL) - 1) == 0)
1578                         physical_hp = false;
1579                 else
1580                         return -EINVAL;
1581         }
1582
1583         /**
1584          * We change the values on regular and HP NRS heads separately, so that
1585          * we do not exit early from ptlrpc_nrs_policy_control() with an error
1586          * returned by nrs_policy_ctl_locked(), in cases where the user has not
1587          * started the policy on either the regular or HP NRS head; i.e. we are
1588          * ignoring -ENODEV within nrs_policy_ctl_locked(). -ENODEV is returned
1589          * only if the operation fails with -ENODEV on all heads that have been
1590          * specified by the command; if at least one operation succeeds,
1591          * success is returned.
1592          */
1593         if ((queue & PTLRPC_NRS_QUEUE_REG) != 0) {
1594                 rc = ptlrpc_nrs_policy_control(svc, PTLRPC_NRS_QUEUE_REG,
1595                                                orr_data->name,
1596                                                NRS_CTL_ORR_WR_OFF_TYPE, false,
1597                                                &physical_reg);
1598                 if ((rc < 0 && rc != -ENODEV) ||
1599                     (rc == -ENODEV && queue == PTLRPC_NRS_QUEUE_REG))
1600                         return rc;
1601         }
1602
1603         if ((queue & PTLRPC_NRS_QUEUE_HP) != 0) {
1604                 rc2 = ptlrpc_nrs_policy_control(svc, PTLRPC_NRS_QUEUE_HP,
1605                                                 orr_data->name,
1606                                                 NRS_CTL_ORR_WR_OFF_TYPE, false,
1607                                                 &physical_hp);
1608                 if ((rc2 < 0 && rc2 != -ENODEV) ||
1609                     (rc2 == -ENODEV && queue == PTLRPC_NRS_QUEUE_HP))
1610                         return rc2;
1611         }
1612
1613         return rc == -ENODEV && rc2 == -ENODEV ? -ENODEV : count;
1614 }
1615
1616 #define NRS_LPROCFS_REQ_SUPP_NAME_REG           "reg_supported:"
1617 #define NRS_LPROCFS_REQ_SUPP_NAME_HP            "hp_supported:"
1618
1619 #define LPROCFS_NRS_SUPP_NAME_READS             "reads"
1620 #define LPROCFS_NRS_SUPP_NAME_WRITES            "writes"
1621 #define LPROCFS_NRS_SUPP_NAME_READWRITES        "reads_and_writes"
1622
1623 /**
1624  * Translates enum nrs_orr_supp values to a corresponding string.
1625  */
1626 static const char *nrs_orr_supp2str(enum nrs_orr_supp supp)
1627 {
1628         switch(supp) {
1629         default:
1630                 LBUG();
1631         case NOS_OST_READ:
1632                 return LPROCFS_NRS_SUPP_NAME_READS;
1633         case NOS_OST_WRITE:
1634                 return LPROCFS_NRS_SUPP_NAME_WRITES;
1635         case NOS_OST_RW:
1636                 return LPROCFS_NRS_SUPP_NAME_READWRITES;
1637         }
1638 }
1639
1640 /**
1641  * Translates strings to the corresponding enum nrs_orr_supp value
1642  */
1643 static enum nrs_orr_supp nrs_orr_str2supp(const char *val)
1644 {
1645         if (strncmp(val, LPROCFS_NRS_SUPP_NAME_READWRITES,
1646                     sizeof(LPROCFS_NRS_SUPP_NAME_READWRITES) - 1) == 0)
1647                 return NOS_OST_RW;
1648         else if (strncmp(val, LPROCFS_NRS_SUPP_NAME_READS,
1649                          sizeof(LPROCFS_NRS_SUPP_NAME_READS) - 1) == 0)
1650                 return NOS_OST_READ;
1651         else if (strncmp(val, LPROCFS_NRS_SUPP_NAME_WRITES,
1652                          sizeof(LPROCFS_NRS_SUPP_NAME_WRITES) - 1) == 0)
1653                 return NOS_OST_WRITE;
1654         else
1655                 return -EINVAL;
1656 }
1657
1658 /**
1659  * Retrieves the type of RPCs handled at the point of invocation by ORR/TRR
1660  * policy instances on both the regular and high-priority NRS head of a service,
1661  * as long as a policy instance is not in the
1662  * ptlrpc_nrs_pol_state::NRS_POL_STATE_STOPPED state; policy instances in this
1663  * state are skipped later by nrs_orr_ctl().
1664  *
1665  * Supported RPC type information is a (reads|writes|reads_and_writes) string,
1666  * and output is in YAML format.
1667  *
1668  * For example:
1669  *
1670  *      reg_supported:reads
1671  *      hp_supported:reads_and_writes
1672  */
1673 static int ptlrpc_lprocfs_rd_nrs_orr_supported(char *page, char **start,
1674                                                off_t off, int count, int *eof,
1675                                                void *data)
1676 {
1677         struct nrs_lprocfs_orr_data *orr_data = data;
1678         struct ptlrpc_service       *svc = orr_data->svc;
1679         enum nrs_orr_supp            supported;
1680         int                          rc;
1681         int                          rc2 = 0;
1682
1683         /**
1684          * Perform two separate calls to this as only one of the NRS heads'
1685          * policies may be in the ptlrpc_nrs_pol_state::NRS_POL_STATE_STARTED
1686          * or ptlrpc_nrs_pol_state::NRS_POL_STATE_STOPPING state.
1687          */
1688         rc = ptlrpc_nrs_policy_control(svc, PTLRPC_NRS_QUEUE_REG,
1689                                        orr_data->name,
1690                                        NRS_CTL_ORR_RD_SUPP_REQ, true,
1691                                        &supported);
1692
1693         if (rc == 0) {
1694                 *eof = 1;
1695                 rc2 = snprintf(page, count,
1696                                NRS_LPROCFS_REQ_SUPP_NAME_REG"%s\n",
1697                                nrs_orr_supp2str(supported));
1698                 /**
1699                  * Ignore -ENODEV as the regular NRS head's policy may be in the
1700                  * ptlrpc_nrs_pol_state::NRS_POL_STATE_STOPPED state.
1701                  */
1702         } else if (rc != -ENODEV) {
1703                 return rc;
1704         }
1705
1706         /**
1707          * We know the ost_io service which is the only one ORR/TRR policies are
1708          * compatible with, do have an HP NRS head, but it may be best to guard
1709          * against a possible change of this in the future.
1710          */
1711         if (!nrs_svc_has_hp(svc))
1712                 goto no_hp;
1713
1714         rc = ptlrpc_nrs_policy_control(svc, PTLRPC_NRS_QUEUE_HP,
1715                                        orr_data->name,
1716                                        NRS_CTL_ORR_RD_SUPP_REQ, true,
1717                                        &supported);
1718         if (rc == 0) {
1719                 *eof = 1;
1720                 rc2 += snprintf(page + rc2, count - rc2,
1721                                NRS_LPROCFS_REQ_SUPP_NAME_HP"%s\n",
1722                                nrs_orr_supp2str(supported));
1723                 /**
1724                  * Ignore -ENODEV as the high priority NRS head's policy may be
1725                  * in the ptlrpc_nrs_pol_state::NRS_POL_STATE_STOPPED state.
1726                  */
1727         } else if (rc != -ENODEV) {
1728                 return rc;
1729         }
1730
1731 no_hp:
1732
1733         return rc2 ? : rc;
1734 }
1735
1736 /**
1737  * Max valid command string is the size of the labels, plus "reads_and_writes"
1738  * twice, plus a separating ' '
1739  */
1740 #define LPROCFS_NRS_WR_REQ_SUPP_MAX_CMD                                        \
1741         sizeof(NRS_LPROCFS_REQ_SUPP_NAME_REG LPROCFS_NRS_SUPP_NAME_READWRITES  \
1742                NRS_LPROCFS_REQ_SUPP_NAME_HP LPROCFS_NRS_SUPP_NAME_READWRITES   \
1743                " ")
1744
1745 /**
1746  * Sets the type of RPCs handled by ORR/TRR policy instances. The user can
1747  * modify this setting for the regular or high priority NRS heads separately, or
1748  * both together in a single invocation.
1749  *
1750  * For example:
1751  *
1752  * lctl set_param ost.OSS.ost_io.nrs_orr_supported=
1753  * "reg_supported:reads", to enable the ORR policy instance on the regular NRS
1754  * head of the ost_io service to handle OST_READ RPCs.
1755  *
1756  * lctl set_param ost.OSS.ost_io.nrs_trr_supported=reads_and_writes, to enable
1757  * the TRR policy instances on both the regular ang high priority NRS heads of
1758  * the ost_io service to use handle OST_READ and OST_WRITE RPCs.
1759  *
1760  * policy instances in the ptlrpc_nrs_pol_state::NRS_POL_STATE_STOPPED state are
1761  * are skipped later by nrs_orr_ctl().
1762  */
1763 static int ptlrpc_lprocfs_wr_nrs_orr_supported(struct file *file,
1764                                                const char *buffer,
1765                                                unsigned long count, void *data)
1766 {
1767         struct nrs_lprocfs_orr_data *orr_data = data;
1768         struct ptlrpc_service       *svc = orr_data->svc;
1769         enum ptlrpc_nrs_queue_type   queue = 0;
1770         char                         kernbuf[LPROCFS_NRS_WR_REQ_SUPP_MAX_CMD];
1771         char                        *val_reg;
1772         char                        *val_hp;
1773         enum nrs_orr_supp            supp_reg;
1774         enum nrs_orr_supp            supp_hp;
1775         size_t                       count_copy;
1776         int                          rc = 0;
1777         int                          rc2 = 0;
1778
1779         if (count > (sizeof(kernbuf) - 1))
1780                 return -EINVAL;
1781
1782         if (copy_from_user(kernbuf, buffer, count))
1783                 return -EFAULT;
1784
1785         kernbuf[count] = '\0';
1786
1787         count_copy = count;
1788
1789         /**
1790          * Check if the regular supported requests setting has been specified
1791          */
1792         val_reg = lprocfs_find_named_value(kernbuf,
1793                                            NRS_LPROCFS_REQ_SUPP_NAME_REG,
1794                                            &count_copy);
1795         if (val_reg != kernbuf)
1796                 queue |= PTLRPC_NRS_QUEUE_REG;
1797
1798         count_copy = count;
1799
1800         /**
1801          * Check if the high priority supported requests setting has been
1802          * specified
1803          */
1804         val_hp = lprocfs_find_named_value(kernbuf, NRS_LPROCFS_REQ_SUPP_NAME_HP,
1805                                           &count_copy);
1806         if (val_hp != kernbuf) {
1807                 if (!nrs_svc_has_hp(svc))
1808                         return -ENODEV;
1809
1810                 queue |= PTLRPC_NRS_QUEUE_HP;
1811         }
1812
1813         /**
1814          * If none of the queues has been specified, there may be a valid
1815          * command string at the start of the buffer.
1816          */
1817         if (queue == 0) {
1818                 queue = PTLRPC_NRS_QUEUE_REG;
1819
1820                 if (nrs_svc_has_hp(svc))
1821                         queue |= PTLRPC_NRS_QUEUE_HP;
1822         }
1823
1824         if ((queue & PTLRPC_NRS_QUEUE_REG) != 0) {
1825                 supp_reg = nrs_orr_str2supp(val_reg);
1826                 if (supp_reg == -EINVAL)
1827                         return -EINVAL;
1828         }
1829
1830         if ((queue & PTLRPC_NRS_QUEUE_HP) != 0) {
1831                 supp_hp = nrs_orr_str2supp(val_hp);
1832                 if (supp_hp == -EINVAL)
1833                         return -EINVAL;
1834         }
1835
1836         /**
1837          * We change the values on regular and HP NRS heads separately, so that
1838          * we do not exit early from ptlrpc_nrs_policy_control() with an error
1839          * returned by nrs_policy_ctl_locked(), in cases where the user has not
1840          * started the policy on either the regular or HP NRS head; i.e. we are
1841          * ignoring -ENODEV within nrs_policy_ctl_locked(). -ENODEV is returned
1842          * only if the operation fails with -ENODEV on all heads that have been
1843          * specified by the command; if at least one operation succeeds,
1844          * success is returned.
1845          */
1846         if ((queue & PTLRPC_NRS_QUEUE_REG) != 0) {
1847                 rc = ptlrpc_nrs_policy_control(svc, PTLRPC_NRS_QUEUE_REG,
1848                                                orr_data->name,
1849                                                NRS_CTL_ORR_WR_SUPP_REQ, false,
1850                                                &supp_reg);
1851                 if ((rc < 0 && rc != -ENODEV) ||
1852                     (rc == -ENODEV && queue == PTLRPC_NRS_QUEUE_REG))
1853                         return rc;
1854         }
1855
1856         if ((queue & PTLRPC_NRS_QUEUE_HP) != 0) {
1857                 rc2 = ptlrpc_nrs_policy_control(svc, PTLRPC_NRS_QUEUE_HP,
1858                                                 orr_data->name,
1859                                                 NRS_CTL_ORR_WR_SUPP_REQ, false,
1860                                                 &supp_hp);
1861                 if ((rc2 < 0 && rc2 != -ENODEV) ||
1862                     (rc2 == -ENODEV && queue == PTLRPC_NRS_QUEUE_HP))
1863                         return rc2;
1864         }
1865
1866         return rc == -ENODEV && rc2 == -ENODEV ? -ENODEV : count;
1867 }
1868
1869 int nrs_orr_lprocfs_init(struct ptlrpc_service *svc)
1870 {
1871         int     rc;
1872         int     i;
1873
1874         struct lprocfs_vars nrs_orr_lprocfs_vars[] = {
1875                 { .name         = "nrs_orr_quantum",
1876                   .read_fptr    = ptlrpc_lprocfs_rd_nrs_orr_quantum,
1877                   .write_fptr   = ptlrpc_lprocfs_wr_nrs_orr_quantum },
1878                 { .name         = "nrs_orr_offset_type",
1879                   .read_fptr    = ptlrpc_lprocfs_rd_nrs_orr_offset_type,
1880                   .write_fptr   = ptlrpc_lprocfs_wr_nrs_orr_offset_type },
1881                 { .name         = "nrs_orr_supported",
1882                   .read_fptr    = ptlrpc_lprocfs_rd_nrs_orr_supported,
1883                   .write_fptr   = ptlrpc_lprocfs_wr_nrs_orr_supported },
1884                 { NULL }
1885         };
1886
1887         if (svc->srv_procroot == NULL)
1888                 return 0;
1889
1890         lprocfs_orr_data.svc = svc;
1891
1892         for (i = 0; i < ARRAY_SIZE(nrs_orr_lprocfs_vars); i++)
1893                 nrs_orr_lprocfs_vars[i].data = &lprocfs_orr_data;
1894
1895         rc = lprocfs_add_vars(svc->srv_procroot, nrs_orr_lprocfs_vars, NULL);
1896
1897         return rc;
1898 }
1899
1900 void nrs_orr_lprocfs_fini(struct ptlrpc_service *svc)
1901 {
1902         if (svc->srv_procroot == NULL)
1903                 return;
1904
1905         lprocfs_remove_proc_entry("nrs_orr_quantum", svc->srv_procroot);
1906         lprocfs_remove_proc_entry("nrs_orr_offset_type", svc->srv_procroot);
1907         lprocfs_remove_proc_entry("nrs_orr_supported", svc->srv_procroot);
1908 }
1909
1910 #endif /* LPROCFS */
1911
1912 static const struct ptlrpc_nrs_pol_ops nrs_orr_ops = {
1913         .op_policy_init         = nrs_orr_init,
1914         .op_policy_start        = nrs_orr_start,
1915         .op_policy_stop         = nrs_orr_stop,
1916         .op_policy_ctl          = nrs_orr_ctl,
1917         .op_res_get             = nrs_orr_res_get,
1918         .op_res_put             = nrs_orr_res_put,
1919         .op_req_get             = nrs_orr_req_get,
1920         .op_req_enqueue         = nrs_orr_req_add,
1921         .op_req_dequeue         = nrs_orr_req_del,
1922         .op_req_stop            = nrs_orr_req_stop,
1923 #ifdef LPROCFS
1924         .op_lprocfs_init        = nrs_orr_lprocfs_init,
1925         .op_lprocfs_fini        = nrs_orr_lprocfs_fini,
1926 #endif
1927 };
1928
1929 struct ptlrpc_nrs_pol_conf nrs_conf_orr = {
1930         .nc_name                = NRS_POL_NAME_ORR,
1931         .nc_ops                 = &nrs_orr_ops,
1932         .nc_compat              = nrs_policy_compat_one,
1933         .nc_compat_svc_name     = "ost_io",
1934 };
1935
1936 /**
1937  * TRR, Target-based Round Robin policy
1938  *
1939  * TRR reuses much of the functions and data structures of ORR
1940  */
1941
1942 #ifdef LPROCFS
1943
1944 int nrs_trr_lprocfs_init(struct ptlrpc_service *svc)
1945 {
1946         int     rc;
1947         int     i;
1948
1949         struct lprocfs_vars nrs_trr_lprocfs_vars[] = {
1950                 { .name         = "nrs_trr_quantum",
1951                   .read_fptr    = ptlrpc_lprocfs_rd_nrs_orr_quantum,
1952                   .write_fptr   = ptlrpc_lprocfs_wr_nrs_orr_quantum },
1953                 { .name         = "nrs_trr_offset_type",
1954                   .read_fptr    = ptlrpc_lprocfs_rd_nrs_orr_offset_type,
1955                   .write_fptr   = ptlrpc_lprocfs_wr_nrs_orr_offset_type },
1956                 { .name         = "nrs_trr_supported",
1957                   .read_fptr    = ptlrpc_lprocfs_rd_nrs_orr_supported,
1958                   .write_fptr   = ptlrpc_lprocfs_wr_nrs_orr_supported },
1959                 { NULL }
1960         };
1961
1962         if (svc->srv_procroot == NULL)
1963                 return 0;
1964
1965         lprocfs_trr_data.svc = svc;
1966
1967         for (i = 0; i < ARRAY_SIZE(nrs_trr_lprocfs_vars); i++)
1968                 nrs_trr_lprocfs_vars[i].data = &lprocfs_trr_data;
1969
1970         rc = lprocfs_add_vars(svc->srv_procroot, nrs_trr_lprocfs_vars, NULL);
1971
1972         return rc;
1973 }
1974
1975 void nrs_trr_lprocfs_fini(struct ptlrpc_service *svc)
1976 {
1977         if (svc->srv_procroot == NULL)
1978                 return;
1979
1980         lprocfs_remove_proc_entry("nrs_trr_quantum", svc->srv_procroot);
1981         lprocfs_remove_proc_entry("nrs_trr_offset_type", svc->srv_procroot);
1982         lprocfs_remove_proc_entry("nrs_trr_supported", svc->srv_procroot);
1983 }
1984
1985 #endif /* LPROCFS */
1986
1987 /**
1988  * Reuse much of the ORR functionality for TRR.
1989  */
1990 static const struct ptlrpc_nrs_pol_ops nrs_trr_ops = {
1991         .op_policy_init         = nrs_orr_init,
1992         .op_policy_start        = nrs_orr_start,
1993         .op_policy_stop         = nrs_orr_stop,
1994         .op_policy_ctl          = nrs_orr_ctl,
1995         .op_res_get             = nrs_orr_res_get,
1996         .op_res_put             = nrs_orr_res_put,
1997         .op_req_get             = nrs_orr_req_get,
1998         .op_req_enqueue         = nrs_orr_req_add,
1999         .op_req_dequeue         = nrs_orr_req_del,
2000         .op_req_stop            = nrs_orr_req_stop,
2001 #ifdef LPROCFS
2002         .op_lprocfs_init        = nrs_trr_lprocfs_init,
2003         .op_lprocfs_fini        = nrs_trr_lprocfs_fini,
2004 #endif
2005 };
2006
2007 struct ptlrpc_nrs_pol_conf nrs_conf_trr = {
2008         .nc_name                = NRS_POL_NAME_TRR,
2009         .nc_ops                 = &nrs_trr_ops,
2010         .nc_compat              = nrs_policy_compat_one,
2011         .nc_compat_svc_name     = "ost_io",
2012 };
2013
2014 /** @} ORR/TRR policy */
2015
2016 /** @} nrs */
2017
2018 #endif /* HAVE_SERVER_SUPPORT */