Optimize IO handling by draining FDs on every wake-up.

A single IO wake-up can correspond to multiple actual IO events/waiting IO.
Currently, after handling a single event we go back to waiting on the FD, where we will
be immediatly woke again because of the already waiting IO.

This increases context switches and can increase latency.

By handling all the IO possible on every wakeup before waiting again we can reduce both of these.

Author: rrb3942

ipc.c

@@ -238,26 +238,29 @@ void ipc_handle_job(int fd)
 	ipc_job job;
 	int n;
 
-	/* read one IPC job from the pipe; even if the read is blocking,
-	 * we are here triggered from the reactor, on a READ event, so 
-	 * we shouldn;t ever block */
-	n = read(fd, &job, sizeof(job) );
-	if (n==-1) {
-		if (errno==EAGAIN || errno==EINTR || errno==EWOULDBLOCK )
+	//Process all jobs until handle is drained
+	while (1) {
+		/* read one IPC job from the pipe; even if the read is blocking,
+		 * we are here triggered from the reactor, on a READ event, so 
+		 * we shouldn;t ever block */
+		n = read(fd, &job, sizeof(job) );
+		if (n==-1) {
+			if (errno==EAGAIN || errno==EINTR || errno==EWOULDBLOCK )
+				return;
+			LM_ERR("read failed:[%d] %s\n", errno, strerror(errno));
 			return;
-		LM_ERR("read failed:[%d] %s\n", errno, strerror(errno));
-		return;
-	}
+		}
 
-	LM_DBG("received job type %d[%s] from process %d\n",
-		job.handler_type, ipc_handlers[job.handler_type].name, job.snd_proc);
+		LM_DBG("received job type %d[%s] from process %d\n",
+			job.handler_type, ipc_handlers[job.handler_type].name, job.snd_proc);
 
-	/* custom handling for RPC type */
-	if (job.handler_type==ipc_rpc_type) {
-		((ipc_rpc_f*)job.payload1)( job.snd_proc, job.payload2);
-	} else {
-		/* generic registered type */
-		ipc_handlers[job.handler_type].func( job.snd_proc, job.payload1);
+		/* custom handling for RPC type */
+		if (job.handler_type==ipc_rpc_type) {
+			((ipc_rpc_f*)job.payload1)( job.snd_proc, job.payload2);
+		} else {
+			/* generic registered type */
+			ipc_handlers[job.handler_type].func( job.snd_proc, job.payload1);
+		}
 	}
 
 	return;

net/net_udp.c

@@ -269,8 +269,11 @@ inline static int handle_io(struct fd_map* fm, int idx,int event_type)
 
 	switch(fm->type){
 		case F_UDP_READ:
-			n = protos[((struct socket_info*)fm->data)->proto].net.
-				read( fm->data /*si*/, &read);
+			do {
+				n = protos[((struct socket_info*)fm->data)->proto].net.
+					read( fm->data /*si*/, &read);
+			//Continue reading packets until we get an error
+			} while (n == 0);
 			break;
 		case F_TIMER_JOB:
 			handle_timer_job();
@@ -304,6 +307,10 @@ inline static int handle_io(struct fd_map* fm, int idx,int event_type)
 	post_run_handle_script_reload();
 
 	pt_become_idle();
+
+	if (n == 1) {
+		n = 0;
+	}
 	return n;
 }
 

net/proto_udp/proto_udp.c

@@ -133,9 +133,9 @@ static int udp_read_req(struct socket_info *si, int* bytes_read)
 	fromlen=sockaddru_len(si->su);
 	len=recvfrom(bind_address->socket, buf, BUF_SIZE,0,&ri.src_su.s,&fromlen);
 	if (len==-1){
-		if (errno==EAGAIN)
-			return 0;
-		if ((errno==EINTR)||(errno==EWOULDBLOCK)|| (errno==ECONNREFUSED))
+		if (errno==EAGAIN || errno==EWOULDBLOCK || errno==EINTR)
+			return 1;
+		if (errno==ECONNREFUSED)
 			return -1;
 		LM_ERR("recvfrom:[%d] %s\n", errno, strerror(errno));
 		return -2;

timer.c

@@ -842,32 +842,35 @@ void handle_timer_job(void)
 	struct os_timer *t;
 	ssize_t l;
 
-	/* read one "os_timer" pointer from the pipe (non-blocking) */
-	l = read( timer_fd_out, &t, sizeof(t) );
-	if (l==-1) {
-		if (errno==EAGAIN || errno==EINTR || errno==EWOULDBLOCK )
+	/* Read events until epipe is empty */
+	while(1) {
+		/* read one "os_timer" pointer from the pipe (non-blocking) */
+		l = read( timer_fd_out, &t, sizeof(t) );
+		if (l==-1) {
+			if (errno==EAGAIN || errno==EINTR || errno==EWOULDBLOCK )
+				return;
+			LM_ERR("read failed:[%d] %s\n", errno, strerror(errno));
 			return;
-		LM_ERR("read failed:[%d] %s\n", errno, strerror(errno));
-		return;
-	}
+		}
 
-	/* run the handler */
-	if (t->flags&TIMER_FLAG_IS_UTIMER) {
+		/* run the handler */
+		if (t->flags&TIMER_FLAG_IS_UTIMER) {
 
-		if (t->trigger_time<(*ijiffies-ITIMER_TICK) )
-			LM_WARN("utimer job <%s> has a %lld us delay in execution\n",
-				t->label, *ijiffies-t->trigger_time);
-		t->u.utimer_f( t->time , t->t_param);
-		t->trigger_time = 0;
+			if (t->trigger_time<(*ijiffies-ITIMER_TICK) )
+				LM_WARN("utimer job <%s> has a %lld us delay in execution\n",
+					t->label, *ijiffies-t->trigger_time);
+			t->u.utimer_f( t->time , t->t_param);
+			t->trigger_time = 0;
 
-	} else {
+		} else {
 
-		if (t->trigger_time<(*ijiffies-ITIMER_TICK) )
-			LM_WARN("timer job <%s> has a %lld us delay in execution\n",
-				t->label, *ijiffies-t->trigger_time);
-		t->u.timer_f( (unsigned int)t->time , t->t_param);
-		t->trigger_time = 0;
+			if (t->trigger_time<(*ijiffies-ITIMER_TICK) )
+				LM_WARN("timer job <%s> has a %lld us delay in execution\n",
+					t->label, *ijiffies-t->trigger_time);
+			t->u.timer_f( (unsigned int)t->time , t->t_param);
+			t->trigger_time = 0;
 
+		}
 	}
 
 	return;