MEDIUM: thread: add a new per-thread flag TH_FL_NOTIFIED to remember wakeups

Right now when an inter-thread wakeup happens, we preliminary check if the
thread was asleep, and if so we wake the poller up and remove its bit from
the sleeping mask. That's not very clean since the sleeping mask cannot be
entirely trusted since a thread that's about to wake up will already have
its sleeping bit removed.

This patch adds a new per-thread flag (TH_FL_NOTIFIED) to remember that a
thread was notified to wake up. It's cleared before checking the task lists
last, so that new wakeups can be considered again (since wake_thread() is
only used to notify about task wakeups and FD polling changes). This way
we do not need to modify a remote thread's sleeping mask anymore. As such
wake_thread() now only tests and sets the TH_FL_NOTIFIED flag but doesn't
clear sleeping anymore.

include/haproxy/fd.h

@@ -370,11 +370,15 @@ static inline unsigned int hap_fd_isset(int fd, unsigned int *evts)
 	return evts[fd / (8*sizeof(*evts))] & (1U << (fd & (8*sizeof(*evts) - 1)));
 }
 
+/* send a wake-up event to this thread, only if it's asleep and not notified yet */
 static inline void wake_thread(int thr)
 {
-	if (sleeping_thread_mask & (1UL << thr)) {
+	struct thread_ctx *ctx = &ha_thread_ctx[thr];
+
+	if (sleeping_thread_mask & (1UL << thr) &&
+	    (_HA_ATOMIC_LOAD(&ctx->flags) & TH_FL_NOTIFIED) == 0) {
 		char c = 'c';
-		_HA_ATOMIC_AND(&sleeping_thread_mask, ~(1UL << thr));
+		_HA_ATOMIC_OR(&ctx->flags, TH_FL_NOTIFIED);
 		DISGUISE(write(poller_wr_pipe[thr], &c, 1));
 	}
 }

include/haproxy/tinfo-t.h

@@ -42,6 +42,8 @@ enum {
  */
 #define TH_FL_STUCK             0x00000001
 #define TH_FL_TASK_PROFILING    0x00000002
+#define TH_FL_NOTIFIED          0x00000004  /* task was notified about the need to wake up */
+
 
 /* Thread group information. This defines a base and a count of global thread
  * IDs which belong to it, and which can be looked up into thread_info/ctx. It

src/haproxy.c

@@ -2805,6 +2805,7 @@ void run_poll_loop()
 			activity[tid].wake_tasks++;
 		else {
 			_HA_ATOMIC_OR(&sleeping_thread_mask, tid_bit);
+			_HA_ATOMIC_AND(&th_ctx->flags, ~TH_FL_NOTIFIED);
 			__ha_barrier_atomic_store();
 			if (thread_has_tasks()) {
 				activity[tid].wake_tasks++;