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/src/ev_poll.c b/src/ev_poll.c
index 6093b65..155ac82 100644
--- a/src/ev_poll.c
+++ b/src/ev_poll.c
@@ -45,6 +45,44 @@
hap_fd_clr(fd, fd_evts[DIR_WR]);
}
+static void _update_fd(int fd, int *max_add_fd)
+{
+ int en;
+
+ en = fdtab[fd].state;
+
+ /* we have a single state for all threads, which is why we
+ * don't check the tid_bit. First thread to see the update
+ * takes it for every other one.
+ */
+ if (!(en & FD_EV_POLLED_RW)) {
+ if (!fdtab[fd].polled_mask) {
+ /* fd was not watched, it's still not */
+ return;
+ }
+ /* fd totally removed from poll list */
+ hap_fd_clr(fd, fd_evts[DIR_RD]);
+ hap_fd_clr(fd, fd_evts[DIR_WR]);
+ HA_ATOMIC_AND(&fdtab[fd].polled_mask, 0);
+ }
+ else {
+ /* OK fd has to be monitored, it was either added or changed */
+ if (!(en & FD_EV_POLLED_R))
+ hap_fd_clr(fd, fd_evts[DIR_RD]);
+ else
+ hap_fd_set(fd, fd_evts[DIR_RD]);
+
+ if (!(en & FD_EV_POLLED_W))
+ hap_fd_clr(fd, fd_evts[DIR_WR]);
+ else
+ hap_fd_set(fd, fd_evts[DIR_WR]);
+
+ HA_ATOMIC_OR(&fdtab[fd].polled_mask, tid_bit);
+ if (fd > *max_add_fd)
+ *max_add_fd = fd;
+ }
+}
+
/*
* Poll() poller
*/
@@ -53,11 +91,12 @@
int status;
int fd;
int wait_time;
- int updt_idx, en;
+ int updt_idx;
int fds, count;
int sr, sw;
int old_maxfd, new_maxfd, max_add_fd;
unsigned rn, wn; /* read new, write new */
+ int old_fd;
max_add_fd = -1;
@@ -70,39 +109,31 @@
activity[tid].poll_drop++;
continue;
}
+ _update_fd(fd, &max_add_fd);
+ }
- en = fdtab[fd].state;
-
- /* we have a single state for all threads, which is why we
- * don't check the tid_bit. First thread to see the update
- * takes it for every other one.
- */
- if (!(en & FD_EV_POLLED_RW)) {
- if (!fdtab[fd].polled_mask) {
- /* fd was not watched, it's still not */
- continue;
- }
- /* fd totally removed from poll list */
- hap_fd_clr(fd, fd_evts[DIR_RD]);
- hap_fd_clr(fd, fd_evts[DIR_WR]);
- HA_ATOMIC_AND(&fdtab[fd].polled_mask, 0);
- }
- else {
- /* OK fd has to be monitored, it was either added or changed */
- if (!(en & FD_EV_POLLED_R))
- hap_fd_clr(fd, fd_evts[DIR_RD]);
- else
- hap_fd_set(fd, fd_evts[DIR_RD]);
-
- if (!(en & FD_EV_POLLED_W))
- hap_fd_clr(fd, fd_evts[DIR_WR]);
- else
- hap_fd_set(fd, fd_evts[DIR_WR]);
-
- HA_ATOMIC_OR(&fdtab[fd].polled_mask, tid_bit);
- if (fd > max_add_fd)
- max_add_fd = fd;
+ /* Now scan the global update list */
+ for (old_fd = fd = update_list.first; fd != -1; fd = fdtab[fd].update.next) {
+ if (fd == -2) {
+ fd = old_fd;
+ continue;
}
+ else if (fd <= -3)
+ fd = -fd -4;
+ if (fd == -1)
+ break;
+ if (fdtab[fd].update_mask & tid_bit) {
+ /* Cheat a bit, as the state is global to all pollers
+ * we don't need every thread ot take care of the
+ * update.
+ */
+ HA_ATOMIC_AND(&fdtab[fd].update_mask, ~all_threads_mask);
+ done_update_polling(fd);
+ } else
+ continue;
+ if (!fdtab[fd].owner)
+ continue;
+ _update_fd(fd, &max_add_fd);
}
/* maybe we added at least one fd larger than maxfd */