BUG/MEDIUM: pollers: Use a global list for fd shared between threads.
With the old model, any fd shared by multiple threads, such as listeners
or dns sockets, would only be updated on one threads, so that could lead
to missed event, or spurious wakeups.
To avoid this, add a global list for fd that are shared, using the same
implementation as the fd cache, and only remove entries from this list
when every thread as updated its poller.
[wt: this will need to be backported to 1.8 but differently so this patch
must not be backported as-is]
diff --git a/include/common/hathreads.h b/include/common/hathreads.h
index 0f10b48..e27ecc6 100644
--- a/include/common/hathreads.h
+++ b/include/common/hathreads.h
@@ -256,6 +256,8 @@
int thread_no_sync(void);
int thread_need_sync(void);
+extern unsigned long all_threads_mask;
+
#if defined(DEBUG_THREAD) || defined(DEBUG_FULL)
/* WARNING!!! if you update this enum, please also keep lock_label() up to date below */
diff --git a/include/proto/fd.h b/include/proto/fd.h
index 543a420..da09731 100644
--- a/include/proto/fd.h
+++ b/include/proto/fd.h
@@ -36,6 +36,8 @@
extern volatile struct fdlist fd_cache;
extern volatile struct fdlist fd_cache_local[MAX_THREADS];
+extern volatile struct fdlist update_list;
+
extern unsigned long fd_cache_mask; // Mask of threads with events in the cache
extern THREAD_LOCAL int *fd_updt; // FD updates list
@@ -101,15 +103,57 @@
*/
static inline void updt_fd_polling(const int fd)
{
- unsigned int oldupdt;
+ if (fdtab[fd].thread_mask == tid_bit) {
+ unsigned int oldupdt;
+
+ /* note: we don't have a test-and-set yet in hathreads */
- /* note: we don't have a test-and-set yet in hathreads */
+ if (HA_ATOMIC_BTS(&fdtab[fd].update_mask, tid))
+ return;
+
+ oldupdt = HA_ATOMIC_ADD(&fd_nbupdt, 1) - 1;
+ fd_updt[oldupdt] = fd;
+ } else {
+ unsigned long update_mask = fdtab[fd].update_mask;
+ do {
+ if (update_mask == fdtab[fd].thread_mask)
+ return;
+ } while (!HA_ATOMIC_CAS(&fdtab[fd].update_mask, &update_mask,
+ fdtab[fd].thread_mask));
+ fd_add_to_fd_list(&update_list, fd, offsetof(struct fdtab, update));
+ }
+
+}
- if (HA_ATOMIC_BTS(&fdtab[fd].update_mask, tid))
- return;
+/* Called from the poller to acknoledge we read an entry from the global
+ * update list, to remove our bit from the update_mask, and remove it from
+ * the list if we were the last one.
+ */
+static inline void done_update_polling(int fd)
+{
+ unsigned long update_mask;
+
+ update_mask = HA_ATOMIC_AND(&fdtab[fd].update_mask, ~tid_bit);
+ while ((update_mask & all_threads_mask)== 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, fd, offsetof(struct fdtab, update));
+ if (update_list.first == fd)
+ abort();
+ update_mask = (volatile unsigned long)fdtab[fd].update_mask;
+ if ((update_mask & all_threads_mask) != 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, fd, offsetof(struct fdtab, update));
+ update_mask = (volatile unsigned long)(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
+ * care of it yet */
+ } else
+ break;
- oldupdt = HA_ATOMIC_ADD(&fd_nbupdt, 1) - 1;
- fd_updt[oldupdt] = fd;
+ }
}
/* Allocates a cache entry for a file descriptor if it does not yet have one.
diff --git a/include/types/fd.h b/include/types/fd.h
index 0902e7f..aa18ebe 100644
--- a/include/types/fd.h
+++ b/include/types/fd.h
@@ -117,6 +117,7 @@
unsigned long polled_mask; /* mask of thread IDs currently polling this fd */
unsigned long update_mask; /* mask of thread IDs having an update for fd */
struct fdlist_entry cache; /* Entry in the fdcache */
+ struct fdlist_entry update; /* Entry in the global update list */
void (*iocb)(int fd); /* I/O handler */
void *owner; /* the connection or listener associated with this fd, NULL if closed */
unsigned char state; /* FD state for read and write directions (2*3 bits) */