MEDIUM: listener: add the "shards" bind keyword20211012-bind-shards-1

In multi-threaded mode, on operating systems supporting multiple listeners on
the same IP:port, this will automatically create this number of multiple
identical listeners for the same line, all bound to a fair share of the number
of the threads attached to this listener. This can sometimes be useful when
using very large thread counts where the in-kernel locking on a single socket
starts to cause a significant overhead. In this case the incoming traffic is
distributed over multiple sockets and the contention is reduced. Note that
doing this can easily increase the CPU usage by making more threads work a
little bit.

If the number of shards is higher than the number of available threads, it
will automatically be trimmed to the number of threads. A special value
"by-thread" will automatically assign one shard per thread.

1 files changed, 48 insertions, 2 deletions

diff --git a/src/cfgparse.c b/src/cfgparse.c
index ada344d17..173dcdb32 100644
--- a/src/cfgparse.c
+++ b/src/cfgparse.c
@@ -2566,8 +2566,54 @@ int check_config_validity()
 
 			/* 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;
+				if (bind_conf->settings.shards <= 1) {
+					li->rx.bind_thread = bind_conf->bind_thread;
+					li->rx.bind_tgroup = bind_conf->bind_tgroup;
+				} else {
+					struct listener *new_li;
+					int shard, shards, todo, done, bit;
+					ulong mask;
+
+					shards = bind_conf->settings.shards;
+					todo = my_popcountl(bind_conf->bind_thread);
+
+					/* no more shards than total threads */
+					if (shards > todo)
+						shards = todo;
+
+					shard = done = bit = 0;
+					new_li = li;
+
+					while (1) {
+						mask = 0;
+						while (done < todo) {
+							/* enlarge mask to cover next bit of bind_thread */
+							while (!(bind_conf->bind_thread & (1UL << bit)))
+								bit++;
+							mask |= (1UL << bit);
+							bit++;
+							done += shards;
+						}
+
+						new_li->rx.bind_thread = bind_conf->bind_thread & mask;
+						new_li->rx.bind_tgroup = bind_conf->bind_tgroup;
+						done -= todo;
+
+						shard++;
+						if (shard >= shards)
+							break;
+
+						/* create another listener for new shards */
+						new_li = clone_listener(li);
+						if (!new_li) {
+							ha_alert("Out of memory while trying to allocate extra listener for shard %d in %s %s\n",
+								 shard, proxy_type_str(curproxy), curproxy->id);
+							cfgerr++;
+							err_code |= ERR_FATAL | ERR_ALERT;
+							goto out;
+						}
+					}
+				}
 			}
 		}