MEDIUM: listeners: split the thread mask between receiver and bind_conf

With groups at some point we'll have to have distinct masks/groups in the
receiver and the bind_conf, because a single bind_conf might require to
instantiate multiple receivers (one per group).

Let's split the thread mask and group to have one for the bind_conf and
another one for the receiver while it remains easy to do. This will later
allow to use different storage for the bind_conf if needed (e.g. support
multiple groups).

include/haproxy/listener-t.h

@@ -185,6 +185,8 @@ struct bind_conf {
 	char *file;                /* file where the section appears */
 	int line;                  /* line where the section appears */
 	__decl_thread(HA_RWLOCK_T sni_lock); /* lock the SNI trees during add/del operations */
+	unsigned long bind_thread; /* bitmask of threads allowed on this bind_conf */
+	uint bind_tgroup;          /* thread group ID: 0=global IDs, non-zero=local IDs */
 	struct rx_settings settings; /* all the settings needed for the listening socket */
 };
 

include/haproxy/receiver-t.h

@@ -41,8 +41,6 @@
 
 /* All the settings that are used to configure a receiver */
 struct rx_settings {
-	unsigned long bind_thread;        /* bitmask of threads allowed to use these listeners */
-	uint bind_tgroup;                 /* thread group ID: 0=global IDs, non-zero=local IDs */
 	struct {                          /* UNIX socket permissions */
 		uid_t uid;                /* -1 to leave unchanged */
 		gid_t gid;                /* -1 to leave unchanged */
@@ -60,6 +58,8 @@ struct receiver {
 	struct protocol *proto;          /* protocol this receiver belongs to */
 	void *owner;                     /* receiver's owner (usually a listener) */
 	void (*iocb)(int fd);            /* generic I/O handler (typically accept callback) */
+	unsigned long bind_thread;       /* bitmask of threads allowed on this receiver */
+	uint bind_tgroup;                /* thread group ID: 0=global IDs, non-zero=local IDs */
 	struct rx_settings *settings;    /* points to the settings used by this receiver */
 	struct list proto_list;          /* list in the protocol header */
 #ifdef USE_QUIC

src/cfgparse.c

@@ -2517,6 +2517,7 @@ int check_config_validity()
 		/* check and reduce the bind-proc of each listener */
 		list_for_each_entry(bind_conf, &curproxy->conf.bind, by_fe) {
 			unsigned long mask;
+			struct listener *li;
 
 			/* HTTP frontends with "h2" as ALPN/NPN will work in
 			 * HTTP/2 and absolutely require buffers 16kB or larger.
@@ -2544,13 +2545,13 @@ int check_config_validity()
 
 			/* detect and address thread affinity inconsistencies */
 			err = NULL;
-			if (thread_resolve_group_mask(bind_conf->settings.bind_tgroup, bind_conf->settings.bind_thread,
+			if (thread_resolve_group_mask(bind_conf->bind_tgroup, bind_conf->bind_thread,
-			                              &bind_conf->settings.bind_tgroup, &bind_conf->settings.bind_thread, &err) < 0) {
+			                              &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 (!((mask = bind_conf->settings.bind_thread) & all_threads_mask)) {
+			} else if (!((mask = bind_conf->bind_thread) & all_threads_mask)) {
 				unsigned long new_mask = 0;
 
 				while (mask) {
@@ -2558,10 +2559,16 @@ int check_config_validity()
 					mask >>= global.nbthread;
 				}
 
-				bind_conf->settings.bind_thread = new_mask;
+				bind_conf->bind_thread = new_mask;
 				ha_warning("Proxy '%s': the thread range specified on the 'thread' directive of 'bind %s' at [%s:%d] only refers to thread numbers out of the range defined by the global 'nbthread' directive. The thread numbers were remapped to existing threads instead (mask 0x%lx).\n",
 					   curproxy->id, bind_conf->arg, bind_conf->file, bind_conf->line, new_mask);
 			}
+
+			/* apply thread masks and groups to all receivers */
+			list_for_each_entry(li, &bind_conf->listeners, by_bind) {
+				li->rx.bind_thread = bind_conf->bind_thread;
+				li->rx.bind_tgroup = bind_conf->bind_tgroup;
+			}
 		}
 
 		switch (curproxy->mode) {

src/listener.c

@@ -906,7 +906,7 @@ void listener_accept(struct listener *l)
 
 
 #if defined(USE_THREAD)
-		mask = thread_mask(l->rx.settings->bind_thread) & all_threads_mask;
+		mask = thread_mask(l->rx.bind_thread) & all_threads_mask;
 		if (atleast2(mask) && (global.tune.options & GTUNE_LISTENER_MQ) && !stopping) {
 			struct accept_queue_ring *ring;
 			unsigned int t, t0, t1, t2;
@@ -1563,7 +1563,7 @@ static int bind_parse_process(char **args, int cur_arg, struct proxy *px, struct
 		*slash = '/';
 	}
 
-	conf->settings.bind_thread |= thread;
+	conf->bind_thread |= thread;
 
 	memprintf(err, "'process %s' on 'bind' lines is deprecated and will be removed in 2.7.", args[cur_arg+1]);
 	if (slash)
@@ -1612,8 +1612,8 @@ static int bind_parse_thread(char **args, int cur_arg, struct proxy *px, struct
 		sep = args[cur_arg + 1];
 	}
 
-	if ((conf->settings.bind_tgroup || conf->settings.bind_thread) &&
+	if ((conf->bind_tgroup || conf->bind_thread) &&
-	    conf->settings.bind_tgroup != tgroup) {
+	    conf->bind_tgroup != tgroup) {
 		memprintf(err, "'%s' multiple thread-groups are not supported", args[cur_arg + 1]);
 		return ERR_ALERT | ERR_FATAL;
 	}
@@ -1623,8 +1623,8 @@ static int bind_parse_thread(char **args, int cur_arg, struct proxy *px, struct
 		return ERR_ALERT | ERR_FATAL;
 	}
 
-	conf->settings.bind_thread |= thread;
+	conf->bind_thread |= thread;
-	conf->settings.bind_tgroup  = tgroup;
+	conf->bind_tgroup  = tgroup;
 	return 0;
 }
 

src/proto_sockpair.c

@@ -157,7 +157,7 @@ int sockpair_bind_receiver(struct receiver *rx, char **errmsg)
 
 	rx->flags |= RX_F_BOUND;
 
-	fd_insert(rx->fd, rx->owner, rx->iocb, thread_mask(rx->settings->bind_thread) & all_threads_mask);
+	fd_insert(rx->fd, rx->owner, rx->iocb, thread_mask(rx->bind_thread) & all_threads_mask);
 	return err;
 
  bind_return:

src/sock_inet.c

@@ -391,7 +391,7 @@ int sock_inet_bind_receiver(struct receiver *rx, char **errmsg)
 	rx->fd = fd;
 	rx->flags |= RX_F_BOUND;
 
-	fd_insert(fd, rx->owner, rx->iocb, thread_mask(rx->settings->bind_thread) & all_threads_mask);
+	fd_insert(fd, rx->owner, rx->iocb, thread_mask(rx->bind_thread) & all_threads_mask);
 
 	/* for now, all regularly bound TCP listeners are exportable */
 	if (!(rx->flags & RX_F_INHERITED))

src/sock_unix.c

@@ -285,7 +285,7 @@ int sock_unix_bind_receiver(struct receiver *rx, char **errmsg)
 	rx->fd = fd;
 	rx->flags |= RX_F_BOUND;
 
-	fd_insert(fd, rx->owner, rx->iocb, thread_mask(rx->settings->bind_thread) & all_threads_mask);
+	fd_insert(fd, rx->owner, rx->iocb, thread_mask(rx->bind_thread) & all_threads_mask);
 
 	/* for now, all regularly bound TCP listeners are exportable */
 	if (!(rx->flags & RX_F_INHERITED))