git push --force batsheva rsocket_upstream_epolllibrdmacm: Add support for epoll_wait

Implement repoll_wait as the rsocket equivalent of epoll_wait.
Snapshots the fd array under the lock, calls rpoll without holding
it, then collects events back under the lock. Wakes the monitor
thread on timeout so it can resume background polling.

Future option: the monitor thread can be disabled entirely for a
threadless mode where repoll_wait calls rpoll directly each time.

Signed-off-by: Batsheva Black <bblack@nvidia.com>

Author: BatshevaBlack

debian/librdmacm1.symbols

@@ -9,6 +9,7 @@ librdmacm.so.1 librdmacm1 #MINVER#
  raccept@RDMACM_1.0 1.0.16
  repoll_create@RDMACM_1.5 61
  repoll_ctl@RDMACM_1.5 61
+ repoll_wait@RDMACM_1.5 61
  rbind@RDMACM_1.0 1.0.16
  rclose@RDMACM_1.0 1.0.16
  rconnect@RDMACM_1.0 1.0.16

librdmacm/librdmacm.map

@@ -99,4 +99,5 @@ RDMACM_1.5 {
 	global:
 		repoll_create;
 		repoll_ctl;
+		repoll_wait;
 } RDMACM_1.4;

librdmacm/librspreload.map

@@ -36,5 +36,6 @@
 		epoll_create;
 		epoll_create1;
 		epoll_ctl;
+		epoll_wait;
 	local: *;
 };

librdmacm/man/rsocket.7.in

@@ -31,7 +31,7 @@ rgetpeername, rgetsockname
 .P
 rsetsockopt, rgetsockopt, rfcntl
 .P
-repoll_create, repoll_ctl
+repoll_create, repoll_ctl, repoll_wait
 .P
 Functions take the same parameters as that used for sockets.  The
 follow capabilities and flags are supported at this time:
@@ -153,6 +153,8 @@ wake_up_interval - maximum number of milliseconds to block in poll.
 This value is used to safe guard against potential application hangs
 in rpoll().
 .P
+max_events - maximum number of events for the epoll thread to handle for an epoll_instance.
+.P
 All configuration files should contain a single integer value.  Values may
 be set by issuing a command similar to the following example.
 .P

librdmacm/preload.c

@@ -1364,3 +1364,8 @@ int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event)
 
 	return repoll_ctl(epfd, op, internal_fd, event);
 }
+
+int epoll_wait(int epfd, struct epoll_event *events, int maxevents, int timeout)
+{
+	return repoll_wait(epfd, events, maxevents, timeout);
+}

librdmacm/rsocket.c

@@ -5170,3 +5170,93 @@ int repoll_ctl(int epfd, int op, int fd, struct epoll_event *event)
 	pthread_mutex_unlock(&ri->lock);
 	return ret;
 }
+
+static uint32_t poll_to_epoll_events(uint32_t poll_revents)
+{
+	uint32_t epoll_events = 0;
+
+	if (poll_revents & POLLIN)
+		epoll_events |= EPOLLIN;
+	if (poll_revents & POLLOUT)
+		epoll_events |= EPOLLOUT;
+	if (poll_revents & POLLERR)
+		epoll_events |= EPOLLERR;
+	if (poll_revents & POLLHUP)
+		epoll_events |= EPOLLHUP;
+	return epoll_events;
+}
+
+static void repoll_wake_monitor(struct repoll_info *ri)
+{
+	atomic_store(&ri->monitor_state, REPOLL_MONITOR_RUNNING);
+	pthread_cond_signal(&ri->monitor_wake);
+}
+
+static int repoll_collect_events(struct repoll_info *ri,
+				 struct pollfd *polled_fds, int polled_nfds,
+				 struct epoll_event *events, int maxevents)
+{
+	int j = 0;
+
+	for (int i = 0; j < maxevents && i < polled_nfds; ++i) {
+		int slot = i + 1;
+
+		if (polled_fds[i].revents && slot < ri->slot_count &&
+		    ri->fds[slot].fd == polled_fds[i].fd) {
+			memcpy(&events[j].data, &ri->user_data[slot],
+			       sizeof(events->data));
+			events[j].events =
+				poll_to_epoll_events(polled_fds[i].revents);
+			++j;
+		}
+	}
+	return j;
+}
+
+/* Future option: disable the monitor thread for a threadless mode
+ * where repoll_wait calls rpoll directly.
+ */
+
+int repoll_wait(int epfd, struct epoll_event *events, int maxevents,
+		int timeout)
+{
+	struct repoll_info *ri;
+	struct pollfd *local_fds;
+	int nfds, ret;
+
+	if (epfd < 0)
+		return EBADF;
+
+	ri = idm_lookup(&repoll_idm, epfd);
+	if (!ri)
+		return EINVAL;
+
+	pthread_mutex_lock(&ri->lock);
+	nfds = ri->slot_count - 1;
+	if (nfds <= 0) {
+		pthread_mutex_unlock(&ri->lock);
+		return 0;
+	}
+	local_fds = malloc(sizeof(*local_fds) * nfds);
+	if (!local_fds) {
+		pthread_mutex_unlock(&ri->lock);
+		return ENOMEM;
+	}
+	memcpy(local_fds, ri->fds + 1, sizeof(*local_fds) * nfds);
+	pthread_mutex_unlock(&ri->lock);
+
+	ret = rpoll(local_fds, nfds, timeout);
+
+	if (ret > 0) {
+		pthread_mutex_lock(&ri->lock);
+		ret = repoll_collect_events(ri, local_fds, nfds, events,
+					    maxevents);
+		pthread_mutex_unlock(&ri->lock);
+	} else if (ret == 0) {
+		if (atomic_load(&ri->monitor_state) == REPOLL_MONITOR_SLEEPING)
+			repoll_wake_monitor(ri);
+	}
+
+	free(local_fds);
+	return ret;
+}

librdmacm/rsocket.h

@@ -95,6 +95,8 @@ size_t riowrite(int socket, const void *buf, size_t count, off_t offset, int fla
 
 int repoll_create(int flags);
 int repoll_ctl(int epfd, int op, int fd, struct epoll_event *event);
+int repoll_wait(int epfd, struct epoll_event *events, int maxevents,
+		int timeout);
 
 #ifdef __cplusplus
 }