Whamcloud - gitweb
LU-11014 mdc: remove obsolete intent opcodes
[fs/lustre-release.git] / 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 <lustre_req_layout.h>
46 #include "ptlrpc_internal.h"
47
48 /**
49  * \name tbf
50  *
51  * Token Bucket Filter over client NIDs
52  *
53  * @{
54  */
55
56 #define NRS_POL_NAME_TBF        "tbf"
57
58 static int tbf_jobid_cache_size = 8192;
59 module_param(tbf_jobid_cache_size, int, 0644);
60 MODULE_PARM_DESC(tbf_jobid_cache_size, "The size of jobid cache");
61
62 static int tbf_rate = 10000;
63 module_param(tbf_rate, int, 0644);
64 MODULE_PARM_DESC(tbf_rate, "Default rate limit in RPCs/s");
65
66 static int tbf_depth = 3;
67 module_param(tbf_depth, int, 0644);
68 MODULE_PARM_DESC(tbf_depth, "How many tokens that a client can save up");
69
70 static enum hrtimer_restart nrs_tbf_timer_cb(struct hrtimer *timer)
71 {
72         struct nrs_tbf_head *head = container_of(timer, struct nrs_tbf_head,
73                                                  th_timer);
74         struct ptlrpc_nrs   *nrs = head->th_res.res_policy->pol_nrs;
75         struct ptlrpc_service_part *svcpt = nrs->nrs_svcpt;
76
77         nrs->nrs_throttling = 0;
78         wake_up(&svcpt->scp_waitq);
79
80         return HRTIMER_NORESTART;
81 }
82
83 #define NRS_TBF_DEFAULT_RULE "default"
84
85 static void nrs_tbf_rule_fini(struct nrs_tbf_rule *rule)
86 {
87         LASSERT(atomic_read(&rule->tr_ref) == 0);
88         LASSERT(list_empty(&rule->tr_cli_list));
89         LASSERT(list_empty(&rule->tr_linkage));
90
91         rule->tr_head->th_ops->o_rule_fini(rule);
92         OBD_FREE_PTR(rule);
93 }
94
95 /**
96  * Decreases the rule's usage reference count, and stops the rule in case it
97  * was already stopping and have no more outstanding usage references (which
98  * indicates it has no more queued or started requests, and can be safely
99  * stopped).
100  */
101 static void nrs_tbf_rule_put(struct nrs_tbf_rule *rule)
102 {
103         if (atomic_dec_and_test(&rule->tr_ref))
104                 nrs_tbf_rule_fini(rule);
105 }
106
107 /**
108  * Increases the rule's usage reference count.
109  */
110 static inline void nrs_tbf_rule_get(struct nrs_tbf_rule *rule)
111 {
112         atomic_inc(&rule->tr_ref);
113 }
114
115 static void
116 nrs_tbf_cli_rule_put(struct nrs_tbf_client *cli)
117 {
118         LASSERT(!list_empty(&cli->tc_linkage));
119         LASSERT(cli->tc_rule);
120         spin_lock(&cli->tc_rule->tr_rule_lock);
121         list_del_init(&cli->tc_linkage);
122         spin_unlock(&cli->tc_rule->tr_rule_lock);
123         nrs_tbf_rule_put(cli->tc_rule);
124         cli->tc_rule = NULL;
125 }
126
127 static void
128 nrs_tbf_cli_reset_value(struct nrs_tbf_head *head,
129                         struct nrs_tbf_client *cli)
130
131 {
132         struct nrs_tbf_rule *rule = cli->tc_rule;
133
134         cli->tc_rpc_rate = rule->tr_rpc_rate;
135         cli->tc_nsecs = rule->tr_nsecs;
136         cli->tc_depth = rule->tr_depth;
137         cli->tc_ntoken = rule->tr_depth;
138         cli->tc_check_time = ktime_to_ns(ktime_get());
139         cli->tc_rule_sequence = atomic_read(&head->th_rule_sequence);
140         cli->tc_rule_generation = rule->tr_generation;
141
142         if (cli->tc_in_heap)
143                 cfs_binheap_relocate(head->th_binheap,
144                                      &cli->tc_node);
145 }
146
147 static void
148 nrs_tbf_cli_reset(struct nrs_tbf_head *head,
149                   struct nrs_tbf_rule *rule,
150                   struct nrs_tbf_client *cli)
151 {
152         spin_lock(&cli->tc_rule_lock);
153         if (cli->tc_rule != NULL && !list_empty(&cli->tc_linkage)) {
154                 LASSERT(rule != cli->tc_rule);
155                 nrs_tbf_cli_rule_put(cli);
156         }
157         LASSERT(cli->tc_rule == NULL);
158         LASSERT(list_empty(&cli->tc_linkage));
159         /* Rule's ref is added before called */
160         cli->tc_rule = rule;
161         spin_lock(&rule->tr_rule_lock);
162         list_add_tail(&cli->tc_linkage, &rule->tr_cli_list);
163         spin_unlock(&rule->tr_rule_lock);
164         spin_unlock(&cli->tc_rule_lock);
165         nrs_tbf_cli_reset_value(head, cli);
166 }
167
168 static int
169 nrs_tbf_rule_dump(struct nrs_tbf_rule *rule, struct seq_file *m)
170 {
171         return rule->tr_head->th_ops->o_rule_dump(rule, m);
172 }
173
174 static int
175 nrs_tbf_rule_dump_all(struct nrs_tbf_head *head, struct seq_file *m)
176 {
177         struct nrs_tbf_rule *rule;
178         int rc = 0;
179
180         LASSERT(head != NULL);
181         spin_lock(&head->th_rule_lock);
182         /* List the rules from newest to oldest */
183         list_for_each_entry(rule, &head->th_list, tr_linkage) {
184                 LASSERT((rule->tr_flags & NTRS_STOPPING) == 0);
185                 rc = nrs_tbf_rule_dump(rule, m);
186                 if (rc) {
187                         rc = -ENOSPC;
188                         break;
189                 }
190         }
191         spin_unlock(&head->th_rule_lock);
192
193         return rc;
194 }
195
196 static struct nrs_tbf_rule *
197 nrs_tbf_rule_find_nolock(struct nrs_tbf_head *head,
198                          const char *name)
199 {
200         struct nrs_tbf_rule *rule;
201
202         LASSERT(head != NULL);
203         list_for_each_entry(rule, &head->th_list, tr_linkage) {
204                 LASSERT((rule->tr_flags & NTRS_STOPPING) == 0);
205                 if (strcmp(rule->tr_name, name) == 0) {
206                         nrs_tbf_rule_get(rule);
207                         return rule;
208                 }
209         }
210         return NULL;
211 }
212
213 static struct nrs_tbf_rule *
214 nrs_tbf_rule_find(struct nrs_tbf_head *head,
215                   const char *name)
216 {
217         struct nrs_tbf_rule *rule;
218
219         LASSERT(head != NULL);
220         spin_lock(&head->th_rule_lock);
221         rule = nrs_tbf_rule_find_nolock(head, name);
222         spin_unlock(&head->th_rule_lock);
223         return rule;
224 }
225
226 static struct nrs_tbf_rule *
227 nrs_tbf_rule_match(struct nrs_tbf_head *head,
228                    struct nrs_tbf_client *cli)
229 {
230         struct nrs_tbf_rule *rule = NULL;
231         struct nrs_tbf_rule *tmp_rule;
232
233         spin_lock(&head->th_rule_lock);
234         /* Match the newest rule in the list */
235         list_for_each_entry(tmp_rule, &head->th_list, tr_linkage) {
236                 LASSERT((tmp_rule->tr_flags & NTRS_STOPPING) == 0);
237                 if (head->th_ops->o_rule_match(tmp_rule, cli)) {
238                         rule = tmp_rule;
239                         break;
240                 }
241         }
242
243         if (rule == NULL)
244                 rule = head->th_rule;
245
246         nrs_tbf_rule_get(rule);
247         spin_unlock(&head->th_rule_lock);
248         return rule;
249 }
250
251 static void
252 nrs_tbf_cli_init(struct nrs_tbf_head *head,
253                  struct nrs_tbf_client *cli,
254                  struct ptlrpc_request *req)
255 {
256         struct nrs_tbf_rule *rule;
257
258         memset(cli, 0, sizeof(*cli));
259         cli->tc_in_heap = false;
260         head->th_ops->o_cli_init(cli, req);
261         INIT_LIST_HEAD(&cli->tc_list);
262         INIT_LIST_HEAD(&cli->tc_linkage);
263         spin_lock_init(&cli->tc_rule_lock);
264         atomic_set(&cli->tc_ref, 1);
265         rule = nrs_tbf_rule_match(head, cli);
266         nrs_tbf_cli_reset(head, rule, cli);
267 }
268
269 static void
270 nrs_tbf_cli_fini(struct nrs_tbf_client *cli)
271 {
272         LASSERT(list_empty(&cli->tc_list));
273         LASSERT(!cli->tc_in_heap);
274         LASSERT(atomic_read(&cli->tc_ref) == 0);
275         spin_lock(&cli->tc_rule_lock);
276         nrs_tbf_cli_rule_put(cli);
277         spin_unlock(&cli->tc_rule_lock);
278         OBD_FREE_PTR(cli);
279 }
280
281 static int
282 nrs_tbf_rule_start(struct ptlrpc_nrs_policy *policy,
283                    struct nrs_tbf_head *head,
284                    struct nrs_tbf_cmd *start)
285 {
286         struct nrs_tbf_rule     *rule;
287         struct nrs_tbf_rule     *tmp_rule;
288         struct nrs_tbf_rule     *next_rule;
289         char                    *next_name = start->u.tc_start.ts_next_name;
290         int                      rc;
291
292         rule = nrs_tbf_rule_find(head, start->tc_name);
293         if (rule) {
294                 nrs_tbf_rule_put(rule);
295                 return -EEXIST;
296         }
297
298         OBD_CPT_ALLOC_PTR(rule, nrs_pol2cptab(policy), nrs_pol2cptid(policy));
299         if (rule == NULL)
300                 return -ENOMEM;
301
302         memcpy(rule->tr_name, start->tc_name, strlen(start->tc_name));
303         rule->tr_rpc_rate = start->u.tc_start.ts_rpc_rate;
304         rule->tr_flags = start->u.tc_start.ts_rule_flags;
305         rule->tr_nsecs = NSEC_PER_SEC;
306         do_div(rule->tr_nsecs, rule->tr_rpc_rate);
307         rule->tr_depth = tbf_depth;
308         atomic_set(&rule->tr_ref, 1);
309         INIT_LIST_HEAD(&rule->tr_cli_list);
310         INIT_LIST_HEAD(&rule->tr_nids);
311         INIT_LIST_HEAD(&rule->tr_linkage);
312         spin_lock_init(&rule->tr_rule_lock);
313         rule->tr_head = head;
314
315         rc = head->th_ops->o_rule_init(policy, rule, start);
316         if (rc) {
317                 OBD_FREE_PTR(rule);
318                 return rc;
319         }
320
321         /* Add as the newest rule */
322         spin_lock(&head->th_rule_lock);
323         tmp_rule = nrs_tbf_rule_find_nolock(head, start->tc_name);
324         if (tmp_rule) {
325                 spin_unlock(&head->th_rule_lock);
326                 nrs_tbf_rule_put(tmp_rule);
327                 nrs_tbf_rule_put(rule);
328                 return -EEXIST;
329         }
330
331         if (next_name) {
332                 next_rule = nrs_tbf_rule_find_nolock(head, next_name);
333                 if (!next_rule) {
334                         spin_unlock(&head->th_rule_lock);
335                         nrs_tbf_rule_put(rule);
336                         return -ENOENT;
337                 }
338
339                 list_add(&rule->tr_linkage, next_rule->tr_linkage.prev);
340                 nrs_tbf_rule_put(next_rule);
341         } else {
342                 /* Add on the top of the rule list */
343                 list_add(&rule->tr_linkage, &head->th_list);
344         }
345         spin_unlock(&head->th_rule_lock);
346         atomic_inc(&head->th_rule_sequence);
347         if (start->u.tc_start.ts_rule_flags & NTRS_DEFAULT) {
348                 rule->tr_flags |= NTRS_DEFAULT;
349                 LASSERT(head->th_rule == NULL);
350                 head->th_rule = rule;
351         }
352
353         CDEBUG(D_RPCTRACE, "TBF starts rule@%p rate %llu gen %llu\n",
354                rule, rule->tr_rpc_rate, rule->tr_generation);
355
356         return 0;
357 }
358
359 /**
360  * Change the rank of a rule in the rule list
361  *
362  * The matched rule will be moved to the position right before another
363  * given rule.
364  *
365  * \param[in] policy    the policy instance
366  * \param[in] head      the TBF policy instance
367  * \param[in] name      the rule name to be moved
368  * \param[in] next_name the rule name before which the matched rule will be
369  *                      moved
370  *
371  */
372 static int
373 nrs_tbf_rule_change_rank(struct ptlrpc_nrs_policy *policy,
374                          struct nrs_tbf_head *head,
375                          char *name,
376                          char *next_name)
377 {
378         struct nrs_tbf_rule     *rule = NULL;
379         struct nrs_tbf_rule     *next_rule = NULL;
380         int                      rc = 0;
381
382         LASSERT(head != NULL);
383
384         spin_lock(&head->th_rule_lock);
385         rule = nrs_tbf_rule_find_nolock(head, name);
386         if (!rule)
387                 GOTO(out, rc = -ENOENT);
388
389         if (strcmp(name, next_name) == 0)
390                 GOTO(out_put, rc);
391
392         next_rule = nrs_tbf_rule_find_nolock(head, next_name);
393         if (!next_rule)
394                 GOTO(out_put, rc = -ENOENT);
395
396         list_move(&rule->tr_linkage, next_rule->tr_linkage.prev);
397         nrs_tbf_rule_put(next_rule);
398 out_put:
399         nrs_tbf_rule_put(rule);
400 out:
401         spin_unlock(&head->th_rule_lock);
402         return rc;
403 }
404
405 static int
406 nrs_tbf_rule_change_rate(struct ptlrpc_nrs_policy *policy,
407                          struct nrs_tbf_head *head,
408                          char *name,
409                          __u64 rate)
410 {
411         struct nrs_tbf_rule *rule;
412
413         assert_spin_locked(&policy->pol_nrs->nrs_lock);
414
415         rule = nrs_tbf_rule_find(head, name);
416         if (rule == NULL)
417                 return -ENOENT;
418
419         rule->tr_rpc_rate = rate;
420         rule->tr_nsecs = NSEC_PER_SEC;
421         do_div(rule->tr_nsecs, rule->tr_rpc_rate);
422         rule->tr_generation++;
423         nrs_tbf_rule_put(rule);
424
425         return 0;
426 }
427
428 static int
429 nrs_tbf_rule_change(struct ptlrpc_nrs_policy *policy,
430                     struct nrs_tbf_head *head,
431                     struct nrs_tbf_cmd *change)
432 {
433         __u64    rate = change->u.tc_change.tc_rpc_rate;
434         char    *next_name = change->u.tc_change.tc_next_name;
435         int      rc;
436
437         if (rate != 0) {
438                 rc = nrs_tbf_rule_change_rate(policy, head, change->tc_name,
439                                               rate);
440                 if (rc)
441                         return rc;
442         }
443
444         if (next_name) {
445                 rc = nrs_tbf_rule_change_rank(policy, head, change->tc_name,
446                                               next_name);
447                 if (rc)
448                         return rc;
449         }
450
451         return 0;
452 }
453
454 static int
455 nrs_tbf_rule_stop(struct ptlrpc_nrs_policy *policy,
456                   struct nrs_tbf_head *head,
457                   struct nrs_tbf_cmd *stop)
458 {
459         struct nrs_tbf_rule *rule;
460
461         assert_spin_locked(&policy->pol_nrs->nrs_lock);
462
463         if (strcmp(stop->tc_name, NRS_TBF_DEFAULT_RULE) == 0)
464                 return -EPERM;
465
466         rule = nrs_tbf_rule_find(head, stop->tc_name);
467         if (rule == NULL)
468                 return -ENOENT;
469
470         list_del_init(&rule->tr_linkage);
471         rule->tr_flags |= NTRS_STOPPING;
472         nrs_tbf_rule_put(rule);
473         nrs_tbf_rule_put(rule);
474
475         return 0;
476 }
477
478 static int
479 nrs_tbf_command(struct ptlrpc_nrs_policy *policy,
480                 struct nrs_tbf_head *head,
481                 struct nrs_tbf_cmd *cmd)
482 {
483         int rc;
484
485         assert_spin_locked(&policy->pol_nrs->nrs_lock);
486
487         switch (cmd->tc_cmd) {
488         case NRS_CTL_TBF_START_RULE:
489                 if (cmd->u.tc_start.ts_valid_type != head->th_type_flag)
490                         return -EINVAL;
491
492                 spin_unlock(&policy->pol_nrs->nrs_lock);
493                 rc = nrs_tbf_rule_start(policy, head, cmd);
494                 spin_lock(&policy->pol_nrs->nrs_lock);
495                 return rc;
496         case NRS_CTL_TBF_CHANGE_RULE:
497                 rc = nrs_tbf_rule_change(policy, head, cmd);
498                 return rc;
499         case NRS_CTL_TBF_STOP_RULE:
500                 rc = nrs_tbf_rule_stop(policy, head, cmd);
501                 /* Take it as a success, if not exists at all */
502                 return rc == -ENOENT ? 0 : rc;
503         default:
504                 return -EFAULT;
505         }
506 }
507
508 /**
509  * Binary heap predicate.
510  *
511  * \param[in] e1 the first binheap node to compare
512  * \param[in] e2 the second binheap node to compare
513  *
514  * \retval 0 e1 > e2
515  * \retval 1 e1 < e2
516  */
517 static int
518 tbf_cli_compare(struct cfs_binheap_node *e1, struct cfs_binheap_node *e2)
519 {
520         struct nrs_tbf_client *cli1;
521         struct nrs_tbf_client *cli2;
522
523         cli1 = container_of(e1, struct nrs_tbf_client, tc_node);
524         cli2 = container_of(e2, struct nrs_tbf_client, tc_node);
525
526         if (cli1->tc_deadline < cli2->tc_deadline)
527                 return 1;
528         else if (cli1->tc_deadline > cli2->tc_deadline)
529                 return 0;
530
531         if (cli1->tc_check_time < cli2->tc_check_time)
532                 return 1;
533         else if (cli1->tc_check_time > cli2->tc_check_time)
534                 return 0;
535
536         /* Maybe need more comparasion, e.g. request number in the rules */
537         return 1;
538 }
539
540 /**
541  * TBF binary heap operations
542  */
543 static struct cfs_binheap_ops nrs_tbf_heap_ops = {
544         .hop_enter      = NULL,
545         .hop_exit       = NULL,
546         .hop_compare    = tbf_cli_compare,
547 };
548
549 static unsigned nrs_tbf_jobid_hop_hash(struct cfs_hash *hs, const void *key,
550                                   unsigned mask)
551 {
552         return cfs_hash_djb2_hash(key, strlen(key), mask);
553 }
554
555 static int nrs_tbf_jobid_hop_keycmp(const void *key, struct hlist_node *hnode)
556 {
557         struct nrs_tbf_client *cli = hlist_entry(hnode,
558                                                      struct nrs_tbf_client,
559                                                      tc_hnode);
560
561         return (strcmp(cli->tc_jobid, key) == 0);
562 }
563
564 static void *nrs_tbf_jobid_hop_key(struct hlist_node *hnode)
565 {
566         struct nrs_tbf_client *cli = hlist_entry(hnode,
567                                                      struct nrs_tbf_client,
568                                                      tc_hnode);
569
570         return cli->tc_jobid;
571 }
572
573 static void *nrs_tbf_hop_object(struct hlist_node *hnode)
574 {
575         return hlist_entry(hnode, struct nrs_tbf_client, tc_hnode);
576 }
577
578 static void nrs_tbf_jobid_hop_get(struct cfs_hash *hs, struct hlist_node *hnode)
579 {
580         struct nrs_tbf_client *cli = hlist_entry(hnode,
581                                                      struct nrs_tbf_client,
582                                                      tc_hnode);
583
584         atomic_inc(&cli->tc_ref);
585 }
586
587 static void nrs_tbf_jobid_hop_put(struct cfs_hash *hs, struct hlist_node *hnode)
588 {
589         struct nrs_tbf_client *cli = hlist_entry(hnode,
590                                                      struct nrs_tbf_client,
591                                                      tc_hnode);
592
593         atomic_dec(&cli->tc_ref);
594 }
595
596 static void
597 nrs_tbf_jobid_hop_exit(struct cfs_hash *hs, struct hlist_node *hnode)
598
599 {
600         struct nrs_tbf_client *cli = hlist_entry(hnode,
601                                                  struct nrs_tbf_client,
602                                                  tc_hnode);
603
604         LASSERT(atomic_read(&cli->tc_ref) == 0);
605         nrs_tbf_cli_fini(cli);
606 }
607
608 static struct cfs_hash_ops nrs_tbf_jobid_hash_ops = {
609         .hs_hash        = nrs_tbf_jobid_hop_hash,
610         .hs_keycmp      = nrs_tbf_jobid_hop_keycmp,
611         .hs_key         = nrs_tbf_jobid_hop_key,
612         .hs_object      = nrs_tbf_hop_object,
613         .hs_get         = nrs_tbf_jobid_hop_get,
614         .hs_put         = nrs_tbf_jobid_hop_put,
615         .hs_put_locked  = nrs_tbf_jobid_hop_put,
616         .hs_exit        = nrs_tbf_jobid_hop_exit,
617 };
618
619 #define NRS_TBF_JOBID_HASH_FLAGS (CFS_HASH_SPIN_BKTLOCK | \
620                                   CFS_HASH_NO_ITEMREF | \
621                                   CFS_HASH_DEPTH)
622
623 static struct nrs_tbf_client *
624 nrs_tbf_jobid_hash_lookup(struct cfs_hash *hs,
625                           struct cfs_hash_bd *bd,
626                           const char *jobid)
627 {
628         struct hlist_node *hnode;
629         struct nrs_tbf_client *cli;
630
631         hnode = cfs_hash_bd_lookup_locked(hs, bd, (void *)jobid);
632         if (hnode == NULL)
633                 return NULL;
634
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         if (rc) {
791                 cfs_hash_putref(head->th_cli_hash);
792                 head->th_cli_hash = NULL;
793         }
794
795         return rc;
796 }
797
798 /**
799  * Frees jobid of \a list.
800  *
801  */
802 static void
803 nrs_tbf_jobid_list_free(struct list_head *jobid_list)
804 {
805         struct nrs_tbf_jobid *jobid, *n;
806
807         list_for_each_entry_safe(jobid, n, jobid_list, tj_linkage) {
808                 OBD_FREE(jobid->tj_id, strlen(jobid->tj_id) + 1);
809                 list_del(&jobid->tj_linkage);
810                 OBD_FREE(jobid, sizeof(struct nrs_tbf_jobid));
811         }
812 }
813
814 static int
815 nrs_tbf_jobid_list_add(struct cfs_lstr *id, struct list_head *jobid_list)
816 {
817         struct nrs_tbf_jobid *jobid;
818         char *ptr;
819
820         OBD_ALLOC(jobid, sizeof(struct nrs_tbf_jobid));
821         if (jobid == NULL)
822                 return -ENOMEM;
823
824         OBD_ALLOC(jobid->tj_id, id->ls_len + 1);
825         if (jobid->tj_id == NULL) {
826                 OBD_FREE(jobid, sizeof(struct nrs_tbf_jobid));
827                 return -ENOMEM;
828         }
829
830         memcpy(jobid->tj_id, id->ls_str, id->ls_len);
831         ptr = lprocfs_strnstr(id->ls_str, "*", id->ls_len);
832         if (ptr == NULL)
833                 jobid->tj_match_flag = NRS_TBF_MATCH_FULL;
834         else
835                 jobid->tj_match_flag = NRS_TBF_MATCH_WILDCARD;
836
837         list_add_tail(&jobid->tj_linkage, jobid_list);
838         return 0;
839 }
840
841 static bool
842 cfs_match_wildcard(const char *pattern, const char *content)
843 {
844         if (*pattern == '\0' && *content == '\0')
845                 return true;
846
847         if (*pattern == '*' && *(pattern + 1) != '\0' && *content == '\0')
848                 return false;
849
850         while (*pattern == *content) {
851                 pattern++;
852                 content++;
853                 if (*pattern == '\0' && *content == '\0')
854                         return true;
855
856                 if (*pattern == '*' && *(pattern + 1) != '\0' &&
857                     *content == '\0')
858                         return false;
859         }
860
861         if (*pattern == '*')
862                 return (cfs_match_wildcard(pattern + 1, content) ||
863                         cfs_match_wildcard(pattern, content + 1));
864
865         return false;
866 }
867
868 static inline bool
869 nrs_tbf_jobid_match(const struct nrs_tbf_jobid *jobid, const char *id)
870 {
871         if (jobid->tj_match_flag == NRS_TBF_MATCH_FULL)
872                 return strcmp(jobid->tj_id, id) == 0;
873
874         if (jobid->tj_match_flag == NRS_TBF_MATCH_WILDCARD)
875                 return cfs_match_wildcard(jobid->tj_id, id);
876
877         return false;
878 }
879
880 static int
881 nrs_tbf_jobid_list_match(struct list_head *jobid_list, char *id)
882 {
883         struct nrs_tbf_jobid *jobid;
884
885         list_for_each_entry(jobid, jobid_list, tj_linkage) {
886                 if (nrs_tbf_jobid_match(jobid, id))
887                         return 1;
888         }
889         return 0;
890 }
891
892 static int
893 nrs_tbf_jobid_list_parse(char *str, int len, struct list_head *jobid_list)
894 {
895         struct cfs_lstr src;
896         struct cfs_lstr res;
897         int rc = 0;
898         ENTRY;
899
900         src.ls_str = str;
901         src.ls_len = len;
902         INIT_LIST_HEAD(jobid_list);
903         while (src.ls_str) {
904                 rc = cfs_gettok(&src, ' ', &res);
905                 if (rc == 0) {
906                         rc = -EINVAL;
907                         break;
908                 }
909                 rc = nrs_tbf_jobid_list_add(&res, jobid_list);
910                 if (rc)
911                         break;
912         }
913         if (rc)
914                 nrs_tbf_jobid_list_free(jobid_list);
915         RETURN(rc);
916 }
917
918 static void nrs_tbf_jobid_cmd_fini(struct nrs_tbf_cmd *cmd)
919 {
920         if (!list_empty(&cmd->u.tc_start.ts_jobids))
921                 nrs_tbf_jobid_list_free(&cmd->u.tc_start.ts_jobids);
922         if (cmd->u.tc_start.ts_jobids_str)
923                 OBD_FREE(cmd->u.tc_start.ts_jobids_str,
924                          strlen(cmd->u.tc_start.ts_jobids_str) + 1);
925 }
926
927 static int nrs_tbf_check_id_value(struct cfs_lstr *src, char *key)
928 {
929         struct cfs_lstr res;
930         int keylen = strlen(key);
931         int rc;
932
933         rc = cfs_gettok(src, '=', &res);
934         if (rc == 0 || res.ls_len != keylen ||
935             strncmp(res.ls_str, key, keylen) != 0 ||
936             src->ls_len <= 2 || src->ls_str[0] != '{' ||
937             src->ls_str[src->ls_len - 1] != '}')
938                 return -EINVAL;
939
940         /* Skip '{' and '}' */
941         src->ls_str++;
942         src->ls_len -= 2;
943         return 0;
944 }
945
946 static int nrs_tbf_jobid_parse(struct nrs_tbf_cmd *cmd, char *id)
947 {
948         struct cfs_lstr src;
949         int rc;
950
951         src.ls_str = id;
952         src.ls_len = strlen(id);
953         rc = nrs_tbf_check_id_value(&src, "jobid");
954         if (rc)
955                 return rc;
956
957         OBD_ALLOC(cmd->u.tc_start.ts_jobids_str, src.ls_len + 1);
958         if (cmd->u.tc_start.ts_jobids_str == NULL)
959                 return -ENOMEM;
960
961         memcpy(cmd->u.tc_start.ts_jobids_str, src.ls_str, src.ls_len);
962
963         /* parse jobid list */
964         rc = nrs_tbf_jobid_list_parse(cmd->u.tc_start.ts_jobids_str,
965                                       strlen(cmd->u.tc_start.ts_jobids_str),
966                                       &cmd->u.tc_start.ts_jobids);
967         if (rc)
968                 nrs_tbf_jobid_cmd_fini(cmd);
969
970         return rc;
971 }
972
973 static int nrs_tbf_jobid_rule_init(struct ptlrpc_nrs_policy *policy,
974                                    struct nrs_tbf_rule *rule,
975                                    struct nrs_tbf_cmd *start)
976 {
977         int rc = 0;
978
979         LASSERT(start->u.tc_start.ts_jobids_str);
980         OBD_ALLOC(rule->tr_jobids_str,
981                   strlen(start->u.tc_start.ts_jobids_str) + 1);
982         if (rule->tr_jobids_str == NULL)
983                 return -ENOMEM;
984
985         memcpy(rule->tr_jobids_str,
986                start->u.tc_start.ts_jobids_str,
987                strlen(start->u.tc_start.ts_jobids_str));
988
989         INIT_LIST_HEAD(&rule->tr_jobids);
990         if (!list_empty(&start->u.tc_start.ts_jobids)) {
991                 rc = nrs_tbf_jobid_list_parse(rule->tr_jobids_str,
992                                               strlen(rule->tr_jobids_str),
993                                               &rule->tr_jobids);
994                 if (rc)
995                         CERROR("jobids {%s} illegal\n", rule->tr_jobids_str);
996         }
997         if (rc)
998                 OBD_FREE(rule->tr_jobids_str,
999                          strlen(start->u.tc_start.ts_jobids_str) + 1);
1000         return rc;
1001 }
1002
1003 static int
1004 nrs_tbf_jobid_rule_dump(struct nrs_tbf_rule *rule, struct seq_file *m)
1005 {
1006         seq_printf(m, "%s {%s} %llu, ref %d\n", rule->tr_name,
1007                    rule->tr_jobids_str, rule->tr_rpc_rate,
1008                    atomic_read(&rule->tr_ref) - 1);
1009         return 0;
1010 }
1011
1012 static int
1013 nrs_tbf_jobid_rule_match(struct nrs_tbf_rule *rule,
1014                          struct nrs_tbf_client *cli)
1015 {
1016         return nrs_tbf_jobid_list_match(&rule->tr_jobids, cli->tc_jobid);
1017 }
1018
1019 static void nrs_tbf_jobid_rule_fini(struct nrs_tbf_rule *rule)
1020 {
1021         if (!list_empty(&rule->tr_jobids))
1022                 nrs_tbf_jobid_list_free(&rule->tr_jobids);
1023         LASSERT(rule->tr_jobids_str != NULL);
1024         OBD_FREE(rule->tr_jobids_str, strlen(rule->tr_jobids_str) + 1);
1025 }
1026
1027 static struct nrs_tbf_ops nrs_tbf_jobid_ops = {
1028         .o_name = NRS_TBF_TYPE_JOBID,
1029         .o_startup = nrs_tbf_jobid_startup,
1030         .o_cli_find = nrs_tbf_jobid_cli_find,
1031         .o_cli_findadd = nrs_tbf_jobid_cli_findadd,
1032         .o_cli_put = nrs_tbf_jobid_cli_put,
1033         .o_cli_init = nrs_tbf_jobid_cli_init,
1034         .o_rule_init = nrs_tbf_jobid_rule_init,
1035         .o_rule_dump = nrs_tbf_jobid_rule_dump,
1036         .o_rule_match = nrs_tbf_jobid_rule_match,
1037         .o_rule_fini = nrs_tbf_jobid_rule_fini,
1038 };
1039
1040 /**
1041  * libcfs_hash operations for nrs_tbf_net::cn_cli_hash
1042  *
1043  * This uses ptlrpc_request::rq_peer.nid as its key, in order to hash
1044  * nrs_tbf_client objects.
1045  */
1046 #define NRS_TBF_NID_BKT_BITS    8
1047 #define NRS_TBF_NID_BITS        16
1048
1049 static unsigned nrs_tbf_nid_hop_hash(struct cfs_hash *hs, const void *key,
1050                                   unsigned mask)
1051 {
1052         return cfs_hash_djb2_hash(key, sizeof(lnet_nid_t), mask);
1053 }
1054
1055 static int nrs_tbf_nid_hop_keycmp(const void *key, struct hlist_node *hnode)
1056 {
1057         lnet_nid_t            *nid = (lnet_nid_t *)key;
1058         struct nrs_tbf_client *cli = hlist_entry(hnode,
1059                                                      struct nrs_tbf_client,
1060                                                      tc_hnode);
1061
1062         return *nid == cli->tc_nid;
1063 }
1064
1065 static void *nrs_tbf_nid_hop_key(struct hlist_node *hnode)
1066 {
1067         struct nrs_tbf_client *cli = hlist_entry(hnode,
1068                                                      struct nrs_tbf_client,
1069                                                      tc_hnode);
1070
1071         return &cli->tc_nid;
1072 }
1073
1074 static void nrs_tbf_nid_hop_get(struct cfs_hash *hs, struct hlist_node *hnode)
1075 {
1076         struct nrs_tbf_client *cli = hlist_entry(hnode,
1077                                                      struct nrs_tbf_client,
1078                                                      tc_hnode);
1079
1080         atomic_inc(&cli->tc_ref);
1081 }
1082
1083 static void nrs_tbf_nid_hop_put(struct cfs_hash *hs, struct hlist_node *hnode)
1084 {
1085         struct nrs_tbf_client *cli = hlist_entry(hnode,
1086                                                      struct nrs_tbf_client,
1087                                                      tc_hnode);
1088
1089         atomic_dec(&cli->tc_ref);
1090 }
1091
1092 static void nrs_tbf_nid_hop_exit(struct cfs_hash *hs, struct hlist_node *hnode)
1093 {
1094         struct nrs_tbf_client *cli = hlist_entry(hnode,
1095                                                      struct nrs_tbf_client,
1096                                                      tc_hnode);
1097
1098         LASSERTF(atomic_read(&cli->tc_ref) == 0,
1099                  "Busy TBF object from client with NID %s, with %d refs\n",
1100                  libcfs_nid2str(cli->tc_nid), atomic_read(&cli->tc_ref));
1101
1102         nrs_tbf_cli_fini(cli);
1103 }
1104
1105 static struct cfs_hash_ops nrs_tbf_nid_hash_ops = {
1106         .hs_hash        = nrs_tbf_nid_hop_hash,
1107         .hs_keycmp      = nrs_tbf_nid_hop_keycmp,
1108         .hs_key         = nrs_tbf_nid_hop_key,
1109         .hs_object      = nrs_tbf_hop_object,
1110         .hs_get         = nrs_tbf_nid_hop_get,
1111         .hs_put         = nrs_tbf_nid_hop_put,
1112         .hs_put_locked  = nrs_tbf_nid_hop_put,
1113         .hs_exit        = nrs_tbf_nid_hop_exit,
1114 };
1115
1116 static struct nrs_tbf_client *
1117 nrs_tbf_nid_cli_find(struct nrs_tbf_head *head,
1118                      struct ptlrpc_request *req)
1119 {
1120         return cfs_hash_lookup(head->th_cli_hash, &req->rq_peer.nid);
1121 }
1122
1123 static struct nrs_tbf_client *
1124 nrs_tbf_nid_cli_findadd(struct nrs_tbf_head *head,
1125                         struct nrs_tbf_client *cli)
1126 {
1127         return cfs_hash_findadd_unique(head->th_cli_hash, &cli->tc_nid,
1128                                        &cli->tc_hnode);
1129 }
1130
1131 static void
1132 nrs_tbf_nid_cli_put(struct nrs_tbf_head *head,
1133                       struct nrs_tbf_client *cli)
1134 {
1135         cfs_hash_put(head->th_cli_hash, &cli->tc_hnode);
1136 }
1137
1138 static int
1139 nrs_tbf_nid_startup(struct ptlrpc_nrs_policy *policy,
1140                     struct nrs_tbf_head *head)
1141 {
1142         struct nrs_tbf_cmd      start;
1143         int rc;
1144
1145         head->th_cli_hash = cfs_hash_create("nrs_tbf_hash",
1146                                             NRS_TBF_NID_BITS,
1147                                             NRS_TBF_NID_BITS,
1148                                             NRS_TBF_NID_BKT_BITS, 0,
1149                                             CFS_HASH_MIN_THETA,
1150                                             CFS_HASH_MAX_THETA,
1151                                             &nrs_tbf_nid_hash_ops,
1152                                             CFS_HASH_RW_BKTLOCK);
1153         if (head->th_cli_hash == NULL)
1154                 return -ENOMEM;
1155
1156         memset(&start, 0, sizeof(start));
1157         start.u.tc_start.ts_nids_str = "*";
1158
1159         start.u.tc_start.ts_rpc_rate = tbf_rate;
1160         start.u.tc_start.ts_rule_flags = NTRS_DEFAULT;
1161         start.tc_name = NRS_TBF_DEFAULT_RULE;
1162         INIT_LIST_HEAD(&start.u.tc_start.ts_nids);
1163         rc = nrs_tbf_rule_start(policy, head, &start);
1164         if (rc) {
1165                 cfs_hash_putref(head->th_cli_hash);
1166                 head->th_cli_hash = NULL;
1167         }
1168
1169         return rc;
1170 }
1171
1172 static void
1173 nrs_tbf_nid_cli_init(struct nrs_tbf_client *cli,
1174                              struct ptlrpc_request *req)
1175 {
1176         cli->tc_nid = req->rq_peer.nid;
1177 }
1178
1179 static int nrs_tbf_nid_rule_init(struct ptlrpc_nrs_policy *policy,
1180                                  struct nrs_tbf_rule *rule,
1181                                  struct nrs_tbf_cmd *start)
1182 {
1183         LASSERT(start->u.tc_start.ts_nids_str);
1184         OBD_ALLOC(rule->tr_nids_str,
1185                   strlen(start->u.tc_start.ts_nids_str) + 1);
1186         if (rule->tr_nids_str == NULL)
1187                 return -ENOMEM;
1188
1189         memcpy(rule->tr_nids_str,
1190                start->u.tc_start.ts_nids_str,
1191                strlen(start->u.tc_start.ts_nids_str));
1192
1193         INIT_LIST_HEAD(&rule->tr_nids);
1194         if (!list_empty(&start->u.tc_start.ts_nids)) {
1195                 if (cfs_parse_nidlist(rule->tr_nids_str,
1196                                       strlen(rule->tr_nids_str),
1197                                       &rule->tr_nids) <= 0) {
1198                         CERROR("nids {%s} illegal\n",
1199                                rule->tr_nids_str);
1200                         OBD_FREE(rule->tr_nids_str,
1201                                  strlen(start->u.tc_start.ts_nids_str) + 1);
1202                         return -EINVAL;
1203                 }
1204         }
1205         return 0;
1206 }
1207
1208 static int
1209 nrs_tbf_nid_rule_dump(struct nrs_tbf_rule *rule, struct seq_file *m)
1210 {
1211         seq_printf(m, "%s {%s} %llu, ref %d\n", rule->tr_name,
1212                    rule->tr_nids_str, rule->tr_rpc_rate,
1213                    atomic_read(&rule->tr_ref) - 1);
1214         return 0;
1215 }
1216
1217 static int
1218 nrs_tbf_nid_rule_match(struct nrs_tbf_rule *rule,
1219                        struct nrs_tbf_client *cli)
1220 {
1221         return cfs_match_nid(cli->tc_nid, &rule->tr_nids);
1222 }
1223
1224 static void nrs_tbf_nid_rule_fini(struct nrs_tbf_rule *rule)
1225 {
1226         if (!list_empty(&rule->tr_nids))
1227                 cfs_free_nidlist(&rule->tr_nids);
1228         LASSERT(rule->tr_nids_str != NULL);
1229         OBD_FREE(rule->tr_nids_str, strlen(rule->tr_nids_str) + 1);
1230 }
1231
1232 static void nrs_tbf_nid_cmd_fini(struct nrs_tbf_cmd *cmd)
1233 {
1234         if (!list_empty(&cmd->u.tc_start.ts_nids))
1235                 cfs_free_nidlist(&cmd->u.tc_start.ts_nids);
1236         if (cmd->u.tc_start.ts_nids_str)
1237                 OBD_FREE(cmd->u.tc_start.ts_nids_str,
1238                          strlen(cmd->u.tc_start.ts_nids_str) + 1);
1239 }
1240
1241 static int nrs_tbf_nid_parse(struct nrs_tbf_cmd *cmd, char *id)
1242 {
1243         struct cfs_lstr src;
1244         int rc;
1245
1246         src.ls_str = id;
1247         src.ls_len = strlen(id);
1248         rc = nrs_tbf_check_id_value(&src, "nid");
1249         if (rc)
1250                 return rc;
1251
1252         OBD_ALLOC(cmd->u.tc_start.ts_nids_str, src.ls_len + 1);
1253         if (cmd->u.tc_start.ts_nids_str == NULL)
1254                 return -ENOMEM;
1255
1256         memcpy(cmd->u.tc_start.ts_nids_str, src.ls_str, src.ls_len);
1257
1258         /* parse NID list */
1259         if (cfs_parse_nidlist(cmd->u.tc_start.ts_nids_str,
1260                               strlen(cmd->u.tc_start.ts_nids_str),
1261                               &cmd->u.tc_start.ts_nids) <= 0) {
1262                 nrs_tbf_nid_cmd_fini(cmd);
1263                 return -EINVAL;
1264         }
1265
1266         return 0;
1267 }
1268
1269 static struct nrs_tbf_ops nrs_tbf_nid_ops = {
1270         .o_name = NRS_TBF_TYPE_NID,
1271         .o_startup = nrs_tbf_nid_startup,
1272         .o_cli_find = nrs_tbf_nid_cli_find,
1273         .o_cli_findadd = nrs_tbf_nid_cli_findadd,
1274         .o_cli_put = nrs_tbf_nid_cli_put,
1275         .o_cli_init = nrs_tbf_nid_cli_init,
1276         .o_rule_init = nrs_tbf_nid_rule_init,
1277         .o_rule_dump = nrs_tbf_nid_rule_dump,
1278         .o_rule_match = nrs_tbf_nid_rule_match,
1279         .o_rule_fini = nrs_tbf_nid_rule_fini,
1280 };
1281
1282 static unsigned nrs_tbf_hop_hash(struct cfs_hash *hs, const void *key,
1283                                  unsigned mask)
1284 {
1285         return cfs_hash_djb2_hash(key, strlen(key), mask);
1286 }
1287
1288 static int nrs_tbf_hop_keycmp(const void *key, struct hlist_node *hnode)
1289 {
1290         struct nrs_tbf_client *cli = hlist_entry(hnode,
1291                                                  struct nrs_tbf_client,
1292                                                  tc_hnode);
1293
1294         return (strcmp(cli->tc_key, key) == 0);
1295 }
1296
1297 static void *nrs_tbf_hop_key(struct hlist_node *hnode)
1298 {
1299         struct nrs_tbf_client *cli = hlist_entry(hnode,
1300                                                  struct nrs_tbf_client,
1301                                                  tc_hnode);
1302         return cli->tc_key;
1303 }
1304
1305 static void nrs_tbf_hop_get(struct cfs_hash *hs, struct hlist_node *hnode)
1306 {
1307         struct nrs_tbf_client *cli = hlist_entry(hnode,
1308                                                  struct nrs_tbf_client,
1309                                                  tc_hnode);
1310
1311         atomic_inc(&cli->tc_ref);
1312 }
1313
1314 static void nrs_tbf_hop_put(struct cfs_hash *hs, struct hlist_node *hnode)
1315 {
1316         struct nrs_tbf_client *cli = hlist_entry(hnode,
1317                                                  struct nrs_tbf_client,
1318                                                  tc_hnode);
1319
1320         atomic_dec(&cli->tc_ref);
1321 }
1322
1323 static void nrs_tbf_hop_exit(struct cfs_hash *hs, struct hlist_node *hnode)
1324
1325 {
1326         struct nrs_tbf_client *cli = hlist_entry(hnode,
1327                                                  struct nrs_tbf_client,
1328                                                  tc_hnode);
1329
1330         LASSERT(atomic_read(&cli->tc_ref) == 0);
1331         nrs_tbf_cli_fini(cli);
1332 }
1333
1334 static struct cfs_hash_ops nrs_tbf_hash_ops = {
1335         .hs_hash        = nrs_tbf_hop_hash,
1336         .hs_keycmp      = nrs_tbf_hop_keycmp,
1337         .hs_key         = nrs_tbf_hop_key,
1338         .hs_object      = nrs_tbf_hop_object,
1339         .hs_get         = nrs_tbf_hop_get,
1340         .hs_put         = nrs_tbf_hop_put,
1341         .hs_put_locked  = nrs_tbf_hop_put,
1342         .hs_exit        = nrs_tbf_hop_exit,
1343 };
1344
1345 #define NRS_TBF_GENERIC_BKT_BITS        10
1346 #define NRS_TBF_GENERIC_HASH_FLAGS      (CFS_HASH_SPIN_BKTLOCK | \
1347                                         CFS_HASH_NO_ITEMREF | \
1348                                         CFS_HASH_DEPTH)
1349
1350 static int
1351 nrs_tbf_startup(struct ptlrpc_nrs_policy *policy, struct nrs_tbf_head *head)
1352 {
1353         struct nrs_tbf_cmd       start;
1354         struct nrs_tbf_bucket   *bkt;
1355         int                      bits;
1356         int                      i;
1357         int                      rc;
1358         struct cfs_hash_bd       bd;
1359
1360         bits = nrs_tbf_jobid_hash_order();
1361         if (bits < NRS_TBF_GENERIC_BKT_BITS)
1362                 bits = NRS_TBF_GENERIC_BKT_BITS;
1363         head->th_cli_hash = cfs_hash_create("nrs_tbf_hash",
1364                                             bits, bits,
1365                                             NRS_TBF_GENERIC_BKT_BITS,
1366                                             sizeof(*bkt), 0, 0,
1367                                             &nrs_tbf_hash_ops,
1368                                             NRS_TBF_GENERIC_HASH_FLAGS);
1369         if (head->th_cli_hash == NULL)
1370                 return -ENOMEM;
1371
1372         cfs_hash_for_each_bucket(head->th_cli_hash, &bd, i) {
1373                 bkt = cfs_hash_bd_extra_get(head->th_cli_hash, &bd);
1374                 INIT_LIST_HEAD(&bkt->ntb_lru);
1375         }
1376
1377         memset(&start, 0, sizeof(start));
1378         start.u.tc_start.ts_conds_str = "*";
1379
1380         start.u.tc_start.ts_rpc_rate = tbf_rate;
1381         start.u.tc_start.ts_rule_flags = NTRS_DEFAULT;
1382         start.tc_name = NRS_TBF_DEFAULT_RULE;
1383         INIT_LIST_HEAD(&start.u.tc_start.ts_conds);
1384         rc = nrs_tbf_rule_start(policy, head, &start);
1385         if (rc)
1386                 cfs_hash_putref(head->th_cli_hash);
1387
1388         return rc;
1389 }
1390
1391 static struct nrs_tbf_client *
1392 nrs_tbf_cli_hash_lookup(struct cfs_hash *hs, struct cfs_hash_bd *bd,
1393                         const char *key)
1394 {
1395         struct hlist_node *hnode;
1396         struct nrs_tbf_client *cli;
1397
1398         hnode = cfs_hash_bd_lookup_locked(hs, bd, (void *)key);
1399         if (hnode == NULL)
1400                 return NULL;
1401
1402         cli = container_of0(hnode, struct nrs_tbf_client, tc_hnode);
1403         if (!list_empty(&cli->tc_lru))
1404                 list_del_init(&cli->tc_lru);
1405         return cli;
1406 }
1407
1408 /**
1409  * ONLY opcode presented in this function will be checked in
1410  * nrs_tbf_id_cli_set(). That means, we can add or remove an
1411  * opcode to enable or disable requests handled in nrs_tbf
1412  */
1413 static struct req_format *req_fmt(__u32 opcode)
1414 {
1415         switch (opcode) {
1416         case OST_GETATTR:
1417                 return &RQF_OST_GETATTR;
1418         case OST_SETATTR:
1419                 return &RQF_OST_SETATTR;
1420         case OST_READ:
1421                 return &RQF_OST_BRW_READ;
1422         case OST_WRITE:
1423                 return &RQF_OST_BRW_WRITE;
1424         /* FIXME: OST_CREATE and OST_DESTROY comes from MDS
1425          * in most case. Should they be removed? */
1426         case OST_CREATE:
1427                 return &RQF_OST_CREATE;
1428         case OST_DESTROY:
1429                 return &RQF_OST_DESTROY;
1430         case OST_PUNCH:
1431                 return &RQF_OST_PUNCH;
1432         case OST_SYNC:
1433                 return &RQF_OST_SYNC;
1434         case OST_LADVISE:
1435                 return &RQF_OST_LADVISE;
1436         case MDS_GETATTR:
1437                 return &RQF_MDS_GETATTR;
1438         case MDS_GETATTR_NAME:
1439                 return &RQF_MDS_GETATTR_NAME;
1440         /* close is skipped to avoid LDLM cancel slowness */
1441 #if 0
1442         case MDS_CLOSE:
1443                 return &RQF_MDS_CLOSE;
1444 #endif
1445         case MDS_REINT:
1446                 return &RQF_MDS_REINT;
1447         case MDS_READPAGE:
1448                 return &RQF_MDS_READPAGE;
1449         case MDS_GET_ROOT:
1450                 return &RQF_MDS_GET_ROOT;
1451         case MDS_STATFS:
1452                 return &RQF_MDS_STATFS;
1453         case MDS_SYNC:
1454                 return &RQF_MDS_SYNC;
1455         case MDS_QUOTACTL:
1456                 return &RQF_MDS_QUOTACTL;
1457         case MDS_GETXATTR:
1458                 return &RQF_MDS_GETXATTR;
1459         case MDS_GET_INFO:
1460                 return &RQF_MDS_GET_INFO;
1461         /* HSM op is skipped */
1462 #if 0 
1463         case MDS_HSM_STATE_GET:
1464                 return &RQF_MDS_HSM_STATE_GET;
1465         case MDS_HSM_STATE_SET:
1466                 return &RQF_MDS_HSM_STATE_SET;
1467         case MDS_HSM_ACTION:
1468                 return &RQF_MDS_HSM_ACTION;
1469         case MDS_HSM_CT_REGISTER:
1470                 return &RQF_MDS_HSM_CT_REGISTER;
1471         case MDS_HSM_CT_UNREGISTER:
1472                 return &RQF_MDS_HSM_CT_UNREGISTER;
1473 #endif
1474         case MDS_SWAP_LAYOUTS:
1475                 return &RQF_MDS_SWAP_LAYOUTS;
1476         case LDLM_ENQUEUE:
1477                 return &RQF_LDLM_ENQUEUE;
1478         default:
1479                 return NULL;
1480         }
1481 }
1482
1483 static struct req_format *intent_req_fmt(__u32 it_opc)
1484 {
1485         if (it_opc & (IT_OPEN | IT_CREAT))
1486                 return &RQF_LDLM_INTENT_OPEN;
1487         else if (it_opc & (IT_GETATTR | IT_LOOKUP))
1488                 return &RQF_LDLM_INTENT_GETATTR;
1489         else if (it_opc & IT_GETXATTR)
1490                 return &RQF_LDLM_INTENT_GETXATTR;
1491         else if (it_opc & (IT_GLIMPSE | IT_BRW))
1492                 return &RQF_LDLM_INTENT;
1493         else
1494                 return NULL;
1495 }
1496
1497 static int ost_tbf_id_cli_set(struct ptlrpc_request *req,
1498                               struct tbf_id *id)
1499 {
1500         struct ost_body *body;
1501
1502         body = req_capsule_client_get(&req->rq_pill, &RMF_OST_BODY);
1503         if (body != NULL) {
1504                 id->ti_uid = body->oa.o_uid;
1505                 id->ti_gid = body->oa.o_gid;
1506                 return 0;
1507         }
1508
1509         return -EINVAL;
1510 }
1511
1512 static void unpack_ugid_from_mdt_body(struct ptlrpc_request *req,
1513                                       struct tbf_id *id)
1514 {
1515         struct mdt_body *b = req_capsule_client_get(&req->rq_pill,
1516                                                     &RMF_MDT_BODY);
1517         LASSERT(b != NULL);
1518
1519         /* TODO: nodemaping feature converts {ug}id from individual
1520          * clients to the actual ones of the file system. Some work
1521          * may be needed to fix this. */
1522         id->ti_uid = b->mbo_uid;
1523         id->ti_gid = b->mbo_gid;
1524 }
1525
1526 static void unpack_ugid_from_mdt_rec_reint(struct ptlrpc_request *req,
1527                                            struct tbf_id *id)
1528 {
1529         struct mdt_rec_reint *rec;
1530
1531         rec = req_capsule_client_get(&req->rq_pill, &RMF_REC_REINT);
1532         LASSERT(rec != NULL);
1533
1534         /* use the fs{ug}id as {ug}id of the process */
1535         id->ti_uid = rec->rr_fsuid;
1536         id->ti_gid = rec->rr_fsgid;
1537 }
1538
1539 static int mdt_tbf_id_cli_set(struct ptlrpc_request *req,
1540                               struct tbf_id *id)
1541 {
1542         u32 opc = lustre_msg_get_opc(req->rq_reqmsg);
1543         int rc = 0;
1544
1545         switch (opc) {
1546         case MDS_GETATTR:
1547         case MDS_GETATTR_NAME:
1548         case MDS_GET_ROOT:
1549         case MDS_READPAGE:
1550         case MDS_SYNC:
1551         case MDS_GETXATTR:
1552         case MDS_HSM_STATE_GET ... MDS_SWAP_LAYOUTS:
1553                 unpack_ugid_from_mdt_body(req, id);
1554                 break;
1555         case MDS_CLOSE:
1556         case MDS_REINT:
1557                 unpack_ugid_from_mdt_rec_reint(req, id);
1558                 break;
1559         default:
1560                 rc = -EINVAL;
1561                 break;
1562         }
1563         return rc;
1564 }
1565
1566 static int ldlm_tbf_id_cli_set(struct ptlrpc_request *req,
1567                               struct tbf_id *id)
1568 {
1569         struct ldlm_intent *lit;
1570         struct req_format *fmt;
1571
1572         if (req->rq_reqmsg->lm_bufcount <= DLM_INTENT_IT_OFF)
1573                 return -EINVAL;
1574
1575         req_capsule_extend(&req->rq_pill, &RQF_LDLM_INTENT_BASIC);
1576         lit = req_capsule_client_get(&req->rq_pill, &RMF_LDLM_INTENT);
1577         if (lit == NULL)
1578                 return -EINVAL;
1579
1580         fmt = intent_req_fmt(lit->opc);
1581         if (fmt == NULL)
1582                 return -EINVAL;
1583
1584         req_capsule_extend(&req->rq_pill, fmt);
1585
1586         if (lit->opc & (IT_GETXATTR | IT_GETATTR | IT_LOOKUP))
1587                 unpack_ugid_from_mdt_body(req, id);
1588         else if (lit->opc & (IT_OPEN | IT_OPEN | IT_GLIMPSE | IT_BRW))
1589                 unpack_ugid_from_mdt_rec_reint(req, id);
1590         else
1591                 return -EINVAL;
1592         return 0;
1593 }
1594
1595 static int nrs_tbf_id_cli_set(struct ptlrpc_request *req, struct tbf_id *id,
1596                               enum nrs_tbf_flag ti_type)
1597 {
1598         u32 opc = lustre_msg_get_opc(req->rq_reqmsg);
1599         struct req_format *fmt = req_fmt(opc);
1600         bool fmt_unset = false;
1601         int rc;
1602
1603         memset(id, 0, sizeof(struct tbf_id));
1604         id->ti_type = ti_type;
1605
1606         if (fmt == NULL)
1607                 return -EINVAL;
1608         req_capsule_init(&req->rq_pill, req, RCL_SERVER);
1609         if (req->rq_pill.rc_fmt == NULL) {
1610                 req_capsule_set(&req->rq_pill, fmt);
1611                 fmt_unset = true;
1612         }
1613
1614         if (opc < OST_LAST_OPC)
1615                 rc = ost_tbf_id_cli_set(req, id);
1616         else if (opc >= MDS_FIRST_OPC && opc < MDS_LAST_OPC)
1617                 rc = mdt_tbf_id_cli_set(req, id);
1618         else if (opc == LDLM_ENQUEUE)
1619                 rc = ldlm_tbf_id_cli_set(req, id);
1620         else
1621                 rc = -EINVAL;
1622
1623         /* restore it to the initialized state */
1624         if (fmt_unset)
1625                 req->rq_pill.rc_fmt = NULL;
1626         return rc;
1627 }
1628
1629 static inline void nrs_tbf_cli_gen_key(struct nrs_tbf_client *cli,
1630                                        struct ptlrpc_request *req,
1631                                        char *keystr, size_t keystr_sz)
1632 {
1633         const char *jobid;
1634         u32 opc = lustre_msg_get_opc(req->rq_reqmsg);
1635         struct tbf_id id;
1636
1637         nrs_tbf_id_cli_set(req, &id, NRS_TBF_FLAG_UID | NRS_TBF_FLAG_GID);
1638         jobid = lustre_msg_get_jobid(req->rq_reqmsg);
1639         if (jobid == NULL)
1640                 jobid = NRS_TBF_JOBID_NULL;
1641
1642         snprintf(keystr, keystr_sz, "%s_%s_%d_%u_%u", jobid,
1643                  libcfs_nid2str(req->rq_peer.nid), opc, id.ti_uid,
1644                  id.ti_gid);
1645
1646         if (cli) {
1647                 INIT_LIST_HEAD(&cli->tc_lru);
1648                 strlcpy(cli->tc_key, keystr, sizeof(cli->tc_key));
1649                 strlcpy(cli->tc_jobid, jobid, sizeof(cli->tc_jobid));
1650                 cli->tc_nid = req->rq_peer.nid;
1651                 cli->tc_opcode = opc;
1652                 cli->tc_id = id;
1653         }
1654 }
1655
1656 static struct nrs_tbf_client *
1657 nrs_tbf_cli_find(struct nrs_tbf_head *head, struct ptlrpc_request *req)
1658 {
1659         struct nrs_tbf_client *cli;
1660         struct cfs_hash *hs = head->th_cli_hash;
1661         struct cfs_hash_bd bd;
1662         char keystr[NRS_TBF_KEY_LEN];
1663
1664         nrs_tbf_cli_gen_key(NULL, req, keystr, sizeof(keystr));
1665         cfs_hash_bd_get_and_lock(hs, (void *)keystr, &bd, 1);
1666         cli = nrs_tbf_cli_hash_lookup(hs, &bd, keystr);
1667         cfs_hash_bd_unlock(hs, &bd, 1);
1668
1669         return cli;
1670 }
1671
1672 static struct nrs_tbf_client *
1673 nrs_tbf_cli_findadd(struct nrs_tbf_head *head,
1674                     struct nrs_tbf_client *cli)
1675 {
1676         const char              *key;
1677         struct nrs_tbf_client   *ret;
1678         struct cfs_hash         *hs = head->th_cli_hash;
1679         struct cfs_hash_bd       bd;
1680
1681         key = cli->tc_key;
1682         cfs_hash_bd_get_and_lock(hs, (void *)key, &bd, 1);
1683         ret = nrs_tbf_cli_hash_lookup(hs, &bd, key);
1684         if (ret == NULL) {
1685                 cfs_hash_bd_add_locked(hs, &bd, &cli->tc_hnode);
1686                 ret = cli;
1687         }
1688         cfs_hash_bd_unlock(hs, &bd, 1);
1689
1690         return ret;
1691 }
1692
1693 static void
1694 nrs_tbf_cli_put(struct nrs_tbf_head *head, struct nrs_tbf_client *cli)
1695 {
1696         struct cfs_hash_bd       bd;
1697         struct cfs_hash         *hs = head->th_cli_hash;
1698         struct nrs_tbf_bucket   *bkt;
1699         int                      hw;
1700         struct list_head         zombies;
1701
1702         INIT_LIST_HEAD(&zombies);
1703         cfs_hash_bd_get(hs, &cli->tc_key, &bd);
1704         bkt = cfs_hash_bd_extra_get(hs, &bd);
1705         if (!cfs_hash_bd_dec_and_lock(hs, &bd, &cli->tc_ref))
1706                 return;
1707         LASSERT(list_empty(&cli->tc_lru));
1708         list_add_tail(&cli->tc_lru, &bkt->ntb_lru);
1709
1710         /**
1711          * Check and purge the LRU, there is at least one client in the LRU.
1712          */
1713         hw = tbf_jobid_cache_size >> (hs->hs_cur_bits - hs->hs_bkt_bits);
1714         while (cfs_hash_bd_count_get(&bd) > hw) {
1715                 if (unlikely(list_empty(&bkt->ntb_lru)))
1716                         break;
1717                 cli = list_entry(bkt->ntb_lru.next,
1718                                  struct nrs_tbf_client,
1719                                  tc_lru);
1720                 LASSERT(atomic_read(&cli->tc_ref) == 0);
1721                 cfs_hash_bd_del_locked(hs, &bd, &cli->tc_hnode);
1722                 list_move(&cli->tc_lru, &zombies);
1723         }
1724         cfs_hash_bd_unlock(head->th_cli_hash, &bd, 1);
1725
1726         while (!list_empty(&zombies)) {
1727                 cli = container_of0(zombies.next,
1728                                     struct nrs_tbf_client, tc_lru);
1729                 list_del_init(&cli->tc_lru);
1730                 nrs_tbf_cli_fini(cli);
1731         }
1732 }
1733
1734 static void
1735 nrs_tbf_generic_cli_init(struct nrs_tbf_client *cli,
1736                          struct ptlrpc_request *req)
1737 {
1738         char keystr[NRS_TBF_KEY_LEN];
1739
1740         nrs_tbf_cli_gen_key(cli, req, keystr, sizeof(keystr));
1741 }
1742
1743 static void
1744 nrs_tbf_id_list_free(struct list_head *uid_list)
1745 {
1746         struct nrs_tbf_id *nti_id, *n;
1747
1748         list_for_each_entry_safe(nti_id, n, uid_list, nti_linkage) {
1749                 list_del_init(&nti_id->nti_linkage);
1750                 OBD_FREE_PTR(nti_id);
1751         }
1752 }
1753
1754 static void
1755 nrs_tbf_expression_free(struct nrs_tbf_expression *expr)
1756 {
1757         LASSERT(expr->te_field >= NRS_TBF_FIELD_NID &&
1758                 expr->te_field < NRS_TBF_FIELD_MAX);
1759         switch (expr->te_field) {
1760         case NRS_TBF_FIELD_NID:
1761                 cfs_free_nidlist(&expr->te_cond);
1762                 break;
1763         case NRS_TBF_FIELD_JOBID:
1764                 nrs_tbf_jobid_list_free(&expr->te_cond);
1765                 break;
1766         case NRS_TBF_FIELD_OPCODE:
1767                 CFS_FREE_BITMAP(expr->te_opcodes);
1768                 break;
1769         case NRS_TBF_FIELD_UID:
1770         case NRS_TBF_FIELD_GID:
1771                 nrs_tbf_id_list_free(&expr->te_cond);
1772                 break;
1773         default:
1774                 LBUG();
1775         }
1776         OBD_FREE_PTR(expr);
1777 }
1778
1779 static void
1780 nrs_tbf_conjunction_free(struct nrs_tbf_conjunction *conjunction)
1781 {
1782         struct nrs_tbf_expression *expression;
1783         struct nrs_tbf_expression *n;
1784
1785         LASSERT(list_empty(&conjunction->tc_linkage));
1786         list_for_each_entry_safe(expression, n,
1787                                  &conjunction->tc_expressions,
1788                                  te_linkage) {
1789                 list_del_init(&expression->te_linkage);
1790                 nrs_tbf_expression_free(expression);
1791         }
1792         OBD_FREE_PTR(conjunction);
1793 }
1794
1795 static void
1796 nrs_tbf_conds_free(struct list_head *cond_list)
1797 {
1798         struct nrs_tbf_conjunction *conjunction;
1799         struct nrs_tbf_conjunction *n;
1800
1801         list_for_each_entry_safe(conjunction, n, cond_list, tc_linkage) {
1802                 list_del_init(&conjunction->tc_linkage);
1803                 nrs_tbf_conjunction_free(conjunction);
1804         }
1805 }
1806
1807 static void
1808 nrs_tbf_generic_cmd_fini(struct nrs_tbf_cmd *cmd)
1809 {
1810         if (!list_empty(&cmd->u.tc_start.ts_conds))
1811                 nrs_tbf_conds_free(&cmd->u.tc_start.ts_conds);
1812         if (cmd->u.tc_start.ts_conds_str)
1813                 OBD_FREE(cmd->u.tc_start.ts_conds_str,
1814                          strlen(cmd->u.tc_start.ts_conds_str) + 1);
1815 }
1816
1817 #define NRS_TBF_DISJUNCTION_DELIM       (',')
1818 #define NRS_TBF_CONJUNCTION_DELIM       ('&')
1819 #define NRS_TBF_EXPRESSION_DELIM        ('=')
1820
1821 static inline bool
1822 nrs_tbf_check_field(struct cfs_lstr *field, char *str)
1823 {
1824         int len = strlen(str);
1825
1826         return (field->ls_len == len &&
1827                 strncmp(field->ls_str, str, len) == 0);
1828 }
1829
1830 static int
1831 nrs_tbf_opcode_list_parse(char *str, int len, struct cfs_bitmap **bitmaptr);
1832 static int
1833 nrs_tbf_id_list_parse(char *str, int len, struct list_head *id_list,
1834                       enum nrs_tbf_flag tif);
1835
1836 static int
1837 nrs_tbf_expression_parse(struct cfs_lstr *src, struct list_head *cond_list)
1838 {
1839         struct nrs_tbf_expression *expr;
1840         struct cfs_lstr field;
1841         int rc = 0;
1842
1843         OBD_ALLOC(expr, sizeof(struct nrs_tbf_expression));
1844         if (expr == NULL)
1845                 return -ENOMEM;
1846
1847         rc = cfs_gettok(src, NRS_TBF_EXPRESSION_DELIM, &field);
1848         if (rc == 0 || src->ls_len <= 2 || src->ls_str[0] != '{' ||
1849             src->ls_str[src->ls_len - 1] != '}')
1850                 GOTO(out, rc = -EINVAL);
1851
1852         /* Skip '{' and '}' */
1853         src->ls_str++;
1854         src->ls_len -= 2;
1855
1856         if (nrs_tbf_check_field(&field, "nid")) {
1857                 if (cfs_parse_nidlist(src->ls_str,
1858                                       src->ls_len,
1859                                       &expr->te_cond) <= 0)
1860                         GOTO(out, rc = -EINVAL);
1861                 expr->te_field = NRS_TBF_FIELD_NID;
1862         } else if (nrs_tbf_check_field(&field, "jobid")) {
1863                 if (nrs_tbf_jobid_list_parse(src->ls_str,
1864                                              src->ls_len,
1865                                              &expr->te_cond) < 0)
1866                         GOTO(out, rc = -EINVAL);
1867                 expr->te_field = NRS_TBF_FIELD_JOBID;
1868         } else if (nrs_tbf_check_field(&field, "opcode")) {
1869                 if (nrs_tbf_opcode_list_parse(src->ls_str,
1870                                               src->ls_len,
1871                                               &expr->te_opcodes) < 0)
1872                         GOTO(out, rc = -EINVAL);
1873                 expr->te_field = NRS_TBF_FIELD_OPCODE;
1874         } else if (nrs_tbf_check_field(&field, "uid")) {
1875                 if (nrs_tbf_id_list_parse(src->ls_str,
1876                                           src->ls_len,
1877                                           &expr->te_cond,
1878                                           NRS_TBF_FLAG_UID) < 0)
1879                         GOTO(out, rc = -EINVAL);
1880                 expr->te_field = NRS_TBF_FIELD_UID;
1881         } else if (nrs_tbf_check_field(&field, "gid")) {
1882                 if (nrs_tbf_id_list_parse(src->ls_str,
1883                                           src->ls_len,
1884                                           &expr->te_cond,
1885                                           NRS_TBF_FLAG_GID) < 0)
1886                         GOTO(out, rc = -EINVAL);
1887                 expr->te_field = NRS_TBF_FIELD_GID;
1888         } else {
1889                 GOTO(out, rc = -EINVAL);
1890         }
1891
1892         list_add_tail(&expr->te_linkage, cond_list);
1893         return 0;
1894 out:
1895         OBD_FREE_PTR(expr);
1896         return rc;
1897 }
1898
1899 static int
1900 nrs_tbf_conjunction_parse(struct cfs_lstr *src, struct list_head *cond_list)
1901 {
1902         struct nrs_tbf_conjunction *conjunction;
1903         struct cfs_lstr expr;
1904         int rc = 0;
1905
1906         OBD_ALLOC(conjunction, sizeof(struct nrs_tbf_conjunction));
1907         if (conjunction == NULL)
1908                 return -ENOMEM;
1909
1910         INIT_LIST_HEAD(&conjunction->tc_expressions);
1911         list_add_tail(&conjunction->tc_linkage, cond_list);
1912
1913         while (src->ls_str) {
1914                 rc = cfs_gettok(src, NRS_TBF_CONJUNCTION_DELIM, &expr);
1915                 if (rc == 0) {
1916                         rc = -EINVAL;
1917                         break;
1918                 }
1919                 rc = nrs_tbf_expression_parse(&expr,
1920                                               &conjunction->tc_expressions);
1921                 if (rc)
1922                         break;
1923         }
1924         return rc;
1925 }
1926
1927 static int
1928 nrs_tbf_conds_parse(char *str, int len, struct list_head *cond_list)
1929 {
1930         struct cfs_lstr src;
1931         struct cfs_lstr res;
1932         int rc = 0;
1933
1934         src.ls_str = str;
1935         src.ls_len = len;
1936         INIT_LIST_HEAD(cond_list);
1937         while (src.ls_str) {
1938                 rc = cfs_gettok(&src, NRS_TBF_DISJUNCTION_DELIM, &res);
1939                 if (rc == 0) {
1940                         rc = -EINVAL;
1941                         break;
1942                 }
1943                 rc = nrs_tbf_conjunction_parse(&res, cond_list);
1944                 if (rc)
1945                         break;
1946         }
1947         return rc;
1948 }
1949
1950 static int
1951 nrs_tbf_generic_parse(struct nrs_tbf_cmd *cmd, const char *id)
1952 {
1953         int rc;
1954
1955         OBD_ALLOC(cmd->u.tc_start.ts_conds_str, strlen(id) + 1);
1956         if (cmd->u.tc_start.ts_conds_str == NULL)
1957                 return -ENOMEM;
1958
1959         memcpy(cmd->u.tc_start.ts_conds_str, id, strlen(id));
1960
1961         /* Parse hybird NID and JOBID conditions */
1962         rc = nrs_tbf_conds_parse(cmd->u.tc_start.ts_conds_str,
1963                                  strlen(cmd->u.tc_start.ts_conds_str),
1964                                  &cmd->u.tc_start.ts_conds);
1965         if (rc)
1966                 nrs_tbf_generic_cmd_fini(cmd);
1967
1968         return rc;
1969 }
1970
1971 static int
1972 nrs_tbf_id_list_match(struct list_head *id_list, struct tbf_id id);
1973
1974 static int
1975 nrs_tbf_expression_match(struct nrs_tbf_expression *expr,
1976                          struct nrs_tbf_rule *rule,
1977                          struct nrs_tbf_client *cli)
1978 {
1979         switch (expr->te_field) {
1980         case NRS_TBF_FIELD_NID:
1981                 return cfs_match_nid(cli->tc_nid, &expr->te_cond);
1982         case NRS_TBF_FIELD_JOBID:
1983                 return nrs_tbf_jobid_list_match(&expr->te_cond, cli->tc_jobid);
1984         case NRS_TBF_FIELD_OPCODE:
1985                 return cfs_bitmap_check(expr->te_opcodes, cli->tc_opcode);
1986         case NRS_TBF_FIELD_UID:
1987         case NRS_TBF_FIELD_GID:
1988                 return nrs_tbf_id_list_match(&expr->te_cond, cli->tc_id);
1989         default:
1990                 return 0;
1991         }
1992 }
1993
1994 static int
1995 nrs_tbf_conjunction_match(struct nrs_tbf_conjunction *conjunction,
1996                           struct nrs_tbf_rule *rule,
1997                           struct nrs_tbf_client *cli)
1998 {
1999         struct nrs_tbf_expression *expr;
2000         int matched;
2001
2002         list_for_each_entry(expr, &conjunction->tc_expressions, te_linkage) {
2003                 matched = nrs_tbf_expression_match(expr, rule, cli);
2004                 if (!matched)
2005                         return 0;
2006         }
2007
2008         return 1;
2009 }
2010
2011 static int
2012 nrs_tbf_cond_match(struct nrs_tbf_rule *rule, struct nrs_tbf_client *cli)
2013 {
2014         struct nrs_tbf_conjunction *conjunction;
2015         int matched;
2016
2017         list_for_each_entry(conjunction, &rule->tr_conds, tc_linkage) {
2018                 matched = nrs_tbf_conjunction_match(conjunction, rule, cli);
2019                 if (matched)
2020                         return 1;
2021         }
2022
2023         return 0;
2024 }
2025
2026 static void
2027 nrs_tbf_generic_rule_fini(struct nrs_tbf_rule *rule)
2028 {
2029         if (!list_empty(&rule->tr_conds))
2030                 nrs_tbf_conds_free(&rule->tr_conds);
2031         LASSERT(rule->tr_conds_str != NULL);
2032         OBD_FREE(rule->tr_conds_str, strlen(rule->tr_conds_str) + 1);
2033 }
2034
2035 static int
2036 nrs_tbf_rule_init(struct ptlrpc_nrs_policy *policy,
2037                   struct nrs_tbf_rule *rule, struct nrs_tbf_cmd *start)
2038 {
2039         int rc = 0;
2040
2041         LASSERT(start->u.tc_start.ts_conds_str);
2042         OBD_ALLOC(rule->tr_conds_str,
2043                   strlen(start->u.tc_start.ts_conds_str) + 1);
2044         if (rule->tr_conds_str == NULL)
2045                 return -ENOMEM;
2046
2047         memcpy(rule->tr_conds_str,
2048                start->u.tc_start.ts_conds_str,
2049                strlen(start->u.tc_start.ts_conds_str));
2050
2051         INIT_LIST_HEAD(&rule->tr_conds);
2052         if (!list_empty(&start->u.tc_start.ts_conds)) {
2053                 rc = nrs_tbf_conds_parse(rule->tr_conds_str,
2054                                          strlen(rule->tr_conds_str),
2055                                          &rule->tr_conds);
2056         }
2057         if (rc)
2058                 nrs_tbf_generic_rule_fini(rule);
2059
2060         return rc;
2061 }
2062
2063 static int
2064 nrs_tbf_generic_rule_dump(struct nrs_tbf_rule *rule, struct seq_file *m)
2065 {
2066         seq_printf(m, "%s %s %llu, ref %d\n", rule->tr_name,
2067                    rule->tr_conds_str, rule->tr_rpc_rate,
2068                    atomic_read(&rule->tr_ref) - 1);
2069         return 0;
2070 }
2071
2072 static int
2073 nrs_tbf_generic_rule_match(struct nrs_tbf_rule *rule,
2074                            struct nrs_tbf_client *cli)
2075 {
2076         return nrs_tbf_cond_match(rule, cli);
2077 }
2078
2079 static struct nrs_tbf_ops nrs_tbf_generic_ops = {
2080         .o_name = NRS_TBF_TYPE_GENERIC,
2081         .o_startup = nrs_tbf_startup,
2082         .o_cli_find = nrs_tbf_cli_find,
2083         .o_cli_findadd = nrs_tbf_cli_findadd,
2084         .o_cli_put = nrs_tbf_cli_put,
2085         .o_cli_init = nrs_tbf_generic_cli_init,
2086         .o_rule_init = nrs_tbf_rule_init,
2087         .o_rule_dump = nrs_tbf_generic_rule_dump,
2088         .o_rule_match = nrs_tbf_generic_rule_match,
2089         .o_rule_fini = nrs_tbf_generic_rule_fini,
2090 };
2091
2092 static void nrs_tbf_opcode_rule_fini(struct nrs_tbf_rule *rule)
2093 {
2094         if (rule->tr_opcodes != NULL)
2095                 CFS_FREE_BITMAP(rule->tr_opcodes);
2096
2097         LASSERT(rule->tr_opcodes_str != NULL);
2098         OBD_FREE(rule->tr_opcodes_str, strlen(rule->tr_opcodes_str) + 1);
2099 }
2100
2101 static unsigned nrs_tbf_opcode_hop_hash(struct cfs_hash *hs, const void *key,
2102                                         unsigned mask)
2103 {
2104         return cfs_hash_djb2_hash(key, sizeof(__u32), mask);
2105 }
2106
2107 static int nrs_tbf_opcode_hop_keycmp(const void *key, struct hlist_node *hnode)
2108 {
2109         const __u32     *opc = key;
2110         struct nrs_tbf_client *cli = hlist_entry(hnode,
2111                                                  struct nrs_tbf_client,
2112                                                  tc_hnode);
2113
2114         return *opc == cli->tc_opcode;
2115 }
2116
2117 static void *nrs_tbf_opcode_hop_key(struct hlist_node *hnode)
2118 {
2119         struct nrs_tbf_client *cli = hlist_entry(hnode,
2120                                                  struct nrs_tbf_client,
2121                                                  tc_hnode);
2122
2123         return &cli->tc_opcode;
2124 }
2125
2126 static void nrs_tbf_opcode_hop_get(struct cfs_hash *hs,
2127                                    struct hlist_node *hnode)
2128 {
2129         struct nrs_tbf_client *cli = hlist_entry(hnode,
2130                                                  struct nrs_tbf_client,
2131                                                  tc_hnode);
2132
2133         atomic_inc(&cli->tc_ref);
2134 }
2135
2136 static void nrs_tbf_opcode_hop_put(struct cfs_hash *hs,
2137                                    struct hlist_node *hnode)
2138 {
2139         struct nrs_tbf_client *cli = hlist_entry(hnode,
2140                                                  struct nrs_tbf_client,
2141                                                  tc_hnode);
2142
2143         atomic_dec(&cli->tc_ref);
2144 }
2145
2146 static void nrs_tbf_opcode_hop_exit(struct cfs_hash *hs,
2147                                     struct hlist_node *hnode)
2148 {
2149         struct nrs_tbf_client *cli = hlist_entry(hnode,
2150                                                  struct nrs_tbf_client,
2151                                                  tc_hnode);
2152
2153         LASSERTF(atomic_read(&cli->tc_ref) == 0,
2154                  "Busy TBF object from client with opcode %s, with %d refs\n",
2155                  ll_opcode2str(cli->tc_opcode),
2156                  atomic_read(&cli->tc_ref));
2157
2158         nrs_tbf_cli_fini(cli);
2159 }
2160 static struct cfs_hash_ops nrs_tbf_opcode_hash_ops = {
2161         .hs_hash        = nrs_tbf_opcode_hop_hash,
2162         .hs_keycmp      = nrs_tbf_opcode_hop_keycmp,
2163         .hs_key         = nrs_tbf_opcode_hop_key,
2164         .hs_object      = nrs_tbf_hop_object,
2165         .hs_get         = nrs_tbf_opcode_hop_get,
2166         .hs_put         = nrs_tbf_opcode_hop_put,
2167         .hs_put_locked  = nrs_tbf_opcode_hop_put,
2168         .hs_exit        = nrs_tbf_opcode_hop_exit,
2169 };
2170
2171 static int
2172 nrs_tbf_opcode_startup(struct ptlrpc_nrs_policy *policy,
2173                     struct nrs_tbf_head *head)
2174 {
2175         struct nrs_tbf_cmd      start = { 0 };
2176         int rc;
2177
2178         head->th_cli_hash = cfs_hash_create("nrs_tbf_hash",
2179                                             NRS_TBF_NID_BITS,
2180                                             NRS_TBF_NID_BITS,
2181                                             NRS_TBF_NID_BKT_BITS, 0,
2182                                             CFS_HASH_MIN_THETA,
2183                                             CFS_HASH_MAX_THETA,
2184                                             &nrs_tbf_opcode_hash_ops,
2185                                             CFS_HASH_RW_BKTLOCK);
2186         if (head->th_cli_hash == NULL)
2187                 return -ENOMEM;
2188
2189         start.u.tc_start.ts_opcodes = NULL;
2190         start.u.tc_start.ts_opcodes_str = "*";
2191
2192         start.u.tc_start.ts_rpc_rate = tbf_rate;
2193         start.u.tc_start.ts_rule_flags = NTRS_DEFAULT;
2194         start.tc_name = NRS_TBF_DEFAULT_RULE;
2195         rc = nrs_tbf_rule_start(policy, head, &start);
2196
2197         return rc;
2198 }
2199
2200 static struct nrs_tbf_client *
2201 nrs_tbf_opcode_cli_find(struct nrs_tbf_head *head,
2202                         struct ptlrpc_request *req)
2203 {
2204         __u32 opc;
2205
2206         opc = lustre_msg_get_opc(req->rq_reqmsg);
2207         return cfs_hash_lookup(head->th_cli_hash, &opc);
2208 }
2209
2210 static struct nrs_tbf_client *
2211 nrs_tbf_opcode_cli_findadd(struct nrs_tbf_head *head,
2212                            struct nrs_tbf_client *cli)
2213 {
2214         return cfs_hash_findadd_unique(head->th_cli_hash, &cli->tc_opcode,
2215                                        &cli->tc_hnode);
2216 }
2217
2218 static void
2219 nrs_tbf_opcode_cli_init(struct nrs_tbf_client *cli,
2220                         struct ptlrpc_request *req)
2221 {
2222         cli->tc_opcode = lustre_msg_get_opc(req->rq_reqmsg);
2223 }
2224
2225 #define MAX_OPCODE_LEN  32
2226 static int
2227 nrs_tbf_opcode_set_bit(const struct cfs_lstr *id, struct cfs_bitmap *opcodes)
2228 {
2229         int     op = 0;
2230         char    opcode_str[MAX_OPCODE_LEN];
2231
2232         if (id->ls_len + 1 > MAX_OPCODE_LEN)
2233                 return -EINVAL;
2234
2235         memcpy(opcode_str, id->ls_str, id->ls_len);
2236         opcode_str[id->ls_len] = '\0';
2237
2238         op = ll_str2opcode(opcode_str);
2239         if (op < 0)
2240                 return -EINVAL;
2241
2242         cfs_bitmap_set(opcodes, op);
2243         return 0;
2244 }
2245
2246 static int
2247 nrs_tbf_opcode_list_parse(char *str, int len, struct cfs_bitmap **bitmaptr)
2248 {
2249         struct cfs_bitmap *opcodes;
2250         struct cfs_lstr src;
2251         struct cfs_lstr res;
2252         int rc = 0;
2253         ENTRY;
2254
2255         opcodes = CFS_ALLOCATE_BITMAP(LUSTRE_MAX_OPCODES);
2256         if (opcodes == NULL)
2257                 return -ENOMEM;
2258
2259         src.ls_str = str;
2260         src.ls_len = len;
2261         while (src.ls_str) {
2262                 rc = cfs_gettok(&src, ' ', &res);
2263                 if (rc == 0) {
2264                         rc = -EINVAL;
2265                         break;
2266                 }
2267                 rc = nrs_tbf_opcode_set_bit(&res, opcodes);
2268                 if (rc)
2269                         break;
2270         }
2271
2272         if (rc == 0)
2273                 *bitmaptr = opcodes;
2274         else
2275                 CFS_FREE_BITMAP(opcodes);
2276
2277         RETURN(rc);
2278 }
2279
2280 static void nrs_tbf_opcode_cmd_fini(struct nrs_tbf_cmd *cmd)
2281 {
2282         if (cmd->u.tc_start.ts_opcodes)
2283                 CFS_FREE_BITMAP(cmd->u.tc_start.ts_opcodes);
2284
2285         if (cmd->u.tc_start.ts_opcodes_str)
2286                 OBD_FREE(cmd->u.tc_start.ts_opcodes_str,
2287                          strlen(cmd->u.tc_start.ts_opcodes_str) + 1);
2288
2289 }
2290
2291 static int nrs_tbf_opcode_parse(struct nrs_tbf_cmd *cmd, char *id)
2292 {
2293         struct cfs_lstr src;
2294         int rc;
2295
2296         src.ls_str = id;
2297         src.ls_len = strlen(id);
2298         rc = nrs_tbf_check_id_value(&src, "opcode");
2299         if (rc)
2300                 return rc;
2301
2302         OBD_ALLOC(cmd->u.tc_start.ts_opcodes_str, src.ls_len + 1);
2303         if (cmd->u.tc_start.ts_opcodes_str == NULL)
2304                 return -ENOMEM;
2305
2306         memcpy(cmd->u.tc_start.ts_opcodes_str, src.ls_str, src.ls_len);
2307
2308         /* parse opcode list */
2309         rc = nrs_tbf_opcode_list_parse(cmd->u.tc_start.ts_opcodes_str,
2310                                        strlen(cmd->u.tc_start.ts_opcodes_str),
2311                                        &cmd->u.tc_start.ts_opcodes);
2312         if (rc)
2313                 nrs_tbf_opcode_cmd_fini(cmd);
2314
2315         return rc;
2316 }
2317
2318 static int
2319 nrs_tbf_opcode_rule_match(struct nrs_tbf_rule *rule,
2320                           struct nrs_tbf_client *cli)
2321 {
2322         if (rule->tr_opcodes == NULL)
2323                 return 0;
2324
2325         return cfs_bitmap_check(rule->tr_opcodes, cli->tc_opcode);
2326 }
2327
2328 static int nrs_tbf_opcode_rule_init(struct ptlrpc_nrs_policy *policy,
2329                                     struct nrs_tbf_rule *rule,
2330                                     struct nrs_tbf_cmd *start)
2331 {
2332         int rc = 0;
2333
2334         LASSERT(start->u.tc_start.ts_opcodes_str != NULL);
2335         OBD_ALLOC(rule->tr_opcodes_str,
2336                   strlen(start->u.tc_start.ts_opcodes_str) + 1);
2337         if (rule->tr_opcodes_str == NULL)
2338                 return -ENOMEM;
2339
2340         strncpy(rule->tr_opcodes_str, start->u.tc_start.ts_opcodes_str,
2341                 strlen(start->u.tc_start.ts_opcodes_str) + 1);
2342
2343         /* Default rule '*' */
2344         if (start->u.tc_start.ts_opcodes == NULL)
2345                 return 0;
2346
2347         rc = nrs_tbf_opcode_list_parse(rule->tr_opcodes_str,
2348                                        strlen(rule->tr_opcodes_str),
2349                                        &rule->tr_opcodes);
2350         if (rc)
2351                 OBD_FREE(rule->tr_opcodes_str,
2352                          strlen(start->u.tc_start.ts_opcodes_str) + 1);
2353
2354         return rc;
2355 }
2356
2357 static int
2358 nrs_tbf_opcode_rule_dump(struct nrs_tbf_rule *rule, struct seq_file *m)
2359 {
2360         seq_printf(m, "%s {%s} %llu, ref %d\n", rule->tr_name,
2361                    rule->tr_opcodes_str, rule->tr_rpc_rate,
2362                    atomic_read(&rule->tr_ref) - 1);
2363         return 0;
2364 }
2365
2366
2367 struct nrs_tbf_ops nrs_tbf_opcode_ops = {
2368         .o_name = NRS_TBF_TYPE_OPCODE,
2369         .o_startup = nrs_tbf_opcode_startup,
2370         .o_cli_find = nrs_tbf_opcode_cli_find,
2371         .o_cli_findadd = nrs_tbf_opcode_cli_findadd,
2372         .o_cli_put = nrs_tbf_nid_cli_put,
2373         .o_cli_init = nrs_tbf_opcode_cli_init,
2374         .o_rule_init = nrs_tbf_opcode_rule_init,
2375         .o_rule_dump = nrs_tbf_opcode_rule_dump,
2376         .o_rule_match = nrs_tbf_opcode_rule_match,
2377         .o_rule_fini = nrs_tbf_opcode_rule_fini,
2378 };
2379
2380 static unsigned nrs_tbf_id_hop_hash(struct cfs_hash *hs, const void *key,
2381                                     unsigned mask)
2382 {
2383         return cfs_hash_djb2_hash(key, sizeof(struct tbf_id), mask);
2384 }
2385
2386 static int nrs_tbf_id_hop_keycmp(const void *key, struct hlist_node *hnode)
2387 {
2388         const struct tbf_id *opc = key;
2389         enum nrs_tbf_flag ntf;
2390         struct nrs_tbf_client *cli = hlist_entry(hnode, struct nrs_tbf_client,
2391                                                  tc_hnode);
2392         ntf = opc->ti_type & cli->tc_id.ti_type;
2393         if ((ntf & NRS_TBF_FLAG_UID) && opc->ti_uid != cli->tc_id.ti_uid)
2394                 return 0;
2395
2396         if ((ntf & NRS_TBF_FLAG_GID) && opc->ti_gid != cli->tc_id.ti_gid)
2397                 return 0;
2398
2399         return 1;
2400 }
2401
2402 static void *nrs_tbf_id_hop_key(struct hlist_node *hnode)
2403 {
2404         struct nrs_tbf_client *cli = hlist_entry(hnode,
2405                                                  struct nrs_tbf_client,
2406                                                  tc_hnode);
2407         return &cli->tc_id;
2408 }
2409
2410 static void nrs_tbf_id_hop_get(struct cfs_hash *hs, struct hlist_node *hnode)
2411 {
2412         struct nrs_tbf_client *cli = hlist_entry(hnode,
2413                                                  struct nrs_tbf_client,
2414                                                  tc_hnode);
2415
2416         atomic_inc(&cli->tc_ref);
2417 }
2418
2419 static void nrs_tbf_id_hop_put(struct cfs_hash *hs, struct hlist_node *hnode)
2420 {
2421         struct nrs_tbf_client *cli = hlist_entry(hnode,
2422                                                  struct nrs_tbf_client,
2423                                                  tc_hnode);
2424
2425         atomic_dec(&cli->tc_ref);
2426 }
2427
2428 static void
2429 nrs_tbf_id_hop_exit(struct cfs_hash *hs, struct hlist_node *hnode)
2430
2431 {
2432         struct nrs_tbf_client *cli = hlist_entry(hnode,
2433                                                  struct nrs_tbf_client,
2434                                                  tc_hnode);
2435
2436         LASSERT(atomic_read(&cli->tc_ref) == 0);
2437         nrs_tbf_cli_fini(cli);
2438 }
2439
2440 static struct cfs_hash_ops nrs_tbf_id_hash_ops = {
2441         .hs_hash        = nrs_tbf_id_hop_hash,
2442         .hs_keycmp      = nrs_tbf_id_hop_keycmp,
2443         .hs_key         = nrs_tbf_id_hop_key,
2444         .hs_object      = nrs_tbf_hop_object,
2445         .hs_get         = nrs_tbf_id_hop_get,
2446         .hs_put         = nrs_tbf_id_hop_put,
2447         .hs_put_locked  = nrs_tbf_id_hop_put,
2448         .hs_exit        = nrs_tbf_id_hop_exit,
2449 };
2450
2451 static int
2452 nrs_tbf_id_startup(struct ptlrpc_nrs_policy *policy,
2453                    struct nrs_tbf_head *head)
2454 {
2455         struct nrs_tbf_cmd start;
2456         int rc;
2457
2458         head->th_cli_hash = cfs_hash_create("nrs_tbf_id_hash",
2459                                             NRS_TBF_NID_BITS,
2460                                             NRS_TBF_NID_BITS,
2461                                             NRS_TBF_NID_BKT_BITS, 0,
2462                                             CFS_HASH_MIN_THETA,
2463                                             CFS_HASH_MAX_THETA,
2464                                             &nrs_tbf_id_hash_ops,
2465                                             CFS_HASH_RW_BKTLOCK);
2466         if (head->th_cli_hash == NULL)
2467                 return -ENOMEM;
2468
2469         memset(&start, 0, sizeof(start));
2470         start.u.tc_start.ts_ids_str = "*";
2471         start.u.tc_start.ts_rpc_rate = tbf_rate;
2472         start.u.tc_start.ts_rule_flags = NTRS_DEFAULT;
2473         start.tc_name = NRS_TBF_DEFAULT_RULE;
2474         INIT_LIST_HEAD(&start.u.tc_start.ts_ids);
2475         rc = nrs_tbf_rule_start(policy, head, &start);
2476         if (rc) {
2477                 cfs_hash_putref(head->th_cli_hash);
2478                 head->th_cli_hash = NULL;
2479         }
2480
2481         return rc;
2482 }
2483
2484 static struct nrs_tbf_client *
2485 nrs_tbf_id_cli_find(struct nrs_tbf_head *head,
2486                     struct ptlrpc_request *req)
2487 {
2488         struct tbf_id id;
2489
2490         LASSERT(head->th_type_flag == NRS_TBF_FLAG_UID ||
2491                 head->th_type_flag == NRS_TBF_FLAG_GID);
2492
2493         nrs_tbf_id_cli_set(req, &id, head->th_type_flag);
2494         return cfs_hash_lookup(head->th_cli_hash, &id);
2495 }
2496
2497 static struct nrs_tbf_client *
2498 nrs_tbf_id_cli_findadd(struct nrs_tbf_head *head,
2499                        struct nrs_tbf_client *cli)
2500 {
2501         return cfs_hash_findadd_unique(head->th_cli_hash, &cli->tc_id,
2502                                        &cli->tc_hnode);
2503 }
2504
2505 static void
2506 nrs_tbf_uid_cli_init(struct nrs_tbf_client *cli,
2507                      struct ptlrpc_request *req)
2508 {
2509         nrs_tbf_id_cli_set(req, &cli->tc_id, NRS_TBF_FLAG_UID);
2510 }
2511
2512 static void
2513 nrs_tbf_gid_cli_init(struct nrs_tbf_client *cli,
2514                      struct ptlrpc_request *req)
2515 {
2516         nrs_tbf_id_cli_set(req, &cli->tc_id, NRS_TBF_FLAG_GID);
2517 }
2518
2519 static int
2520 nrs_tbf_id_list_match(struct list_head *id_list, struct tbf_id id)
2521 {
2522         struct nrs_tbf_id *nti_id;
2523         enum nrs_tbf_flag flag;
2524
2525         list_for_each_entry(nti_id, id_list, nti_linkage) {
2526                 flag = id.ti_type & nti_id->nti_id.ti_type;
2527                 if (!flag)
2528                         continue;
2529
2530                 if ((flag & NRS_TBF_FLAG_UID) &&
2531                     (id.ti_uid != nti_id->nti_id.ti_uid))
2532                         continue;
2533
2534                 if ((flag & NRS_TBF_FLAG_GID) &&
2535                     (id.ti_gid != nti_id->nti_id.ti_gid))
2536                         continue;
2537
2538                 return 1;
2539         }
2540         return 0;
2541 }
2542
2543 static int
2544 nrs_tbf_id_rule_match(struct nrs_tbf_rule *rule,
2545                       struct nrs_tbf_client *cli)
2546 {
2547         return nrs_tbf_id_list_match(&rule->tr_ids, cli->tc_id);
2548 }
2549
2550 static void nrs_tbf_id_cmd_fini(struct nrs_tbf_cmd *cmd)
2551 {
2552         nrs_tbf_id_list_free(&cmd->u.tc_start.ts_ids);
2553
2554         if (cmd->u.tc_start.ts_ids_str)
2555                 OBD_FREE(cmd->u.tc_start.ts_ids_str,
2556                          strlen(cmd->u.tc_start.ts_ids_str) + 1);
2557 }
2558
2559 static int
2560 nrs_tbf_id_list_parse(char *str, int len, struct list_head *id_list,
2561                       enum nrs_tbf_flag tif)
2562 {
2563         struct cfs_lstr src;
2564         struct cfs_lstr res;
2565         int rc = 0;
2566         struct tbf_id id = { 0 };
2567         ENTRY;
2568
2569         if (tif != NRS_TBF_FLAG_UID && tif != NRS_TBF_FLAG_GID)
2570                 RETURN(-EINVAL);
2571
2572         src.ls_str = str;
2573         src.ls_len = len;
2574         INIT_LIST_HEAD(id_list);
2575         while (src.ls_str) {
2576                 struct nrs_tbf_id *nti_id;
2577
2578                 if (cfs_gettok(&src, ' ', &res) == 0)
2579                         GOTO(out, rc = -EINVAL);
2580
2581                 id.ti_type = tif;
2582                 if (tif == NRS_TBF_FLAG_UID) {
2583                         if (!cfs_str2num_check(res.ls_str, res.ls_len,
2584                                                &id.ti_uid, 0, (u32)~0U))
2585                                 GOTO(out, rc = -EINVAL);
2586                 } else {
2587                         if (!cfs_str2num_check(res.ls_str, res.ls_len,
2588                                                &id.ti_gid, 0, (u32)~0U))
2589                                 GOTO(out, rc = -EINVAL);
2590                 }
2591
2592                 OBD_ALLOC_PTR(nti_id);
2593                 if (nti_id == NULL)
2594                         GOTO(out, rc = -ENOMEM);
2595
2596                 nti_id->nti_id = id;
2597                 list_add_tail(&nti_id->nti_linkage, id_list);
2598         }
2599 out:
2600         if (rc)
2601                 nrs_tbf_id_list_free(id_list);
2602         RETURN(rc);
2603 }
2604
2605 static int nrs_tbf_ug_id_parse(struct nrs_tbf_cmd *cmd, char *id)
2606 {
2607         struct cfs_lstr src;
2608         int rc;
2609         enum nrs_tbf_flag tif;
2610
2611         tif = cmd->u.tc_start.ts_valid_type;
2612
2613         src.ls_str = id;
2614         src.ls_len = strlen(id);
2615
2616         rc = nrs_tbf_check_id_value(&src,
2617                                     tif == NRS_TBF_FLAG_UID ? "uid" : "gid");
2618         if (rc)
2619                 return rc;
2620
2621         OBD_ALLOC(cmd->u.tc_start.ts_ids_str, src.ls_len + 1);
2622         if (cmd->u.tc_start.ts_ids_str == NULL)
2623                 return -ENOMEM;
2624
2625         strlcpy(cmd->u.tc_start.ts_ids_str, src.ls_str, src.ls_len + 1);
2626
2627         rc = nrs_tbf_id_list_parse(cmd->u.tc_start.ts_ids_str,
2628                                    strlen(cmd->u.tc_start.ts_ids_str),
2629                                    &cmd->u.tc_start.ts_ids, tif);
2630         if (rc)
2631                 nrs_tbf_id_cmd_fini(cmd);
2632
2633         return rc;
2634 }
2635
2636 static int
2637 nrs_tbf_id_rule_init(struct ptlrpc_nrs_policy *policy,
2638                      struct nrs_tbf_rule *rule,
2639                      struct nrs_tbf_cmd *start)
2640 {
2641         struct nrs_tbf_head *head = rule->tr_head;
2642         int rc = 0;
2643         enum nrs_tbf_flag tif = head->th_type_flag;
2644         int ids_len = strlen(start->u.tc_start.ts_ids_str) + 1;
2645
2646         LASSERT(start->u.tc_start.ts_ids_str);
2647         INIT_LIST_HEAD(&rule->tr_ids);
2648
2649         OBD_ALLOC(rule->tr_ids_str, ids_len);
2650         if (rule->tr_ids_str == NULL)
2651                 return -ENOMEM;
2652
2653         strlcpy(rule->tr_ids_str, start->u.tc_start.ts_ids_str,
2654                 ids_len);
2655
2656         if (!list_empty(&start->u.tc_start.ts_ids)) {
2657                 rc = nrs_tbf_id_list_parse(rule->tr_ids_str,
2658                                            strlen(rule->tr_ids_str),
2659                                            &rule->tr_ids, tif);
2660                 if (rc)
2661                         CERROR("%ss {%s} illegal\n",
2662                                tif == NRS_TBF_FLAG_UID ? "uid" : "gid",
2663                                rule->tr_ids_str);
2664         }
2665         if (rc) {
2666                 OBD_FREE(rule->tr_ids_str, ids_len);
2667                 rule->tr_ids_str = NULL;
2668         }
2669         return rc;
2670 }
2671
2672 static int
2673 nrs_tbf_id_rule_dump(struct nrs_tbf_rule *rule, struct seq_file *m)
2674 {
2675         seq_printf(m, "%s {%s} %llu, ref %d\n", rule->tr_name,
2676                    rule->tr_ids_str, rule->tr_rpc_rate,
2677                    atomic_read(&rule->tr_ref) - 1);
2678         return 0;
2679 }
2680
2681 static void nrs_tbf_id_rule_fini(struct nrs_tbf_rule *rule)
2682 {
2683         nrs_tbf_id_list_free(&rule->tr_ids);
2684         if (rule->tr_ids_str != NULL)
2685                 OBD_FREE(rule->tr_ids_str, strlen(rule->tr_ids_str) + 1);
2686 }
2687
2688 struct nrs_tbf_ops nrs_tbf_uid_ops = {
2689         .o_name = NRS_TBF_TYPE_UID,
2690         .o_startup = nrs_tbf_id_startup,
2691         .o_cli_find = nrs_tbf_id_cli_find,
2692         .o_cli_findadd = nrs_tbf_id_cli_findadd,
2693         .o_cli_put = nrs_tbf_nid_cli_put,
2694         .o_cli_init = nrs_tbf_uid_cli_init,
2695         .o_rule_init = nrs_tbf_id_rule_init,
2696         .o_rule_dump = nrs_tbf_id_rule_dump,
2697         .o_rule_match = nrs_tbf_id_rule_match,
2698         .o_rule_fini = nrs_tbf_id_rule_fini,
2699 };
2700
2701 struct nrs_tbf_ops nrs_tbf_gid_ops = {
2702         .o_name = NRS_TBF_TYPE_GID,
2703         .o_startup = nrs_tbf_id_startup,
2704         .o_cli_find = nrs_tbf_id_cli_find,
2705         .o_cli_findadd = nrs_tbf_id_cli_findadd,
2706         .o_cli_put = nrs_tbf_nid_cli_put,
2707         .o_cli_init = nrs_tbf_gid_cli_init,
2708         .o_rule_init = nrs_tbf_id_rule_init,
2709         .o_rule_dump = nrs_tbf_id_rule_dump,
2710         .o_rule_match = nrs_tbf_id_rule_match,
2711         .o_rule_fini = nrs_tbf_id_rule_fini,
2712 };
2713
2714 static struct nrs_tbf_type nrs_tbf_types[] = {
2715         {
2716                 .ntt_name = NRS_TBF_TYPE_JOBID,
2717                 .ntt_flag = NRS_TBF_FLAG_JOBID,
2718                 .ntt_ops = &nrs_tbf_jobid_ops,
2719         },
2720         {
2721                 .ntt_name = NRS_TBF_TYPE_NID,
2722                 .ntt_flag = NRS_TBF_FLAG_NID,
2723                 .ntt_ops = &nrs_tbf_nid_ops,
2724         },
2725         {
2726                 .ntt_name = NRS_TBF_TYPE_OPCODE,
2727                 .ntt_flag = NRS_TBF_FLAG_OPCODE,
2728                 .ntt_ops = &nrs_tbf_opcode_ops,
2729         },
2730         {
2731                 .ntt_name = NRS_TBF_TYPE_GENERIC,
2732                 .ntt_flag = NRS_TBF_FLAG_GENERIC,
2733                 .ntt_ops = &nrs_tbf_generic_ops,
2734         },
2735         {
2736                 .ntt_name = NRS_TBF_TYPE_UID,
2737                 .ntt_flag = NRS_TBF_FLAG_UID,
2738                 .ntt_ops = &nrs_tbf_uid_ops,
2739         },
2740         {
2741                 .ntt_name = NRS_TBF_TYPE_GID,
2742                 .ntt_flag = NRS_TBF_FLAG_GID,
2743                 .ntt_ops = &nrs_tbf_gid_ops,
2744         },
2745 };
2746
2747 /**
2748  * Is called before the policy transitions into
2749  * ptlrpc_nrs_pol_state::NRS_POL_STATE_STARTED; allocates and initializes a
2750  * policy-specific private data structure.
2751  *
2752  * \param[in] policy The policy to start
2753  *
2754  * \retval -ENOMEM OOM error
2755  * \retval  0      success
2756  *
2757  * \see nrs_policy_register()
2758  * \see nrs_policy_ctl()
2759  */
2760 static int nrs_tbf_start(struct ptlrpc_nrs_policy *policy, char *arg)
2761 {
2762         struct nrs_tbf_head     *head;
2763         struct nrs_tbf_ops      *ops;
2764         __u32                    type;
2765         char                    *name;
2766         int found = 0;
2767         int i;
2768         int rc = 0;
2769
2770         if (arg == NULL)
2771                 name = NRS_TBF_TYPE_GENERIC;
2772         else if (strlen(arg) < NRS_TBF_TYPE_MAX_LEN)
2773                 name = arg;
2774         else
2775                 GOTO(out, rc = -EINVAL);
2776
2777         for (i = 0; i < ARRAY_SIZE(nrs_tbf_types); i++) {
2778                 if (strcmp(name, nrs_tbf_types[i].ntt_name) == 0) {
2779                         ops = nrs_tbf_types[i].ntt_ops;
2780                         type = nrs_tbf_types[i].ntt_flag;
2781                         found = 1;
2782                         break;
2783                 }
2784         }
2785         if (found == 0)
2786                 GOTO(out, rc = -ENOTSUPP);
2787
2788         OBD_CPT_ALLOC_PTR(head, nrs_pol2cptab(policy), nrs_pol2cptid(policy));
2789         if (head == NULL)
2790                 GOTO(out, rc = -ENOMEM);
2791
2792         memcpy(head->th_type, name, strlen(name));
2793         head->th_type[strlen(name)] = '\0';
2794         head->th_ops = ops;
2795         head->th_type_flag = type;
2796
2797         head->th_binheap = cfs_binheap_create(&nrs_tbf_heap_ops,
2798                                               CBH_FLAG_ATOMIC_GROW, 4096, NULL,
2799                                               nrs_pol2cptab(policy),
2800                                               nrs_pol2cptid(policy));
2801         if (head->th_binheap == NULL)
2802                 GOTO(out_free_head, rc = -ENOMEM);
2803
2804         atomic_set(&head->th_rule_sequence, 0);
2805         spin_lock_init(&head->th_rule_lock);
2806         INIT_LIST_HEAD(&head->th_list);
2807         hrtimer_init(&head->th_timer, CLOCK_MONOTONIC, HRTIMER_MODE_ABS);
2808         head->th_timer.function = nrs_tbf_timer_cb;
2809         rc = head->th_ops->o_startup(policy, head);
2810         if (rc)
2811                 GOTO(out_free_heap, rc);
2812
2813         policy->pol_private = head;
2814         return 0;
2815 out_free_heap:
2816         cfs_binheap_destroy(head->th_binheap);
2817 out_free_head:
2818         OBD_FREE_PTR(head);
2819 out:
2820         return rc;
2821 }
2822
2823 /**
2824  * Is called before the policy transitions into
2825  * ptlrpc_nrs_pol_state::NRS_POL_STATE_STOPPED; deallocates the policy-specific
2826  * private data structure.
2827  *
2828  * \param[in] policy The policy to stop
2829  *
2830  * \see nrs_policy_stop0()
2831  */
2832 static void nrs_tbf_stop(struct ptlrpc_nrs_policy *policy)
2833 {
2834         struct nrs_tbf_head *head = policy->pol_private;
2835         struct ptlrpc_nrs *nrs = policy->pol_nrs;
2836         struct nrs_tbf_rule *rule, *n;
2837
2838         LASSERT(head != NULL);
2839         LASSERT(head->th_cli_hash != NULL);
2840         hrtimer_cancel(&head->th_timer);
2841         /* Should cleanup hash first before free rules */
2842         cfs_hash_putref(head->th_cli_hash);
2843         list_for_each_entry_safe(rule, n, &head->th_list, tr_linkage) {
2844                 list_del_init(&rule->tr_linkage);
2845                 nrs_tbf_rule_put(rule);
2846         }
2847         LASSERT(list_empty(&head->th_list));
2848         LASSERT(head->th_binheap != NULL);
2849         LASSERT(cfs_binheap_is_empty(head->th_binheap));
2850         cfs_binheap_destroy(head->th_binheap);
2851         OBD_FREE_PTR(head);
2852         nrs->nrs_throttling = 0;
2853         wake_up(&policy->pol_nrs->nrs_svcpt->scp_waitq);
2854 }
2855
2856 /**
2857  * Performs a policy-specific ctl function on TBF policy instances; similar
2858  * to ioctl.
2859  *
2860  * \param[in]     policy the policy instance
2861  * \param[in]     opc    the opcode
2862  * \param[in,out] arg    used for passing parameters and information
2863  *
2864  * \pre assert_spin_locked(&policy->pol_nrs->->nrs_lock)
2865  * \post assert_spin_locked(&policy->pol_nrs->->nrs_lock)
2866  *
2867  * \retval 0   operation carried out successfully
2868  * \retval -ve error
2869  */
2870 static int nrs_tbf_ctl(struct ptlrpc_nrs_policy *policy,
2871                        enum ptlrpc_nrs_ctl opc,
2872                        void *arg)
2873 {
2874         int rc = 0;
2875         ENTRY;
2876
2877         assert_spin_locked(&policy->pol_nrs->nrs_lock);
2878
2879         switch ((enum nrs_ctl_tbf)opc) {
2880         default:
2881                 RETURN(-EINVAL);
2882
2883         /**
2884          * Read RPC rate size of a policy instance.
2885          */
2886         case NRS_CTL_TBF_RD_RULE: {
2887                 struct nrs_tbf_head *head = policy->pol_private;
2888                 struct seq_file *m = (struct seq_file *) arg;
2889                 struct ptlrpc_service_part *svcpt;
2890
2891                 svcpt = policy->pol_nrs->nrs_svcpt;
2892                 seq_printf(m, "CPT %d:\n", svcpt->scp_cpt);
2893
2894                 rc = nrs_tbf_rule_dump_all(head, m);
2895                 }
2896                 break;
2897
2898         /**
2899          * Write RPC rate of a policy instance.
2900          */
2901         case NRS_CTL_TBF_WR_RULE: {
2902                 struct nrs_tbf_head *head = policy->pol_private;
2903                 struct nrs_tbf_cmd *cmd;
2904
2905                 cmd = (struct nrs_tbf_cmd *)arg;
2906                 rc = nrs_tbf_command(policy,
2907                                      head,
2908                                      cmd);
2909                 }
2910                 break;
2911         /**
2912          * Read the TBF policy type of a policy instance.
2913          */
2914         case NRS_CTL_TBF_RD_TYPE_FLAG: {
2915                 struct nrs_tbf_head *head = policy->pol_private;
2916
2917                 *(__u32 *)arg = head->th_type_flag;
2918                 }
2919                 break;
2920         }
2921
2922         RETURN(rc);
2923 }
2924
2925 /**
2926  * Is called for obtaining a TBF policy resource.
2927  *
2928  * \param[in]  policy     The policy on which the request is being asked for
2929  * \param[in]  nrq        The request for which resources are being taken
2930  * \param[in]  parent     Parent resource, unused in this policy
2931  * \param[out] resp       Resources references are placed in this array
2932  * \param[in]  moving_req Signifies limited caller context; unused in this
2933  *                        policy
2934  *
2935  *
2936  * \see nrs_resource_get_safe()
2937  */
2938 static int nrs_tbf_res_get(struct ptlrpc_nrs_policy *policy,
2939                            struct ptlrpc_nrs_request *nrq,
2940                            const struct ptlrpc_nrs_resource *parent,
2941                            struct ptlrpc_nrs_resource **resp,
2942                            bool moving_req)
2943 {
2944         struct nrs_tbf_head   *head;
2945         struct nrs_tbf_client *cli;
2946         struct nrs_tbf_client *tmp;
2947         struct ptlrpc_request *req;
2948
2949         if (parent == NULL) {
2950                 *resp = &((struct nrs_tbf_head *)policy->pol_private)->th_res;
2951                 return 0;
2952         }
2953
2954         head = container_of(parent, struct nrs_tbf_head, th_res);
2955         req = container_of(nrq, struct ptlrpc_request, rq_nrq);
2956         cli = head->th_ops->o_cli_find(head, req);
2957         if (cli != NULL) {
2958                 spin_lock(&policy->pol_nrs->nrs_svcpt->scp_req_lock);
2959                 LASSERT(cli->tc_rule);
2960                 if (cli->tc_rule_sequence !=
2961                     atomic_read(&head->th_rule_sequence) ||
2962                     cli->tc_rule->tr_flags & NTRS_STOPPING) {
2963                         struct nrs_tbf_rule *rule;
2964
2965                         CDEBUG(D_RPCTRACE,
2966                                "TBF class@%p rate %llu sequence %d, "
2967                                "rule flags %d, head sequence %d\n",
2968                                cli, cli->tc_rpc_rate,
2969                                cli->tc_rule_sequence,
2970                                cli->tc_rule->tr_flags,
2971                                atomic_read(&head->th_rule_sequence));
2972                         rule = nrs_tbf_rule_match(head, cli);
2973                         if (rule != cli->tc_rule) {
2974                                 nrs_tbf_cli_reset(head, rule, cli);
2975                         } else {
2976                                 if (cli->tc_rule_generation != rule->tr_generation)
2977                                         nrs_tbf_cli_reset_value(head, cli);
2978                                 nrs_tbf_rule_put(rule);
2979                         }
2980                 } else if (cli->tc_rule_generation !=
2981                            cli->tc_rule->tr_generation) {
2982                         nrs_tbf_cli_reset_value(head, cli);
2983                 }
2984                 spin_unlock(&policy->pol_nrs->nrs_svcpt->scp_req_lock);
2985                 goto out;
2986         }
2987
2988         OBD_CPT_ALLOC_GFP(cli, nrs_pol2cptab(policy), nrs_pol2cptid(policy),
2989                           sizeof(*cli), moving_req ? GFP_ATOMIC : __GFP_IO);
2990         if (cli == NULL)
2991                 return -ENOMEM;
2992
2993         nrs_tbf_cli_init(head, cli, req);
2994         tmp = head->th_ops->o_cli_findadd(head, cli);
2995         if (tmp != cli) {
2996                 atomic_dec(&cli->tc_ref);
2997                 nrs_tbf_cli_fini(cli);
2998                 cli = tmp;
2999         }
3000 out:
3001         *resp = &cli->tc_res;
3002
3003         return 1;
3004 }
3005
3006 /**
3007  * Called when releasing references to the resource hierachy obtained for a
3008  * request for scheduling using the TBF policy.
3009  *
3010  * \param[in] policy   the policy the resource belongs to
3011  * \param[in] res      the resource to be released
3012  */
3013 static void nrs_tbf_res_put(struct ptlrpc_nrs_policy *policy,
3014                             const struct ptlrpc_nrs_resource *res)
3015 {
3016         struct nrs_tbf_head   *head;
3017         struct nrs_tbf_client *cli;
3018
3019         /**
3020          * Do nothing for freeing parent, nrs_tbf_net resources
3021          */
3022         if (res->res_parent == NULL)
3023                 return;
3024
3025         cli = container_of(res, struct nrs_tbf_client, tc_res);
3026         head = container_of(res->res_parent, struct nrs_tbf_head, th_res);
3027
3028         head->th_ops->o_cli_put(head, cli);
3029 }
3030
3031 /**
3032  * Called when getting a request from the TBF policy for handling, or just
3033  * peeking; removes the request from the policy when it is to be handled.
3034  *
3035  * \param[in] policy The policy
3036  * \param[in] peek   When set, signifies that we just want to examine the
3037  *                   request, and not handle it, so the request is not removed
3038  *                   from the policy.
3039  * \param[in] force  Force the policy to return a request; unused in this
3040  *                   policy
3041  *
3042  * \retval The request to be handled; this is the next request in the TBF
3043  *         rule
3044  *
3045  * \see ptlrpc_nrs_req_get_nolock()
3046  * \see nrs_request_get()
3047  */
3048 static
3049 struct ptlrpc_nrs_request *nrs_tbf_req_get(struct ptlrpc_nrs_policy *policy,
3050                                            bool peek, bool force)
3051 {
3052         struct nrs_tbf_head       *head = policy->pol_private;
3053         struct ptlrpc_nrs_request *nrq = NULL;
3054         struct nrs_tbf_client     *cli;
3055         struct cfs_binheap_node   *node;
3056
3057         assert_spin_locked(&policy->pol_nrs->nrs_svcpt->scp_req_lock);
3058
3059         if (!peek && policy->pol_nrs->nrs_throttling)
3060                 return NULL;
3061
3062         node = cfs_binheap_root(head->th_binheap);
3063         if (unlikely(node == NULL))
3064                 return NULL;
3065
3066         cli = container_of(node, struct nrs_tbf_client, tc_node);
3067         LASSERT(cli->tc_in_heap);
3068         if (peek) {
3069                 nrq = list_entry(cli->tc_list.next,
3070                                      struct ptlrpc_nrs_request,
3071                                      nr_u.tbf.tr_list);
3072         } else {
3073                 struct nrs_tbf_rule *rule = cli->tc_rule;
3074                 __u64 now = ktime_to_ns(ktime_get());
3075                 __u64 passed;
3076                 __u64 ntoken;
3077                 __u64 deadline;
3078                 __u64 old_resid = 0;
3079
3080                 deadline = cli->tc_check_time +
3081                           cli->tc_nsecs;
3082                 LASSERT(now >= cli->tc_check_time);
3083                 passed = now - cli->tc_check_time;
3084                 ntoken = passed * cli->tc_rpc_rate;
3085                 do_div(ntoken, NSEC_PER_SEC);
3086
3087                 ntoken += cli->tc_ntoken;
3088                 if (rule->tr_flags & NTRS_REALTIME) {
3089                         LASSERT(cli->tc_nsecs_resid < cli->tc_nsecs);
3090                         old_resid = cli->tc_nsecs_resid;
3091                         cli->tc_nsecs_resid += passed % cli->tc_nsecs;
3092                         if (cli->tc_nsecs_resid > cli->tc_nsecs) {
3093                                 ntoken++;
3094                                 cli->tc_nsecs_resid -= cli->tc_nsecs;
3095                         }
3096                 } else if (ntoken > cli->tc_depth)
3097                         ntoken = cli->tc_depth;
3098
3099                 if (ntoken > 0) {
3100                         struct ptlrpc_request *req;
3101                         nrq = list_entry(cli->tc_list.next,
3102                                              struct ptlrpc_nrs_request,
3103                                              nr_u.tbf.tr_list);
3104                         req = container_of(nrq,
3105                                            struct ptlrpc_request,
3106                                            rq_nrq);
3107                         ntoken--;
3108                         cli->tc_ntoken = ntoken;
3109                         cli->tc_check_time = now;
3110                         list_del_init(&nrq->nr_u.tbf.tr_list);
3111                         if (list_empty(&cli->tc_list)) {
3112                                 cfs_binheap_remove(head->th_binheap,
3113                                                    &cli->tc_node);
3114                                 cli->tc_in_heap = false;
3115                         } else {
3116                                 if (!(rule->tr_flags & NTRS_REALTIME))
3117                                         cli->tc_deadline = now + cli->tc_nsecs;
3118                                 cfs_binheap_relocate(head->th_binheap,
3119                                                      &cli->tc_node);
3120                         }
3121                         CDEBUG(D_RPCTRACE,
3122                                "TBF dequeues: class@%p rate %llu gen %llu "
3123                                "token %llu, rule@%p rate %llu gen %llu\n",
3124                                cli, cli->tc_rpc_rate,
3125                                cli->tc_rule_generation, cli->tc_ntoken,
3126                                cli->tc_rule, cli->tc_rule->tr_rpc_rate,
3127                                cli->tc_rule->tr_generation);
3128                 } else {
3129                         ktime_t time;
3130
3131                         if (rule->tr_flags & NTRS_REALTIME) {
3132                                 cli->tc_deadline = deadline;
3133                                 cli->tc_nsecs_resid = old_resid;
3134                                 cfs_binheap_relocate(head->th_binheap,
3135                                                      &cli->tc_node);
3136                                 if (node != cfs_binheap_root(head->th_binheap))
3137                                         return nrs_tbf_req_get(policy,
3138                                                                peek, force);
3139                         }
3140                         policy->pol_nrs->nrs_throttling = 1;
3141                         head->th_deadline = deadline;
3142                         time = ktime_set(0, 0);
3143                         time = ktime_add_ns(time, deadline);
3144                         hrtimer_start(&head->th_timer, time, HRTIMER_MODE_ABS);
3145                 }
3146         }
3147
3148         return nrq;
3149 }
3150
3151 /**
3152  * Adds request \a nrq to \a policy's list of queued requests
3153  *
3154  * \param[in] policy The policy
3155  * \param[in] nrq    The request to add
3156  *
3157  * \retval 0 success; nrs_request_enqueue() assumes this function will always
3158  *                    succeed
3159  */
3160 static int nrs_tbf_req_add(struct ptlrpc_nrs_policy *policy,
3161                            struct ptlrpc_nrs_request *nrq)
3162 {
3163         struct nrs_tbf_head   *head;
3164         struct nrs_tbf_client *cli;
3165         int                    rc = 0;
3166
3167         assert_spin_locked(&policy->pol_nrs->nrs_svcpt->scp_req_lock);
3168
3169         cli = container_of(nrs_request_resource(nrq),
3170                            struct nrs_tbf_client, tc_res);
3171         head = container_of(nrs_request_resource(nrq)->res_parent,
3172                             struct nrs_tbf_head, th_res);
3173         if (list_empty(&cli->tc_list)) {
3174                 LASSERT(!cli->tc_in_heap);
3175                 cli->tc_deadline = cli->tc_check_time + cli->tc_nsecs;
3176                 rc = cfs_binheap_insert(head->th_binheap, &cli->tc_node);
3177                 if (rc == 0) {
3178                         cli->tc_in_heap = true;
3179                         nrq->nr_u.tbf.tr_sequence = head->th_sequence++;
3180                         list_add_tail(&nrq->nr_u.tbf.tr_list,
3181                                           &cli->tc_list);
3182                         if (policy->pol_nrs->nrs_throttling) {
3183                                 __u64 deadline = cli->tc_deadline;
3184                                 if ((head->th_deadline > deadline) &&
3185                                     (hrtimer_try_to_cancel(&head->th_timer)
3186                                      >= 0)) {
3187                                         ktime_t time;
3188                                         head->th_deadline = deadline;
3189                                         time = ktime_set(0, 0);
3190                                         time = ktime_add_ns(time, deadline);
3191                                         hrtimer_start(&head->th_timer, time,
3192                                                       HRTIMER_MODE_ABS);
3193                                 }
3194                         }
3195                 }
3196         } else {
3197                 LASSERT(cli->tc_in_heap);
3198                 nrq->nr_u.tbf.tr_sequence = head->th_sequence++;
3199                 list_add_tail(&nrq->nr_u.tbf.tr_list,
3200                                   &cli->tc_list);
3201         }
3202
3203         if (rc == 0)
3204                 CDEBUG(D_RPCTRACE,
3205                        "TBF enqueues: class@%p rate %llu gen %llu "
3206                        "token %llu, rule@%p rate %llu gen %llu\n",
3207                        cli, cli->tc_rpc_rate,
3208                        cli->tc_rule_generation, cli->tc_ntoken,
3209                        cli->tc_rule, cli->tc_rule->tr_rpc_rate,
3210                        cli->tc_rule->tr_generation);
3211
3212         return rc;
3213 }
3214
3215 /**
3216  * Removes request \a nrq from \a policy's list of queued requests.
3217  *
3218  * \param[in] policy The policy
3219  * \param[in] nrq    The request to remove
3220  */
3221 static void nrs_tbf_req_del(struct ptlrpc_nrs_policy *policy,
3222                              struct ptlrpc_nrs_request *nrq)
3223 {
3224         struct nrs_tbf_head   *head;
3225         struct nrs_tbf_client *cli;
3226
3227         assert_spin_locked(&policy->pol_nrs->nrs_svcpt->scp_req_lock);
3228
3229         cli = container_of(nrs_request_resource(nrq),
3230                            struct nrs_tbf_client, tc_res);
3231         head = container_of(nrs_request_resource(nrq)->res_parent,
3232                             struct nrs_tbf_head, th_res);
3233
3234         LASSERT(!list_empty(&nrq->nr_u.tbf.tr_list));
3235         list_del_init(&nrq->nr_u.tbf.tr_list);
3236         if (list_empty(&cli->tc_list)) {
3237                 cfs_binheap_remove(head->th_binheap,
3238                                    &cli->tc_node);
3239                 cli->tc_in_heap = false;
3240         } else {
3241                 cfs_binheap_relocate(head->th_binheap,
3242                                      &cli->tc_node);
3243         }
3244 }
3245
3246 /**
3247  * Prints a debug statement right before the request \a nrq stops being
3248  * handled.
3249  *
3250  * \param[in] policy The policy handling the request
3251  * \param[in] nrq    The request being handled
3252  *
3253  * \see ptlrpc_server_finish_request()
3254  * \see ptlrpc_nrs_req_stop_nolock()
3255  */
3256 static void nrs_tbf_req_stop(struct ptlrpc_nrs_policy *policy,
3257                               struct ptlrpc_nrs_request *nrq)
3258 {
3259         struct ptlrpc_request *req = container_of(nrq, struct ptlrpc_request,
3260                                                   rq_nrq);
3261
3262         assert_spin_locked(&policy->pol_nrs->nrs_svcpt->scp_req_lock);
3263
3264         CDEBUG(D_RPCTRACE, "NRS stop %s request from %s, seq: %llu\n",
3265                policy->pol_desc->pd_name, libcfs_id2str(req->rq_peer),
3266                nrq->nr_u.tbf.tr_sequence);
3267 }
3268
3269 /**
3270  * debugfs interface
3271  */
3272
3273 /**
3274  * The maximum RPC rate.
3275  */
3276 #define LPROCFS_NRS_RATE_MAX            65535
3277
3278 static int
3279 ptlrpc_lprocfs_nrs_tbf_rule_seq_show(struct seq_file *m, void *data)
3280 {
3281         struct ptlrpc_service       *svc = m->private;
3282         int                          rc;
3283
3284         seq_printf(m, "regular_requests:\n");
3285         /**
3286          * Perform two separate calls to this as only one of the NRS heads'
3287          * policies may be in the ptlrpc_nrs_pol_state::NRS_POL_STATE_STARTED or
3288          * ptlrpc_nrs_pol_state::NRS_POL_STATE_STOPPING state.
3289          */
3290         rc = ptlrpc_nrs_policy_control(svc, PTLRPC_NRS_QUEUE_REG,
3291                                        NRS_POL_NAME_TBF,
3292                                        NRS_CTL_TBF_RD_RULE,
3293                                        false, m);
3294         if (rc == 0) {
3295                 /**
3296                  * -ENOSPC means buf in the parameter m is overflow, return 0
3297                  * here to let upper layer function seq_read alloc a larger
3298                  * memory area and do this process again.
3299                  */
3300         } else if (rc == -ENOSPC) {
3301                 return 0;
3302
3303                 /**
3304                  * Ignore -ENODEV as the regular NRS head's policy may be in the
3305                  * ptlrpc_nrs_pol_state::NRS_POL_STATE_STOPPED state.
3306                  */
3307         } else if (rc != -ENODEV) {
3308                 return rc;
3309         }
3310
3311         if (!nrs_svc_has_hp(svc))
3312                 goto no_hp;
3313
3314         seq_printf(m, "high_priority_requests:\n");
3315         rc = ptlrpc_nrs_policy_control(svc, PTLRPC_NRS_QUEUE_HP,
3316                                        NRS_POL_NAME_TBF,
3317                                        NRS_CTL_TBF_RD_RULE,
3318                                        false, m);
3319         if (rc == 0) {
3320                 /**
3321                  * -ENOSPC means buf in the parameter m is overflow, return 0
3322                  * here to let upper layer function seq_read alloc a larger
3323                  * memory area and do this process again.
3324                  */
3325         } else if (rc == -ENOSPC) {
3326                 return 0;
3327         }
3328
3329 no_hp:
3330
3331         return rc;
3332 }
3333
3334 static int nrs_tbf_id_parse(struct nrs_tbf_cmd *cmd, char *token)
3335 {
3336         int rc;
3337         ENTRY;
3338
3339         switch (cmd->u.tc_start.ts_valid_type) {
3340         case NRS_TBF_FLAG_JOBID:
3341                 rc = nrs_tbf_jobid_parse(cmd, token);
3342                 break;
3343         case NRS_TBF_FLAG_NID:
3344                 rc = nrs_tbf_nid_parse(cmd, token);
3345                 break;
3346         case NRS_TBF_FLAG_OPCODE:
3347                 rc = nrs_tbf_opcode_parse(cmd, token);
3348                 break;
3349         case NRS_TBF_FLAG_GENERIC:
3350                 rc = nrs_tbf_generic_parse(cmd, token);
3351                 break;
3352         case NRS_TBF_FLAG_UID:
3353         case NRS_TBF_FLAG_GID:
3354                 rc = nrs_tbf_ug_id_parse(cmd, token);
3355                 break;
3356         default:
3357                 RETURN(-EINVAL);
3358         }
3359
3360         RETURN(rc);
3361 }
3362
3363 static void nrs_tbf_cmd_fini(struct nrs_tbf_cmd *cmd)
3364 {
3365         if (cmd->tc_cmd == NRS_CTL_TBF_START_RULE) {
3366                 switch (cmd->u.tc_start.ts_valid_type) {
3367                 case NRS_TBF_FLAG_JOBID: