Enables the use of EPOLLEXCLUSIVE for UDP workers where supported.

Normal behaviour for multiple processes using EPOLL to listen on a single FD is for every
process to be woken on every IO event. This can cause a thundering herd effect, increasing context
switches and cpu usage.

With EPOLLEXLUSIVE only a single UDP worker will be woken to handle an IO request greatly
reducing context switching and contention, especially as the number of processes grow.

One potential downside to using EPOLLEXCLUSIVE is that EPOLL may coalesce multiple events on
a file descriptor into a single wakeup. This has the potential to increase latency if only a single
process is woken to handle potentially multiple SIP messages. To help balance latency and reduced
thundering this patch causes the first worker for a socket to not use EPOLLEXCLUSIVE and thus ALWAYS
get woken for events. If present, at least one other worker using EPOLLEXCLUSIVE will also be woken.

Author: rrb3942

io_wait.h

@@ -347,7 +347,8 @@ inline static int io_watch_add(	io_wait_h* h, // lgtm [cpp/use-of-goto]
 								void* data,
 								int prio,
 								unsigned int timeout,
-								int flags)
+								int flags,
+								int exclusive)
 {
 
 	/* helper macros */
@@ -506,16 +507,9 @@ inline static int io_watch_add(	io_wait_h* h, // lgtm [cpp/use-of-goto]
 				ep_event.events|=EPOLLOUT;
 			if (!already) {
 again1:
-#if 0
-/* This is currently broken, because when using EPOLLEXCLUSIVE, the OS will
- * send sequential events to the same process - thus our pseudo-dispatcher
- * will no longer work, since events on a pipe will be queued by a single
- * process. - razvanc
- */
-#if (defined __OS_linux) && (__GLIBC__ >= 2) && (__GLIBC_MINOR__ >= 24)
-				if (e->flags & IO_WATCH_READ)
+#ifdef EPOLLEXCLUSIVE
+				if (e->flags & IO_WATCH_READ && exclusive == 1)
 					ep_event.events|=EPOLLEXCLUSIVE;
-#endif
 #endif
 				n=epoll_ctl(h->epfd, EPOLL_CTL_ADD, fd, &ep_event);
 				if (n==-1){

net/net_udp.c

@@ -331,7 +331,7 @@ inline static int handle_io(struct fd_map* fm, int idx,int event_type)
 }
 
 
-int udp_proc_reactor_init( struct socket_info *si )
+int udp_proc_reactor_init( struct socket_info *si, int si_rank )
 {
 
 	/* create the reactor for UDP proc */
@@ -359,9 +359,18 @@ int udp_proc_reactor_init( struct socket_info *si )
 	}
 
 	/* init: start watching the SIP UDP fd */
-	if (reactor_add_reader( si->socket, F_UDP_READ, RCT_PRIO_NET, si)<0) {
-		LM_CRIT("failed to add UDP listen socket to reactor\n");
-		goto error;
+	//First child per socket becomes 'Master', will wake on every event
+	if (si_rank == 0) {
+		if (reactor_add_reader( si->socket, F_UDP_READ, RCT_PRIO_NET, si)<0) {
+			LM_CRIT("failed to add UDP listen socket to reactor\n");
+			goto error;
+		}
+	} else {
+		//Subsequent processes are helpers, only one should be woken to help at a time
+		if (reactor_add_reader_exclusive( si->socket, F_UDP_READ, RCT_PRIO_NET, si)<0) {
+			LM_CRIT("failed to add UDP listen socket to reactor\n");
+			goto error;
+		}
 	}
 
 	return 0;
@@ -389,7 +398,9 @@ static int fork_dynamic_udp_process(void *si_filter)
 		bind_address=si; /* shortcut */
 		/* we first need to init the reactor to be able to add fd
 		 * into it in child_init routines */
-		if (udp_proc_reactor_init(si) < 0 ||
+		/* Since this is in addition to the master process, si_rank should be > 0 to enable
+		 * exlusive polling with EPOLL */
+		if (udp_proc_reactor_init(si, 1) < 0 ||
 		init_child(10000/*FIXME*/) < 0) {
 			goto error;
 		}
@@ -486,7 +497,7 @@ int udp_start_processes(int *chd_rank, int *startup_done)
 					bind_address=si; /* shortcut */
 					/* we first need to init the reactor to be able to add fd
 					 * into it in child_init routines */
-					if (udp_proc_reactor_init(si) < 0 ||
+					if (udp_proc_reactor_init(si, i) < 0 ||
 							init_child(*chd_rank) < 0) {
 						report_failure_status();
 						if (*chd_rank == 1 && startup_done)

reactor_defs.h

@@ -78,13 +78,16 @@ int init_reactor_size(void);
 	init_io_wait(&_worker_io, _name, reactor_size, io_poll_method, _prio_max)
 
 #define reactor_add_reader( _fd, _type, _prio, _data) \
-	io_watch_add(&_worker_io, _fd, _type, _data, _prio, 0, IO_WATCH_READ)
+	io_watch_add(&_worker_io, _fd, _type, _data, _prio, 0, IO_WATCH_READ, 0)
+
+#define reactor_add_reader_exclusive( _fd, _type, _prio, _data) \
+	io_watch_add(&_worker_io, _fd, _type, _data, _prio, 0, IO_WATCH_READ, 1)
 
 #define reactor_add_reader_with_timeout( _fd, _type, _prio, _t, _data) \
-	io_watch_add(&_worker_io, _fd, _type, _data, _prio, _t, IO_WATCH_READ)
+	io_watch_add(&_worker_io, _fd, _type, _data, _prio, _t, IO_WATCH_READ, 0)
 
 #define reactor_add_writer( _fd, _type, _prio, _data) \
-	io_watch_add(&_worker_io, _fd, _type, _data, _prio, 0, IO_WATCH_WRITE)
+	io_watch_add(&_worker_io, _fd, _type, _data, _prio, 0, IO_WATCH_WRITE, 0)
 
 #define reactor_del_reader( _fd, _idx, _io_flags) \
 	io_watch_del(&_worker_io, _fd, _idx, _io_flags, IO_WATCH_READ)