From: Willy Tarreau <w@1wt.eu>
Date: Tue, 5 Jul 2022 17:21:06 +0000 (+0200)
Subject: MEDIUM: fd/poller: turn update_mask to group-local IDs
X-Git-Tag: v2.7-dev2~17
X-Git-Url: http://git.ipfire.org/cgi-bin/gitweb.cgi?a=commitdiff_plain;h=6d3c501c080d8f307f633898c8d4d54169343402;p=thirdparty%2Fhaproxy.git

MEDIUM: fd/poller: turn update_mask to group-local IDs

From now on, the FD's update_mask only refers to local thread IDs. However,
there remains a limitation, in updt_fd_polling(), we temporarily have to
check and set shared FDs against .thread_mask, which still contains global
ones. As such, nbtgroups > 1 may break (but this is not yet supported without
special build options).
---

diff --git a/include/haproxy/fd.h b/include/haproxy/fd.h
index 601419e660..9321777767 100644
--- a/include/haproxy/fd.h
+++ b/include/haproxy/fd.h
@@ -131,17 +131,17 @@ static inline void done_update_polling(int fd)
 {
 	unsigned long update_mask;
 
-	update_mask = _HA_ATOMIC_AND_FETCH(&fdtab[fd].update_mask, ~tid_bit);
-	while ((update_mask & all_threads_mask)== 0) {
+	update_mask = _HA_ATOMIC_AND_FETCH(&fdtab[fd].update_mask, ~ti->ltid_bit);
+	while ((update_mask & tg->threads_enabled) == 0) {
 		/* If we were the last one that had to update that entry, remove it from the list */
 		fd_rm_from_fd_list(&update_list[tgid - 1], fd);
-		update_mask = (volatile unsigned long)fdtab[fd].update_mask;
-		if ((update_mask & all_threads_mask) != 0) {
+		update_mask = _HA_ATOMIC_LOAD(&fdtab[fd].update_mask);
+		if ((update_mask & tg->threads_enabled) != 0) {
 			/* Maybe it's been re-updated in the meanwhile, and we
 			 * wrongly removed it from the list, if so, re-add it
 			 */
 			fd_add_to_fd_list(&update_list[tgid - 1], fd);
-			update_mask = (volatile unsigned long)(fdtab[fd].update_mask);
+			update_mask = _HA_ATOMIC_LOAD(&fdtab[fd].update_mask);
 			/* And then check again, just in case after all it
 			 * should be removed, even if it's very unlikely, given
 			 * the current thread wouldn't have been able to take
diff --git a/src/ev_epoll.c b/src/ev_epoll.c
index 6a71650e8b..13ebc4f4d4 100644
--- a/src/ev_epoll.c
+++ b/src/ev_epoll.c
@@ -169,7 +169,7 @@ static void _do_poll(struct poller *p, int exp, int wake)
 	for (updt_idx = 0; updt_idx < fd_nbupdt; updt_idx++) {
 		fd = fd_updt[updt_idx];
 
-		_HA_ATOMIC_AND(&fdtab[fd].update_mask, ~tid_bit);
+		_HA_ATOMIC_AND(&fdtab[fd].update_mask, ~ti->ltid_bit);
 		if (!fdtab[fd].owner) {
 			activity[tid].poll_drop_fd++;
 			continue;
@@ -188,7 +188,7 @@ static void _do_poll(struct poller *p, int exp, int wake)
 			fd = -fd -4;
 		if (fd == -1)
 			break;
-		if (fdtab[fd].update_mask & tid_bit)
+		if (fdtab[fd].update_mask & ti->ltid_bit)
 			done_update_polling(fd);
 		else
 			continue;
diff --git a/src/ev_evports.c b/src/ev_evports.c
index 8129e9e4e9..a8fbc13fc4 100644
--- a/src/ev_evports.c
+++ b/src/ev_evports.c
@@ -126,7 +126,7 @@ static void _do_poll(struct poller *p, int exp, int wake)
 	for (i = 0; i < fd_nbupdt; i++) {
 		fd = fd_updt[i];
 
-		_HA_ATOMIC_AND(&fdtab[fd].update_mask, ~tid_bit);
+		_HA_ATOMIC_AND(&fdtab[fd].update_mask, ~ti->ltid_bit);
 		if (fdtab[fd].owner == NULL) {
 			activity[tid].poll_drop_fd++;
 			continue;
@@ -145,7 +145,7 @@ static void _do_poll(struct poller *p, int exp, int wake)
 			fd = -fd -4;
 		if (fd == -1)
 			break;
-		if (fdtab[fd].update_mask & tid_bit)
+		if (fdtab[fd].update_mask & ti->ltid_bit)
 			done_update_polling(fd);
 		else
 			continue;
diff --git a/src/ev_kqueue.c b/src/ev_kqueue.c
index 79ddf7d517..70ee2ade1d 100644
--- a/src/ev_kqueue.c
+++ b/src/ev_kqueue.c
@@ -102,7 +102,7 @@ static void _do_poll(struct poller *p, int exp, int wake)
 	for (updt_idx = 0; updt_idx < fd_nbupdt; updt_idx++) {
 		fd = fd_updt[updt_idx];
 
-		_HA_ATOMIC_AND(&fdtab[fd].update_mask, ~tid_bit);
+		_HA_ATOMIC_AND(&fdtab[fd].update_mask, ~ti->ltid_bit);
 		if (!fdtab[fd].owner) {
 			activity[tid].poll_drop_fd++;
 			continue;
@@ -119,7 +119,7 @@ static void _do_poll(struct poller *p, int exp, int wake)
 			fd = -fd -4;
 		if (fd == -1)
 			break;
-		if (fdtab[fd].update_mask & tid_bit)
+		if (fdtab[fd].update_mask & ti->ltid_bit)
 			done_update_polling(fd);
 		else
 			continue;
diff --git a/src/ev_poll.c b/src/ev_poll.c
index a7fadeab92..2147813704 100644
--- a/src/ev_poll.c
+++ b/src/ev_poll.c
@@ -116,7 +116,7 @@ static void _do_poll(struct poller *p, int exp, int wake)
 	for (updt_idx = 0; updt_idx < fd_nbupdt; updt_idx++) {
 		fd = fd_updt[updt_idx];
 
-		_HA_ATOMIC_AND(&fdtab[fd].update_mask, ~tid_bit);
+		_HA_ATOMIC_AND(&fdtab[fd].update_mask, ~ti->ltid_bit);
 		if (!fdtab[fd].owner) {
 			activity[tid].poll_drop_fd++;
 			continue;
@@ -134,12 +134,12 @@ static void _do_poll(struct poller *p, int exp, int wake)
 			fd = -fd -4;
 		if (fd == -1)
 			break;
-		if (fdtab[fd].update_mask & tid_bit) {
+		if (fdtab[fd].update_mask & ti->ltid_bit) {
 			/* Cheat a bit, as the state is global to all pollers
 			 * we don't need every thread to take care of the
 			 * update.
 			 */
-			_HA_ATOMIC_AND(&fdtab[fd].update_mask, ~all_threads_mask);
+			_HA_ATOMIC_AND(&fdtab[fd].update_mask, ~tg->threads_enabled);
 			done_update_polling(fd);
 		} else
 			continue;
diff --git a/src/ev_select.c b/src/ev_select.c
index 2b39a81bb1..eadd5888ab 100644
--- a/src/ev_select.c
+++ b/src/ev_select.c
@@ -108,7 +108,7 @@ static void _do_poll(struct poller *p, int exp, int wake)
 	for (updt_idx = 0; updt_idx < fd_nbupdt; updt_idx++) {
 		fd = fd_updt[updt_idx];
 
-		_HA_ATOMIC_AND(&fdtab[fd].update_mask, ~tid_bit);
+		_HA_ATOMIC_AND(&fdtab[fd].update_mask, ~ti->ltid_bit);
 		if (!fdtab[fd].owner) {
 			activity[tid].poll_drop_fd++;
 			continue;
@@ -125,12 +125,12 @@ static void _do_poll(struct poller *p, int exp, int wake)
 			fd = -fd -4;
 		if (fd == -1)
 			break;
-		if (fdtab[fd].update_mask & tid_bit) {
+		if (fdtab[fd].update_mask & ti->ltid_bit) {
 			/* Cheat a bit, as the state is global to all pollers
 			 * we don't need every thread to take care of the
 			 * update.
 			 */
-			_HA_ATOMIC_AND(&fdtab[fd].update_mask, ~all_threads_mask);
+			_HA_ATOMIC_AND(&fdtab[fd].update_mask, ~tg->threads_enabled);
 			done_update_polling(fd);
 		} else
 			continue;
diff --git a/src/fd.c b/src/fd.c
index 6a7043c896..45cc5f46e9 100644
--- a/src/fd.c
+++ b/src/fd.c
@@ -459,14 +459,14 @@ int fd_takeover(int fd, void *expected_owner)
 void updt_fd_polling(const int fd)
 {
 	if (all_threads_mask == 1UL || (fdtab[fd].thread_mask & all_threads_mask) == tid_bit) {
-		if (HA_ATOMIC_BTS(&fdtab[fd].update_mask, tid))
+		if (HA_ATOMIC_BTS(&fdtab[fd].update_mask, ti->ltid))
 			return;
 
 		fd_updt[fd_nbupdt++] = fd;
 	} else {
 		unsigned long update_mask = fdtab[fd].update_mask;
 		do {
-			if (update_mask == fdtab[fd].thread_mask)
+			if (update_mask == fdtab[fd].thread_mask) // FIXME: this works only on thread-groups 1
 				return;
 		} while (!_HA_ATOMIC_CAS(&fdtab[fd].update_mask, &update_mask, fdtab[fd].thread_mask));
 
@@ -525,7 +525,7 @@ int fd_update_events(int fd, uint evts)
 			activity[tid].poll_skip_fd++;
 
 			/* Let the poller know this FD was lost */
-			if (!HA_ATOMIC_BTS(&fdtab[fd].update_mask, tid))
+			if (!HA_ATOMIC_BTS(&fdtab[fd].update_mask, ti->ltid))
 				fd_updt[fd_nbupdt++] = fd;
 
 			fd_drop_tgid(fd);
@@ -603,10 +603,10 @@ int fd_update_events(int fd, uint evts)
 	/* we had to stop this FD and it still must be stopped after the I/O
 	 * cb's changes, so let's program an update for this.
 	 */
-	if (must_stop && !(fdtab[fd].update_mask & tid_bit)) {
+	if (must_stop && !(fdtab[fd].update_mask & ti->ltid_bit)) {
 		if (((must_stop & FD_POLL_IN)  && !fd_recv_active(fd)) ||
 		    ((must_stop & FD_POLL_OUT) && !fd_send_active(fd)))
-			if (!HA_ATOMIC_BTS(&fdtab[fd].update_mask, tid))
+			if (!HA_ATOMIC_BTS(&fdtab[fd].update_mask, ti->ltid))
 				fd_updt[fd_nbupdt++] = fd;
 	}
 
diff --git a/src/stream.c b/src/stream.c
index 66529d4d77..4208885f9d 100644
--- a/src/stream.c
+++ b/src/stream.c
@@ -3330,7 +3330,7 @@ static int stats_dump_full_strm_to_buffer(struct stconn *sc, struct stream *strm
 			              conn->flags,
 			              conn_fd(conn),
 			              conn_fd(conn) >= 0 ? fdtab[conn->handle.fd].state : 0,
-			              conn_fd(conn) >= 0 ? !!(fdtab[conn->handle.fd].update_mask & tid_bit) : 0,
+			              conn_fd(conn) >= 0 ? !!(fdtab[conn->handle.fd].update_mask & ti->ltid_bit) : 0,
 				      conn_fd(conn) >= 0 ? fdtab[conn->handle.fd].thread_mask: 0);
 
 		}
@@ -3368,7 +3368,7 @@ static int stats_dump_full_strm_to_buffer(struct stconn *sc, struct stream *strm
 			              conn->flags,
 			              conn_fd(conn),
 			              conn_fd(conn) >= 0 ? fdtab[conn->handle.fd].state : 0,
-			              conn_fd(conn) >= 0 ? !!(fdtab[conn->handle.fd].update_mask & tid_bit) : 0,
+			              conn_fd(conn) >= 0 ? !!(fdtab[conn->handle.fd].update_mask & ti->ltid_bit) : 0,
 				      conn_fd(conn) >= 0 ? fdtab[conn->handle.fd].thread_mask: 0);
 
 		}