MEDIUM: fd: make thread_mask now represent group-local IDs

With the change that was started on other masks, the thread mask was
still not fully converted, sometimes being used as a global mask and
sometimes as a local one. This finishes the code modifications so that
the mask is always considered as a group-local mask. This doesn't
change anything as long as there's a single group, but is necessary
for groups 2 and above since it's used against running_mask and so on.

diff --git a/include/haproxy/fd.h b/include/haproxy/fd.h
index 9fbc358e27a6660cb1a358f6bd0625b3df354629..8c558189d6d1e7a69f99c0fb0e38abee1b46e8ea 100644 (file)
--- a/include/haproxy/fd.h
+++ b/include/haproxy/fd.h
@@ -431,7 +431,7 @@ static inline void fd_insert(int fd, void *owner, void (*iocb)(int fd), int tgid
        BUG_ON(fdtab[fd].state != 0);
        BUG_ON(fdtab[fd].refc_tgid != 0);
 
-       thread_mask &= all_threads_mask;
+       thread_mask &= tg->threads_enabled;
        BUG_ON(thread_mask == 0);
 
        fdtab[fd].owner = owner;

diff --git a/src/ev_epoll.c b/src/ev_epoll.c
index 13ebc4f4d4a6b6608414309770e32f5d5005b996..77ac0c06777723cbf0ce2983306f36b90007c96e 100644 (file)
--- a/src/ev_epoll.c
+++ b/src/ev_epoll.c
@@ -99,7 +99,7 @@ static void _update_fd(int fd)
                en |= FD_EV_ACTIVE_R;
 
        if ((ps | pr) & ti->ltid_bit) {
-               if (!(fdtab[fd].thread_mask & tid_bit) || !(en & FD_EV_ACTIVE_RW)) {
+               if (!(fdtab[fd].thread_mask & ti->ltid_bit) || !(en & FD_EV_ACTIVE_RW)) {
                        /* fd removed from poll list */
                        opcode = EPOLL_CTL_DEL;
                        if (pr & ti->ltid_bit)
@@ -129,7 +129,7 @@ static void _update_fd(int fd)
                        opcode = EPOLL_CTL_MOD;
                }
        }
-       else if ((fdtab[fd].thread_mask & tid_bit) && (en & FD_EV_ACTIVE_RW)) {
+       else if ((fdtab[fd].thread_mask & ti->ltid_bit) && (en & FD_EV_ACTIVE_RW)) {
                /* new fd in the poll list */
                opcode = EPOLL_CTL_ADD;
                if (en & FD_EV_ACTIVE_R)

diff --git a/src/ev_evports.c b/src/ev_evports.c
index a8fbc13fc477546335354d64b616ccc0221fc2f6..06e7a01c619f62bc2b9b7ab13cd4ad50ec008a26 100644 (file)
--- a/src/ev_evports.c
+++ b/src/ev_evports.c
@@ -71,7 +71,7 @@ static void _update_fd(int fd)
        pr = _HA_ATOMIC_LOAD(&polled_mask[fd].poll_recv);
        ps = _HA_ATOMIC_LOAD(&polled_mask[fd].poll_send);
 
-       if (!(fdtab[fd].thread_mask & tid_bit) || !(en & FD_EV_ACTIVE_RW)) {
+       if (!(fdtab[fd].thread_mask & ti->ltid_bit) || !(en & FD_EV_ACTIVE_RW)) {
                if (!((pr | ps) & ti->ltid_bit)) {
                        /* fd was not watched, it's still not */
                        return;

diff --git a/src/ev_kqueue.c b/src/ev_kqueue.c
index 70ee2ade1d4b895c4ac7a39981b8ea30930a11b2..c37d99b1ad0dbc3e0cb68ca52daf5743f546350b 100644 (file)
--- a/src/ev_kqueue.c
+++ b/src/ev_kqueue.c
@@ -42,7 +42,7 @@ static int _update_fd(int fd, int start)
        pr = _HA_ATOMIC_LOAD(&polled_mask[fd].poll_recv);
        ps = _HA_ATOMIC_LOAD(&polled_mask[fd].poll_send);
 
-       if (!(fdtab[fd].thread_mask & tid_bit) || !(en & FD_EV_ACTIVE_RW)) {
+       if (!(fdtab[fd].thread_mask & ti->ltid_bit) || !(en & FD_EV_ACTIVE_RW)) {
                if (!((pr | ps) & ti->ltid_bit)) {
                        /* fd was not watched, it's still not */
                        return changes;

diff --git a/src/ev_poll.c b/src/ev_poll.c
index 21478137048ca7a84ca147c6660b7f22ccc0e6a3..790ea5bfaf8ee73009ee5cd4eb42072b16e2e372 100644 (file)
--- a/src/ev_poll.c
+++ b/src/ev_poll.c
@@ -192,7 +192,7 @@ static void _do_poll(struct poller *p, int exp, int wake)
                                        continue;
                                }
 
-                               if (!(fdtab[fd].thread_mask & tid_bit)) {
+                               if (!(fdtab[fd].thread_mask & ti->ltid_bit)) {
                                        continue;
                                }
 

diff --git a/src/fd.c b/src/fd.c
index 56a771175f145c02782dbc2045cff532ba30ad48..d412c28589d5e617d5433a98f5a16b8b99d548a8 100644 (file)
--- a/src/fd.c
+++ b/src/fd.c
@@ -440,7 +440,7 @@ int fd_takeover(int fd, void *expected_owner)
        }
 
        /* success, from now on it's ours */
-       HA_ATOMIC_STORE(&fdtab[fd].thread_mask, tid_bit);
+       HA_ATOMIC_STORE(&fdtab[fd].thread_mask, ti->ltid_bit);
 
        /* Make sure the FD doesn't have the active bit. It is possible that
         * the fd is polled by the thread that used to own it, the new thread
@@ -458,7 +458,7 @@ 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 (tg->threads_enabled == 1UL || (fdtab[fd].thread_mask & tg->threads_enabled) == ti->ltid_bit) {
                if (HA_ATOMIC_BTS(&fdtab[fd].update_mask, ti->ltid))
                        return;
 
@@ -472,12 +472,13 @@ void updt_fd_polling(const int fd)
 
                fd_add_to_fd_list(&update_list[tgid - 1], fd);
 
-               if (fd_active(fd) && !(fdtab[fd].thread_mask & tid_bit)) {
+               if (fd_active(fd) && !(fdtab[fd].thread_mask & ti->ltid_bit)) {
                        /* we need to wake up another thread to handle it immediately, any will fit,
                         * so let's pick a random one so that it doesn't always end up on the same.
                         */
-                       int thr = one_among_mask(fdtab[fd].thread_mask & all_threads_mask,
+                       int thr = one_among_mask(fdtab[fd].thread_mask & tg->threads_enabled,
                                                 statistical_prng_range(MAX_THREADS));
+                       thr += ha_tgroup_info[tgid - 1].base;
                        wake_thread(thr);
                }
        }
@@ -520,7 +521,7 @@ int fd_update_events(int fd, uint evts)
                        tmask = _HA_ATOMIC_LOAD(&fdtab[fd].thread_mask);
                } while (rmask & ~tmask);
 
-               if (!(tmask & tid_bit)) {
+               if (!(tmask & ti->ltid_bit)) {
                        /* a takeover has started */
                        activity[tid].poll_skip_fd++;
 
@@ -536,7 +537,7 @@ int fd_update_events(int fd, uint evts)
        /* with running we're safe now, we can drop the reference */
        fd_drop_tgid(fd);
 
-       locked = (tmask != tid_bit);
+       locked = (tmask != ti->ltid_bit);
 
        /* OK now we are guaranteed that our thread_mask was present and
         * that we're allowed to update the FD.