lustre/ptlrpc/nrs_tbf.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 DataDirect Networks, Inc.
  24  *
  25  * Copyright (c) 2014, 2016, Intel Corporation.
  26  */
  27 /*
  28  * lustre/ptlrpc/nrs_tbf.c
  29  *
  30  * Network Request Scheduler (NRS) Token Bucket Filter(TBF) policy
  31  *
  32  */
  33
  34 #ifdef HAVE_SERVER_SUPPORT
  35
  36 /**
  37  * \addtogoup nrs
  38  * @{
  39  */
  40
  41 #define DEBUG_SUBSYSTEM S_RPC
  42 #include <obd_support.h>
  43 #include <obd_class.h>
  44 #include <libcfs/libcfs.h>
  45 #include "ptlrpc_internal.h"
  46
  47 /**
  48  * \name tbf
  49  *
  50  * Token Bucket Filter over client NIDs
  51  *
  52  * @{
  53  */
  54
  55 #define NRS_POL_NAME_TBF        "tbf"
  56
  57 static int tbf_jobid_cache_size = 8192;
  58 module_param(tbf_jobid_cache_size, int, 0644);
  59 MODULE_PARM_DESC(tbf_jobid_cache_size, "The size of jobid cache");
  60
  61 static int tbf_rate = 10000;
  62 module_param(tbf_rate, int, 0644);
  63 MODULE_PARM_DESC(tbf_rate, "Default rate limit in RPCs/s");
  64
  65 static int tbf_depth = 3;
  66 module_param(tbf_depth, int, 0644);
  67 MODULE_PARM_DESC(tbf_depth, "How many tokens that a client can save up");
  68
  69 static enum hrtimer_restart nrs_tbf_timer_cb(struct hrtimer *timer)
  70 {
  71         struct nrs_tbf_head *head = container_of(timer, struct nrs_tbf_head,
  72                                                  th_timer);
  73         struct ptlrpc_nrs   *nrs = head->th_res.res_policy->pol_nrs;
  74         struct ptlrpc_service_part *svcpt = nrs->nrs_svcpt;
  75
  76         nrs->nrs_throttling = 0;
  77         wake_up(&svcpt->scp_waitq);
  78
  79         return HRTIMER_NORESTART;
  80 }
  81
  82 #define NRS_TBF_DEFAULT_RULE "default"
  83
  84 static void nrs_tbf_rule_fini(struct nrs_tbf_rule *rule)
  85 {
  86         LASSERT(atomic_read(&rule->tr_ref) == 0);
  87         LASSERT(list_empty(&rule->tr_cli_list));
  88         LASSERT(list_empty(&rule->tr_linkage));
  89
  90         rule->tr_head->th_ops->o_rule_fini(rule);
  91         OBD_FREE_PTR(rule);
  92 }
  93
  94 /**
  95  * Decreases the rule's usage reference count, and stops the rule in case it
  96  * was already stopping and have no more outstanding usage references (which
  97  * indicates it has no more queued or started requests, and can be safely
  98  * stopped).
  99  */
 100 static void nrs_tbf_rule_put(struct nrs_tbf_rule *rule)
 101 {
 102         if (atomic_dec_and_test(&rule->tr_ref))
 103                 nrs_tbf_rule_fini(rule);
 104 }
 105
 106 /**
 107  * Increases the rule's usage reference count.
 108  */
 109 static inline void nrs_tbf_rule_get(struct nrs_tbf_rule *rule)
 110 {
 111         atomic_inc(&rule->tr_ref);
 112 }
 113
 114 static void
 115 nrs_tbf_cli_rule_put(struct nrs_tbf_client *cli)
 116 {
 117         LASSERT(!list_empty(&cli->tc_linkage));
 118         LASSERT(cli->tc_rule);
 119         spin_lock(&cli->tc_rule->tr_rule_lock);
 120         list_del_init(&cli->tc_linkage);
 121         spin_unlock(&cli->tc_rule->tr_rule_lock);
 122         nrs_tbf_rule_put(cli->tc_rule);
 123         cli->tc_rule = NULL;
 124 }
 125
 126 static void
 127 nrs_tbf_cli_reset_value(struct nrs_tbf_head *head,
 128                         struct nrs_tbf_client *cli)
 129
 130 {
 131         struct nrs_tbf_rule *rule = cli->tc_rule;
 132
 133         cli->tc_rpc_rate = rule->tr_rpc_rate;
 134         cli->tc_nsecs = rule->tr_nsecs;
 135         cli->tc_depth = rule->tr_depth;
 136         cli->tc_ntoken = rule->tr_depth;
 137         cli->tc_check_time = ktime_to_ns(ktime_get());
 138         cli->tc_rule_sequence = atomic_read(&head->th_rule_sequence);
 139         cli->tc_rule_generation = rule->tr_generation;
 140
 141         if (cli->tc_in_heap)
 142                 cfs_binheap_relocate(head->th_binheap,
 143                                      &cli->tc_node);
 144 }
 145
 146 static void
 147 nrs_tbf_cli_reset(struct nrs_tbf_head *head,
 148                   struct nrs_tbf_rule *rule,
 149                   struct nrs_tbf_client *cli)
 150 {
 151         spin_lock(&cli->tc_rule_lock);
 152         if (cli->tc_rule != NULL && !list_empty(&cli->tc_linkage)) {
 153                 LASSERT(rule != cli->tc_rule);
 154                 nrs_tbf_cli_rule_put(cli);
 155         }
 156         LASSERT(cli->tc_rule == NULL);
 157         LASSERT(list_empty(&cli->tc_linkage));
 158         /* Rule's ref is added before called */
 159         cli->tc_rule = rule;
 160         spin_lock(&rule->tr_rule_lock);
 161         list_add_tail(&cli->tc_linkage, &rule->tr_cli_list);
 162         spin_unlock(&rule->tr_rule_lock);
 163         spin_unlock(&cli->tc_rule_lock);
 164         nrs_tbf_cli_reset_value(head, cli);
 165 }
 166
 167 static int
 168 nrs_tbf_rule_dump(struct nrs_tbf_rule *rule, struct seq_file *m)
 169 {
 170         return rule->tr_head->th_ops->o_rule_dump(rule, m);
 171 }
 172
 173 static int
 174 nrs_tbf_rule_dump_all(struct nrs_tbf_head *head, struct seq_file *m)
 175 {
 176         struct nrs_tbf_rule *rule;
 177         int rc = 0;
 178
 179         LASSERT(head != NULL);
 180         spin_lock(&head->th_rule_lock);
 181         /* List the rules from newest to oldest */
 182         list_for_each_entry(rule, &head->th_list, tr_linkage) {
 183                 LASSERT((rule->tr_flags & NTRS_STOPPING) == 0);
 184                 rc = nrs_tbf_rule_dump(rule, m);
 185                 if (rc) {
 186                         rc = -ENOSPC;
 187                         break;
 188                 }
 189         }
 190         spin_unlock(&head->th_rule_lock);
 191
 192         return rc;
 193 }
 194
 195 static struct nrs_tbf_rule *
 196 nrs_tbf_rule_find_nolock(struct nrs_tbf_head *head,
 197                          const char *name)
 198 {
 199         struct nrs_tbf_rule *rule;
 200
 201         LASSERT(head != NULL);
 202         list_for_each_entry(rule, &head->th_list, tr_linkage) {
 203                 LASSERT((rule->tr_flags & NTRS_STOPPING) == 0);
 204                 if (strcmp(rule->tr_name, name) == 0) {
 205                         nrs_tbf_rule_get(rule);
 206                         return rule;
 207                 }
 208         }
 209         return NULL;
 210 }
 211
 212 static struct nrs_tbf_rule *
 213 nrs_tbf_rule_find(struct nrs_tbf_head *head,
 214                   const char *name)
 215 {
 216         struct nrs_tbf_rule *rule;
 217
 218         LASSERT(head != NULL);
 219         spin_lock(&head->th_rule_lock);
 220         rule = nrs_tbf_rule_find_nolock(head, name);
 221         spin_unlock(&head->th_rule_lock);
 222         return rule;
 223 }
 224
 225 static struct nrs_tbf_rule *
 226 nrs_tbf_rule_match(struct nrs_tbf_head *head,
 227                    struct nrs_tbf_client *cli)
 228 {
 229         struct nrs_tbf_rule *rule = NULL;
 230         struct nrs_tbf_rule *tmp_rule;
 231
 232         spin_lock(&head->th_rule_lock);
 233         /* Match the newest rule in the list */
 234         list_for_each_entry(tmp_rule, &head->th_list, tr_linkage) {
 235                 LASSERT((tmp_rule->tr_flags & NTRS_STOPPING) == 0);
 236                 if (head->th_ops->o_rule_match(tmp_rule, cli)) {
 237                         rule = tmp_rule;
 238                         break;
 239                 }
 240         }
 241
 242         if (rule == NULL)
 243                 rule = head->th_rule;
 244
 245         nrs_tbf_rule_get(rule);
 246         spin_unlock(&head->th_rule_lock);
 247         return rule;
 248 }
 249
 250 static void
 251 nrs_tbf_cli_init(struct nrs_tbf_head *head,
 252                  struct nrs_tbf_client *cli,
 253                  struct ptlrpc_request *req)
 254 {
 255         struct nrs_tbf_rule *rule;
 256
 257         memset(cli, 0, sizeof(*cli));
 258         cli->tc_in_heap = false;
 259         head->th_ops->o_cli_init(cli, req);
 260         INIT_LIST_HEAD(&cli->tc_list);
 261         INIT_LIST_HEAD(&cli->tc_linkage);
 262         spin_lock_init(&cli->tc_rule_lock);
 263         atomic_set(&cli->tc_ref, 1);
 264         rule = nrs_tbf_rule_match(head, cli);
 265         nrs_tbf_cli_reset(head, rule, cli);
 266 }
 267
 268 static void
 269 nrs_tbf_cli_fini(struct nrs_tbf_client *cli)
 270 {
 271         LASSERT(list_empty(&cli->tc_list));
 272         LASSERT(!cli->tc_in_heap);
 273         LASSERT(atomic_read(&cli->tc_ref) == 0);
 274         spin_lock(&cli->tc_rule_lock);
 275         nrs_tbf_cli_rule_put(cli);
 276         spin_unlock(&cli->tc_rule_lock);
 277         OBD_FREE_PTR(cli);
 278 }
 279
 280 static int
 281 nrs_tbf_rule_start(struct ptlrpc_nrs_policy *policy,
 282                    struct nrs_tbf_head *head,
 283                    struct nrs_tbf_cmd *start)
 284 {
 285         struct nrs_tbf_rule     *rule;
 286         struct nrs_tbf_rule     *tmp_rule;
 287         struct nrs_tbf_rule     *next_rule;
 288         char                    *next_name = start->u.tc_start.ts_next_name;
 289         int                      rc;
 290
 291         rule = nrs_tbf_rule_find(head, start->tc_name);
 292         if (rule) {
 293                 nrs_tbf_rule_put(rule);
 294                 return -EEXIST;
 295         }
 296
 297         OBD_CPT_ALLOC_PTR(rule, nrs_pol2cptab(policy), nrs_pol2cptid(policy));
 298         if (rule == NULL)
 299                 return -ENOMEM;
 300
 301         memcpy(rule->tr_name, start->tc_name, strlen(start->tc_name));
 302         rule->tr_rpc_rate = start->u.tc_start.ts_rpc_rate;
 303         rule->tr_nsecs = NSEC_PER_SEC;
 304         do_div(rule->tr_nsecs, rule->tr_rpc_rate);
 305         rule->tr_depth = tbf_depth;
 306         atomic_set(&rule->tr_ref, 1);
 307         INIT_LIST_HEAD(&rule->tr_cli_list);
 308         INIT_LIST_HEAD(&rule->tr_nids);
 309         INIT_LIST_HEAD(&rule->tr_linkage);
 310         spin_lock_init(&rule->tr_rule_lock);
 311         rule->tr_head = head;
 312
 313         rc = head->th_ops->o_rule_init(policy, rule, start);
 314         if (rc) {
 315                 OBD_FREE_PTR(rule);
 316                 return rc;
 317         }
 318
 319         /* Add as the newest rule */
 320         spin_lock(&head->th_rule_lock);
 321         tmp_rule = nrs_tbf_rule_find_nolock(head, start->tc_name);
 322         if (tmp_rule) {
 323                 spin_unlock(&head->th_rule_lock);
 324                 nrs_tbf_rule_put(tmp_rule);
 325                 nrs_tbf_rule_put(rule);
 326                 return -EEXIST;
 327         }
 328
 329         if (next_name) {
 330                 next_rule = nrs_tbf_rule_find_nolock(head, next_name);
 331                 if (!next_rule) {
 332                         spin_unlock(&head->th_rule_lock);
 333                         nrs_tbf_rule_put(rule);
 334                         return -ENOENT;
 335                 }
 336
 337                 list_add(&rule->tr_linkage, next_rule->tr_linkage.prev);
 338                 nrs_tbf_rule_put(next_rule);
 339         } else {
 340                 /* Add on the top of the rule list */
 341                 list_add(&rule->tr_linkage, &head->th_list);
 342         }
 343         spin_unlock(&head->th_rule_lock);
 344         atomic_inc(&head->th_rule_sequence);
 345         if (start->u.tc_start.ts_rule_flags & NTRS_DEFAULT) {
 346                 rule->tr_flags |= NTRS_DEFAULT;
 347                 LASSERT(head->th_rule == NULL);
 348                 head->th_rule = rule;
 349         }
 350
 351         return 0;
 352 }
 353
 354 /**
 355  * Change the rank of a rule in the rule list
 356  *
 357  * The matched rule will be moved to the position right before another
 358  * given rule.
 359  *
 360  * \param[in] policy    the policy instance
 361  * \param[in] head      the TBF policy instance
 362  * \param[in] name      the rule name to be moved
 363  * \param[in] next_name the rule name before which the matched rule will be
 364  *                      moved
 365  *
 366  */
 367 static int
 368 nrs_tbf_rule_change_rank(struct ptlrpc_nrs_policy *policy,
 369                          struct nrs_tbf_head *head,
 370                          char *name,
 371                          char *next_name)
 372 {
 373         struct nrs_tbf_rule     *rule = NULL;
 374         struct nrs_tbf_rule     *next_rule = NULL;
 375         int                      rc = 0;
 376
 377         LASSERT(head != NULL);
 378
 379         spin_lock(&head->th_rule_lock);
 380         rule = nrs_tbf_rule_find_nolock(head, name);
 381         if (!rule)
 382                 GOTO(out, rc = -ENOENT);
 383
 384         if (strcmp(name, next_name) == 0)
 385                 GOTO(out_put, rc);
 386
 387         next_rule = nrs_tbf_rule_find_nolock(head, next_name);
 388         if (!next_rule)
 389                 GOTO(out_put, rc = -ENOENT);
 390
 391         list_move(&rule->tr_linkage, next_rule->tr_linkage.prev);
 392         nrs_tbf_rule_put(next_rule);
 393 out_put:
 394         nrs_tbf_rule_put(rule);
 395 out:
 396         spin_unlock(&head->th_rule_lock);
 397         return rc;
 398 }
 399
 400 static int
 401 nrs_tbf_rule_change_rate(struct ptlrpc_nrs_policy *policy,
 402                          struct nrs_tbf_head *head,
 403                          char *name,
 404                          __u64 rate)
 405 {
 406         struct nrs_tbf_rule *rule;
 407
 408         assert_spin_locked(&policy->pol_nrs->nrs_lock);
 409
 410         rule = nrs_tbf_rule_find(head, name);
 411         if (rule == NULL)
 412                 return -ENOENT;
 413
 414         rule->tr_rpc_rate = rate;
 415         rule->tr_nsecs = NSEC_PER_SEC;
 416         do_div(rule->tr_nsecs, rule->tr_rpc_rate);
 417         rule->tr_generation++;
 418         nrs_tbf_rule_put(rule);
 419
 420         return 0;
 421 }
 422
 423 static int
 424 nrs_tbf_rule_change(struct ptlrpc_nrs_policy *policy,
 425                     struct nrs_tbf_head *head,
 426                     struct nrs_tbf_cmd *change)
 427 {
 428         __u64    rate = change->u.tc_change.tc_rpc_rate;
 429         char    *next_name = change->u.tc_change.tc_next_name;
 430         int      rc;
 431
 432         if (rate != 0) {
 433                 rc = nrs_tbf_rule_change_rate(policy, head, change->tc_name,
 434                                               rate);
 435                 if (rc)
 436                         return rc;
 437         }
 438
 439         if (next_name) {
 440                 rc = nrs_tbf_rule_change_rank(policy, head, change->tc_name,
 441                                               next_name);
 442                 if (rc)
 443                         return rc;
 444         }
 445
 446         return 0;
 447 }
 448
 449 static int
 450 nrs_tbf_rule_stop(struct ptlrpc_nrs_policy *policy,
 451                   struct nrs_tbf_head *head,
 452                   struct nrs_tbf_cmd *stop)
 453 {
 454         struct nrs_tbf_rule *rule;
 455
 456         assert_spin_locked(&policy->pol_nrs->nrs_lock);
 457
 458         if (strcmp(stop->tc_name, NRS_TBF_DEFAULT_RULE) == 0)
 459                 return -EPERM;
 460
 461         rule = nrs_tbf_rule_find(head, stop->tc_name);
 462         if (rule == NULL)
 463                 return -ENOENT;
 464
 465         list_del_init(&rule->tr_linkage);
 466         rule->tr_flags |= NTRS_STOPPING;
 467         nrs_tbf_rule_put(rule);
 468         nrs_tbf_rule_put(rule);
 469
 470         return 0;
 471 }
 472
 473 static int
 474 nrs_tbf_command(struct ptlrpc_nrs_policy *policy,
 475                 struct nrs_tbf_head *head,
 476                 struct nrs_tbf_cmd *cmd)
 477 {
 478         int rc;
 479
 480         assert_spin_locked(&policy->pol_nrs->nrs_lock);
 481
 482         switch (cmd->tc_cmd) {
 483         case NRS_CTL_TBF_START_RULE:
 484                 if (cmd->u.tc_start.ts_valid_type != head->th_type_flag)
 485                         return -EINVAL;
 486
 487                 spin_unlock(&policy->pol_nrs->nrs_lock);
 488                 rc = nrs_tbf_rule_start(policy, head, cmd);
 489                 spin_lock(&policy->pol_nrs->nrs_lock);
 490                 return rc;
 491         case NRS_CTL_TBF_CHANGE_RULE:
 492                 rc = nrs_tbf_rule_change(policy, head, cmd);
 493                 return rc;
 494         case NRS_CTL_TBF_STOP_RULE:
 495                 rc = nrs_tbf_rule_stop(policy, head, cmd);
 496                 /* Take it as a success, if not exists at all */
 497                 return rc == -ENOENT ? 0 : rc;
 498         default:
 499                 return -EFAULT;
 500         }
 501 }
 502
 503 /**
 504  * Binary heap predicate.
 505  *
 506  * \param[in] e1 the first binheap node to compare
 507  * \param[in] e2 the second binheap node to compare
 508  *
 509  * \retval 0 e1 > e2
 510  * \retval 1 e1 < e2
 511  */
 512 static int
 513 tbf_cli_compare(struct cfs_binheap_node *e1, struct cfs_binheap_node *e2)
 514 {
 515         struct nrs_tbf_client *cli1;
 516         struct nrs_tbf_client *cli2;
 517
 518         cli1 = container_of(e1, struct nrs_tbf_client, tc_node);
 519         cli2 = container_of(e2, struct nrs_tbf_client, tc_node);
 520
 521         if (cli1->tc_check_time + cli1->tc_nsecs <
 522             cli2->tc_check_time + cli2->tc_nsecs)
 523                 return 1;
 524         else if (cli1->tc_check_time + cli1->tc_nsecs >
 525                  cli2->tc_check_time + cli2->tc_nsecs)
 526                 return 0;
 527
 528         if (cli1->tc_check_time < cli2->tc_check_time)
 529                 return 1;
 530         else if (cli1->tc_check_time > cli2->tc_check_time)
 531                 return 0;
 532
 533         /* Maybe need more comparasion, e.g. request number in the rules */
 534         return 1;
 535 }
 536
 537 /**
 538  * TBF binary heap operations
 539  */
 540 static struct cfs_binheap_ops nrs_tbf_heap_ops = {
 541         .hop_enter      = NULL,
 542         .hop_exit       = NULL,
 543         .hop_compare    = tbf_cli_compare,
 544 };
 545
 546 static unsigned nrs_tbf_jobid_hop_hash(struct cfs_hash *hs, const void *key,
 547                                   unsigned mask)
 548 {
 549         return cfs_hash_djb2_hash(key, strlen(key), mask);
 550 }
 551
 552 static int nrs_tbf_jobid_hop_keycmp(const void *key, struct hlist_node *hnode)
 553 {
 554         struct nrs_tbf_client *cli = hlist_entry(hnode,
 555                                                      struct nrs_tbf_client,
 556                                                      tc_hnode);
 557
 558         return (strcmp(cli->tc_jobid, key) == 0);
 559 }
 560
 561 static void *nrs_tbf_jobid_hop_key(struct hlist_node *hnode)
 562 {
 563         struct nrs_tbf_client *cli = hlist_entry(hnode,
 564                                                      struct nrs_tbf_client,
 565                                                      tc_hnode);
 566
 567         return cli->tc_jobid;
 568 }
 569
 570 static void *nrs_tbf_jobid_hop_object(struct hlist_node *hnode)
 571 {
 572         return hlist_entry(hnode, struct nrs_tbf_client, tc_hnode);
 573 }
 574
 575 static void nrs_tbf_jobid_hop_get(struct cfs_hash *hs, struct hlist_node *hnode)
 576 {
 577         struct nrs_tbf_client *cli = hlist_entry(hnode,
 578                                                      struct nrs_tbf_client,
 579                                                      tc_hnode);
 580
 581         atomic_inc(&cli->tc_ref);
 582 }
 583
 584 static void nrs_tbf_jobid_hop_put(struct cfs_hash *hs, struct hlist_node *hnode)
 585 {
 586         struct nrs_tbf_client *cli = hlist_entry(hnode,
 587                                                      struct nrs_tbf_client,
 588                                                      tc_hnode);
 589
 590         atomic_dec(&cli->tc_ref);
 591 }
 592
 593 static void
 594 nrs_tbf_jobid_hop_exit(struct cfs_hash *hs, struct hlist_node *hnode)
 595
 596 {
 597         struct nrs_tbf_client *cli = hlist_entry(hnode,
 598                                                  struct nrs_tbf_client,
 599                                                  tc_hnode);
 600
 601         LASSERT(atomic_read(&cli->tc_ref) == 0);
 602         nrs_tbf_cli_fini(cli);
 603 }
 604
 605 static struct cfs_hash_ops nrs_tbf_jobid_hash_ops = {
 606         .hs_hash        = nrs_tbf_jobid_hop_hash,
 607         .hs_keycmp      = nrs_tbf_jobid_hop_keycmp,
 608         .hs_key         = nrs_tbf_jobid_hop_key,
 609         .hs_object      = nrs_tbf_jobid_hop_object,
 610         .hs_get         = nrs_tbf_jobid_hop_get,
 611         .hs_put         = nrs_tbf_jobid_hop_put,
 612         .hs_put_locked  = nrs_tbf_jobid_hop_put,
 613         .hs_exit        = nrs_tbf_jobid_hop_exit,
 614 };
 615
 616 #define NRS_TBF_JOBID_HASH_FLAGS (CFS_HASH_SPIN_BKTLOCK | \
 617                                   CFS_HASH_NO_ITEMREF | \
 618                                   CFS_HASH_DEPTH)
 619
 620 static struct nrs_tbf_client *
 621 nrs_tbf_jobid_hash_lookup(struct cfs_hash *hs,
 622                           struct cfs_hash_bd *bd,
 623                           const char *jobid)
 624 {
 625         struct hlist_node *hnode;
 626         struct nrs_tbf_client *cli;
 627
 628         /* cfs_hash_bd_peek_locked is a somehow "internal" function
 629          * of cfs_hash, it doesn't add refcount on object. */
 630         hnode = cfs_hash_bd_peek_locked(hs, bd, (void *)jobid);
 631         if (hnode == NULL)
 632                 return NULL;
 633
 634         cfs_hash_get(hs, hnode);
 635         cli = container_of0(hnode, struct nrs_tbf_client, tc_hnode);
 636         if (!list_empty(&cli->tc_lru))
 637                 list_del_init(&cli->tc_lru);
 638         return cli;
 639 }
 640
 641 #define NRS_TBF_JOBID_NULL ""
 642
 643 static struct nrs_tbf_client *
 644 nrs_tbf_jobid_cli_find(struct nrs_tbf_head *head,
 645                        struct ptlrpc_request *req)
 646 {
 647         const char              *jobid;
 648         struct nrs_tbf_client   *cli;
 649         struct cfs_hash         *hs = head->th_cli_hash;
 650         struct cfs_hash_bd               bd;
 651
 652         jobid = lustre_msg_get_jobid(req->rq_reqmsg);
 653         if (jobid == NULL)
 654                 jobid = NRS_TBF_JOBID_NULL;
 655         cfs_hash_bd_get_and_lock(hs, (void *)jobid, &bd, 1);
 656         cli = nrs_tbf_jobid_hash_lookup(hs, &bd, jobid);
 657         cfs_hash_bd_unlock(hs, &bd, 1);
 658
 659         return cli;
 660 }
 661
 662 static struct nrs_tbf_client *
 663 nrs_tbf_jobid_cli_findadd(struct nrs_tbf_head *head,
 664                           struct nrs_tbf_client *cli)
 665 {
 666         const char              *jobid;
 667         struct nrs_tbf_client   *ret;
 668         struct cfs_hash         *hs = head->th_cli_hash;
 669         struct cfs_hash_bd               bd;
 670
 671         jobid = cli->tc_jobid;
 672         cfs_hash_bd_get_and_lock(hs, (void *)jobid, &bd, 1);
 673         ret = nrs_tbf_jobid_hash_lookup(hs, &bd, jobid);
 674         if (ret == NULL) {
 675                 cfs_hash_bd_add_locked(hs, &bd, &cli->tc_hnode);
 676                 ret = cli;
 677         }
 678         cfs_hash_bd_unlock(hs, &bd, 1);
 679
 680         return ret;
 681 }
 682
 683 static void
 684 nrs_tbf_jobid_cli_put(struct nrs_tbf_head *head,
 685                       struct nrs_tbf_client *cli)
 686 {
 687         struct cfs_hash_bd               bd;
 688         struct cfs_hash         *hs = head->th_cli_hash;
 689         struct nrs_tbf_bucket   *bkt;
 690         int                      hw;
 691         struct list_head        zombies;
 692
 693         INIT_LIST_HEAD(&zombies);
 694         cfs_hash_bd_get(hs, &cli->tc_jobid, &bd);
 695         bkt = cfs_hash_bd_extra_get(hs, &bd);
 696         if (!cfs_hash_bd_dec_and_lock(hs, &bd, &cli->tc_ref))
 697                 return;
 698         LASSERT(list_empty(&cli->tc_lru));
 699         list_add_tail(&cli->tc_lru, &bkt->ntb_lru);
 700
 701         /*
 702          * Check and purge the LRU, there is at least one client in the LRU.
 703          */
 704         hw = tbf_jobid_cache_size >>
 705              (hs->hs_cur_bits - hs->hs_bkt_bits);
 706         while (cfs_hash_bd_count_get(&bd) > hw) {
 707                 if (unlikely(list_empty(&bkt->ntb_lru)))
 708                         break;
 709                 cli = list_entry(bkt->ntb_lru.next,
 710                                      struct nrs_tbf_client,
 711                                      tc_lru);
 712                 LASSERT(atomic_read(&cli->tc_ref) == 0);
 713                 cfs_hash_bd_del_locked(hs, &bd, &cli->tc_hnode);
 714                 list_move(&cli->tc_lru, &zombies);
 715         }
 716         cfs_hash_bd_unlock(head->th_cli_hash, &bd, 1);
 717
 718         while (!list_empty(&zombies)) {
 719                 cli = container_of0(zombies.next,
 720                                     struct nrs_tbf_client, tc_lru);
 721                 list_del_init(&cli->tc_lru);
 722                 nrs_tbf_cli_fini(cli);
 723         }
 724 }
 725
 726 static void
 727 nrs_tbf_jobid_cli_init(struct nrs_tbf_client *cli,
 728                        struct ptlrpc_request *req)
 729 {
 730         char *jobid = lustre_msg_get_jobid(req->rq_reqmsg);
 731
 732         if (jobid == NULL)
 733                 jobid = NRS_TBF_JOBID_NULL;
 734         LASSERT(strlen(jobid) < LUSTRE_JOBID_SIZE);
 735         INIT_LIST_HEAD(&cli->tc_lru);
 736         memcpy(cli->tc_jobid, jobid, strlen(jobid));
 737 }
 738
 739 static int nrs_tbf_jobid_hash_order(void)
 740 {
 741         int bits;
 742
 743         for (bits = 1; (1 << bits) < tbf_jobid_cache_size; ++bits)
 744                 ;
 745
 746         return bits;
 747 }
 748
 749 #define NRS_TBF_JOBID_BKT_BITS 10
 750
 751 static int
 752 nrs_tbf_jobid_startup(struct ptlrpc_nrs_policy *policy,
 753                       struct nrs_tbf_head *head)
 754 {
 755         struct nrs_tbf_cmd       start;
 756         struct nrs_tbf_bucket   *bkt;
 757         int                      bits;
 758         int                      i;
 759         int                      rc;
 760         struct cfs_hash_bd       bd;
 761
 762         bits = nrs_tbf_jobid_hash_order();
 763         if (bits < NRS_TBF_JOBID_BKT_BITS)
 764                 bits = NRS_TBF_JOBID_BKT_BITS;
 765         head->th_cli_hash = cfs_hash_create("nrs_tbf_hash",
 766                                             bits,
 767                                             bits,
 768                                             NRS_TBF_JOBID_BKT_BITS,
 769                                             sizeof(*bkt),
 770                                             0,
 771                                             0,
 772                                             &nrs_tbf_jobid_hash_ops,
 773                                             NRS_TBF_JOBID_HASH_FLAGS);
 774         if (head->th_cli_hash == NULL)
 775                 return -ENOMEM;
 776
 777         cfs_hash_for_each_bucket(head->th_cli_hash, &bd, i) {
 778                 bkt = cfs_hash_bd_extra_get(head->th_cli_hash, &bd);
 779                 INIT_LIST_HEAD(&bkt->ntb_lru);
 780         }
 781
 782         memset(&start, 0, sizeof(start));
 783         start.u.tc_start.ts_jobids_str = "*";
 784
 785         start.u.tc_start.ts_rpc_rate = tbf_rate;
 786         start.u.tc_start.ts_rule_flags = NTRS_DEFAULT;
 787         start.tc_name = NRS_TBF_DEFAULT_RULE;
 788         INIT_LIST_HEAD(&start.u.tc_start.ts_jobids);
 789         rc = nrs_tbf_rule_start(policy, head, &start);
 790
 791         return rc;
 792 }
 793
 794 /**
 795  * Frees jobid of \a list.
 796  *
 797  */
 798 static void
 799 nrs_tbf_jobid_list_free(struct list_head *jobid_list)
 800 {
 801         struct nrs_tbf_jobid *jobid, *n;
 802
 803         list_for_each_entry_safe(jobid, n, jobid_list, tj_linkage) {
 804                 OBD_FREE(jobid->tj_id, strlen(jobid->tj_id) + 1);
 805                 list_del(&jobid->tj_linkage);
 806                 OBD_FREE(jobid, sizeof(struct nrs_tbf_jobid));
 807         }
 808 }
 809
 810 static int
 811 nrs_tbf_jobid_list_add(const struct cfs_lstr *id, struct list_head *jobid_list)
 812 {
 813         struct nrs_tbf_jobid *jobid;
 814
 815         OBD_ALLOC(jobid, sizeof(struct nrs_tbf_jobid));
 816         if (jobid == NULL)
 817                 return -ENOMEM;
 818
 819         OBD_ALLOC(jobid->tj_id, id->ls_len + 1);
 820         if (jobid->tj_id == NULL) {
 821                 OBD_FREE(jobid, sizeof(struct nrs_tbf_jobid));
 822                 return -ENOMEM;
 823         }
 824
 825         memcpy(jobid->tj_id, id->ls_str, id->ls_len);
 826         list_add_tail(&jobid->tj_linkage, jobid_list);
 827         return 0;
 828 }
 829
 830 static int
 831 nrs_tbf_jobid_list_match(struct list_head *jobid_list, char *id)
 832 {
 833         struct nrs_tbf_jobid *jobid;
 834
 835         list_for_each_entry(jobid, jobid_list, tj_linkage) {
 836                 if (strcmp(id, jobid->tj_id) == 0)
 837                         return 1;
 838         }
 839         return 0;
 840 }
 841
 842 static int
 843 nrs_tbf_jobid_list_parse(char *str, int len, struct list_head *jobid_list)
 844 {
 845         struct cfs_lstr src;
 846         struct cfs_lstr res;
 847         int rc = 0;
 848         ENTRY;
 849
 850         src.ls_str = str;
 851         src.ls_len = len;
 852         INIT_LIST_HEAD(jobid_list);
 853         while (src.ls_str) {
 854                 rc = cfs_gettok(&src, ' ', &res);
 855                 if (rc == 0) {
 856                         rc = -EINVAL;
 857                         break;
 858                 }
 859                 rc = nrs_tbf_jobid_list_add(&res, jobid_list);
 860                 if (rc)
 861                         break;
 862         }
 863         if (rc)
 864                 nrs_tbf_jobid_list_free(jobid_list);
 865         RETURN(rc);
 866 }
 867
 868 static void nrs_tbf_jobid_cmd_fini(struct nrs_tbf_cmd *cmd)
 869 {
 870         if (!list_empty(&cmd->u.tc_start.ts_jobids))
 871                 nrs_tbf_jobid_list_free(&cmd->u.tc_start.ts_jobids);
 872         if (cmd->u.tc_start.ts_jobids_str)
 873                 OBD_FREE(cmd->u.tc_start.ts_jobids_str,
 874                          strlen(cmd->u.tc_start.ts_jobids_str) + 1);
 875 }
 876
 877 static int nrs_tbf_check_id_value(struct cfs_lstr *src, char *key)
 878 {
 879         struct cfs_lstr res;
 880         int keylen = strlen(key);
 881         int rc;
 882
 883         rc = cfs_gettok(src, '=', &res);
 884         if (rc == 0 || res.ls_len != keylen ||
 885             strncmp(res.ls_str, key, keylen) != 0 ||
 886             src->ls_len <= 2 || src->ls_str[0] != '{' ||
 887             src->ls_str[src->ls_len - 1] != '}')
 888                 return -EINVAL;
 889
 890         /* Skip '{' and '}' */
 891         src->ls_str++;
 892         src->ls_len -= 2;
 893         return 0;
 894 }
 895
 896 static int nrs_tbf_jobid_parse(struct nrs_tbf_cmd *cmd, char *id)
 897 {
 898         struct cfs_lstr src;
 899         int rc;
 900
 901         src.ls_str = id;
 902         src.ls_len = strlen(id);
 903         rc = nrs_tbf_check_id_value(&src, "jobid");
 904         if (rc)
 905                 return rc;
 906
 907         OBD_ALLOC(cmd->u.tc_start.ts_jobids_str, src.ls_len + 1);
 908         if (cmd->u.tc_start.ts_jobids_str == NULL)
 909                 return -ENOMEM;
 910
 911         memcpy(cmd->u.tc_start.ts_jobids_str, src.ls_str, src.ls_len);
 912
 913         /* parse jobid list */
 914         rc = nrs_tbf_jobid_list_parse(cmd->u.tc_start.ts_jobids_str,
 915                                       strlen(cmd->u.tc_start.ts_jobids_str),
 916                                       &cmd->u.tc_start.ts_jobids);
 917         if (rc)
 918                 nrs_tbf_jobid_cmd_fini(cmd);
 919
 920         return rc;
 921 }
 922
 923 static int nrs_tbf_jobid_rule_init(struct ptlrpc_nrs_policy *policy,
 924                                    struct nrs_tbf_rule *rule,
 925                                    struct nrs_tbf_cmd *start)
 926 {
 927         int rc = 0;
 928
 929         LASSERT(start->u.tc_start.ts_jobids_str);
 930         OBD_ALLOC(rule->tr_jobids_str,
 931                   strlen(start->u.tc_start.ts_jobids_str) + 1);
 932         if (rule->tr_jobids_str == NULL)
 933                 return -ENOMEM;
 934
 935         memcpy(rule->tr_jobids_str,
 936                start->u.tc_start.ts_jobids_str,
 937                strlen(start->u.tc_start.ts_jobids_str));
 938
 939         INIT_LIST_HEAD(&rule->tr_jobids);
 940         if (!list_empty(&start->u.tc_start.ts_jobids)) {
 941                 rc = nrs_tbf_jobid_list_parse(rule->tr_jobids_str,
 942                                               strlen(rule->tr_jobids_str),
 943                                               &rule->tr_jobids);
 944                 if (rc)
 945                         CERROR("jobids {%s} illegal\n", rule->tr_jobids_str);
 946         }
 947         if (rc)
 948                 OBD_FREE(rule->tr_jobids_str,
 949                          strlen(start->u.tc_start.ts_jobids_str) + 1);
 950         return rc;
 951 }
 952
 953 static int
 954 nrs_tbf_jobid_rule_dump(struct nrs_tbf_rule *rule, struct seq_file *m)
 955 {
 956         seq_printf(m, "%s {%s} %llu, ref %d\n", rule->tr_name,
 957                    rule->tr_jobids_str, rule->tr_rpc_rate,
 958                    atomic_read(&rule->tr_ref) - 1);
 959         return 0;
 960 }
 961
 962 static int
 963 nrs_tbf_jobid_rule_match(struct nrs_tbf_rule *rule,
 964                          struct nrs_tbf_client *cli)
 965 {
 966         return nrs_tbf_jobid_list_match(&rule->tr_jobids, cli->tc_jobid);
 967 }
 968
 969 static void nrs_tbf_jobid_rule_fini(struct nrs_tbf_rule *rule)
 970 {
 971         if (!list_empty(&rule->tr_jobids))
 972                 nrs_tbf_jobid_list_free(&rule->tr_jobids);
 973         LASSERT(rule->tr_jobids_str != NULL);
 974         OBD_FREE(rule->tr_jobids_str, strlen(rule->tr_jobids_str) + 1);
 975 }
 976
 977 static struct nrs_tbf_ops nrs_tbf_jobid_ops = {
 978         .o_name = NRS_TBF_TYPE_JOBID,
 979         .o_startup = nrs_tbf_jobid_startup,
 980         .o_cli_find = nrs_tbf_jobid_cli_find,
 981         .o_cli_findadd = nrs_tbf_jobid_cli_findadd,
 982         .o_cli_put = nrs_tbf_jobid_cli_put,
 983         .o_cli_init = nrs_tbf_jobid_cli_init,
 984         .o_rule_init = nrs_tbf_jobid_rule_init,
 985         .o_rule_dump = nrs_tbf_jobid_rule_dump,
 986         .o_rule_match = nrs_tbf_jobid_rule_match,
 987         .o_rule_fini = nrs_tbf_jobid_rule_fini,
 988 };
 989
 990 /**
 991  * libcfs_hash operations for nrs_tbf_net::cn_cli_hash
 992  *
 993  * This uses ptlrpc_request::rq_peer.nid as its key, in order to hash
 994  * nrs_tbf_client objects.
 995  */
 996 #define NRS_TBF_NID_BKT_BITS    8
 997 #define NRS_TBF_NID_BITS        16
 998
 999 static unsigned nrs_tbf_nid_hop_hash(struct cfs_hash *hs, const void *key,
1000                                   unsigned mask)
1001 {
1002         return cfs_hash_djb2_hash(key, sizeof(lnet_nid_t), mask);
1003 }
1004
1005 static int nrs_tbf_nid_hop_keycmp(const void *key, struct hlist_node *hnode)
1006 {
1007         lnet_nid_t            *nid = (lnet_nid_t *)key;
1008         struct nrs_tbf_client *cli = hlist_entry(hnode,
1009                                                      struct nrs_tbf_client,
1010                                                      tc_hnode);
1011
1012         return *nid == cli->tc_nid;
1013 }
1014
1015 static void *nrs_tbf_nid_hop_key(struct hlist_node *hnode)
1016 {
1017         struct nrs_tbf_client *cli = hlist_entry(hnode,
1018                                                      struct nrs_tbf_client,
1019                                                      tc_hnode);
1020
1021         return &cli->tc_nid;
1022 }
1023
1024 static void *nrs_tbf_nid_hop_object(struct hlist_node *hnode)
1025 {
1026         return hlist_entry(hnode, struct nrs_tbf_client, tc_hnode);
1027 }
1028
1029 static void nrs_tbf_nid_hop_get(struct cfs_hash *hs, struct hlist_node *hnode)
1030 {
1031         struct nrs_tbf_client *cli = hlist_entry(hnode,
1032                                                      struct nrs_tbf_client,
1033                                                      tc_hnode);
1034
1035         atomic_inc(&cli->tc_ref);
1036 }
1037
1038 static void nrs_tbf_nid_hop_put(struct cfs_hash *hs, struct hlist_node *hnode)
1039 {
1040         struct nrs_tbf_client *cli = hlist_entry(hnode,
1041                                                      struct nrs_tbf_client,
1042                                                      tc_hnode);
1043
1044         atomic_dec(&cli->tc_ref);
1045 }
1046
1047 static void nrs_tbf_nid_hop_exit(struct cfs_hash *hs, struct hlist_node *hnode)
1048 {
1049         struct nrs_tbf_client *cli = hlist_entry(hnode,
1050                                                      struct nrs_tbf_client,
1051                                                      tc_hnode);
1052
1053         LASSERTF(atomic_read(&cli->tc_ref) == 0,
1054                  "Busy TBF object from client with NID %s, with %d refs\n",
1055                  libcfs_nid2str(cli->tc_nid), atomic_read(&cli->tc_ref));
1056
1057         nrs_tbf_cli_fini(cli);
1058 }
1059
1060 static struct cfs_hash_ops nrs_tbf_nid_hash_ops = {
1061         .hs_hash        = nrs_tbf_nid_hop_hash,
1062         .hs_keycmp      = nrs_tbf_nid_hop_keycmp,
1063         .hs_key         = nrs_tbf_nid_hop_key,
1064         .hs_object      = nrs_tbf_nid_hop_object,
1065         .hs_get         = nrs_tbf_nid_hop_get,
1066         .hs_put         = nrs_tbf_nid_hop_put,
1067         .hs_put_locked  = nrs_tbf_nid_hop_put,
1068         .hs_exit        = nrs_tbf_nid_hop_exit,
1069 };
1070
1071 static struct nrs_tbf_client *
1072 nrs_tbf_nid_cli_find(struct nrs_tbf_head *head,
1073                      struct ptlrpc_request *req)
1074 {
1075         return cfs_hash_lookup(head->th_cli_hash, &req->rq_peer.nid);
1076 }
1077
1078 static struct nrs_tbf_client *
1079 nrs_tbf_nid_cli_findadd(struct nrs_tbf_head *head,
1080                         struct nrs_tbf_client *cli)
1081 {
1082         return cfs_hash_findadd_unique(head->th_cli_hash, &cli->tc_nid,
1083                                        &cli->tc_hnode);
1084 }
1085
1086 static void
1087 nrs_tbf_nid_cli_put(struct nrs_tbf_head *head,
1088                       struct nrs_tbf_client *cli)
1089 {
1090         cfs_hash_put(head->th_cli_hash, &cli->tc_hnode);
1091 }
1092
1093 static int
1094 nrs_tbf_nid_startup(struct ptlrpc_nrs_policy *policy,
1095                     struct nrs_tbf_head *head)
1096 {
1097         struct nrs_tbf_cmd      start;
1098         int rc;
1099
1100         head->th_cli_hash = cfs_hash_create("nrs_tbf_hash",
1101                                             NRS_TBF_NID_BITS,
1102                                             NRS_TBF_NID_BITS,
1103                                             NRS_TBF_NID_BKT_BITS, 0,
1104                                             CFS_HASH_MIN_THETA,
1105                                             CFS_HASH_MAX_THETA,
1106                                             &nrs_tbf_nid_hash_ops,
1107                                             CFS_HASH_RW_BKTLOCK);
1108         if (head->th_cli_hash == NULL)
1109                 return -ENOMEM;
1110
1111         memset(&start, 0, sizeof(start));
1112         start.u.tc_start.ts_nids_str = "*";
1113
1114         start.u.tc_start.ts_rpc_rate = tbf_rate;
1115         start.u.tc_start.ts_rule_flags = NTRS_DEFAULT;
1116         start.tc_name = NRS_TBF_DEFAULT_RULE;
1117         INIT_LIST_HEAD(&start.u.tc_start.ts_nids);
1118         rc = nrs_tbf_rule_start(policy, head, &start);
1119
1120         return rc;
1121 }
1122
1123 static void
1124 nrs_tbf_nid_cli_init(struct nrs_tbf_client *cli,
1125                              struct ptlrpc_request *req)
1126 {
1127         cli->tc_nid = req->rq_peer.nid;
1128 }
1129
1130 static int nrs_tbf_nid_rule_init(struct ptlrpc_nrs_policy *policy,
1131                                  struct nrs_tbf_rule *rule,
1132                                  struct nrs_tbf_cmd *start)
1133 {
1134         LASSERT(start->u.tc_start.ts_nids_str);
1135         OBD_ALLOC(rule->tr_nids_str,
1136                   strlen(start->u.tc_start.ts_nids_str) + 1);
1137         if (rule->tr_nids_str == NULL)
1138                 return -ENOMEM;
1139
1140         memcpy(rule->tr_nids_str,
1141                start->u.tc_start.ts_nids_str,
1142                strlen(start->u.tc_start.ts_nids_str));
1143
1144         INIT_LIST_HEAD(&rule->tr_nids);
1145         if (!list_empty(&start->u.tc_start.ts_nids)) {
1146                 if (cfs_parse_nidlist(rule->tr_nids_str,
1147                                       strlen(rule->tr_nids_str),
1148                                       &rule->tr_nids) <= 0) {
1149                         CERROR("nids {%s} illegal\n",
1150                                rule->tr_nids_str);
1151                         OBD_FREE(rule->tr_nids_str,
1152                                  strlen(start->u.tc_start.ts_nids_str) + 1);
1153                         return -EINVAL;
1154                 }
1155         }
1156         return 0;
1157 }
1158
1159 static int
1160 nrs_tbf_nid_rule_dump(struct nrs_tbf_rule *rule, struct seq_file *m)
1161 {
1162         seq_printf(m, "%s {%s} %llu, ref %d\n", rule->tr_name,
1163                    rule->tr_nids_str, rule->tr_rpc_rate,
1164                    atomic_read(&rule->tr_ref) - 1);
1165         return 0;
1166 }
1167
1168 static int
1169 nrs_tbf_nid_rule_match(struct nrs_tbf_rule *rule,
1170                        struct nrs_tbf_client *cli)
1171 {
1172         return cfs_match_nid(cli->tc_nid, &rule->tr_nids);
1173 }
1174
1175 static void nrs_tbf_nid_rule_fini(struct nrs_tbf_rule *rule)
1176 {
1177         if (!list_empty(&rule->tr_nids))
1178                 cfs_free_nidlist(&rule->tr_nids);
1179         LASSERT(rule->tr_nids_str != NULL);
1180         OBD_FREE(rule->tr_nids_str, strlen(rule->tr_nids_str) + 1);
1181 }
1182
1183 static void nrs_tbf_nid_cmd_fini(struct nrs_tbf_cmd *cmd)
1184 {
1185         if (!list_empty(&cmd->u.tc_start.ts_nids))
1186                 cfs_free_nidlist(&cmd->u.tc_start.ts_nids);
1187         if (cmd->u.tc_start.ts_nids_str)
1188                 OBD_FREE(cmd->u.tc_start.ts_nids_str,
1189                          strlen(cmd->u.tc_start.ts_nids_str) + 1);
1190 }
1191
1192 static int nrs_tbf_nid_parse(struct nrs_tbf_cmd *cmd, char *id)
1193 {
1194         struct cfs_lstr src;
1195         int rc;
1196
1197         src.ls_str = id;
1198         src.ls_len = strlen(id);
1199         rc = nrs_tbf_check_id_value(&src, "nid");
1200         if (rc)
1201                 return rc;
1202
1203         OBD_ALLOC(cmd->u.tc_start.ts_nids_str, src.ls_len + 1);
1204         if (cmd->u.tc_start.ts_nids_str == NULL)
1205                 return -ENOMEM;
1206
1207         memcpy(cmd->u.tc_start.ts_nids_str, src.ls_str, src.ls_len);
1208
1209         /* parse NID list */
1210         if (cfs_parse_nidlist(cmd->u.tc_start.ts_nids_str,
1211                               strlen(cmd->u.tc_start.ts_nids_str),
1212                               &cmd->u.tc_start.ts_nids) <= 0) {
1213                 nrs_tbf_nid_cmd_fini(cmd);
1214                 return -EINVAL;
1215         }
1216
1217         return 0;
1218 }
1219
1220 static struct nrs_tbf_ops nrs_tbf_nid_ops = {
1221         .o_name = NRS_TBF_TYPE_NID,
1222         .o_startup = nrs_tbf_nid_startup,
1223         .o_cli_find = nrs_tbf_nid_cli_find,
1224         .o_cli_findadd = nrs_tbf_nid_cli_findadd,
1225         .o_cli_put = nrs_tbf_nid_cli_put,
1226         .o_cli_init = nrs_tbf_nid_cli_init,
1227         .o_rule_init = nrs_tbf_nid_rule_init,
1228         .o_rule_dump = nrs_tbf_nid_rule_dump,
1229         .o_rule_match = nrs_tbf_nid_rule_match,
1230         .o_rule_fini = nrs_tbf_nid_rule_fini,
1231 };
1232
1233 static void nrs_tbf_opcode_rule_fini(struct nrs_tbf_rule *rule)
1234 {
1235         if (rule->tr_opcodes != NULL)
1236                 CFS_FREE_BITMAP(rule->tr_opcodes);
1237
1238         LASSERT(rule->tr_opcodes_str != NULL);
1239         OBD_FREE(rule->tr_opcodes_str, strlen(rule->tr_opcodes_str) + 1);
1240 }
1241
1242 static unsigned nrs_tbf_opcode_hop_hash(struct cfs_hash *hs, const void *key,
1243                                         unsigned mask)
1244 {
1245         return cfs_hash_djb2_hash(key, sizeof(__u32), mask);
1246 }
1247
1248 static int nrs_tbf_opcode_hop_keycmp(const void *key, struct hlist_node *hnode)
1249 {
1250         const __u32     *opc = key;
1251         struct nrs_tbf_client *cli = hlist_entry(hnode,
1252                                                  struct nrs_tbf_client,
1253                                                  tc_hnode);
1254
1255         return *opc == cli->tc_opcode;
1256 }
1257
1258 static void *nrs_tbf_opcode_hop_key(struct hlist_node *hnode)
1259 {
1260         struct nrs_tbf_client *cli = hlist_entry(hnode,
1261                                                  struct nrs_tbf_client,
1262                                                  tc_hnode);
1263
1264         return &cli->tc_opcode;
1265 }
1266
1267 static void *nrs_tbf_opcode_hop_object(struct hlist_node *hnode)
1268 {
1269         return hlist_entry(hnode, struct nrs_tbf_client, tc_hnode);
1270 }
1271
1272 static void nrs_tbf_opcode_hop_get(struct cfs_hash *hs,
1273                                    struct hlist_node *hnode)
1274 {
1275         struct nrs_tbf_client *cli = hlist_entry(hnode,
1276                                                  struct nrs_tbf_client,
1277                                                  tc_hnode);
1278
1279         atomic_inc(&cli->tc_ref);
1280 }
1281
1282 static void nrs_tbf_opcode_hop_put(struct cfs_hash *hs,
1283                                    struct hlist_node *hnode)
1284 {
1285         struct nrs_tbf_client *cli = hlist_entry(hnode,
1286                                                  struct nrs_tbf_client,
1287                                                  tc_hnode);
1288
1289         atomic_dec(&cli->tc_ref);
1290 }
1291
1292 static void nrs_tbf_opcode_hop_exit(struct cfs_hash *hs,
1293                                     struct hlist_node *hnode)
1294 {
1295         struct nrs_tbf_client *cli = hlist_entry(hnode,
1296                                                  struct nrs_tbf_client,
1297                                                  tc_hnode);
1298
1299         LASSERTF(atomic_read(&cli->tc_ref) == 0,
1300                  "Busy TBF object from client with opcode %s, with %d refs\n",
1301                  ll_opcode2str(cli->tc_opcode),
1302                  atomic_read(&cli->tc_ref));
1303
1304         nrs_tbf_cli_fini(cli);
1305 }
1306 static struct cfs_hash_ops nrs_tbf_opcode_hash_ops = {
1307         .hs_hash        = nrs_tbf_opcode_hop_hash,
1308         .hs_keycmp      = nrs_tbf_opcode_hop_keycmp,
1309         .hs_key         = nrs_tbf_opcode_hop_key,
1310         .hs_object      = nrs_tbf_opcode_hop_object,
1311         .hs_get         = nrs_tbf_opcode_hop_get,
1312         .hs_put         = nrs_tbf_opcode_hop_put,
1313         .hs_put_locked  = nrs_tbf_opcode_hop_put,
1314         .hs_exit        = nrs_tbf_opcode_hop_exit,
1315 };
1316
1317 static int
1318 nrs_tbf_opcode_startup(struct ptlrpc_nrs_policy *policy,
1319                     struct nrs_tbf_head *head)
1320 {
1321         struct nrs_tbf_cmd      start = { 0 };
1322         int rc;
1323
1324         head->th_cli_hash = cfs_hash_create("nrs_tbf_hash",
1325                                             NRS_TBF_NID_BITS,
1326                                             NRS_TBF_NID_BITS,
1327                                             NRS_TBF_NID_BKT_BITS, 0,
1328                                             CFS_HASH_MIN_THETA,
1329                                             CFS_HASH_MAX_THETA,
1330                                             &nrs_tbf_opcode_hash_ops,
1331                                             CFS_HASH_RW_BKTLOCK);
1332         if (head->th_cli_hash == NULL)
1333                 return -ENOMEM;
1334
1335         start.u.tc_start.ts_opcodes = NULL;
1336         start.u.tc_start.ts_opcodes_str = "*";
1337
1338         start.u.tc_start.ts_rpc_rate = tbf_rate;
1339         start.u.tc_start.ts_rule_flags = NTRS_DEFAULT;
1340         start.tc_name = NRS_TBF_DEFAULT_RULE;
1341         rc = nrs_tbf_rule_start(policy, head, &start);
1342
1343         return rc;
1344 }
1345
1346 static struct nrs_tbf_client *
1347 nrs_tbf_opcode_cli_find(struct nrs_tbf_head *head,
1348                         struct ptlrpc_request *req)
1349 {
1350         __u32 opc;
1351
1352         opc = lustre_msg_get_opc(req->rq_reqmsg);
1353         return cfs_hash_lookup(head->th_cli_hash, &opc);
1354 }
1355
1356 static struct nrs_tbf_client *
1357 nrs_tbf_opcode_cli_findadd(struct nrs_tbf_head *head,
1358                            struct nrs_tbf_client *cli)
1359 {
1360         return cfs_hash_findadd_unique(head->th_cli_hash, &cli->tc_opcode,
1361                                        &cli->tc_hnode);
1362 }
1363
1364 static void
1365 nrs_tbf_opcode_cli_init(struct nrs_tbf_client *cli,
1366                         struct ptlrpc_request *req)
1367 {
1368         cli->tc_opcode = lustre_msg_get_opc(req->rq_reqmsg);
1369 }
1370
1371 #define MAX_OPCODE_LEN  32
1372 static int
1373 nrs_tbf_opcode_set_bit(const struct cfs_lstr *id, struct cfs_bitmap *opcodes)
1374 {
1375         int     op = 0;
1376         char    opcode_str[MAX_OPCODE_LEN];
1377
1378         if (id->ls_len + 1 > MAX_OPCODE_LEN)
1379                 return -EINVAL;
1380
1381         memcpy(opcode_str, id->ls_str, id->ls_len);
1382         opcode_str[id->ls_len] = '\0';
1383
1384         op = ll_str2opcode(opcode_str);
1385         if (op < 0)
1386                 return -EINVAL;
1387
1388         cfs_bitmap_set(opcodes, op);
1389         return 0;
1390 }
1391
1392 static int
1393 nrs_tbf_opcode_list_parse(char *str, int len, struct cfs_bitmap *opcodes)
1394 {
1395         struct cfs_lstr src;
1396         struct cfs_lstr res;
1397         int rc = 0;
1398
1399         ENTRY;
1400
1401         src.ls_str = str;
1402         src.ls_len = len;
1403         while (src.ls_str) {
1404                 rc = cfs_gettok(&src, ' ', &res);
1405                 if (rc == 0) {
1406                         rc = -EINVAL;
1407                         break;
1408                 }
1409                 rc = nrs_tbf_opcode_set_bit(&res, opcodes);
1410                 if (rc)
1411                         break;
1412         }
1413
1414         RETURN(rc);
1415 }
1416
1417 static void nrs_tbf_opcode_cmd_fini(struct nrs_tbf_cmd *cmd)
1418 {
1419         if (cmd->u.tc_start.ts_opcodes)
1420                 CFS_FREE_BITMAP(cmd->u.tc_start.ts_opcodes);
1421
1422         if (cmd->u.tc_start.ts_opcodes_str)
1423                 OBD_FREE(cmd->u.tc_start.ts_opcodes_str,
1424                          strlen(cmd->u.tc_start.ts_opcodes_str) + 1);
1425
1426 }
1427
1428 static int nrs_tbf_opcode_parse(struct nrs_tbf_cmd *cmd, char *id)
1429 {
1430         struct cfs_lstr src;
1431         int rc;
1432
1433         cmd->u.tc_start.ts_opcodes = CFS_ALLOCATE_BITMAP(LUSTRE_MAX_OPCODES);
1434         if (cmd->u.tc_start.ts_opcodes == NULL)
1435                 return -ENOMEM;
1436
1437         src.ls_str = id;
1438         src.ls_len = strlen(id);
1439         rc = nrs_tbf_check_id_value(&src, "opcode");
1440         if (rc)
1441                 GOTO(out, rc);
1442
1443         OBD_ALLOC(cmd->u.tc_start.ts_opcodes_str, src.ls_len + 1);
1444         if (cmd->u.tc_start.ts_opcodes_str == NULL)
1445                 GOTO(out, rc = -ENOMEM);
1446
1447         memcpy(cmd->u.tc_start.ts_opcodes_str, src.ls_str, src.ls_len);
1448
1449         /* parse opcode list */
1450         rc = nrs_tbf_opcode_list_parse(cmd->u.tc_start.ts_opcodes_str,
1451                                        strlen(cmd->u.tc_start.ts_opcodes_str),
1452                                        cmd->u.tc_start.ts_opcodes);
1453 out:
1454         if (rc != 0)
1455                 nrs_tbf_opcode_cmd_fini(cmd);
1456
1457         return rc;
1458 }
1459
1460 static int
1461 nrs_tbf_opcode_rule_match(struct nrs_tbf_rule *rule,
1462                           struct nrs_tbf_client *cli)
1463 {
1464         if (rule->tr_opcodes == NULL)
1465                 return 0;
1466
1467         return cfs_bitmap_check(rule->tr_opcodes, cli->tc_opcode);
1468 }
1469
1470 static int nrs_tbf_opcode_rule_init(struct ptlrpc_nrs_policy *policy,
1471                                     struct nrs_tbf_rule *rule,
1472                                     struct nrs_tbf_cmd *start)
1473 {
1474         LASSERT(start->u.tc_start.ts_opcodes_str != NULL);
1475         OBD_ALLOC(rule->tr_opcodes_str,
1476                   strlen(start->u.tc_start.ts_opcodes_str) + 1);
1477         if (rule->tr_opcodes_str == NULL)
1478                 return -ENOMEM;
1479
1480         strncpy(rule->tr_opcodes_str, start->u.tc_start.ts_opcodes_str,
1481                 strlen(start->u.tc_start.ts_opcodes_str) + 1);
1482
1483         if (start->u.tc_start.ts_opcodes == NULL)
1484                 return 0;
1485
1486         rule->tr_opcodes = CFS_ALLOCATE_BITMAP(LUSTRE_MAX_OPCODES);
1487         if (rule->tr_opcodes == NULL) {
1488                 OBD_FREE(rule->tr_opcodes_str,
1489                          strlen(start->u.tc_start.ts_opcodes_str) + 1);
1490                 return -ENOMEM;
1491         }
1492
1493         cfs_bitmap_copy(rule->tr_opcodes, start->u.tc_start.ts_opcodes);
1494
1495         return 0;
1496 }
1497
1498 static int
1499 nrs_tbf_opcode_rule_dump(struct nrs_tbf_rule *rule, struct seq_file *m)
1500 {
1501         seq_printf(m, "%s {%s} %llu, ref %d\n", rule->tr_name,
1502                    rule->tr_opcodes_str, rule->tr_rpc_rate,
1503                    atomic_read(&rule->tr_ref) - 1);
1504         return 0;
1505 }
1506
1507
1508 struct nrs_tbf_ops nrs_tbf_opcode_ops = {
1509         .o_name = NRS_TBF_TYPE_OPCODE,
1510         .o_startup = nrs_tbf_opcode_startup,
1511         .o_cli_find = nrs_tbf_opcode_cli_find,
1512         .o_cli_findadd = nrs_tbf_opcode_cli_findadd,
1513         .o_cli_put = nrs_tbf_nid_cli_put,
1514         .o_cli_init = nrs_tbf_opcode_cli_init,
1515         .o_rule_init = nrs_tbf_opcode_rule_init,
1516         .o_rule_dump = nrs_tbf_opcode_rule_dump,
1517         .o_rule_match = nrs_tbf_opcode_rule_match,
1518         .o_rule_fini = nrs_tbf_opcode_rule_fini,
1519 };
1520
1521 static struct nrs_tbf_type nrs_tbf_types[] = {
1522         {
1523                 .ntt_name = NRS_TBF_TYPE_JOBID,
1524                 .ntt_flag = NRS_TBF_FLAG_JOBID,
1525                 .ntt_ops = &nrs_tbf_jobid_ops,
1526         },
1527         {
1528                 .ntt_name = NRS_TBF_TYPE_NID,
1529                 .ntt_flag = NRS_TBF_FLAG_NID,
1530                 .ntt_ops = &nrs_tbf_nid_ops,
1531         },
1532         {
1533                 .ntt_name = NRS_TBF_TYPE_OPCODE,
1534                 .ntt_flag = NRS_TBF_FLAG_OPCODE,
1535                 .ntt_ops = &nrs_tbf_opcode_ops,
1536         },
1537 };
1538
1539 /**
1540  * Is called before the policy transitions into
1541  * ptlrpc_nrs_pol_state::NRS_POL_STATE_STARTED; allocates and initializes a
1542  * policy-specific private data structure.
1543  *
1544  * \param[in] policy The policy to start
1545  *
1546  * \retval -ENOMEM OOM error
1547  * \retval  0      success
1548  *
1549  * \see nrs_policy_register()
1550  * \see nrs_policy_ctl()
1551  */
1552 static int nrs_tbf_start(struct ptlrpc_nrs_policy *policy, char *arg)
1553 {
1554         struct nrs_tbf_head     *head;
1555         struct nrs_tbf_ops      *ops;
1556         __u32                    type;
1557         int found = 0;
1558         int i;
1559         int rc = 0;
1560
1561         if (arg == NULL || strlen(arg) > NRS_TBF_TYPE_MAX_LEN)
1562                 GOTO(out, rc = -EINVAL);
1563
1564         for (i = 0; i < ARRAY_SIZE(nrs_tbf_types); i++) {
1565                 if (strcmp(arg, nrs_tbf_types[i].ntt_name) == 0) {
1566                         ops = nrs_tbf_types[i].ntt_ops;
1567                         type = nrs_tbf_types[i].ntt_flag;
1568                         found = 1;
1569                         break;
1570                 }
1571         }
1572         if (found == 0)
1573                 GOTO(out, rc = -ENOTSUPP);
1574
1575         OBD_CPT_ALLOC_PTR(head, nrs_pol2cptab(policy), nrs_pol2cptid(policy));
1576         if (head == NULL)
1577                 GOTO(out, rc = -ENOMEM);
1578
1579         memcpy(head->th_type, arg, strlen(arg));
1580         head->th_type[strlen(arg)] = '\0';
1581         head->th_ops = ops;
1582         head->th_type_flag = type;
1583
1584         head->th_binheap = cfs_binheap_create(&nrs_tbf_heap_ops,
1585                                               CBH_FLAG_ATOMIC_GROW, 4096, NULL,
1586                                               nrs_pol2cptab(policy),
1587                                               nrs_pol2cptid(policy));
1588         if (head->th_binheap == NULL)
1589                 GOTO(out_free_head, rc = -ENOMEM);
1590
1591         atomic_set(&head->th_rule_sequence, 0);
1592         spin_lock_init(&head->th_rule_lock);
1593         INIT_LIST_HEAD(&head->th_list);
1594         hrtimer_init(&head->th_timer, CLOCK_MONOTONIC, HRTIMER_MODE_ABS);
1595         head->th_timer.function = nrs_tbf_timer_cb;
1596         rc = head->th_ops->o_startup(policy, head);
1597         if (rc)
1598                 GOTO(out_free_heap, rc);
1599
1600         policy->pol_private = head;
1601         return 0;
1602 out_free_heap:
1603         cfs_binheap_destroy(head->th_binheap);
1604 out_free_head:
1605         OBD_FREE_PTR(head);
1606 out:
1607         return rc;
1608 }
1609
1610 /**
1611  * Is called before the policy transitions into
1612  * ptlrpc_nrs_pol_state::NRS_POL_STATE_STOPPED; deallocates the policy-specific
1613  * private data structure.
1614  *
1615  * \param[in] policy The policy to stop
1616  *
1617  * \see nrs_policy_stop0()
1618  */
1619 static void nrs_tbf_stop(struct ptlrpc_nrs_policy *policy)
1620 {
1621         struct nrs_tbf_head *head = policy->pol_private;
1622         struct ptlrpc_nrs *nrs = policy->pol_nrs;
1623         struct nrs_tbf_rule *rule, *n;
1624
1625         LASSERT(head != NULL);
1626         LASSERT(head->th_cli_hash != NULL);
1627         hrtimer_cancel(&head->th_timer);
1628         /* Should cleanup hash first before free rules */
1629         cfs_hash_putref(head->th_cli_hash);
1630         list_for_each_entry_safe(rule, n, &head->th_list, tr_linkage) {
1631                 list_del_init(&rule->tr_linkage);
1632                 nrs_tbf_rule_put(rule);
1633         }
1634         LASSERT(list_empty(&head->th_list));
1635         LASSERT(head->th_binheap != NULL);
1636         LASSERT(cfs_binheap_is_empty(head->th_binheap));
1637         cfs_binheap_destroy(head->th_binheap);
1638         OBD_FREE_PTR(head);
1639         nrs->nrs_throttling = 0;
1640         wake_up(&policy->pol_nrs->nrs_svcpt->scp_waitq);
1641 }
1642
1643 /**
1644  * Performs a policy-specific ctl function on TBF policy instances; similar
1645  * to ioctl.
1646  *
1647  * \param[in]     policy the policy instance
1648  * \param[in]     opc    the opcode
1649  * \param[in,out] arg    used for passing parameters and information
1650  *
1651  * \pre assert_spin_locked(&policy->pol_nrs->->nrs_lock)
1652  * \post assert_spin_locked(&policy->pol_nrs->->nrs_lock)
1653  *
1654  * \retval 0   operation carried out successfully
1655  * \retval -ve error
1656  */
1657 static int nrs_tbf_ctl(struct ptlrpc_nrs_policy *policy,
1658                        enum ptlrpc_nrs_ctl opc,
1659                        void *arg)
1660 {
1661         int rc = 0;
1662         ENTRY;
1663
1664         assert_spin_locked(&policy->pol_nrs->nrs_lock);
1665
1666         switch ((enum nrs_ctl_tbf)opc) {
1667         default:
1668                 RETURN(-EINVAL);
1669
1670         /**
1671          * Read RPC rate size of a policy instance.
1672          */
1673         case NRS_CTL_TBF_RD_RULE: {
1674                 struct nrs_tbf_head *head = policy->pol_private;
1675                 struct seq_file *m = (struct seq_file *) arg;
1676                 struct ptlrpc_service_part *svcpt;
1677
1678                 svcpt = policy->pol_nrs->nrs_svcpt;
1679                 seq_printf(m, "CPT %d:\n", svcpt->scp_cpt);
1680
1681                 rc = nrs_tbf_rule_dump_all(head, m);
1682                 }
1683                 break;
1684
1685         /**
1686          * Write RPC rate of a policy instance.
1687          */
1688         case NRS_CTL_TBF_WR_RULE: {
1689                 struct nrs_tbf_head *head = policy->pol_private;
1690                 struct nrs_tbf_cmd *cmd;
1691
1692                 cmd = (struct nrs_tbf_cmd *)arg;
1693                 rc = nrs_tbf_command(policy,
1694                                      head,
1695                                      cmd);
1696                 }
1697                 break;
1698         /**
1699          * Read the TBF policy type of a policy instance.
1700          */
1701         case NRS_CTL_TBF_RD_TYPE_FLAG: {
1702                 struct nrs_tbf_head *head = policy->pol_private;
1703
1704                 *(__u32 *)arg = head->th_type_flag;
1705                 }
1706                 break;
1707         }
1708
1709         RETURN(rc);
1710 }
1711
1712 /**
1713  * Is called for obtaining a TBF policy resource.
1714  *
1715  * \param[in]  policy     The policy on which the request is being asked for
1716  * \param[in]  nrq        The request for which resources are being taken
1717  * \param[in]  parent     Parent resource, unused in this policy
1718  * \param[out] resp       Resources references are placed in this array
1719  * \param[in]  moving_req Signifies limited caller context; unused in this
1720  *                        policy
1721  *
1722  *
1723  * \see nrs_resource_get_safe()
1724  */
1725 static int nrs_tbf_res_get(struct ptlrpc_nrs_policy *policy,
1726                            struct ptlrpc_nrs_request *nrq,
1727                            const struct ptlrpc_nrs_resource *parent,
1728                            struct ptlrpc_nrs_resource **resp,
1729                            bool moving_req)
1730 {
1731         struct nrs_tbf_head   *head;
1732         struct nrs_tbf_client *cli;
1733         struct nrs_tbf_client *tmp;
1734         struct ptlrpc_request *req;
1735
1736         if (parent == NULL) {
1737                 *resp = &((struct nrs_tbf_head *)policy->pol_private)->th_res;
1738                 return 0;
1739         }
1740
1741         head = container_of(parent, struct nrs_tbf_head, th_res);
1742         req = container_of(nrq, struct ptlrpc_request, rq_nrq);
1743         cli = head->th_ops->o_cli_find(head, req);
1744         if (cli != NULL) {
1745                 spin_lock(&policy->pol_nrs->nrs_svcpt->scp_req_lock);
1746                 LASSERT(cli->tc_rule);
1747                 if (cli->tc_rule_sequence !=
1748                     atomic_read(&head->th_rule_sequence) ||
1749                     cli->tc_rule->tr_flags & NTRS_STOPPING) {
1750                         struct nrs_tbf_rule *rule;
1751
1752                         rule = nrs_tbf_rule_match(head, cli);
1753                         if (rule != cli->tc_rule)
1754                                 nrs_tbf_cli_reset(head, rule, cli);
1755                         else
1756                                 nrs_tbf_rule_put(rule);
1757                 } else if (cli->tc_rule_generation !=
1758                            cli->tc_rule->tr_generation) {
1759                         nrs_tbf_cli_reset_value(head, cli);
1760                 }
1761                 spin_unlock(&policy->pol_nrs->nrs_svcpt->scp_req_lock);
1762                 goto out;
1763         }
1764
1765         OBD_CPT_ALLOC_GFP(cli, nrs_pol2cptab(policy), nrs_pol2cptid(policy),
1766                           sizeof(*cli), moving_req ? GFP_ATOMIC : __GFP_IO);
1767         if (cli == NULL)
1768                 return -ENOMEM;
1769
1770         nrs_tbf_cli_init(head, cli, req);
1771         tmp = head->th_ops->o_cli_findadd(head, cli);
1772         if (tmp != cli) {
1773                 atomic_dec(&cli->tc_ref);
1774                 nrs_tbf_cli_fini(cli);
1775                 cli = tmp;
1776         }
1777 out:
1778         *resp = &cli->tc_res;
1779
1780         return 1;
1781 }
1782
1783 /**
1784  * Called when releasing references to the resource hierachy obtained for a
1785  * request for scheduling using the TBF policy.
1786  *
1787  * \param[in] policy   the policy the resource belongs to
1788  * \param[in] res      the resource to be released
1789  */
1790 static void nrs_tbf_res_put(struct ptlrpc_nrs_policy *policy,
1791                             const struct ptlrpc_nrs_resource *res)
1792 {
1793         struct nrs_tbf_head   *head;
1794         struct nrs_tbf_client *cli;
1795
1796         /**
1797          * Do nothing for freeing parent, nrs_tbf_net resources
1798          */
1799         if (res->res_parent == NULL)
1800                 return;
1801
1802         cli = container_of(res, struct nrs_tbf_client, tc_res);
1803         head = container_of(res->res_parent, struct nrs_tbf_head, th_res);
1804
1805         head->th_ops->o_cli_put(head, cli);
1806 }
1807
1808 /**
1809  * Called when getting a request from the TBF policy for handling, or just
1810  * peeking; removes the request from the policy when it is to be handled.
1811  *
1812  * \param[in] policy The policy
1813  * \param[in] peek   When set, signifies that we just want to examine the
1814  *                   request, and not handle it, so the request is not removed
1815  *                   from the policy.
1816  * \param[in] force  Force the policy to return a request; unused in this
1817  *                   policy
1818  *
1819  * \retval The request to be handled; this is the next request in the TBF
1820  *         rule
1821  *
1822  * \see ptlrpc_nrs_req_get_nolock()
1823  * \see nrs_request_get()
1824  */
1825 static
1826 struct ptlrpc_nrs_request *nrs_tbf_req_get(struct ptlrpc_nrs_policy *policy,
1827                                            bool peek, bool force)
1828 {
1829         struct nrs_tbf_head       *head = policy->pol_private;
1830         struct ptlrpc_nrs_request *nrq = NULL;
1831         struct nrs_tbf_client     *cli;
1832         struct cfs_binheap_node   *node;
1833
1834         assert_spin_locked(&policy->pol_nrs->nrs_svcpt->scp_req_lock);
1835
1836         if (!peek && policy->pol_nrs->nrs_throttling)
1837                 return NULL;
1838
1839         node = cfs_binheap_root(head->th_binheap);
1840         if (unlikely(node == NULL))
1841                 return NULL;
1842
1843         cli = container_of(node, struct nrs_tbf_client, tc_node);
1844         LASSERT(cli->tc_in_heap);
1845         if (peek) {
1846                 nrq = list_entry(cli->tc_list.next,
1847                                      struct ptlrpc_nrs_request,
1848                                      nr_u.tbf.tr_list);
1849         } else {
1850                 __u64 now = ktime_to_ns(ktime_get());
1851                 __u64 passed;
1852                 __u64 ntoken;
1853                 __u64 deadline;
1854
1855                 deadline = cli->tc_check_time +
1856                           cli->tc_nsecs;
1857                 LASSERT(now >= cli->tc_check_time);
1858                 passed = now - cli->tc_check_time;
1859                 ntoken = passed * cli->tc_rpc_rate;
1860                 do_div(ntoken, NSEC_PER_SEC);
1861                 ntoken += cli->tc_ntoken;
1862                 if (ntoken > cli->tc_depth)
1863                         ntoken = cli->tc_depth;
1864                 if (ntoken > 0) {
1865                         struct ptlrpc_request *req;
1866                         nrq = list_entry(cli->tc_list.next,
1867                                              struct ptlrpc_nrs_request,
1868                                              nr_u.tbf.tr_list);
1869                         req = container_of(nrq,
1870                                            struct ptlrpc_request,
1871                                            rq_nrq);
1872                         ntoken--;
1873                         cli->tc_ntoken = ntoken;
1874                         cli->tc_check_time = now;
1875                         list_del_init(&nrq->nr_u.tbf.tr_list);
1876                         if (list_empty(&cli->tc_list)) {
1877                                 cfs_binheap_remove(head->th_binheap,
1878                                                    &cli->tc_node);
1879                                 cli->tc_in_heap = false;
1880                         } else {
1881                                 cfs_binheap_relocate(head->th_binheap,
1882                                                      &cli->tc_node);
1883                         }
1884                         CDEBUG(D_RPCTRACE,
1885                                "NRS start %s request from %s, "
1886                                "seq: %llu\n",
1887                                policy->pol_desc->pd_name,
1888                                libcfs_id2str(req->rq_peer),
1889                                nrq->nr_u.tbf.tr_sequence);
1890                 } else {
1891                         ktime_t time;
1892
1893                         policy->pol_nrs->nrs_throttling = 1;
1894                         head->th_deadline = deadline;
1895                         time = ktime_set(0, 0);
1896                         time = ktime_add_ns(time, deadline);
1897                         hrtimer_start(&head->th_timer, time, HRTIMER_MODE_ABS);
1898                 }
1899         }
1900
1901         return nrq;
1902 }
1903
1904 /**
1905  * Adds request \a nrq to \a policy's list of queued requests
1906  *
1907  * \param[in] policy The policy
1908  * \param[in] nrq    The request to add
1909  *
1910  * \retval 0 success; nrs_request_enqueue() assumes this function will always
1911  *                    succeed
1912  */
1913 static int nrs_tbf_req_add(struct ptlrpc_nrs_policy *policy,
1914                            struct ptlrpc_nrs_request *nrq)
1915 {
1916         struct nrs_tbf_head   *head;
1917         struct nrs_tbf_client *cli;
1918         int                    rc = 0;
1919
1920         assert_spin_locked(&policy->pol_nrs->nrs_svcpt->scp_req_lock);
1921
1922         cli = container_of(nrs_request_resource(nrq),
1923                            struct nrs_tbf_client, tc_res);
1924         head = container_of(nrs_request_resource(nrq)->res_parent,
1925                             struct nrs_tbf_head, th_res);
1926         if (list_empty(&cli->tc_list)) {
1927                 LASSERT(!cli->tc_in_heap);
1928                 rc = cfs_binheap_insert(head->th_binheap, &cli->tc_node);
1929                 if (rc == 0) {
1930                         cli->tc_in_heap = true;
1931                         nrq->nr_u.tbf.tr_sequence = head->th_sequence++;
1932                         list_add_tail(&nrq->nr_u.tbf.tr_list,
1933                                           &cli->tc_list);
1934                         if (policy->pol_nrs->nrs_throttling) {
1935                                 __u64 deadline = cli->tc_check_time +
1936                                                  cli->tc_nsecs;
1937                                 if ((head->th_deadline > deadline) &&
1938                                     (hrtimer_try_to_cancel(&head->th_timer)
1939                                      >= 0)) {
1940                                         ktime_t time;
1941                                         head->th_deadline = deadline;
1942                                         time = ktime_set(0, 0);
1943                                         time = ktime_add_ns(time, deadline);
1944                                         hrtimer_start(&head->th_timer, time,
1945                                                       HRTIMER_MODE_ABS);
1946                                 }
1947                         }
1948                 }
1949         } else {
1950                 LASSERT(cli->tc_in_heap);
1951                 nrq->nr_u.tbf.tr_sequence = head->th_sequence++;
1952                 list_add_tail(&nrq->nr_u.tbf.tr_list,
1953                                   &cli->tc_list);
1954         }
1955         return rc;
1956 }
1957
1958 /**
1959  * Removes request \a nrq from \a policy's list of queued requests.
1960  *
1961  * \param[in] policy The policy
1962  * \param[in] nrq    The request to remove
1963  */
1964 static void nrs_tbf_req_del(struct ptlrpc_nrs_policy *policy,
1965                              struct ptlrpc_nrs_request *nrq)
1966 {
1967         struct nrs_tbf_head   *head;
1968         struct nrs_tbf_client *cli;
1969
1970         assert_spin_locked(&policy->pol_nrs->nrs_svcpt->scp_req_lock);
1971
1972         cli = container_of(nrs_request_resource(nrq),
1973                            struct nrs_tbf_client, tc_res);
1974         head = container_of(nrs_request_resource(nrq)->res_parent,
1975                             struct nrs_tbf_head, th_res);
1976
1977         LASSERT(!list_empty(&nrq->nr_u.tbf.tr_list));
1978         list_del_init(&nrq->nr_u.tbf.tr_list);
1979         if (list_empty(&cli->tc_list)) {
1980                 cfs_binheap_remove(head->th_binheap,
1981                                    &cli->tc_node);
1982                 cli->tc_in_heap = false;
1983         } else {
1984                 cfs_binheap_relocate(head->th_binheap,
1985                                      &cli->tc_node);
1986         }
1987 }
1988
1989 /**
1990  * Prints a debug statement right before the request \a nrq stops being
1991  * handled.
1992  *
1993  * \param[in] policy The policy handling the request
1994  * \param[in] nrq    The request being handled
1995  *
1996  * \see ptlrpc_server_finish_request()
1997  * \see ptlrpc_nrs_req_stop_nolock()
1998  */
1999 static void nrs_tbf_req_stop(struct ptlrpc_nrs_policy *policy,
2000                               struct ptlrpc_nrs_request *nrq)
2001 {
2002         struct ptlrpc_request *req = container_of(nrq, struct ptlrpc_request,
2003                                                   rq_nrq);
2004
2005         assert_spin_locked(&policy->pol_nrs->nrs_svcpt->scp_req_lock);
2006
2007         CDEBUG(D_RPCTRACE, "NRS stop %s request from %s, seq: %llu\n",
2008                policy->pol_desc->pd_name, libcfs_id2str(req->rq_peer),
2009                nrq->nr_u.tbf.tr_sequence);
2010 }
2011
2012 #ifdef CONFIG_PROC_FS
2013
2014 /**
2015  * lprocfs interface
2016  */
2017
2018 /**
2019  * The maximum RPC rate.
2020  */
2021 #define LPROCFS_NRS_RATE_MAX            65535
2022
2023 static int
2024 ptlrpc_lprocfs_nrs_tbf_rule_seq_show(struct seq_file *m, void *data)
2025 {
2026         struct ptlrpc_service       *svc = m->private;
2027         int                          rc;
2028
2029         seq_printf(m, "regular_requests:\n");
2030         /**
2031          * Perform two separate calls to this as only one of the NRS heads'
2032          * policies may be in the ptlrpc_nrs_pol_state::NRS_POL_STATE_STARTED or
2033          * ptlrpc_nrs_pol_state::NRS_POL_STATE_STOPPING state.
2034          */
2035         rc = ptlrpc_nrs_policy_control(svc, PTLRPC_NRS_QUEUE_REG,
2036                                        NRS_POL_NAME_TBF,
2037                                        NRS_CTL_TBF_RD_RULE,
2038                                        false, m);
2039         if (rc == 0) {
2040                 /**
2041                  * -ENOSPC means buf in the parameter m is overflow, return 0
2042                  * here to let upper layer function seq_read alloc a larger
2043                  * memory area and do this process again.
2044                  */
2045         } else if (rc == -ENOSPC) {
2046                 return 0;
2047
2048                 /**
2049                  * Ignore -ENODEV as the regular NRS head's policy may be in the
2050                  * ptlrpc_nrs_pol_state::NRS_POL_STATE_STOPPED state.
2051                  */
2052         } else if (rc != -ENODEV) {
2053                 return rc;
2054         }
2055
2056         if (!nrs_svc_has_hp(svc))
2057                 goto no_hp;
2058
2059         seq_printf(m, "high_priority_requests:\n");
2060         rc = ptlrpc_nrs_policy_control(svc, PTLRPC_NRS_QUEUE_HP,
2061                                        NRS_POL_NAME_TBF,
2062                                        NRS_CTL_TBF_RD_RULE,
2063                                        false, m);
2064         if (rc == 0) {
2065                 /**
2066                  * -ENOSPC means buf in the parameter m is overflow, return 0
2067                  * here to let upper layer function seq_read alloc a larger
2068                  * memory area and do this process again.
2069                  */
2070         } else if (rc == -ENOSPC) {
2071                 return 0;
2072         }
2073
2074 no_hp:
2075
2076         return rc;
2077 }
2078
2079 static int nrs_tbf_id_parse(struct nrs_tbf_cmd *cmd, char *token)
2080 {
2081         int rc;
2082
2083         switch (cmd->u.tc_start.ts_valid_type) {
2084         case NRS_TBF_FLAG_JOBID:
2085                 rc = nrs_tbf_jobid_parse(cmd, token);
2086                 break;
2087         case NRS_TBF_FLAG_NID:
2088                 rc = nrs_tbf_nid_parse(cmd, token);
2089                 break;
2090         case NRS_TBF_FLAG_OPCODE:
2091                 rc = nrs_tbf_opcode_parse(cmd, token);
2092                 break;
2093         default:
2094                 RETURN(-EINVAL);
2095         }
2096
2097         return rc;
2098 }
2099
2100 static void nrs_tbf_cmd_fini(struct nrs_tbf_cmd *cmd)
2101 {
2102         if (cmd->tc_cmd == NRS_CTL_TBF_START_RULE) {
2103                 if (cmd->u.tc_start.ts_valid_type == NRS_TBF_FLAG_JOBID)
2104                         nrs_tbf_jobid_cmd_fini(cmd);
2105                 else if (cmd->u.tc_start.ts_valid_type == NRS_TBF_FLAG_NID)
2106                         nrs_tbf_nid_cmd_fini(cmd);
2107                 else if (cmd->u.tc_start.ts_valid_type == NRS_TBF_FLAG_OPCODE)
2108                         nrs_tbf_opcode_cmd_fini(cmd);
2109         }
2110 }
2111
2112 static bool name_is_valid(const char *name)
2113 {
2114         int i;
2115
2116         for (i = 0; i < strlen(name); i++) {
2117                 if ((!isalnum(name[i])) &&
2118                     (name[i] != '_'))
2119                         return false;
2120         }
2121         return true;
2122 }
2123
2124 static int
2125 nrs_tbf_parse_value_pair(struct nrs_tbf_cmd *cmd, char *buffer)
2126 {
2127         char    *key;
2128         char    *val;
2129         int      rc;
2130         __u64    rate;
2131
2132         val = buffer;
2133         key = strsep(&val, "=");
2134         if (val == NULL || strlen(val) == 0)
2135                 return -EINVAL;
2136
2137         /* Key of the value pair */
2138         if (strcmp(key, "rate") == 0) {
2139                 rc = kstrtoull(val, 10, &rate);
2140                 if (rc)
2141                         return rc;
2142
2143                 if (rate <= 0 || rate >= LPROCFS_NRS_RATE_MAX)
2144                         return -EINVAL;
2145
2146                 if (cmd->tc_cmd == NRS_CTL_TBF_START_RULE)
2147                         cmd->u.tc_start.ts_rpc_rate = rate;
2148                 else if (cmd->tc_cmd == NRS_CTL_TBF_CHANGE_RULE)
2149                         cmd->u.tc_change.tc_rpc_rate = rate;
2150                 else
2151                         return -EINVAL;
2152         }  else if (strcmp(key, "rank") == 0) {
2153                 if (!name_is_valid(val))
2154                         return -EINVAL;
2155
2156                 if (cmd->tc_cmd == NRS_CTL_TBF_START_RULE)
2157                         cmd->u.tc_start.ts_next_name = val;
2158                 else if (cmd->tc_cmd == NRS_CTL_TBF_CHANGE_RULE)
2159                         cmd->u.tc_change.tc_next_name = val;
2160                 else
2161                         return -EINVAL;
2162         } else {
2163                 return -EINVAL;
2164         }
2165         return 0;
2166 }
2167
2168 static int
2169 nrs_tbf_parse_value_pairs(struct nrs_tbf_cmd *cmd, char *buffer)
2170 {
2171         char    *val;
2172         char    *token;
2173         int      rc;
2174
2175         val = buffer;
2176         while (val != NULL && strlen(val) != 0) {
2177                 token = strsep(&val, " ");
2178                 rc = nrs_tbf_parse_value_pair(cmd, token);
2179                 if (rc)
2180                         return rc;
2181         }
2182
2183         switch (cmd->tc_cmd) {
2184         case NRS_CTL_TBF_START_RULE:
2185                 if (cmd->u.tc_start.ts_rpc_rate == 0)
2186                         cmd->u.tc_start.ts_rpc_rate = tbf_rate;
2187                 break;
2188         case NRS_CTL_TBF_CHANGE_RULE:
2189                 if (cmd->u.tc_change.tc_rpc_rate == 0 &&
2190                     cmd->u.tc_change.tc_next_name == NULL)
2191                         return -EINVAL;
2192                 break;
2193         case NRS_CTL_TBF_STOP_RULE:
2194                 break;
2195         default:
2196                 return -EINVAL;
2197         }
2198         return 0;
2199 }
2200
2201 static struct nrs_tbf_cmd *
2202 nrs_tbf_parse_cmd(char *buffer, unsigned long count, __u32 type_flag)
2203 {
2204         static struct nrs_tbf_cmd       *cmd;
2205         char                            *token;
2206         char                            *val;
2207         int                              rc = 0;
2208
2209         OBD_ALLOC_PTR(cmd);
2210         if (cmd == NULL)
2211                 GOTO(out, rc = -ENOMEM);
2212         memset(cmd, 0, sizeof(*cmd));
2213
2214         val = buffer;
2215         token = strsep(&val, " ");
2216         if (val == NULL || strlen(val) == 0)
2217                 GOTO(out_free_cmd, rc = -EINVAL);
2218
2219         /* Type of the command */
2220         if (strcmp(token, "start") == 0) {
2221                 cmd->tc_cmd = NRS_CTL_TBF_START_RULE;
2222                 cmd->u.tc_start.ts_valid_type = type_flag;
2223         } else if (strcmp(token, "stop") == 0)
2224                 cmd->tc_cmd = NRS_CTL_TBF_STOP_RULE;
2225         else if (strcmp(token, "change") == 0)
2226                 cmd->tc_cmd = NRS_CTL_TBF_CHANGE_RULE;
2227         else
2228                 GOTO(out_free_cmd, rc = -EINVAL);
2229
2230         /* Name of the rule */
2231         token = strsep(&val, " ");
2232         if ((val == NULL && cmd->tc_cmd != NRS_CTL_TBF_STOP_RULE) ||
2233             !name_is_valid(token))
2234                 GOTO(out_free_cmd, rc = -EINVAL);
2235         cmd->tc_name = token;
2236
2237         if (cmd->tc_cmd == NRS_CTL_TBF_START_RULE) {
2238                 /* List of ID */
2239                 LASSERT(val);
2240                 token = val;
2241                 val = strrchr(token, '}');
2242                 if (!val)
2243                         GOTO(out_free_cmd, rc = -EINVAL);
2244
2245                 /* Skip '}' */
2246                 val++;
2247                 if (*val == '\0') {
2248                         val = NULL;
2249                 } else if (*val == ' ') {
2250                         *val = '\0';
2251                         val++;
2252                 } else
2253                         GOTO(out_free_cmd, rc = -EINVAL);
2254
2255                 rc = nrs_tbf_id_parse(cmd, token);
2256                 if (rc)
2257                         GOTO(out_free_cmd, rc);
2258         }
2259
2260         rc = nrs_tbf_parse_value_pairs(cmd, val);
2261         if (rc)
2262                 GOTO(out_cmd_fini, rc = -EINVAL);
2263         goto out;
2264 out_cmd_fini:
2265         nrs_tbf_cmd_fini(cmd);
2266 out_free_cmd:
2267         OBD_FREE_PTR(cmd);
2268 out:
2269         if (rc)
2270                 cmd = ERR_PTR(rc);
2271         return cmd;
2272 }
2273
2274 /**
2275  * Get the TBF policy type (nid, jobid, etc) preset by
2276  * proc entry 'nrs_policies' for command buffer parsing.
2277  *
2278  * \param[in] svc the PTLRPC service
2279  * \param[in] queue the NRS queue type
2280  *
2281  * \retval the preset TBF policy type flag
2282  */
2283 static __u32
2284 nrs_tbf_type_flag(struct ptlrpc_service *svc, enum ptlrpc_nrs_queue_type queue)
2285 {
2286         __u32   type;
2287         int     rc;
2288
2289         rc = ptlrpc_nrs_policy_control(svc, queue,
2290                                        NRS_POL_NAME_TBF,
2291                                        NRS_CTL_TBF_RD_TYPE_FLAG,
2292                                        true, &type);
2293         if (rc != 0)
2294                 type = NRS_TBF_FLAG_INVALID;
2295
2296         return type;
2297 }
2298
2299 extern struct nrs_core nrs_core;
2300 #define LPROCFS_WR_NRS_TBF_MAX_CMD (4096)
2301 static ssize_t
2302 ptlrpc_lprocfs_nrs_tbf_rule_seq_write(struct file *file,
2303                                       const char __user *buffer,
2304                                       size_t count, loff_t *off)
2305 {
2306         struct seq_file           *m = file->private_data;
2307         struct ptlrpc_service     *svc = m->private;
2308         char                      *kernbuf;
2309         char                      *val;
2310         int                        rc;
2311         static struct nrs_tbf_cmd *cmd;
2312         enum ptlrpc_nrs_queue_type queue = PTLRPC_NRS_QUEUE_BOTH;
2313         unsigned long              length;
2314         char                      *token;
2315
2316         OBD_ALLOC(kernbuf, LPROCFS_WR_NRS_TBF_MAX_CMD);
2317         if (kernbuf == NULL)
2318                 GOTO(out, rc = -ENOMEM);
2319
2320         if (count > LPROCFS_WR_NRS_TBF_MAX_CMD - 1)
2321                 GOTO(out_free_kernbuff, rc = -EINVAL);
2322
2323         if (copy_from_user(kernbuf, buffer, count))
2324                 GOTO(out_free_kernbuff, rc = -EFAULT);
2325
2326         val = kernbuf;
2327         token = strsep(&val, " ");
2328         if (val == NULL)
2329                 GOTO(out_free_kernbuff, rc = -EINVAL);
2330
2331         if (strcmp(token, "reg") == 0) {
2332                 queue = PTLRPC_NRS_QUEUE_REG;
2333         } else if (strcmp(token, "hp") == 0) {
2334                 queue = PTLRPC_NRS_QUEUE_HP;
2335         } else {
2336                 kernbuf[strlen(token)] = ' ';
2337                 val = kernbuf;
2338         }
2339         length = strlen(val);
2340
2341         if (length == 0)
2342                 GOTO(out_free_kernbuff, rc = -EINVAL);
2343
2344         if (queue == PTLRPC_NRS_QUEUE_HP && !nrs_svc_has_hp(svc))
2345                 GOTO(out_free_kernbuff, rc = -ENODEV);
2346         else if (queue == PTLRPC_NRS_QUEUE_BOTH && !nrs_svc_has_hp(svc))
2347                 queue = PTLRPC_NRS_QUEUE_REG;
2348
2349         cmd = nrs_tbf_parse_cmd(val, length, nrs_tbf_type_flag(svc, queue));
2350         if (IS_ERR(cmd))
2351                 GOTO(out_free_kernbuff, rc = PTR_ERR(cmd));
2352
2353         /**
2354          * Serialize NRS core lprocfs operations with policy registration/
2355          * unregistration.
2356          */
2357         mutex_lock(&nrs_core.nrs_mutex);
2358         rc = ptlrpc_nrs_policy_control(svc, queue,
2359                                        NRS_POL_NAME_TBF,
2360                                        NRS_CTL_TBF_WR_RULE,
2361                                        false, cmd);
2362         mutex_unlock(&nrs_core.nrs_mutex);
2363
2364         nrs_tbf_cmd_fini(cmd);
2365         OBD_FREE_PTR(cmd);
2366 out_free_kernbuff:
2367         OBD_FREE(kernbuf, LPROCFS_WR_NRS_TBF_MAX_CMD);
2368 out:
2369         return rc ? rc : count;
2370 }
2371 LPROC_SEQ_FOPS(ptlrpc_lprocfs_nrs_tbf_rule);
2372
2373 /**
2374  * Initializes a TBF policy's lprocfs interface for service \a svc
2375  *
2376  * \param[in] svc the service
2377  *
2378  * \retval 0    success
2379  * \retval != 0 error
2380  */
2381 static int nrs_tbf_lprocfs_init(struct ptlrpc_service *svc)
2382 {
2383         struct lprocfs_vars nrs_tbf_lprocfs_vars[] = {
2384                 { .name         = "nrs_tbf_rule",
2385                   .fops         = &ptlrpc_lprocfs_nrs_tbf_rule_fops,
2386                   .data = svc },
2387                 { NULL }
2388         };
2389
2390         if (svc->srv_procroot == NULL)
2391                 return 0;
2392
2393         return lprocfs_add_vars(svc->srv_procroot, nrs_tbf_lprocfs_vars, NULL);
2394 }
2395
2396 /**
2397  * Cleans up a TBF policy's lprocfs interface for service \a svc
2398  *
2399  * \param[in] svc the service
2400  */
2401 static void nrs_tbf_lprocfs_fini(struct ptlrpc_service *svc)
2402 {
2403         if (svc->srv_procroot == NULL)
2404                 return;
2405
2406         lprocfs_remove_proc_entry("nrs_tbf_rule", svc->srv_procroot);
2407 }
2408
2409 #endif /* CONFIG_PROC_FS */
2410
2411 /**
2412  * TBF policy operations
2413  */
2414 static const struct ptlrpc_nrs_pol_ops nrs_tbf_ops = {
2415         .op_policy_start        = nrs_tbf_start,
2416         .op_policy_stop         = nrs_tbf_stop,
2417         .op_policy_ctl          = nrs_tbf_ctl,
2418         .op_res_get             = nrs_tbf_res_get,
2419         .op_res_put             = nrs_tbf_res_put,
2420         .op_req_get             = nrs_tbf_req_get,
2421         .op_req_enqueue         = nrs_tbf_req_add,
2422         .op_req_dequeue         = nrs_tbf_req_del,
2423         .op_req_stop            = nrs_tbf_req_stop,
2424 #ifdef CONFIG_PROC_FS
2425         .op_lprocfs_init        = nrs_tbf_lprocfs_init,
2426         .op_lprocfs_fini        = nrs_tbf_lprocfs_fini,
2427 #endif
2428 };
2429
2430 /**
2431  * TBF policy configuration
2432  */
2433 struct ptlrpc_nrs_pol_conf nrs_conf_tbf = {
2434         .nc_name                = NRS_POL_NAME_TBF,
2435         .nc_ops                 = &nrs_tbf_ops,
2436         .nc_compat              = nrs_policy_compat_all,
2437 };
2438
2439 /** @} tbf */
2440
2441 /** @} nrs */
2442
2443 #endif /* HAVE_SERVER_SUPPORT */