MEDIUM: listener/config: make the "thread" parser rely on thread_sets
Instead of reading and storing a single group and a single mask for a
"thread" directive on a bind line, we now store the complete range in
a thread set that's stored in the bind_conf. The bind_parse_thread()
function now just calls parse_thread_set() to complete the current set,
which starts empty, and thread_resolve_group_mask() was updated to
support retrieving thread group numbers or absolute thread numbers
directly from the pre-filled thread_set, and continue to feed bind_tgroup
and bind_thread. The CLI parsers which were pre-initialized to set the
bind_tgroup to 1 cannot do it anymore as it would prevent one from
restricting the thread set. Instead check_config_validity() now detects
the CLI frontend and passes the info down to thread_resolve_group_mask()
that will automatically use only the group 1's threads for these
listeners. The same is done for the peers listeners for now.
At this step it's already possible to start with all previous valid
configs as well as extended ones supporting comma-delimited thread
sets. In addition the parser already accepts large ranges spanning
multiple groups, but since the underlying listeners infrastructure
is not read, for now we're maintaining a specific check against this
at the higher level of the config validity check.
The patch is a bit large because thread resolution is performed in
multiple steps, so we need to adjust all of them at once to preserve
functional and technical consistency.
diff --git a/src/cfgparse.c b/src/cfgparse.c
index 82e6c08..759a469 100644
--- a/src/cfgparse.c
+++ b/src/cfgparse.c
@@ -2954,13 +2954,18 @@
/* detect and address thread affinity inconsistencies */
err = NULL;
- if (thread_resolve_group_mask(bind_conf->bind_tgroup, bind_conf->bind_thread,
+ if (thread_resolve_group_mask(&bind_conf->thread_set, 1,
&bind_conf->bind_tgroup, &bind_conf->bind_thread, &err) < 0) {
ha_alert("Proxy '%s': %s in 'bind %s' at [%s:%d].\n",
curproxy->id, err, bind_conf->arg, bind_conf->file, bind_conf->line);
free(err);
cfgerr++;
}
+ else if (bind_conf->thread_set.nbgrp > 1) {
+ ha_alert("Proxy '%s': 'thread' spans more than one group in 'bind %s' at [%s:%d].\n",
+ curproxy->id, bind_conf->arg, bind_conf->file, bind_conf->line);
+ cfgerr++;
+ }
/* apply thread masks and groups to all receivers */
list_for_each_entry(li, &bind_conf->listeners, by_bind) {
@@ -4431,13 +4436,18 @@
}
err = NULL;
- if (thread_resolve_group_mask(bind_conf->bind_tgroup, bind_conf->bind_thread,
+ if (thread_resolve_group_mask(&bind_conf->thread_set, (curproxy == global.cli_fe) ? 1 : 0,
&bind_conf->bind_tgroup, &bind_conf->bind_thread, &err) < 0) {
ha_alert("Peers section '%s': %s in 'bind %s' at [%s:%d].\n",
curpeers->peers_fe->id, err, bind_conf->arg, bind_conf->file, bind_conf->line);
free(err);
cfgerr++;
}
+ else if (bind_conf->thread_set.nbgrp > 1) {
+ ha_alert("Peers section '%s': 'thread' spans more than one group in 'bind %s' at [%s:%d].\n",
+ curpeers->peers_fe->id, bind_conf->arg, bind_conf->file, bind_conf->line);
+ cfgerr++;
+ }
/* apply thread masks and groups to all receivers */
list_for_each_entry(li, &bind_conf->listeners, by_bind) {
diff --git a/src/cli.c b/src/cli.c
index 679eb5a..a541464 100644
--- a/src/cli.c
+++ b/src/cli.c
@@ -494,7 +494,6 @@
}
bind_conf->level &= ~ACCESS_LVL_MASK;
bind_conf->level |= ACCESS_LVL_OPER; /* default access level */
- bind_conf->bind_tgroup = 1; // bind to a single group in any case
if (!str2listener(args[2], global.cli_fe, bind_conf, file, line, err)) {
memprintf(err, "parsing [%s:%d] : '%s %s' : %s\n",
@@ -2999,7 +2998,6 @@
bind_conf->level &= ~ACCESS_LVL_MASK;
bind_conf->level |= ACCESS_LVL_ADMIN;
bind_conf->level |= ACCESS_MASTER | ACCESS_MASTER_ONLY;
- bind_conf->bind_tgroup = 1; // bind to a single group in any case
if (!str2listener(args[0], mworker_proxy, bind_conf, "master-socket", 0, &err)) {
ha_alert("Cannot create the listener of the master CLI\n");
@@ -3096,7 +3094,6 @@
bind_conf->level &= ~ACCESS_LVL_MASK;
bind_conf->level |= ACCESS_LVL_ADMIN; /* TODO: need to lower the rights with a CLI keyword*/
bind_conf->level |= ACCESS_FD_LISTENERS;
- bind_conf->bind_tgroup = 1; // bind to a single group in any case
if (!memprintf(&path, "sockpair@%d", mworker_proc->ipc_fd[1])) {
ha_alert("Cannot allocate listener.\n");
diff --git a/src/listener.c b/src/listener.c
index b7b3b22..81fdc2f 100644
--- a/src/listener.c
+++ b/src/listener.c
@@ -1801,41 +1801,15 @@
return 0;
}
-/* parse the "thread" bind keyword */
+/* parse the "thread" bind keyword. This will replace any preset thread_set */
static int bind_parse_thread(char **args, int cur_arg, struct proxy *px, struct bind_conf *conf, char **err)
{
- char *sep = NULL;
- ulong thread = 0;
- long tgroup = 0;
-
- tgroup = strtol(args[cur_arg + 1], &sep, 10);
- if (*sep == '/') {
- /* a thread group was present */
- if (tgroup < 1 || tgroup > MAX_TGROUPS) {
- memprintf(err, "'%s' thread-group number must be between 1 and %d (was %ld)", args[cur_arg + 1], MAX_TGROUPS, tgroup);
- return ERR_ALERT | ERR_FATAL;
- }
- sep++;
- }
- else {
- /* no thread group */
- tgroup = 0;
- sep = args[cur_arg + 1];
- }
-
- if ((conf->bind_tgroup || conf->bind_thread) &&
- conf->bind_tgroup != tgroup) {
- memprintf(err, "'%s' multiple thread-groups are not supported", args[cur_arg + 1]);
- return ERR_ALERT | ERR_FATAL;
- }
-
- if (parse_process_number(sep, &thread, LONGBITS, NULL, err)) {
- memprintf(err, "'%s' : %s", sep, *err);
+ /* note that the thread set is zeroed before first call, and we don't
+ * want to reset it so that it remains possible to chain multiple
+ * "thread" directives.
+ */
+ if (parse_thread_set(args[cur_arg+1], &conf->thread_set, err) < 0)
return ERR_ALERT | ERR_FATAL;
- }
-
- conf->bind_thread |= thread;
- conf->bind_tgroup = tgroup;
return 0;
}
diff --git a/src/thread.c b/src/thread.c
index 59e247e..75ce24a 100644
--- a/src/thread.c
+++ b/src/thread.c
@@ -1221,79 +1221,73 @@
return 0;
}
-/* converts a configuration thread num or group+mask to a global group+mask
- * depending on the configured thread group id. This is essentially for use
- * with the "thread" directive on "bind" lines, where "thread 4-6" might be
- * turned to "2/1-3". It cannot be used before the thread mapping above was
- * completed and the thread group number configured. Possible options:
- * - igid == 0: imask represents global IDs. We have to check that all
- * configured threads in the mask belong to the same group. If imask is zero
- * it means everything, so for now we only support this with a single group.
- * - igid > 0, imask = 0: convert global values to local values for this thread
- * - igid > 0, imask > 0: convert global values to local values
- * Note that the output mask is always local to the group.
+/* Converts a configuration thread set based on either absolute or relative
+ * thread numbers into a global group+mask. This is essentially for use with
+ * the "thread" directive on "bind" lines, where "thread 4-6,10-12" might be
+ * turned to "2/1-3,4/1-3". It cannot be used before the thread mapping above
+ * was completed and the thread group numbers configured. The thread_set is
+ * replaced by the resolved group-based one. It is possible to force a single
+ * default group for unspecified sets instead of enabling all groups by passing
+ * this group's non-zero value to defgrp. The output ogid and omask are set,
+ * respectively, to the first non-empty group and its mask. They're used only
+ * for the transition to the new model.
*
* Returns <0 on failure, >=0 on success.
*/
-int thread_resolve_group_mask(uint igid, ulong imask, uint *ogid, ulong *omask, char **err)
+int thread_resolve_group_mask(struct thread_set *ts, int defgrp, uint *ogid, ulong *omask, char **err)
{
- ulong mask;
- uint t;
+ struct thread_set new_ts = { 0 };
+ ulong mask, imask;
+ uint g;
- if (igid == 0) {
+ if (!ts->nbgrp) {
/* unspecified group, IDs are global */
- if (!imask) {
- /* all threads of all groups */
- if (global.nbtgroups > 1) {
- memprintf(err, "'thread' directive spans multiple groups");
- return -1;
+ if (thread_set_is_empty(ts)) {
+ /* all threads of all groups, unless defgrp is set and
+ * we then set it as the only group.
+ */
+ for (g = defgrp ? defgrp-1 : 0; g < (defgrp ? defgrp : global.nbtgroups); g++) {
+ new_ts.rel[g] = ha_tgroup_info[g].threads_enabled;
+ new_ts.nbgrp++;
}
- *ogid = 1; // first and only group
- *omask = ha_tgroup_info[0].threads_enabled;
- return 0;
} else {
- /* some global threads */
- for (t = 0; t < global.nbthread; t++) {
- if (imask & (1UL << t)) {
- if (ha_thread_info[t].tgid != igid) {
- if (!igid)
- igid = ha_thread_info[t].tgid;
- else {
- memprintf(err, "'thread' directive spans multiple groups (at least %u and %u)", igid, ha_thread_info[t].tgid);
- return -1;
- }
- }
- }
- }
+ /* some absolute threads are set, we must remap them to
+ * relative ones. Each group cannot have more than
+ * LONGBITS threads, thus it spans at most two absolute
+ * blocks.
+ */
+ for (g = 0; g < global.nbtgroups; g++) {
+ uint block = ha_tgroup_info[g].base / LONGBITS;
+ uint base = ha_tgroup_info[g].base % LONGBITS;
- if (!igid) {
- memprintf(err, "'thread' directive contains threads that belong to no group");
- return -1;
- }
+ mask = ts->abs[block] >> base;
+ if (base && ha_tgroup_info[g].count > (LONGBITS - base))
+ mask |= ts->abs[block + 1] << (LONGBITS - base);
+ mask &= nbits(ha_tgroup_info[g].count);
+ mask &= ha_tgroup_info[g].threads_enabled;
- /* we have a valid group, convert this to global thread IDs */
- *ogid = igid;
- imask = imask >> ha_tgroup_info[igid - 1].base;
- imask &= ha_tgroup_info[igid - 1].threads_enabled;
- *omask = imask;
- return 0;
+ /* now the mask exactly matches the threads to be enabled
+ * in this group.
+ */
+ if (!new_ts.rel[g] && mask)
+ new_ts.nbgrp++;
+ new_ts.rel[g] |= mask;
+ }
}
} else {
- /* group was specified */
- if (igid > global.nbtgroups) {
- memprintf(err, "'thread' directive references non-existing thread group %u", igid);
- return -1;
- }
+ /* groups were specified */
+ for (g = 0; g < MAX_TGROUPS; g++) {
+ imask = ts->rel[g];
+ if (!imask)
+ continue;
- if (!imask) {
- /* all threads of this groups. Let's make a mask from their count and base. */
- *ogid = igid;
- *omask = nbits(ha_tgroup_info[igid - 1].count);
- return 0;
- } else {
- /* some local threads. Keep only existing ones for this group */
+ if (g >= global.nbtgroups) {
+ memprintf(err, "'thread' directive references non-existing thread group %u", g+1);
+ return -1;
+ }
- mask = nbits(ha_tgroup_info[igid - 1].count);
+ /* some relative threads are set. Keep only existing ones for this group */
+ mask = nbits(ha_tgroup_info[g].count);
if (!(mask & imask)) {
/* no intersection between the thread group's