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