Viewing: nrs_tbf.c
// SPDX-License-Identifier: GPL-2.0
/*
* Copyright (C) 2013 DataDirect Networks, Inc.
*
* Copyright (c) 2014, 2016, Intel Corporation.
*/
/*
* Network Request Scheduler (NRS) Token Bucket Filter(TBF) policy
*/
#define DEBUG_SUBSYSTEM S_RPC
#include <linux/delay.h>
#include <obd_support.h>
#include <obd_class.h>
#include <lustre_req_layout.h>
#include <lustre_nodemap.h>
#include "ptlrpc_internal.h"
/*
* name tbf
*
* Token Bucket Filter over client NIDs
*/
#define NRS_POL_NAME_TBF "tbf"
static int tbf_jobid_cache_size = 8192;
module_param(tbf_jobid_cache_size, int, 0644);
MODULE_PARM_DESC(tbf_jobid_cache_size, "The size of jobid cache");
static int tbf_rate = 10000;
module_param(tbf_rate, int, 0644);
MODULE_PARM_DESC(tbf_rate, "Default rate limit in RPCs/s");
static int tbf_depth = 3;
module_param(tbf_depth, int, 0644);
MODULE_PARM_DESC(tbf_depth, "How many tokens that a client can save up");
static int nrs_tbf_jobid_str(const struct nrs_tbf_key *key, char *str, int len)
{
return snprintf(str, len, "%.*s", LUSTRE_JOBID_SIZE - 1, key->tk_jobid);
}
static int nrs_tbf_nid_str(const struct nrs_tbf_key *key, char *str, int len)
{
char nidstr[LNET_NIDSTR_SIZE];
libcfs_nidstr_r(&key->tk_nid, nidstr, sizeof(nidstr));
return snprintf(str, len, "%s", nidstr);
}
static int nrs_tbf_opc_str(const struct nrs_tbf_key *key, char *str, int len)
{
return snprintf(str, len, "%s", ll_opcode2str(key->tk_opcode));
}
static int nrs_tbf_id_str(u32 id, char *str, int len)
{
if (id == U32_MAX)
return snprintf(str, len, "-1");
return snprintf(str, len, "%u", id);
}
static int nrs_tbf_uid_str(const struct nrs_tbf_key *key, char *str, int len)
{
return nrs_tbf_id_str(key->tk_id.ti_uid, str, len);
}
static int nrs_tbf_gid_str(const struct nrs_tbf_key *key, char *str, int len)
{
return nrs_tbf_id_str(key->tk_id.ti_gid, str, len);
}
static int nrs_tbf_projid_str(const struct nrs_tbf_key *key, char *str, int len)
{
return nrs_tbf_id_str(key->tk_id.ti_projid, str, len);
}
static int nrs_tbf_nodemap_str(const struct nrs_tbf_key *key, char *str,
int len)
{
return nrs_tbf_id_str(key->tk_nmid, str, len);
}
static const struct nrs_tbf_type nrs_tbf_types[] = {
[NRS_TBF_FIELD_JOBID] = {
.ntt_name = NRS_TBF_TYPE_JOBID,
.ntt_flag = NRS_TBF_FLAG_JOBID,
.ntt_str = nrs_tbf_jobid_str,
},
[NRS_TBF_FIELD_NID] = {
.ntt_name = NRS_TBF_TYPE_NID,
.ntt_flag = NRS_TBF_FLAG_NID,
.ntt_str = nrs_tbf_nid_str,
},
[NRS_TBF_FIELD_OPCODE] = {
.ntt_name = NRS_TBF_TYPE_OPCODE,
.ntt_flag = NRS_TBF_FLAG_OPCODE,
.ntt_str = nrs_tbf_opc_str,
},
[NRS_TBF_FIELD_UID] = {
.ntt_name = NRS_TBF_TYPE_UID,
.ntt_flag = NRS_TBF_FLAG_UID,
.ntt_str = nrs_tbf_uid_str,
},
[NRS_TBF_FIELD_GID] = {
.ntt_name = NRS_TBF_TYPE_GID,
.ntt_flag = NRS_TBF_FLAG_GID,
.ntt_str = nrs_tbf_gid_str,
},
[NRS_TBF_FIELD_PROJID] = {
.ntt_name = NRS_TBF_TYPE_PROJID,
.ntt_flag = NRS_TBF_FLAG_PROJID,
.ntt_str = nrs_tbf_projid_str,
},
[NRS_TBF_FIELD_NODEMAP] = {
.ntt_name = NRS_TBF_TYPE_NODEMAP,
.ntt_flag = NRS_TBF_FLAG_NODEMAP,
.ntt_str = nrs_tbf_nodemap_str,
},
};
/* This converts the class key to a string using the TBF rule format.
* e.g: "jobid={kworker.0}&uid={0}&gid={0}"
*/
static int nrs_tbf_cli2str_r(struct nrs_tbf_client *cli, char *str, int len)
{
const struct nrs_tbf_type *type = nrs_tbf_types;
struct nrs_tbf_key *key = &cli->tc_key;
int outl = 0;
int i;
for (i = 0; i < ARRAY_SIZE(nrs_tbf_types); i++, type++) {
if (!(key->tk_flags & type->ntt_flag))
continue;
if (!type->ntt_str)
continue;
outl += snprintf(str + outl, len - outl,
"%s%s={", outl ? "&" : "", type->ntt_name);
if (outl >= len - 1)
break;
outl += type->ntt_str(key, str + outl, len - outl);
if (outl >= len - 2)
break;
str[outl++] = '}';
str[outl] = '\0';
}
outl = min(outl, len - 1);
str[outl] = '\0';
return outl;
}
/* racy print (used for debug logs) */
#define TBF_PRINT_KEY_BUF_SIZE 256
static DEFINE_PER_CPU(char [TBF_PRINT_KEY_BUF_SIZE], nrs_tbf_print_bufs);
static char *nrs_tbf_cli2str(struct nrs_tbf_client *cli)
{
char *buf = raw_cpu_ptr(nrs_tbf_print_bufs);
nrs_tbf_cli2str_r(cli, buf, TBF_PRINT_KEY_BUF_SIZE);
return buf;
}
__printf(3, 4) /* function attribute */
static void __nrs_tbf_cli_debug(struct nrs_tbf_client *cli,
struct libcfs_debug_msg_data *data,
const char *fmt, ...)
{
struct ptlrpc_nrs_policy *pol;
struct va_format vaf;
char *srv_name;
va_list args;
int cpt;
va_start(args, fmt);
vaf.fmt = fmt;
vaf.va = &args;
if (unlikely(!cli)) {
libcfs_debug_msg(data, "%pV: class@%p\n", &vaf, cli);
goto out;
}
pol = cli->tc_res.res_policy;
cpt = pol ? nrs_pol2cptid(pol) : 0;
srv_name = pol ? nrs_pol2svc(pol)->srv_name : "unknown";
libcfs_debug_msg(data, "%s.%d NRS: %pV: class@%p key %s rule@0x%p gen %llu rate %llu depth %llu token %llu deadline %llu time %llu state 0x%lx ref %u\n",
srv_name, cpt, &vaf, cli, nrs_tbf_cli2str(cli),
cli->tc_rule, cli->tc_rule_generation,
cli->tc_rpc_rate, cli->tc_depth, cli->tc_ntoken,
cli->tc_deadline, cli->tc_check_time,
cli->tc_state, refcount_read(&cli->tc_ref));
out:
va_end(args);
}
#define TBF_CLI_DEBUG(cli, fmt, a...) \
do { \
if (cfs_cdebug_show(D_RPCTRACE, DEBUG_SUBSYSTEM)) { \
LIBCFS_DEBUG_MSG_DATA_DECL(msgdata, D_RPCTRACE, NULL); \
__nrs_tbf_cli_debug(cli, &msgdata, fmt, ##a); \
} \
} while (0)
static int nrs_tbf_id_cli_set(struct ptlrpc_request *req, struct tbf_id *id,
enum nrs_tbf_flag need);
#define NRS_TBF_JOBID_NULL ""
static void nrs_tbf_cli_gen_key(struct ptlrpc_request *req,
struct nrs_tbf_key *key,
enum nrs_tbf_flag valid)
{
if (valid & NRS_TBF_FLAG_IDS)
nrs_tbf_id_cli_set(req, &key->tk_id, valid & NRS_TBF_FLAG_IDS);
if (valid & NRS_TBF_FLAG_NID)
key->tk_nid = req->rq_peer.nid;
if (valid & NRS_TBF_FLAG_OPCODE)
key->tk_opcode = lustre_msg_get_opc(req->rq_reqmsg);
if (valid & NRS_TBF_FLAG_JOBID) {
const char *jobid;
jobid = lustre_msg_get_jobid(req->rq_reqmsg);
if (jobid == NULL)
jobid = NRS_TBF_JOBID_NULL;
strscpy(key->tk_jobid, jobid, sizeof(key->tk_jobid));
}
if (valid & NRS_TBF_FLAG_NODEMAP) {
struct lu_nodemap *lnm;
lnm = nodemap_get_from_exp(req->rq_export);
if (IS_ERR_OR_NULL(lnm)) {
key->tk_nmid = LUSTRE_NODEMAP_MAX_ID;
} else {
key->tk_nmid = lnm->nm_id;
nodemap_putref(lnm);
}
}
key->tk_flags = valid;
}
static enum hrtimer_restart nrs_tbf_timer_cb(struct hrtimer *timer)
{
struct nrs_tbf_head *head = container_of(timer, struct nrs_tbf_head,
th_timer);
struct ptlrpc_nrs *nrs = head->th_res.res_policy->pol_nrs;
struct ptlrpc_service_part *svcpt = nrs->nrs_svcpt;
nrs->nrs_throttling = 0;
wake_up(&svcpt->scp_waitq);
return HRTIMER_NORESTART;
}
static void nrs_tbf_conds_free(struct list_head *cond_list);
#define NRS_TBF_DEFAULT_RULE "default"
/* rule's usage reference count is now dropped below one. There is no more
* outstanding usage references left. Stops the rule in case it was already
* stopping.
*/
static void nrs_tbf_rule_fini(struct kref *kref)
{
struct nrs_tbf_rule *rule = container_of(kref, struct nrs_tbf_rule,
tr_ref);
LASSERT(list_empty(&rule->tr_cli_list));
LASSERT(list_empty(&rule->tr_linkage));
if (!list_empty(&rule->tr_conds))
nrs_tbf_conds_free(&rule->tr_conds);
LASSERT(rule->tr_conds_str != NULL);
OBD_FREE_STR(rule->tr_conds_str);
OBD_FREE_PTR(rule);
}
static void
nrs_tbf_cli_rule_put(struct nrs_tbf_client *cli)
{
LASSERT(!list_empty(&cli->tc_linkage));
LASSERT(cli->tc_rule);
spin_lock(&cli->tc_rule->tr_rule_lock);
list_del_init(&cli->tc_linkage);
spin_unlock(&cli->tc_rule->tr_rule_lock);
kref_put(&cli->tc_rule->tr_ref, nrs_tbf_rule_fini);
cli->tc_rule = NULL;
}
static void
nrs_tbf_cli_reset_value(struct nrs_tbf_head *head,
struct nrs_tbf_client *cli)
{
struct nrs_tbf_rule *rule = cli->tc_rule;
cli->tc_rpc_rate = rule->tr_rpc_rate;
cli->tc_nsecs = rule->tr_nsecs_per_rpc;
cli->tc_nsecs_resid = 0;
cli->tc_depth = rule->tr_depth;
cli->tc_ntoken = rule->tr_depth;
cli->tc_check_time = ktime_to_ns(ktime_get());
cli->tc_rule_sequence = atomic_read(&head->th_rule_sequence);
cli->tc_rule_generation = rule->tr_generation;
if (test_bit(NRS_TBF_CLI_HEAP_BIT, &cli->tc_state))
binheap_relocate(head->th_binheap,
&cli->tc_node);
}
static void
nrs_tbf_cli_reset(struct nrs_tbf_head *head,
struct nrs_tbf_rule *rule,
struct nrs_tbf_client *cli)
{
spin_lock(&cli->tc_rule_lock);
if (cli->tc_rule != NULL && !list_empty(&cli->tc_linkage)) {
LASSERT(rule != cli->tc_rule);
nrs_tbf_cli_rule_put(cli);
}
LASSERT(cli->tc_rule == NULL);
LASSERT(list_empty(&cli->tc_linkage));
/* Rule's ref is added before called */
cli->tc_rule = rule;
spin_lock(&rule->tr_rule_lock);
list_add_tail(&cli->tc_linkage, &rule->tr_cli_list);
spin_unlock(&rule->tr_rule_lock);
spin_unlock(&cli->tc_rule_lock);
nrs_tbf_cli_reset_value(head, cli);
}
static int
nrs_tbf_rule_dump(struct nrs_tbf_rule *rule, struct seq_file *m)
{
seq_printf(m, "%s %s %llu, ref %d\n", rule->tr_name,
rule->tr_conds_str, rule->tr_rpc_rate,
kref_read(&rule->tr_ref) - 1);
return 0;
}
static int
nrs_tbf_rule_dump_all(struct nrs_tbf_head *head, struct seq_file *m)
{
struct nrs_tbf_rule *rule;
int rc = 0;
LASSERT(head != NULL);
spin_lock(&head->th_rule_lock);
/* List the rules from newest to oldest */
list_for_each_entry(rule, &head->th_list, tr_linkage) {
LASSERT((rule->tr_flags & NTRS_STOPPING) == 0);
rc = nrs_tbf_rule_dump(rule, m);
if (rc) {
rc = -ENOSPC;
break;
}
}
spin_unlock(&head->th_rule_lock);
return rc;
}
static struct nrs_tbf_rule *
nrs_tbf_rule_find_nolock(struct nrs_tbf_head *head,
const char *name)
{
struct nrs_tbf_rule *rule;
LASSERT(head != NULL);
list_for_each_entry(rule, &head->th_list, tr_linkage) {
LASSERT((rule->tr_flags & NTRS_STOPPING) == 0);
if (strcmp(rule->tr_name, name) == 0) {
kref_get(&rule->tr_ref);
return rule;
}
}
return NULL;
}
static struct nrs_tbf_rule *
nrs_tbf_rule_find(struct nrs_tbf_head *head,
const char *name)
{
struct nrs_tbf_rule *rule;
LASSERT(head != NULL);
spin_lock(&head->th_rule_lock);
rule = nrs_tbf_rule_find_nolock(head, name);
spin_unlock(&head->th_rule_lock);
return rule;
}
static int nrs_tbf_cond_match(struct nrs_tbf_rule *rule,
struct nrs_tbf_client *cli);
static struct nrs_tbf_rule *
nrs_tbf_rule_match(struct nrs_tbf_head *head,
struct nrs_tbf_client *cli)
{
struct nrs_tbf_rule *rule = NULL;
struct nrs_tbf_rule *tmp_rule;
spin_lock(&head->th_rule_lock);
/* Match the newest rule in the list */
list_for_each_entry(tmp_rule, &head->th_list, tr_linkage) {
LASSERT((tmp_rule->tr_flags & NTRS_STOPPING) == 0);
if (nrs_tbf_cond_match(tmp_rule, cli)) {
rule = tmp_rule;
break;
}
}
if (rule == NULL)
rule = head->th_rule;
kref_get(&rule->tr_ref);
spin_unlock(&head->th_rule_lock);
return rule;
}
static
struct nrs_tbf_client *nrs_tbf_cli_alloc(struct nrs_tbf_head *head,
struct ptlrpc_request *req,
gfp_t gfp)
{
struct ptlrpc_nrs_policy *pol = head->th_res.res_policy;
enum nrs_tbf_flag valid = head->th_type_flag;
struct nrs_tbf_client *cli;
struct nrs_tbf_rule *rule;
OBD_CPT_ALLOC_GFP(cli, nrs_pol2cptab(pol), nrs_pol2cptid(pol),
sizeof(*cli), gfp);
if (!cli)
return NULL;
nrs_tbf_cli_gen_key(req, &cli->tc_key, valid);
INIT_LIST_HEAD(&cli->tc_lru);
INIT_LIST_HEAD(&cli->tc_list);
INIT_LIST_HEAD(&cli->tc_linkage);
spin_lock_init(&cli->tc_rule_lock);
refcount_set(&cli->tc_ref, 1);
rule = nrs_tbf_rule_match(head, cli);
nrs_tbf_cli_reset(head, rule, cli);
return cli;
}
static void nrs_tbf_cli_free(struct rcu_head *head)
{
struct nrs_tbf_client *cli = container_of(head, struct nrs_tbf_client,
tc_rcu_head);
OBD_FREE_PTR(cli);
}
static void
nrs_tbf_cli_fini(struct nrs_tbf_client *cli)
{
LASSERT(list_empty(&cli->tc_list));
LASSERT(!test_bit(NRS_TBF_CLI_HEAP_BIT, &cli->tc_state));
LASSERT(!test_and_set_bit(NRS_TBF_CLI_DEL_BIT, &cli->tc_state));
TBF_CLI_DEBUG(cli, "TBF class fini");
spin_lock(&cli->tc_rule_lock);
nrs_tbf_cli_rule_put(cli);
spin_unlock(&cli->tc_rule_lock);
call_rcu(&cli->tc_rcu_head, nrs_tbf_cli_free);
}
static
struct nrs_tbf_rule *nrs_tbf_rule_alloc(struct ptlrpc_nrs_policy *policy,
struct nrs_tbf_head *head,
struct nrs_tbf_cmd *start);
static int
nrs_tbf_rule_start(struct ptlrpc_nrs_policy *policy,
struct nrs_tbf_head *head,
struct nrs_tbf_cmd *start)
{
struct nrs_tbf_rule *rule;
struct nrs_tbf_rule *tmp_rule;
struct nrs_tbf_rule *next_rule;
char *next_name = start->u.tc_start.ts_next_name;
rule = nrs_tbf_rule_find(head, start->tc_name);
if (rule) {
kref_put(&rule->tr_ref, nrs_tbf_rule_fini);
return -EEXIST;
}
rule = nrs_tbf_rule_alloc(policy, head, start);
if (IS_ERR(rule))
return PTR_ERR(rule);
/* Add as the newest rule */
spin_lock(&head->th_rule_lock);
tmp_rule = nrs_tbf_rule_find_nolock(head, start->tc_name);
if (tmp_rule) {
spin_unlock(&head->th_rule_lock);
kref_put(&tmp_rule->tr_ref, nrs_tbf_rule_fini);
kref_put(&rule->tr_ref, nrs_tbf_rule_fini);
return -EEXIST;
}
if (next_name) {
next_rule = nrs_tbf_rule_find_nolock(head, next_name);
if (!next_rule) {
spin_unlock(&head->th_rule_lock);
kref_put(&rule->tr_ref, nrs_tbf_rule_fini);
return -ENOENT;
}
list_add(&rule->tr_linkage, next_rule->tr_linkage.prev);
kref_put(&next_rule->tr_ref, nrs_tbf_rule_fini);
} else {
/* Add on the top of the rule list */
list_add(&rule->tr_linkage, &head->th_list);
}
spin_unlock(&head->th_rule_lock);
atomic_inc(&head->th_rule_sequence);
if (start->u.tc_start.ts_rule_flags & NTRS_DEFAULT) {
rule->tr_flags |= NTRS_DEFAULT;
LASSERT(head->th_rule == NULL);
head->th_rule = rule;
}
CDEBUG(D_RPCTRACE, "TBF starts rule@%p rate %llu gen %llu\n",
rule, rule->tr_rpc_rate, rule->tr_generation);
return 0;
}
/**
* nrs_tbf_rule_change_rank() - Change the rank of a rule in the rule list
* @policy: the policy instance
* @head: the TBF policy instance
* @name: the rule name to be moved
* @next_name: the rule name before which the matched rule will be moved
*
* The matched rule will be moved to the position right before another
* given rule.
*
* Return
* * %0 on success
* * %negative on error
*/
static int
nrs_tbf_rule_change_rank(struct ptlrpc_nrs_policy *policy,
struct nrs_tbf_head *head,
char *name,
char *next_name)
{
struct nrs_tbf_rule *rule = NULL;
struct nrs_tbf_rule *next_rule = NULL;
int rc = 0;
LASSERT(head != NULL);
spin_lock(&head->th_rule_lock);
rule = nrs_tbf_rule_find_nolock(head, name);
if (!rule)
GOTO(out, rc = -ENOENT);
if (strcmp(name, next_name) == 0)
GOTO(out_put, rc);
next_rule = nrs_tbf_rule_find_nolock(head, next_name);
if (!next_rule)
GOTO(out_put, rc = -ENOENT);
/* rules may be adjacent in same list, so list_move() isn't safe here */
list_move_tail(&rule->tr_linkage, &next_rule->tr_linkage);
kref_put(&next_rule->tr_ref, nrs_tbf_rule_fini);
out_put:
kref_put(&rule->tr_ref, nrs_tbf_rule_fini);
out:
spin_unlock(&head->th_rule_lock);
return rc;
}
static int
nrs_tbf_rule_change_rate(struct ptlrpc_nrs_policy *policy,
struct nrs_tbf_head *head,
char *name,
__u64 rate)
{
struct nrs_tbf_rule *rule;
assert_spin_locked(&policy->pol_nrs->nrs_lock);
rule = nrs_tbf_rule_find(head, name);
if (rule == NULL)
return -ENOENT;
rule->tr_rpc_rate = rate;
rule->tr_nsecs_per_rpc = NSEC_PER_SEC / rule->tr_rpc_rate;
rule->tr_generation++;
kref_put(&rule->tr_ref, nrs_tbf_rule_fini);
return 0;
}
static int
nrs_tbf_rule_change(struct ptlrpc_nrs_policy *policy,
struct nrs_tbf_head *head,
struct nrs_tbf_cmd *change)
{
__u64 rate = change->u.tc_change.tc_rpc_rate;
char *next_name = change->u.tc_change.tc_next_name;
int rc;
if (rate != 0) {
rc = nrs_tbf_rule_change_rate(policy, head, change->tc_name,
rate);
if (rc)
return rc;
}
if (next_name) {
rc = nrs_tbf_rule_change_rank(policy, head, change->tc_name,
next_name);
if (rc)
return rc;
}
return 0;
}
static int
nrs_tbf_rule_stop(struct ptlrpc_nrs_policy *policy,
struct nrs_tbf_head *head,
struct nrs_tbf_cmd *stop)
{
struct nrs_tbf_rule *rule;
assert_spin_locked(&policy->pol_nrs->nrs_lock);
if (strcmp(stop->tc_name, NRS_TBF_DEFAULT_RULE) == 0)
return -EPERM;
rule = nrs_tbf_rule_find(head, stop->tc_name);
if (rule == NULL)
return -ENOENT;
list_del_init(&rule->tr_linkage);
rule->tr_flags |= NTRS_STOPPING;
kref_put(&rule->tr_ref, nrs_tbf_rule_fini);
kref_put(&rule->tr_ref, nrs_tbf_rule_fini);
return 0;
}
static int
nrs_tbf_command(struct ptlrpc_nrs_policy *policy,
struct nrs_tbf_head *head,
struct nrs_tbf_cmd *cmd)
{
int rc;
assert_spin_locked(&policy->pol_nrs->nrs_lock);
switch (cmd->tc_cmd) {
case NRS_CTL_TBF_START_RULE:
if (cmd->u.tc_start.ts_valid_type != head->th_type_flag)
return -EINVAL;
spin_unlock(&policy->pol_nrs->nrs_lock);
rc = nrs_tbf_rule_start(policy, head, cmd);
spin_lock(&policy->pol_nrs->nrs_lock);
return rc;
case NRS_CTL_TBF_CHANGE_RULE:
rc = nrs_tbf_rule_change(policy, head, cmd);
return rc;
case NRS_CTL_TBF_STOP_RULE:
rc = nrs_tbf_rule_stop(policy, head, cmd);
/* Take it as a success, if not exists at all */
return rc == -ENOENT ? 0 : rc;
default:
return -EFAULT;
}
}
/**
* tbf_cli_compare() - Binary heap predicate.
* @e1: the first binheap node to compare
* @e2: the second binheap node to compare
*
* Return
* * %0 if e1 > e2
* * %1 if e1 < e2
*/
static int
tbf_cli_compare(struct binheap_node *e1, struct binheap_node *e2)
{
struct nrs_tbf_client *cli1;
struct nrs_tbf_client *cli2;
cli1 = container_of(e1, struct nrs_tbf_client, tc_node);
cli2 = container_of(e2, struct nrs_tbf_client, tc_node);
if (cli1->tc_deadline < cli2->tc_deadline)
return 1;
else if (cli1->tc_deadline > cli2->tc_deadline)
return 0;
if (cli1->tc_check_time < cli2->tc_check_time)
return 1;
else if (cli1->tc_check_time > cli2->tc_check_time)
return 0;
/* Maybe need more comparasion, e.g. request number in the rules */
return 1;
}
/*
* TBF binary heap operations
*/
static struct binheap_ops nrs_tbf_heap_ops = {
.hop_enter = NULL,
.hop_exit = NULL,
.hop_compare = tbf_cli_compare,
};
/*
* Frees jobid of @jobid_list.
*
*/
static void
nrs_tbf_jobid_list_free(struct list_head *jobid_list)
{
struct nrs_tbf_jobid *jobid, *n;
list_for_each_entry_safe(jobid, n, jobid_list, tj_linkage) {
OBD_FREE_STR(jobid->tj_id);
list_del(&jobid->tj_linkage);
OBD_FREE_PTR(jobid);
}
}
static int
nrs_tbf_jobid_list_add(char *id, struct list_head *jobid_list)
{
struct nrs_tbf_jobid *jobid;
char *ptr;
OBD_ALLOC_PTR(jobid);
if (jobid == NULL)
return -ENOMEM;
OBD_STRNDUP(jobid->tj_id, id, strlen(id));
if (jobid->tj_id == NULL) {
OBD_FREE_PTR(jobid);
return -ENOMEM;
}
ptr = strchr(id, '*');
if (ptr == NULL)
jobid->tj_match_flag = NRS_TBF_MATCH_FULL;
else
jobid->tj_match_flag = NRS_TBF_MATCH_WILDCARD;
list_add_tail(&jobid->tj_linkage, jobid_list);
return 0;
}
static bool
cfs_match_wildcard(const char *pattern, const char *content)
{
if (*pattern == '\0' && *content == '\0')
return true;
if (*pattern == '*' && *(pattern + 1) != '\0' && *content == '\0')
return false;
while (*pattern == *content) {
pattern++;
content++;
if (*pattern == '\0' && *content == '\0')
return true;
if (*pattern == '*' && *(pattern + 1) != '\0' &&
*content == '\0')
return false;
}
if (*pattern == '*')
return (cfs_match_wildcard(pattern + 1, content) ||
cfs_match_wildcard(pattern, content + 1));
return false;
}
static inline bool
nrs_tbf_jobid_match(const struct nrs_tbf_jobid *jobid, const char *id)
{
if (jobid->tj_match_flag == NRS_TBF_MATCH_FULL)
return strcmp(jobid->tj_id, id) == 0;
if (jobid->tj_match_flag == NRS_TBF_MATCH_WILDCARD)
return cfs_match_wildcard(jobid->tj_id, id);
return false;
}
static int
nrs_tbf_jobid_list_match(struct list_head *jobid_list, char *id)
{
struct nrs_tbf_jobid *jobid;
list_for_each_entry(jobid, jobid_list, tj_linkage) {
if (nrs_tbf_jobid_match(jobid, id))
return 1;
}
return 0;
}
static int
nrs_tbf_jobid_list_parse(char *orig, struct list_head *jobid_list)
{
char *str, *copy;
int rc = 0;
ENTRY;
copy = kstrdup(orig, GFP_KERNEL);
if (!copy)
RETURN(-ENOMEM);
str = copy;
INIT_LIST_HEAD(jobid_list);
while (str && rc == 0) {
char *tok = strsep(&str, " ");
if (*tok)
rc = nrs_tbf_jobid_list_add(tok, jobid_list);
}
if (list_empty(jobid_list))
rc = -EINVAL;
if (rc)
nrs_tbf_jobid_list_free(jobid_list);
kfree(copy);
RETURN(rc);
}
static inline bool nrs_tbf_flags_valid(u32 flags)
{
return (flags & NRS_TBF_FLAG_ALL) &&
(flags & ~NRS_TBF_FLAG_ALL) == 0;
}
static void nrs_tbf_cli_exit(void *vcli, void *data)
{
struct nrs_tbf_client *cli = vcli;
if (test_bit(NRS_TBF_CLI_LRU_BIT, &cli->tc_state))
list_del(&cli->tc_lru);
else
TBF_CLI_DEBUG(cli, "Busy TBF class");
nrs_tbf_cli_fini(cli);
}
static int
nrs_tbf_startup(struct ptlrpc_nrs_policy *policy, struct nrs_tbf_head *head)
{
struct rhashtable_params *params = &head->th_rhash_params;
struct nrs_tbf_cmd start;
int rc;
memset(params, 0, sizeof(*params));
params->key_len = sizeof(struct nrs_tbf_key);
params->key_offset = offsetof(struct nrs_tbf_client, tc_key);
params->head_offset = offsetof(struct nrs_tbf_client, tc_rhash);
params->min_size = 256;
params->automatic_shrinking = true;
rc = rhashtable_init(&head->th_cli_rhash, params);
if (rc)
return rc;
memset(&start, 0, sizeof(start));
start.u.tc_start.ts_conds_str = "*";
start.u.tc_start.ts_rpc_rate = tbf_rate;
start.u.tc_start.ts_rule_flags = NTRS_DEFAULT;
start.tc_name = NRS_TBF_DEFAULT_RULE;
INIT_LIST_HEAD(&start.u.tc_start.ts_conds);
rc = nrs_tbf_rule_start(policy, head, &start);
if (rc)
rhashtable_free_and_destroy(&head->th_cli_rhash,
nrs_tbf_cli_exit, NULL);
return rc;
}
/*
* ONLY opcode presented in this function will be checked in
* nrs_tbf_id_cli_set(). That means, we can add or remove an
* opcode to enable or disable requests handled in nrs_tbf
*/
static struct req_format *req_fmt(__u32 opcode)
{
switch (opcode) {
case OST_GETATTR:
return &RQF_OST_GETATTR;
case OST_SETATTR:
return &RQF_OST_SETATTR;
case OST_READ:
return &RQF_OST_BRW_READ;
case OST_WRITE:
return &RQF_OST_BRW_WRITE;
/* FIXME: OST_CREATE and OST_DESTROY comes from MDS
* in most case. Should they be removed? */
case OST_CREATE:
return &RQF_OST_CREATE;
case OST_DESTROY:
return &RQF_OST_DESTROY;
case OST_PUNCH:
return &RQF_OST_PUNCH;
case OST_SYNC:
return &RQF_OST_SYNC;
case OST_LADVISE:
return &RQF_OST_LADVISE;
case MDS_GETATTR:
return &RQF_MDS_GETATTR;
case MDS_GETATTR_NAME:
return &RQF_MDS_GETATTR_NAME;
/* close is skipped to avoid LDLM cancel slowness */
#if 0
case MDS_CLOSE:
return &RQF_MDS_CLOSE;
#endif
case MDS_REINT:
return &RQF_MDS_REINT;
case MDS_READPAGE:
return &RQF_MDS_READPAGE;
case MDS_GET_ROOT:
return &RQF_MDS_GET_ROOT;
case MDS_STATFS:
return &RQF_MDS_STATFS;
case MDS_SYNC:
return &RQF_MDS_SYNC;
case MDS_QUOTACTL:
return &RQF_MDS_QUOTACTL;
case MDS_GETXATTR:
return &RQF_MDS_GETXATTR;
case MDS_GET_INFO:
return &RQF_MDS_GET_INFO;
/* HSM op is skipped */
#if 0
case MDS_HSM_STATE_GET:
return &RQF_MDS_HSM_STATE_GET;
case MDS_HSM_STATE_SET:
return &RQF_MDS_HSM_STATE_SET;
case MDS_HSM_ACTION:
return &RQF_MDS_HSM_ACTION;
case MDS_HSM_CT_REGISTER:
return &RQF_MDS_HSM_CT_REGISTER;
case MDS_HSM_CT_UNREGISTER:
return &RQF_MDS_HSM_CT_UNREGISTER;
#endif
case MDS_SWAP_LAYOUTS:
return &RQF_MDS_SWAP_LAYOUTS;
case LDLM_ENQUEUE:
return &RQF_LDLM_ENQUEUE;
default:
return NULL;
}
}
static struct req_format *intent_req_fmt(__u32 it_opc)
{
if (it_opc & (IT_OPEN | IT_CREAT))
return &RQF_LDLM_INTENT_OPEN;
else if (it_opc & (IT_GETATTR | IT_LOOKUP))
return &RQF_LDLM_INTENT_GETATTR;
else if (it_opc & IT_GETXATTR)
return &RQF_LDLM_INTENT_GETXATTR;
else if (it_opc & (IT_GLIMPSE | IT_BRW))
return &RQF_LDLM_INTENT;
else
return NULL;
}
static int ost_tbf_id_cli_set(struct ptlrpc_request *req,
struct tbf_id *id)
{
struct ost_body *body;
body = req_capsule_client_get(&req->rq_pill, &RMF_OST_BODY);
if (body != NULL) {
id->ti_uid = body->oa.o_uid;
id->ti_gid = body->oa.o_gid;
id->ti_projid = body->oa.o_projid;
return 0;
}
return -EINVAL;
}
static void unpack_ugpid_from_mdt_body(struct ptlrpc_request *req,
struct tbf_id *id)
{
struct mdt_body *b = req_capsule_client_get(&req->rq_pill,
&RMF_MDT_BODY);
LASSERT(b != NULL);
/*
* TODO: nodemaping feature converts {ug}id from individual
* clients to the actual ones of the file system. Some work
* may be needed to fix this.
* TODO: Set ProjID in @mdt_body->mbo_projid.
*/
id->ti_uid = b->mbo_uid;
id->ti_gid = b->mbo_gid;
id->ti_projid = b->mbo_projid;
}
static void unpack_ugid_from_mdt_rec_reint(struct ptlrpc_request *req,
struct tbf_id *id)
{
struct mdt_rec_reint *rec;
rec = req_capsule_client_get(&req->rq_pill, &RMF_REC_REINT);
LASSERT(rec != NULL);
/* use the fs{ug}id as {ug}id of the process */
id->ti_uid = rec->rr_fsuid;
id->ti_gid = rec->rr_fsgid;
/* TODO: extend @mdt_rec_reint to store ProjID. */
}
static int mdt_tbf_id_cli_set(struct ptlrpc_request *req,
struct tbf_id *id)
{
u32 opc = lustre_msg_get_opc(req->rq_reqmsg);
int rc = 0;
switch (opc) {
case MDS_GETATTR:
case MDS_GETATTR_NAME:
case MDS_GET_ROOT:
case MDS_READPAGE:
case MDS_SYNC:
case MDS_GETXATTR:
case MDS_HSM_STATE_GET ... MDS_SWAP_LAYOUTS:
unpack_ugpid_from_mdt_body(req, id);
break;
case MDS_CLOSE:
case MDS_REINT:
unpack_ugid_from_mdt_rec_reint(req, id);
break;
default:
rc = -EINVAL;
break;
}
return rc;
}
static int ldlm_tbf_id_cli_set(struct ptlrpc_request *req,
struct tbf_id *id)
{
struct ldlm_intent *lit;
struct req_format *fmt;
if (req->rq_reqmsg->lm_bufcount <= DLM_INTENT_IT_OFF)
return -EINVAL;
req_capsule_extend(&req->rq_pill, &RQF_LDLM_INTENT_BASIC);
lit = req_capsule_client_get(&req->rq_pill, &RMF_LDLM_INTENT);
if (lit == NULL)
return -EINVAL;
fmt = intent_req_fmt(lit->opc);
if (fmt == NULL)
return -EINVAL;
req_capsule_extend(&req->rq_pill, fmt);
if (lit->opc & (IT_GETXATTR | IT_GETATTR | IT_LOOKUP))
unpack_ugpid_from_mdt_body(req, id);
else if (lit->opc & (IT_OPEN | IT_OPEN | IT_GLIMPSE | IT_BRW))
unpack_ugid_from_mdt_rec_reint(req, id);
else
return -EINVAL;
return 0;
}
static int nrs_tbf_id_cli_set(struct ptlrpc_request *req, struct tbf_id *id,
enum nrs_tbf_flag need)
{
struct req_format *fmt;
const struct req_format *old_fmt;
int rc = 0;
u32 opc;
id->ti_uid = (__u32) -1;
id->ti_gid = (__u32) -1;
id->ti_projid = (__u32) -1;
if (need & (NRS_TBF_FLAG_UID | NRS_TBF_FLAG_GID)) {
rc = lustre_msg_get_uid_gid(req->rq_reqmsg, &id->ti_uid,
&id->ti_gid);
if (!rc) {
if (id->ti_uid != (u32) -1)
need &= ~NRS_TBF_FLAG_UID;
if (id->ti_gid != (u32) -1)
need &= ~NRS_TBF_FLAG_GID;
}
}
if (need & NRS_TBF_FLAG_PROJID) {
rc = lustre_msg_get_projid(req->rq_reqmsg, &id->ti_projid);
if (!rc && id->ti_projid != (u32) -1)
need &= ~NRS_TBF_FLAG_PROJID;
}
/* Obtain all ID(s) in need. */
if (need == 0)
return rc;
/* The client req doesn't have uid/gid/projid pack in ptlrpc_body
* --> fallback to the old method.
*/
opc = lustre_msg_get_opc(req->rq_reqmsg);
fmt = req_fmt(opc);
if (fmt == NULL)
return -EINVAL;
req_capsule_init(&req->rq_pill, req, RCL_SERVER);
old_fmt = req->rq_pill.rc_fmt;
if (old_fmt == NULL)
req_capsule_set(&req->rq_pill, fmt);
if (opc < OST_LAST_OPC)
rc = ost_tbf_id_cli_set(req, id);
else if (opc >= MDS_FIRST_OPC && opc < MDS_LAST_OPC)
rc = mdt_tbf_id_cli_set(req, id);
else if (opc == LDLM_ENQUEUE)
rc = ldlm_tbf_id_cli_set(req, id);
else
rc = -EINVAL;
/* restore it to the original state */
if (req->rq_pill.rc_fmt != old_fmt)
req->rq_pill.rc_fmt = old_fmt;
return rc;
}
/* The following nrs_tbf_lru_* function manipulate class object with a 0
* refcount. These need to be accessed with a rcu lock to prevent deletions.
* A TBF class with a 0 refcount means the following things:
* - class is in LRU list (NRS_TBF_CLI_LRU_BIT is set)
* - class is being added in LRU (refs 1->0 then NRS_TBF_CLI_LRU_BIT 0->1)
* - class is being deleted (NRS_TBF_CLI_LRU_BIT 1->0)
* - class is being removed from LRU (NRS_TBF_CLI_LRU_BIT 1->0 then refs 0->1)
*/
static void nrs_tbf_lru_add(struct nrs_tbf_head *head,
struct nrs_tbf_client *cli)
{
LASSERTF(!test_bit(NRS_TBF_CLI_LRU_BIT, &cli->tc_state),
"class@%p already in LRU", cli);
LASSERTF(!refcount_read(&cli->tc_ref), "class@%p ref != 0: %d",
cli, refcount_read(&cli->tc_ref));
spin_lock(&head->th_lru_lock);
list_add_tail_rcu(&cli->tc_lru, &head->th_lru_list);
atomic_inc(&head->th_lru_cnt);
spin_unlock(&head->th_lru_lock);
set_bit(NRS_TBF_CLI_LRU_BIT, &cli->tc_state);
TBF_CLI_DEBUG(cli, "TBF LRU add");
}
static
struct nrs_tbf_client *nrs_tbf_lru_tryhit(struct nrs_tbf_head *head,
struct nrs_tbf_client *cli)
{
if (!test_and_clear_bit(NRS_TBF_CLI_LRU_BIT, &cli->tc_state))
return NULL;
LASSERTF(!refcount_read(&cli->tc_ref), "class@%p ref != 0: %d",
cli, refcount_read(&cli->tc_ref));
spin_lock(&head->th_lru_lock);
atomic_dec(&head->th_lru_cnt);
list_del_rcu(&cli->tc_lru);
spin_unlock(&head->th_lru_lock);
refcount_set(&cli->tc_ref, 1);
TBF_CLI_DEBUG(cli, "TBF LRU hit");
return cli;
}
static int nrs_tbf_lru_shrink(struct nrs_tbf_head *head)
{
struct nrs_tbf_client *cli;
int high = tbf_jobid_cache_size;
int low = 3 * high / 4;
int freed = 0;
if (atomic_read(&head->th_lru_cnt) <= high)
return 0;
/* already running ? */
if (test_and_set_bit(NRS_TBF_SHRINKING_BIT, &head->th_state))
return 0;
rcu_read_lock();
list_for_each_entry_rcu(cli, &head->th_lru_list, tc_lru) {
if (atomic_read(&head->th_lru_cnt) <= low)
break;
/* race with nrs_tbf_lru_tryhit()/nrs_tbf_lru_add ? */
if (!test_and_clear_bit(NRS_TBF_CLI_LRU_BIT, &cli->tc_state))
continue;
LASSERTF(!refcount_read(&cli->tc_ref), "class@%p ref != 0: %d",
cli, refcount_read(&cli->tc_ref));
rhashtable_remove_fast(&head->th_cli_rhash, &cli->tc_rhash,
head->th_rhash_params);
spin_lock(&head->th_lru_lock);
atomic_dec(&head->th_lru_cnt);
list_del_rcu(&cli->tc_lru);
spin_unlock(&head->th_lru_lock);
nrs_tbf_cli_fini(cli);
freed++;
}
rcu_read_unlock();
clear_bit(NRS_TBF_SHRINKING_BIT, &head->th_state);
CDEBUG(D_RPCTRACE,
"%d objects freed from LRU (cur: %u, high: %d, low: %d)\n",
freed, atomic_read(&head->th_lru_cnt), high, low);
return freed;
}
static struct nrs_tbf_client *
nrs_tbf_cli_find(struct nrs_tbf_head *head, struct ptlrpc_request *req)
{
struct nrs_tbf_client *cli;
struct nrs_tbf_key key;
memset(&key, 0, sizeof(key));
nrs_tbf_cli_gen_key(req, &key, head->th_type_flag);
rcu_read_lock();
cli = rhashtable_lookup(&head->th_cli_rhash, &key,
head->th_rhash_params);
if (cli && !refcount_inc_not_zero(&cli->tc_ref))
cli = nrs_tbf_lru_tryhit(head, cli);
rcu_read_unlock();
return cli;
}
static struct nrs_tbf_client *
nrs_tbf_cli_findadd(struct nrs_tbf_head *head,
struct nrs_tbf_client *cli)
{
struct nrs_tbf_client *cli2 = NULL;
try_again:
rcu_read_lock();
cli2 = rhashtable_lookup_get_insert_fast(&head->th_cli_rhash,
&cli->tc_rhash,
head->th_rhash_params);
if (IS_ERR(cli2))
goto out;
if (cli2 && !refcount_inc_not_zero(&cli2->tc_ref)) {
cli2 = nrs_tbf_lru_tryhit(head, cli2);
if (!cli2) {
/* lost race -> retry */
rcu_read_unlock();
schedule();
goto try_again;
}
}
out:
rcu_read_unlock();
if (!cli2)
TBF_CLI_DEBUG(cli, "TBF class insert");
return cli2 ?: cli;
}
static void
nrs_tbf_cli_put(struct nrs_tbf_head *head, struct nrs_tbf_client *cli)
{
if (!refcount_dec_and_test(&cli->tc_ref))
return;
nrs_tbf_lru_add(head, cli);
nrs_tbf_lru_shrink(head);
}
static void
nrs_tbf_id_list_free(struct list_head *id_list)
{
struct nrs_tbf_id *nti_id, *n;
list_for_each_entry_safe(nti_id, n, id_list, nti_linkage) {
list_del_init(&nti_id->nti_linkage);
OBD_FREE_PTR(nti_id);
}
}
static void
nrs_tbf_nodemap_list_free(struct list_head *nm_list)
{
struct nrs_tbf_nodemap *nodemap, *n;
list_for_each_entry_safe(nodemap, n, nm_list, ntn_linkage) {
list_del_init(&nodemap->ntn_linkage);
OBD_FREE_PTR(nodemap);
}
}
static void
nrs_tbf_expression_free(struct nrs_tbf_expression *expr)
{
LASSERT(expr->te_field >= 0 && expr->te_field < NRS_TBF_FIELD_MAX);
switch (expr->te_field) {
case NRS_TBF_FIELD_NID:
cfs_free_nidlist(&expr->te_cond);
break;
case NRS_TBF_FIELD_JOBID:
nrs_tbf_jobid_list_free(&expr->te_cond);
break;
case NRS_TBF_FIELD_OPCODE:
bitmap_free(expr->te_opcodes);
break;
case NRS_TBF_FIELD_UID:
case NRS_TBF_FIELD_GID:
case NRS_TBF_FIELD_PROJID:
nrs_tbf_id_list_free(&expr->te_cond);
break;
case NRS_TBF_FIELD_NODEMAP:
nrs_tbf_nodemap_list_free(&expr->te_cond);
break;
default:
LBUG();
}
OBD_FREE_PTR(expr);
}
static void
nrs_tbf_conjunction_free(struct nrs_tbf_conjunction *conjunction)
{
struct nrs_tbf_expression *expression;
struct nrs_tbf_expression *n;
LASSERT(list_empty(&conjunction->tc_linkage));
list_for_each_entry_safe(expression, n,
&conjunction->tc_expressions,
te_linkage) {
list_del_init(&expression->te_linkage);
nrs_tbf_expression_free(expression);
}
OBD_FREE_PTR(conjunction);
}
static void
nrs_tbf_conds_free(struct list_head *cond_list)
{
struct nrs_tbf_conjunction *conjunction;
struct nrs_tbf_conjunction *n;
list_for_each_entry_safe(conjunction, n, cond_list, tc_linkage) {
list_del_init(&conjunction->tc_linkage);
nrs_tbf_conjunction_free(conjunction);
}
}
#define NRS_TBF_DISJUNCTION_DELIM (",")
#define NRS_TBF_CONJUNCTION_DELIM ("&")
#define NRS_TBF_EXPRESSION_DELIM ("=")
static int
nrs_tbf_opcode_list_parse(char *str, unsigned long **bitmaptr);
static int
nrs_tbf_id_list_parse(char *str, struct list_head *id_list,
enum nrs_tbf_flag tif);
static int
nrs_tbf_nodemap_list_add(char *name, struct list_head *nm_list)
{
struct nrs_tbf_nodemap *nodemap;
struct lu_nodemap *lnm;
ENTRY;
OBD_ALLOC_PTR(nodemap);
if (nodemap == NULL)
RETURN(-ENOMEM);
/*
* If the setting for nodemap and TBF are using Lustre config log,
* it must ensure that nodemap is configured before the TBF.
* Otherwise, the nodemap lookup will return an error and adding
* TBF rule for the nodemap will fail.
*/
lnm = nodemap_lookup_unlocked(name);
if (IS_ERR(lnm)) {
OBD_FREE_PTR(nodemap);
RETURN(PTR_ERR(lnm));
}
nodemap->ntn_nmid = lnm->nm_id;
nodemap_putref(lnm);
list_add_tail(&nodemap->ntn_linkage, nm_list);
RETURN(0);
}
static int
nrs_tbf_nodemap_list_parse(char *orig, struct list_head *nm_list)
{
char *str, *copy;
int rc = 0;
ENTRY;
copy = kstrdup(orig, GFP_KERNEL);
if (!copy)
RETURN(-ENOMEM);
str = copy;
INIT_LIST_HEAD(nm_list);
while (str && rc == 0) {
char *tok = strsep(&str, " ");
if (*tok)
rc = nrs_tbf_nodemap_list_add(tok, nm_list);
}
if (list_empty(nm_list))
rc = -EINVAL;
if (rc)
nrs_tbf_nodemap_list_free(nm_list);
kfree(copy);
RETURN(rc);
}
static int
nrs_tbf_expression_parse(enum nrs_tbf_flag ntf, char *str,
struct list_head *cond_list)
{
struct nrs_tbf_expression *expr;
char *field;
int rc = 0;
int len;
OBD_ALLOC_PTR(expr);
if (expr == NULL)
return -ENOMEM;
field = strim(strsep(&str, NRS_TBF_EXPRESSION_DELIM));
if (!*field || !str)
/* No LHS or no '=' sign */
GOTO(out, rc = -EINVAL);
str = strim(str);
len = strlen(str);
if (len < 2 || str[0] != '{' || str[len-1] != '}')
/* No {} around RHS */
GOTO(out, rc = -EINVAL);
/* Skip '{' and '}' */
str[len-1] = '\0';
str += 1;
len -= 2;
if (strcmp(field, NRS_TBF_TYPE_NID) == 0) {
if (!(ntf & NRS_TBF_FLAG_NID))
GOTO(out, rc = -EINVAL);
if (cfs_parse_nidlist(str, len, &expr->te_cond) < 0)
GOTO(out, rc = -EINVAL);
expr->te_field = NRS_TBF_FIELD_NID;
} else if (strcmp(field, NRS_TBF_TYPE_JOBID) == 0) {
if (!(ntf & NRS_TBF_FLAG_JOBID))
GOTO(out, rc = -EINVAL);
if (nrs_tbf_jobid_list_parse(str, &expr->te_cond) < 0)
GOTO(out, rc = -EINVAL);
expr->te_field = NRS_TBF_FIELD_JOBID;
} else if (strcmp(field, NRS_TBF_TYPE_OPCODE) == 0) {
if (!(ntf & NRS_TBF_FLAG_OPCODE))
GOTO(out, rc = -EINVAL);
if (nrs_tbf_opcode_list_parse(str, &expr->te_opcodes) < 0)
GOTO(out, rc = -EINVAL);
expr->te_field = NRS_TBF_FIELD_OPCODE;
} else if (strcmp(field, NRS_TBF_TYPE_UID) == 0) {
if (!(ntf & NRS_TBF_FLAG_UID))
GOTO(out, rc = -EINVAL);
if (nrs_tbf_id_list_parse(str, &expr->te_cond,
NRS_TBF_FLAG_UID) < 0)
GOTO(out, rc = -EINVAL);
expr->te_field = NRS_TBF_FIELD_UID;
} else if (strcmp(field, NRS_TBF_TYPE_GID) == 0) {
if (!(ntf & NRS_TBF_FLAG_GID))
GOTO(out, rc = -EINVAL);
if (nrs_tbf_id_list_parse(str, &expr->te_cond,
NRS_TBF_FLAG_GID) < 0)
GOTO(out, rc = -EINVAL);
expr->te_field = NRS_TBF_FIELD_GID;
} else if (strcmp(field, NRS_TBF_TYPE_PROJID) == 0) {
if (!(ntf & NRS_TBF_FLAG_PROJID))
GOTO(out, rc = -EINVAL);
if (nrs_tbf_id_list_parse(str, &expr->te_cond,
NRS_TBF_FLAG_PROJID) < 0)
GOTO(out, rc = -EINVAL);
expr->te_field = NRS_TBF_FIELD_PROJID;
} else if (strcmp(field, NRS_TBF_TYPE_NODEMAP) == 0) {
if (!(ntf & NRS_TBF_FLAG_NODEMAP))
GOTO(out, rc = -EINVAL);
if (nrs_tbf_nodemap_list_parse(str, &expr->te_cond) < 0)
GOTO(out, rc = -EINVAL);
expr->te_field = NRS_TBF_FIELD_NODEMAP;
} else {
GOTO(out, rc = -EINVAL);
}
list_add_tail(&expr->te_linkage, cond_list);
return 0;
out:
OBD_FREE_PTR(expr);
return rc;
}
static int
nrs_tbf_conjunction_parse(enum nrs_tbf_flag ntf, char *str,
struct list_head *cond_list)
{
struct nrs_tbf_conjunction *conjunction;
int rc = 0;
OBD_ALLOC_PTR(conjunction);
if (conjunction == NULL)
return -ENOMEM;
INIT_LIST_HEAD(&conjunction->tc_expressions);
list_add_tail(&conjunction->tc_linkage, cond_list);
while (str && !rc) {
char *expr = strsep(&str, NRS_TBF_CONJUNCTION_DELIM);
rc = nrs_tbf_expression_parse(ntf, expr,
&conjunction->tc_expressions);
}
return rc;
}
static int
nrs_tbf_conds_parse(enum nrs_tbf_flag ntf, char *orig,
struct list_head *cond_list)
{
char *str;
int rc = 0;
orig = kstrdup(orig, GFP_KERNEL);
if (!orig)
return -ENOMEM;
str = orig;
INIT_LIST_HEAD(cond_list);
while (str && !rc) {
char *term = strsep(&str, NRS_TBF_DISJUNCTION_DELIM);
rc = nrs_tbf_conjunction_parse(ntf, term, cond_list);
}
kfree(orig);
if (rc)
nrs_tbf_conds_free(cond_list);
return rc;
}
static int
nrs_tbf_id_list_match(struct list_head *id_list, u32 id,
enum nrs_tbf_flag flag);
static int
nrs_tbf_nodemap_list_match(struct list_head *nm_list, unsigned int nmid)
{
struct nrs_tbf_nodemap *nodemap;
list_for_each_entry(nodemap, nm_list, ntn_linkage) {
if (nodemap->ntn_nmid == nmid)
return 1;
}
return 0;
}
static int
nrs_tbf_expression_match(struct nrs_tbf_expression *expr,
struct nrs_tbf_rule *rule,
struct nrs_tbf_client *cli)
{
switch (expr->te_field) {
case NRS_TBF_FIELD_NID:
return cfs_match_nid(&cli->tc_nid, &expr->te_cond);
case NRS_TBF_FIELD_JOBID:
return nrs_tbf_jobid_list_match(&expr->te_cond, cli->tc_jobid);
case NRS_TBF_FIELD_OPCODE:
return test_bit(cli->tc_opcode, expr->te_opcodes);
case NRS_TBF_FIELD_UID:
return nrs_tbf_id_list_match(&expr->te_cond, cli->tc_id.ti_uid,
NRS_TBF_FLAG_UID);
case NRS_TBF_FIELD_GID:
return nrs_tbf_id_list_match(&expr->te_cond, cli->tc_id.ti_gid,
NRS_TBF_FLAG_GID);
case NRS_TBF_FIELD_PROJID:
return nrs_tbf_id_list_match(&expr->te_cond,
cli->tc_id.ti_projid,
NRS_TBF_FLAG_PROJID);
case NRS_TBF_FIELD_NODEMAP:
return nrs_tbf_nodemap_list_match(&expr->te_cond,
cli->tc_nmid);
default:
return 0;
}
}
static int
nrs_tbf_conjunction_match(struct nrs_tbf_conjunction *conjunction,
struct nrs_tbf_rule *rule,
struct nrs_tbf_client *cli)
{
struct nrs_tbf_expression *expr;
int matched;
list_for_each_entry(expr, &conjunction->tc_expressions, te_linkage) {
matched = nrs_tbf_expression_match(expr, rule, cli);
if (!matched)
return 0;
}
return 1;
}
static int
nrs_tbf_cond_match(struct nrs_tbf_rule *rule, struct nrs_tbf_client *cli)
{
struct nrs_tbf_conjunction *conjunction;
int matched;
list_for_each_entry(conjunction, &rule->tr_conds, tc_linkage) {
matched = nrs_tbf_conjunction_match(conjunction, rule, cli);
if (matched)
return 1;
}
return 0;
}
static
struct nrs_tbf_rule *nrs_tbf_rule_alloc(struct ptlrpc_nrs_policy *policy,
struct nrs_tbf_head *head,
struct nrs_tbf_cmd *start)
{
struct nrs_tbf_rule *rule;
int rc = 0;
OBD_CPT_ALLOC_PTR(rule, nrs_pol2cptab(policy), nrs_pol2cptid(policy));
if (!rule)
return ERR_PTR(-ENOMEM);
strscpy(rule->tr_name, start->tc_name, sizeof(rule->tr_name));
rule->tr_rpc_rate = start->u.tc_start.ts_rpc_rate;
rule->tr_flags = start->u.tc_start.ts_rule_flags;
rule->tr_nsecs_per_rpc = NSEC_PER_SEC / rule->tr_rpc_rate;
rule->tr_depth = tbf_depth;
kref_init(&rule->tr_ref);
INIT_LIST_HEAD(&rule->tr_cli_list);
INIT_LIST_HEAD(&rule->tr_nids);
INIT_LIST_HEAD(&rule->tr_linkage);
spin_lock_init(&rule->tr_rule_lock);
rule->tr_head = head;
LASSERT(start->u.tc_start.ts_conds_str);
OBD_STRNDUP(rule->tr_conds_str,
start->u.tc_start.ts_conds_str,
strlen(start->u.tc_start.ts_conds_str));
if (!rule->tr_conds_str)
GOTO(err, rc = -ENOMEM);
INIT_LIST_HEAD(&rule->tr_conds);
if (!list_empty(&start->u.tc_start.ts_conds)) {
rc = nrs_tbf_conds_parse(start->u.tc_start.ts_valid_type,
rule->tr_conds_str,
&rule->tr_conds);
}
if (rc)
GOTO(err, rc);
return rule;
err:
kref_put(&rule->tr_ref, nrs_tbf_rule_fini);
return ERR_PTR(rc);
}
#define MAX_OPCODE_LEN 32
static int
nrs_tbf_opcode_set_bit(char *id, unsigned long *opcodes)
{
int op;
op = ll_str2opcode(id);
if (op < 0)
return -EINVAL;
set_bit(op, opcodes);
return 0;
}
static int
nrs_tbf_opcode_list_parse(char *orig, unsigned long **bitmaptr)
{
unsigned long *opcodes;
char *str;
int cnt = 0;
int rc = 0;
ENTRY;
orig = kstrdup(orig, GFP_KERNEL);
if (!orig)
return -ENOMEM;
opcodes = bitmap_zalloc(LUSTRE_MAX_OPCODES, GFP_KERNEL);
if (!opcodes) {
kfree(orig);
return -ENOMEM;
}
str = orig;
while (str && rc == 0) {
char *tok = strsep(&str, " ");
if (*tok) {
rc = nrs_tbf_opcode_set_bit(tok, opcodes);
cnt += 1;
}
}
if (cnt == 0)
rc = -EINVAL;
kfree(orig);
if (rc == 0 && bitmaptr)
*bitmaptr = opcodes;
else
bitmap_free(opcodes);
RETURN(rc);
}
static int
nrs_tbf_id_list_match(struct list_head *id_list, u32 id,
enum nrs_tbf_flag idtype)
{
struct nrs_tbf_id *nti_id;
list_for_each_entry(nti_id, id_list, nti_linkage) {
u32 curid;
switch (idtype) {
case NRS_TBF_FLAG_UID:
curid = nti_id->nti_id.ti_uid;
break;
case NRS_TBF_FLAG_GID:
curid = nti_id->nti_id.ti_gid;
break;
case NRS_TBF_FLAG_PROJID:
curid = nti_id->nti_id.ti_projid;
break;
default:
return 0;
}
if (id != curid)
continue;
return 1;
}
return 0;
}
static int
nrs_tbf_id_list_parse(char *orig, struct list_head *id_list,
enum nrs_tbf_flag ntf)
{
int rc = 0;
unsigned long val;
char *str;
struct tbf_id id = { 0 };
ENTRY;
if (ntf != NRS_TBF_FLAG_UID && ntf != NRS_TBF_FLAG_GID &&
ntf != NRS_TBF_FLAG_PROJID)
RETURN(-EINVAL);
orig = kstrdup(orig, GFP_KERNEL);
if (!orig)
return -ENOMEM;
INIT_LIST_HEAD(id_list);
for (str = orig; str ; ) {
struct nrs_tbf_id *nti_id;
char *tok;
tok = strsep(&str, " ");
if (!*tok)
/* Empty token - leading, trailing, or
* multiple spaces in list
*/
continue;
rc = kstrtoul(tok, 0, &val);
if (rc < 0)
GOTO(out, rc = -EINVAL);
if (ntf == NRS_TBF_FLAG_UID)
id.ti_uid = val;
else if (ntf == NRS_TBF_FLAG_GID)
id.ti_gid = val;
else
id.ti_projid = val;
OBD_ALLOC_PTR(nti_id);
if (nti_id == NULL)
GOTO(out, rc = -ENOMEM);
nti_id->nti_id = id;
list_add_tail(&nti_id->nti_linkage, id_list);
}
if (list_empty(id_list))
/* Only white space in the list */
GOTO(out, rc = -EINVAL);
out:
kfree(orig);
if (rc)
nrs_tbf_id_list_free(id_list);
RETURN(rc);
}
#define NRS_TBF_TYPE_COMBINATOR "+"
static int nrs_tbf_type_parse(const char *orig, enum nrs_tbf_flag *fl)
{
enum nrs_tbf_flag flags = NRS_TBF_FLAG_INVALID;
char *copy, *str;
int rc = 0;
ENTRY;
copy = kstrdup(orig, GFP_KERNEL);
if (!copy)
RETURN(-ENOMEM);
str = copy;
while (str && rc == 0) {
char *name;
bool found = false;
int i;
name = strsep(&str, NRS_TBF_TYPE_COMBINATOR);
if (!*name)
GOTO(out, rc);
for (i = 0; i < ARRAY_SIZE(nrs_tbf_types); i++) {
if (strcmp(name, nrs_tbf_types[i].ntt_name) == 0) {
flags |= nrs_tbf_types[i].ntt_flag;
found = true;
break;
}
}
if (found == false)
GOTO(out, rc = -EINVAL);
}
out:
kfree(copy);
if (rc == 0) {
if (flags == NRS_TBF_FLAG_INVALID)
RETURN(-EINVAL);
*fl = flags;
}
RETURN(rc);
}
/**
* nrs_tbf_start() - policy to start
* @policy: The policy to start
* @arg: tbf flags
*
* Is called before the policy transitions into
* ptlrpc_nrs_pol_state::NRS_POL_STATE_STARTED; allocates and initializes a
* policy-specific private data structure.
*
* see nrs_policy_register()
* see nrs_policy_ctl()
*
* Return
* * %0 on success
* * %errno on error (-ENOMEM (OOM) error)
*
*/
static int nrs_tbf_start(struct ptlrpc_nrs_policy *policy, char *arg)
{
struct nrs_tbf_head *head;
enum nrs_tbf_flag flags;
char *name;
int rc = 0;
if (arg == NULL) {
flags = NRS_TBF_FLAG_ALL;
name = NRS_TBF_TYPE_GENERIC;
} else if (strlen(arg) < NRS_TBF_TYPE_MAX_LEN) {
name = arg;
rc = nrs_tbf_type_parse(arg, &flags);
if (rc)
GOTO(out, rc);
} else {
GOTO(out, rc = -EINVAL);
}
LASSERT((flags & NRS_TBF_FLAG_ALL) != 0);
LASSERT((flags & ~NRS_TBF_FLAG_ALL) == 0);
if (!nrs_tbf_flags_valid(flags))
GOTO(out, rc = -EINVAL);
OBD_CPT_ALLOC_PTR(head, nrs_pol2cptab(policy), nrs_pol2cptid(policy));
if (head == NULL)
GOTO(out, rc = -ENOMEM);
memcpy(head->th_type, name, strlen(name));
head->th_type[strlen(name)] = '\0';
head->th_type_flag = flags;
head->th_binheap = binheap_create(&nrs_tbf_heap_ops,
CBH_FLAG_ATOMIC_GROW, 4096, NULL,
nrs_pol2cptab(policy),
nrs_pol2cptid(policy));
if (head->th_binheap == NULL)
GOTO(out_free_head, rc = -ENOMEM);
atomic_set(&head->th_rule_sequence, 0);
atomic_set(&head->th_lru_cnt, 0);
spin_lock_init(&head->th_rule_lock);
spin_lock_init(&head->th_lru_lock);
INIT_LIST_HEAD(&head->th_list);
INIT_LIST_HEAD(&head->th_lru_list);
hrtimer_setup(&head->th_timer, nrs_tbf_timer_cb, CLOCK_MONOTONIC,
HRTIMER_MODE_ABS);
rc = nrs_tbf_startup(policy, head);
if (rc)
GOTO(out_free_heap, rc);
policy->pol_private = head;
return 0;
out_free_heap:
binheap_destroy(head->th_binheap);
out_free_head:
OBD_FREE_PTR(head);
out:
return rc;
}
/**
* nrs_tbf_stop() - Stop policy transction
* @policy: The policy to stop
*
* Is called before the policy transitions into
* ptlrpc_nrs_pol_state::NRS_POL_STATE_STOPPED; deallocates the policy-specific
* private data structure.
*
* see __nrs_policy_stop()
*/
static void nrs_tbf_stop(struct ptlrpc_nrs_policy *policy)
{
struct nrs_tbf_head *head = policy->pol_private;
struct ptlrpc_nrs *nrs = policy->pol_nrs;
struct nrs_tbf_rule *rule, *n;
LASSERT(head != NULL);
hrtimer_cancel(&head->th_timer);
/* Should cleanup hash first before free rules */
atomic_set(&head->th_lru_cnt, 0);
rhashtable_free_and_destroy(&head->th_cli_rhash,
nrs_tbf_cli_exit, NULL);
list_for_each_entry_safe(rule, n, &head->th_list, tr_linkage) {
list_del_init(&rule->tr_linkage);
kref_put(&rule->tr_ref, nrs_tbf_rule_fini);
}
/* wait for all the class object to be freed */
rcu_barrier();
LASSERT(list_empty(&head->th_lru_list));
LASSERT(list_empty(&head->th_list));
LASSERT(head->th_binheap != NULL);
LASSERT(binheap_is_empty(head->th_binheap));
binheap_destroy(head->th_binheap);
OBD_FREE_PTR(head);
nrs->nrs_throttling = 0;
wake_up(&policy->pol_nrs->nrs_svcpt->scp_waitq);
}
/**
* nrs_tbf_ctl() - Performs a policy-specific ctl function on TBF policy
* instances; similar to ioctl.
* @policy: the policy instance
* @opc: the opcode
* @arg: used for passing parameters and information[in,out]
*
* pre: assert_spin_locked(&policy->pol_nrs->->nrs_lock)
* post: assert_spin_locked(&policy->pol_nrs->->nrs_lock)
*
* Return
* * %0 on success (operation carried out successfully)
* * %negative on error
*/
static int nrs_tbf_ctl(struct ptlrpc_nrs_policy *policy,
enum ptlrpc_nrs_ctl opc,
void *arg)
{
int rc = 0;
ENTRY;
assert_spin_locked(&policy->pol_nrs->nrs_lock);
switch (opc) {
default:
RETURN(-EINVAL);
/**
* Read RPC rate size of a policy instance.
*/
case NRS_CTL_TBF_RD_RULE: {
struct nrs_tbf_head *head = policy->pol_private;
struct seq_file *m = arg;
struct ptlrpc_service_part *svcpt;
svcpt = policy->pol_nrs->nrs_svcpt;
seq_printf(m, "CPT %d:\n", svcpt->scp_cpt);
rc = nrs_tbf_rule_dump_all(head, m);
}
break;
/**
* Write RPC rate of a policy instance.
*/
case NRS_CTL_TBF_WR_RULE: {
struct nrs_tbf_head *head = policy->pol_private;
struct nrs_tbf_cmd *cmd;
cmd = (struct nrs_tbf_cmd *)arg;
rc = nrs_tbf_command(policy,
head,
cmd);
}
break;
/**
* Read the TBF policy type of a policy instance.
*/
case NRS_CTL_TBF_RD_TYPE_FLAG: {
struct nrs_tbf_head *head = policy->pol_private;
*(__u32 *)arg = head->th_type_flag;
}
break;
}
RETURN(rc);
}
/**
* nrs_tbf_res_get() - Is called for obtaining a TBF policy resource.
* @policy: The policy on which the request is being asked for
* @nrq: The request for which resources are being taken
* @parent: Parent resource, unused in this policy
* @resp: Resources references are placed in this array[out]
* @moving_req: Signifies limited caller context; unused in this policy
*
* see nrs_resource_get_safe()
*
* Return
* * %0 on success
* * %negative on error
*/
static int nrs_tbf_res_get(struct ptlrpc_nrs_policy *policy,
struct ptlrpc_nrs_request *nrq,
const struct ptlrpc_nrs_resource *parent,
struct ptlrpc_nrs_resource **resp,
bool moving_req)
{
struct nrs_tbf_head *head;
struct nrs_tbf_client *cli;
struct nrs_tbf_client *tmp;
struct ptlrpc_request *req;
if (parent == NULL) {
*resp = &((struct nrs_tbf_head *)policy->pol_private)->th_res;
return 0;
}
head = container_of(parent, struct nrs_tbf_head, th_res);
req = container_of(nrq, struct ptlrpc_request, rq_nrq);
cli = nrs_tbf_cli_find(head, req);
if (cli != NULL) {
spin_lock(&policy->pol_nrs->nrs_svcpt->scp_req_lock);
LASSERT(cli->tc_rule);
if (cli->tc_rule_sequence !=
atomic_read(&head->th_rule_sequence) ||
cli->tc_rule->tr_flags & NTRS_STOPPING) {
struct nrs_tbf_rule *rule;
TBF_CLI_DEBUG(cli, "TBF rule update, rule flags 0x%x, head sequence %d",
cli->tc_rule->tr_flags,
atomic_read(&head->th_rule_sequence));
rule = nrs_tbf_rule_match(head, cli);
if (rule != cli->tc_rule) {
nrs_tbf_cli_reset(head, rule, cli);
} else {
if (cli->tc_rule_generation != rule->tr_generation)
nrs_tbf_cli_reset_value(head, cli);
kref_put(&rule->tr_ref, nrs_tbf_rule_fini);
}
} else if (cli->tc_rule_generation !=
cli->tc_rule->tr_generation) {
nrs_tbf_cli_reset_value(head, cli);
}
spin_unlock(&policy->pol_nrs->nrs_svcpt->scp_req_lock);
goto out;
}
cli = nrs_tbf_cli_alloc(head, req, moving_req ? GFP_ATOMIC : __GFP_IO);
if (cli == NULL)
return -ENOMEM;
tmp = nrs_tbf_cli_findadd(head, cli);
if (tmp != cli) {
WARN_ON(!refcount_dec_if_one(&cli->tc_ref));
nrs_tbf_cli_fini(cli);
cli = tmp;
if (IS_ERR(cli))
return PTR_ERR(cli);
}
out:
*resp = &cli->tc_res;
return 1;
}
/**
* nrs_tbf_res_put() - Called when releasing references to the resource hierachy
* obtained for a
* request for scheduling using the TBF policy.
* @policy: the policy the resource belongs to
* @res: the resource to be released
*/
static void nrs_tbf_res_put(struct ptlrpc_nrs_policy *policy,
const struct ptlrpc_nrs_resource *res)
{
struct nrs_tbf_head *head;
struct nrs_tbf_client *cli;
/**
* Do nothing for freeing parent, nrs_tbf_net resources
*/
if (res->res_parent == NULL)
return;
cli = container_of(res, struct nrs_tbf_client, tc_res);
head = container_of(res->res_parent, struct nrs_tbf_head, th_res);
nrs_tbf_cli_put(head, cli);
}
/**
* nrs_tbf_req_get() - Called when getting a request from the TBF policy for
* handling, or just peeking; removes the request from the policy when it is
* to be handled.
* @policy: The policy
* @peek: When set, signifies that we just want to examine the request,
* and not handle it, so the request is not removed from the policy.
* @force: Force the policy to return a request
*
* See: ptlrpc_nrs_req_get_nolock()
* See: nrs_request_get()
*
* Return The request to be handled; this is the next request in the TBF
* rule
*
*/
static
struct ptlrpc_nrs_request *nrs_tbf_req_get(struct ptlrpc_nrs_policy *policy,
bool peek, bool force)
{
struct nrs_tbf_head *head = policy->pol_private;
struct ptlrpc_nrs_request *nrq = NULL;
struct nrs_tbf_client *cli;
struct binheap_node *node;
assert_spin_locked(&policy->pol_nrs->nrs_svcpt->scp_req_lock);
if (likely(!peek && !force) && policy->pol_nrs->nrs_throttling)
return NULL;
node = binheap_root(head->th_binheap);
if (unlikely(node == NULL))
return NULL;
cli = container_of(node, struct nrs_tbf_client, tc_node);
LASSERT(test_bit(NRS_TBF_CLI_HEAP_BIT, &cli->tc_state));
if (unlikely(peek)) {
nrq = list_first_entry(&cli->tc_list,
struct ptlrpc_nrs_request,
nr_u.tbf.tr_list);
} else {
struct nrs_tbf_rule *rule = cli->tc_rule;
__u64 now = ktime_to_ns(ktime_get());
__u64 passed;
__u64 ntoken;
__u64 deadline;
__u64 old_resid = 0;
deadline = cli->tc_check_time +
cli->tc_nsecs;
LASSERT(now >= cli->tc_check_time);
passed = now - cli->tc_check_time;
ntoken = passed * cli->tc_rpc_rate;
do_div(ntoken, NSEC_PER_SEC);
ntoken += cli->tc_ntoken;
if (rule->tr_flags & NTRS_REALTIME) {
LASSERT(cli->tc_nsecs_resid < cli->tc_nsecs);
old_resid = cli->tc_nsecs_resid;
cli->tc_nsecs_resid += passed % cli->tc_nsecs;
if (cli->tc_nsecs_resid > cli->tc_nsecs) {
ntoken++;
cli->tc_nsecs_resid -= cli->tc_nsecs;
}
} else if (ntoken > cli->tc_depth)
ntoken = cli->tc_depth;
/* give an extra token with force mode */
if (unlikely(force) && ntoken == 0)
ntoken = 1;
if (ntoken > 0) {
nrq = list_first_entry(&cli->tc_list,
struct ptlrpc_nrs_request,
nr_u.tbf.tr_list);
ntoken--;
cli->tc_ntoken = ntoken;
cli->tc_check_time = now;
list_del_init(&nrq->nr_u.tbf.tr_list);
if (list_empty(&cli->tc_list)) {
binheap_remove(head->th_binheap,
&cli->tc_node);
clear_bit(NRS_TBF_CLI_HEAP_BIT, &cli->tc_state);
} else {
if (!(rule->tr_flags & NTRS_REALTIME))
cli->tc_deadline = now + cli->tc_nsecs;
binheap_relocate(head->th_binheap,
&cli->tc_node);
}
TBF_CLI_DEBUG(cli, "TBF dequeues");
} else {
ktime_t time;
if (rule->tr_flags & NTRS_REALTIME) {
cli->tc_deadline = deadline;
cli->tc_nsecs_resid = old_resid;
binheap_relocate(head->th_binheap,
&cli->tc_node);
if (node != binheap_root(head->th_binheap))
return nrs_tbf_req_get(policy,
peek, force);
}
policy->pol_nrs->nrs_throttling = 1;
head->th_deadline = deadline;
time = ktime_set(0, 0);
time = ktime_add_ns(time, deadline);
hrtimer_start(&head->th_timer, time, HRTIMER_MODE_ABS);
}
}
return nrq;
}
/**
* nrs_tbf_req_add() - Adds request @nrq to @policy's list of queued requests
* @policy: The policy
* @nrq: The request to add
*
* Return 0 on success; nrs_request_enqueue() assumes this function will always
* succeed
*/
static int nrs_tbf_req_add(struct ptlrpc_nrs_policy *policy,
struct ptlrpc_nrs_request *nrq)
{
struct nrs_tbf_head *head;
struct nrs_tbf_client *cli;
int rc = 0;
assert_spin_locked(&policy->pol_nrs->nrs_svcpt->scp_req_lock);
cli = container_of(nrs_request_resource(nrq),
struct nrs_tbf_client, tc_res);
head = container_of(nrs_request_resource(nrq)->res_parent,
struct nrs_tbf_head, th_res);
if (list_empty(&cli->tc_list)) {
LASSERT(!test_bit(NRS_TBF_CLI_HEAP_BIT, &cli->tc_state));
cli->tc_deadline = cli->tc_check_time + cli->tc_nsecs;
rc = binheap_insert(head->th_binheap, &cli->tc_node);
if (rc == 0) {
set_bit(NRS_TBF_CLI_HEAP_BIT, &cli->tc_state);
nrq->nr_u.tbf.tr_sequence = head->th_sequence++;
list_add_tail(&nrq->nr_u.tbf.tr_list,
&cli->tc_list);
if (policy->pol_nrs->nrs_throttling) {
__u64 deadline = cli->tc_deadline;
if ((head->th_deadline > deadline) &&
(hrtimer_try_to_cancel(&head->th_timer)
>= 0)) {
ktime_t time;
head->th_deadline = deadline;
time = ktime_set(0, 0);
time = ktime_add_ns(time, deadline);
hrtimer_start(&head->th_timer, time,
HRTIMER_MODE_ABS);
}
}
}
} else {
LASSERT(test_bit(NRS_TBF_CLI_HEAP_BIT, &cli->tc_state));
nrq->nr_u.tbf.tr_sequence = head->th_sequence++;
list_add_tail(&nrq->nr_u.tbf.tr_list,
&cli->tc_list);
}
if (rc == 0)
TBF_CLI_DEBUG(cli, "TBF enqueues");
return rc;
}
/**
* nrs_tbf_req_del() - Removes request @nrq from @policy's list of queued
* requests.
* @policy: The policy
* @nrq: The request to remove
*/
static void nrs_tbf_req_del(struct ptlrpc_nrs_policy *policy,
struct ptlrpc_nrs_request *nrq)
{
struct nrs_tbf_head *head;
struct nrs_tbf_client *cli;
assert_spin_locked(&policy->pol_nrs->nrs_svcpt->scp_req_lock);
cli = container_of(nrs_request_resource(nrq),
struct nrs_tbf_client, tc_res);
head = container_of(nrs_request_resource(nrq)->res_parent,
struct nrs_tbf_head, th_res);
LASSERT(!list_empty(&nrq->nr_u.tbf.tr_list));
list_del_init(&nrq->nr_u.tbf.tr_list);
if (list_empty(&cli->tc_list)) {
binheap_remove(head->th_binheap,
&cli->tc_node);
clear_bit(NRS_TBF_CLI_HEAP_BIT, &cli->tc_state);
} else {
binheap_relocate(head->th_binheap,
&cli->tc_node);
}
}
/**
* nrs_tbf_req_stop() - Prints a debug statement right before the request
* @nrq stops being handled.
* @policy: The policy handling the request
* @nrq: The request being handled
*
* ptlrpc_server_finish_request()
* ptlrpc_nrs_req_stop_nolock()
*/
static void nrs_tbf_req_stop(struct ptlrpc_nrs_policy *policy,
struct ptlrpc_nrs_request *nrq)
{
struct ptlrpc_request *req = container_of(nrq, struct ptlrpc_request,
rq_nrq);
assert_spin_locked(&policy->pol_nrs->nrs_svcpt->scp_req_lock);
CDEBUG(D_RPCTRACE, "NRS stop %s request from %s, seq: %llu\n",
policy->pol_desc->pd_name, libcfs_idstr(&req->rq_peer),
nrq->nr_u.tbf.tr_sequence);
}
/*
* debugfs interface
*/
/*
* The maximum RPC rate.
*/
#define LPROCFS_NRS_RATE_MAX 1000000ULL /* 1rpc/us */
static int
ptlrpc_lprocfs_nrs_tbf_rule_seq_show(struct seq_file *m, void *data)
{
struct ptlrpc_service *svc = m->private;
int rc;
seq_printf(m, "regular_requests:\n");
/**
* Perform two separate calls to this as only one of the NRS heads'
* policies may be in the ptlrpc_nrs_pol_state::NRS_POL_STATE_STARTED or
* ptlrpc_nrs_pol_state::NRS_POL_STATE_STOPPING state.
*/
rc = ptlrpc_nrs_policy_control(svc, PTLRPC_NRS_QUEUE_REG,
NRS_POL_NAME_TBF,
NRS_CTL_TBF_RD_RULE,
false, m);
if (rc == 0) {
/**
* -ENOSPC means buf in the parameter m is overflow, return 0
* here to let upper layer function seq_read alloc a larger
* memory area and do this process again.
*/
} else if (rc == -ENOSPC) {
return 0;
/**
* Ignore -ENODEV as the regular NRS head's policy may be in the
* ptlrpc_nrs_pol_state::NRS_POL_STATE_STOPPED state.
*/
} else if (rc != -ENODEV) {
return rc;
}
if (!nrs_svc_has_hp(svc))
goto no_hp;
seq_printf(m, "high_priority_requests:\n");
rc = ptlrpc_nrs_policy_control(svc, PTLRPC_NRS_QUEUE_HP,
NRS_POL_NAME_TBF,
NRS_CTL_TBF_RD_RULE,
false, m);
if (rc == 0) {
/**
* -ENOSPC means buf in the parameter m is overflow, return 0
* here to let upper layer function seq_read alloc a larger
* memory area and do this process again.
*/
} else if (rc == -ENOSPC) {
return 0;
}
no_hp:
return rc;
}
static int nrs_tbf_id_parse(struct nrs_tbf_cmd *cmd, char *token)
{
int rc;
ENTRY;
if (!nrs_tbf_flags_valid(cmd->u.tc_start.ts_valid_type))
RETURN(-EINVAL);
OBD_STRNDUP(cmd->u.tc_start.ts_conds_str, token, strlen(token));
if (cmd->u.tc_start.ts_conds_str == NULL)
return -ENOMEM;
/* Parse hybird NID and JOBID conditions */
rc = nrs_tbf_conds_parse(cmd->u.tc_start.ts_valid_type,
cmd->u.tc_start.ts_conds_str,
&cmd->u.tc_start.ts_conds);
if (rc)
OBD_FREE_STR(cmd->u.tc_start.ts_conds_str);
RETURN(rc);
}
static void nrs_tbf_cmd_fini(struct nrs_tbf_cmd *cmd)
{
if (cmd->tc_cmd != NRS_CTL_TBF_START_RULE)
return;
if (!nrs_tbf_flags_valid(cmd->u.tc_start.ts_valid_type)) {
CWARN("unknown NRS_TBF_FLAGS:0x%x\n",
cmd->u.tc_start.ts_valid_type);
return;
}
if (!list_empty(&cmd->u.tc_start.ts_conds))
nrs_tbf_conds_free(&cmd->u.tc_start.ts_conds);
OBD_FREE_STR(cmd->u.tc_start.ts_conds_str);
}
static int check_rule_name(const char *name)
{
int i;
if (name[0] == '\0')
return -EINVAL;
for (i = 0; name[i] != '\0' && i < MAX_TBF_NAME; i++) {
if (!isalnum(name[i]) && name[i] != '_')
return -EINVAL;
}
if (i == MAX_TBF_NAME)
return -ENAMETOOLONG;
return 0;
}
static int
nrs_tbf_parse_value_pair(struct nrs_tbf_cmd *cmd, char *buffer)
{
char *key;
char *val;
int rc;
__u64 rate;
val = buffer;
key = strsep(&val, "=");
if (val == NULL || strlen(val) == 0)
return -EINVAL;
/* Key of the value pair */
if (strcmp(key, "rate") == 0) {
rc = kstrtoull(val, 10, &rate);
if (rc)
return rc;
if (rate <= 0 || rate >= LPROCFS_NRS_RATE_MAX)
return -EINVAL;
if (cmd->tc_cmd == NRS_CTL_TBF_START_RULE)
cmd->u.tc_start.ts_rpc_rate = rate;
else if (cmd->tc_cmd == NRS_CTL_TBF_CHANGE_RULE)
cmd->u.tc_change.tc_rpc_rate = rate;
else
return -EINVAL;
} else if (strcmp(key, "rank") == 0) {
rc = check_rule_name(val);
if (rc)
return rc;
if (cmd->tc_cmd == NRS_CTL_TBF_START_RULE)
cmd->u.tc_start.ts_next_name = val;
else if (cmd->tc_cmd == NRS_CTL_TBF_CHANGE_RULE)
cmd->u.tc_change.tc_next_name = val;
else
return -EINVAL;
} else if (strcmp(key, "realtime") == 0) {
unsigned long realtime;
rc = kstrtoul(val, 10, &realtime);
if (rc)
return rc;
if (realtime > 0)
cmd->u.tc_start.ts_rule_flags |= NTRS_REALTIME;
} else {
return -EINVAL;
}
return 0;
}
static int
nrs_tbf_parse_value_pairs(struct nrs_tbf_cmd *cmd, char *buffer)
{
char *val;
char *token;
int rc;
val = buffer;
while (val != NULL && strlen(val) != 0) {
token = strsep(&val, " ");
rc = nrs_tbf_parse_value_pair(cmd, token);
if (rc)
return rc;
}
switch (cmd->tc_cmd) {
case NRS_CTL_TBF_START_RULE:
if (cmd->u.tc_start.ts_rpc_rate == 0)
cmd->u.tc_start.ts_rpc_rate = tbf_rate;
break;
case NRS_CTL_TBF_CHANGE_RULE:
if (cmd->u.tc_change.tc_rpc_rate == 0 &&
cmd->u.tc_change.tc_next_name == NULL)
return -EINVAL;
break;
case NRS_CTL_TBF_STOP_RULE:
break;
default:
return -EINVAL;
}
return 0;
}
static struct nrs_tbf_cmd *
nrs_tbf_parse_cmd(char *buffer, unsigned long count, __u32 type_flag)
{
struct nrs_tbf_cmd *cmd;
char *token;
char *val;
int rc = 0;
OBD_ALLOC_PTR(cmd);
if (cmd == NULL)
GOTO(out, rc = -ENOMEM);
memset(cmd, 0, sizeof(*cmd));
val = buffer;
token = strsep(&val, " ");
if (val == NULL || strlen(val) == 0)
GOTO(out_free_cmd, rc = -EINVAL);
/* Type of the command */
if (strcmp(token, "start") == 0) {
cmd->tc_cmd = NRS_CTL_TBF_START_RULE;
cmd->u.tc_start.ts_valid_type = type_flag;
} else if (strcmp(token, "stop") == 0)
cmd->tc_cmd = NRS_CTL_TBF_STOP_RULE;
else if (strcmp(token, "change") == 0)
cmd->tc_cmd = NRS_CTL_TBF_CHANGE_RULE;
else
GOTO(out_free_cmd, rc = -EINVAL);
/* Name of the rule */
token = strsep(&val, " ");
if ((val == NULL && cmd->tc_cmd != NRS_CTL_TBF_STOP_RULE))
GOTO(out_free_cmd, rc = -EINVAL);
rc = check_rule_name(token);
if (rc)
GOTO(out_free_cmd, rc);
cmd->tc_name = token;
if (cmd->tc_cmd == NRS_CTL_TBF_START_RULE) {
/* List of ID */
LASSERT(val);
token = val;
val = strrchr(token, '}');
if (!val)
GOTO(out_free_cmd, rc = -EINVAL);
/* Skip '}' */
val++;
if (*val == '\0') {
val = NULL;
} else if (*val == ' ') {
*val = '\0';
val++;
} else
GOTO(out_free_cmd, rc = -EINVAL);
rc = nrs_tbf_id_parse(cmd, token);
if (rc)
GOTO(out_free_cmd, rc);
}
rc = nrs_tbf_parse_value_pairs(cmd, val);
if (rc)
GOTO(out_cmd_fini, rc = -EINVAL);
goto out;
out_cmd_fini:
nrs_tbf_cmd_fini(cmd);
out_free_cmd:
OBD_FREE_PTR(cmd);
out:
if (rc)
cmd = ERR_PTR(rc);
return cmd;
}
/**
* nrs_tbf_type_flag() - Get the TBF policy type (nid, jobid, etc) preset by
* proc entry 'nrs_policies' for command buffer parsing.
* @svc: the PTLRPC service
* @queue: the NRS queue type
*
* Return the preset TBF policy type flag
*/
static __u32
nrs_tbf_type_flag(struct ptlrpc_service *svc, enum ptlrpc_nrs_queue_type queue)
{
__u32 type;
int rc;
rc = ptlrpc_nrs_policy_control(svc, queue,
NRS_POL_NAME_TBF,
NRS_CTL_TBF_RD_TYPE_FLAG,
true, &type);
if (rc != 0)
type = NRS_TBF_FLAG_INVALID;
return type;
}
#define LPROCFS_WR_NRS_TBF_MAX_CMD (4096)
static ssize_t
ptlrpc_lprocfs_nrs_tbf_rule_seq_write(struct file *file,
const char __user *buffer,
size_t count, loff_t *off)
{
struct seq_file *m = file->private_data;
struct ptlrpc_service *svc = m->private;
char *kernbuf;
char *val;
int rc;
struct nrs_tbf_cmd *cmd;
enum ptlrpc_nrs_queue_type queue = PTLRPC_NRS_QUEUE_BOTH;
unsigned long length;
char *token;
OBD_ALLOC(kernbuf, LPROCFS_WR_NRS_TBF_MAX_CMD);
if (kernbuf == NULL)
GOTO(out, rc = -ENOMEM);
if (count > LPROCFS_WR_NRS_TBF_MAX_CMD - 1)
GOTO(out_free_kernbuff, rc = -EINVAL);
if (copy_from_user(kernbuf, buffer, count))
GOTO(out_free_kernbuff, rc = -EFAULT);
val = kernbuf;
token = strsep(&val, " ");
if (val == NULL)
GOTO(out_free_kernbuff, rc = -EINVAL);
if (strcmp(token, "reg") == 0) {
queue = PTLRPC_NRS_QUEUE_REG;
} else if (strcmp(token, "hp") == 0) {
queue = PTLRPC_NRS_QUEUE_HP;
} else {
kernbuf[strlen(token)] = ' ';
val = kernbuf;
}
length = strlen(val);
if (length == 0)
GOTO(out_free_kernbuff, rc = -EINVAL);
if (queue == PTLRPC_NRS_QUEUE_HP && !nrs_svc_has_hp(svc))
GOTO(out_free_kernbuff, rc = -ENODEV);
else if (queue == PTLRPC_NRS_QUEUE_BOTH && !nrs_svc_has_hp(svc))
queue = PTLRPC_NRS_QUEUE_REG;
cmd = nrs_tbf_parse_cmd(val, length, nrs_tbf_type_flag(svc, queue));
if (IS_ERR(cmd))
GOTO(out_free_kernbuff, rc = PTR_ERR(cmd));
/**
* Serialize NRS core lprocfs operations with policy registration/
* unregistration.
*/
mutex_lock(&nrs_core.nrs_mutex);
rc = ptlrpc_nrs_policy_control(svc, queue,
NRS_POL_NAME_TBF,
NRS_CTL_TBF_WR_RULE,
false, cmd);
mutex_unlock(&nrs_core.nrs_mutex);
nrs_tbf_cmd_fini(cmd);
OBD_FREE_PTR(cmd);
out_free_kernbuff:
OBD_FREE(kernbuf, LPROCFS_WR_NRS_TBF_MAX_CMD);
out:
return rc ? rc : count;
}
LDEBUGFS_SEQ_FOPS(ptlrpc_lprocfs_nrs_tbf_rule);
/**
* nrs_tbf_lprocfs_init() - Initializes a TBF policy's lprocfs interface for
* service @svc
* @svc: the service (Portal RPC)
*
* Return
* * %0 on success
* * %!=0 on error
*/
static int nrs_tbf_lprocfs_init(struct ptlrpc_service *svc)
{
struct ldebugfs_vars nrs_tbf_lprocfs_vars[] = {
{ .name = "nrs_tbf_rule",
.fops = &ptlrpc_lprocfs_nrs_tbf_rule_fops,
.data = svc },
{ NULL }
};
if (!svc->srv_debugfs_entry)
return 0;
ldebugfs_add_vars(svc->srv_debugfs_entry, nrs_tbf_lprocfs_vars, NULL);
return 0;
}
/*
* TBF policy operations
*/
static const struct ptlrpc_nrs_pol_ops nrs_tbf_ops = {
.op_policy_start = nrs_tbf_start,
.op_policy_stop = nrs_tbf_stop,
.op_policy_ctl = nrs_tbf_ctl,
.op_res_get = nrs_tbf_res_get,
.op_res_put = nrs_tbf_res_put,
.op_req_get = nrs_tbf_req_get,
.op_req_enqueue = nrs_tbf_req_add,
.op_req_dequeue = nrs_tbf_req_del,
.op_req_stop = nrs_tbf_req_stop,
.op_lprocfs_init = nrs_tbf_lprocfs_init,
};
/*
* TBF policy configuration
*/
struct ptlrpc_nrs_pol_conf nrs_conf_tbf = {
.nc_name = NRS_POL_NAME_TBF,
.nc_ops = &nrs_tbf_ops,
.nc_compat = nrs_policy_compat_all,
};
