rtpproxy: drain timeout notification sockets

RTPProxy may keep the notification connection open while writing a burst of
timeout events.  The rtpproxy notification callback only performed a single
read per reactor wakeup, so already-buffered notifications could be left
pending with no further event to wake the process.

Set notification sockets nonblocking and drain them until EAGAIN, preserving
partial commands between callbacks.  Also centralize notification connection
cleanup so error, EOF and handler failure paths consistently unregister the fd,
free pending data and close the socket.

Author: sobomax

modules/rtpproxy/notification_process.c

@@ -54,11 +54,44 @@ struct rtpp_notify {
 	int fd;
 	char *remaining;
 	int remaining_len;
+	int linked;
 	union sockaddr_union addr;
 	struct list_head list;
 };
 OSIPS_LIST_HEAD(rtpp_notify_fds);
 
+static void free_rtpp_notify(int fd, struct rtpp_notify *notify)
+{
+	reactor_del_reader(fd, -1, IO_FD_CLOSING);
+	if (notify) {
+		if (notify->linked)
+			list_del(&notify->list);
+		if (notify->remaining)
+			pkg_free(notify->remaining);
+		pkg_free(notify);
+	}
+	shutdown(fd, SHUT_RDWR);
+	close(fd);
+}
+
+static int set_nonblocking(int fd)
+{
+	int flags;
+
+	flags = fcntl(fd, F_GETFL);
+	if (flags == -1) {
+		LM_ERR("fcntl(%d, F_GETFL) failed: %s\n", fd, strerror(errno));
+		return -1;
+	}
+
+	if (fcntl(fd, F_SETFL, flags | O_NONBLOCK) == -1) {
+		LM_ERR("fcntl(%d, F_SETFL) failed: %s\n", fd, strerror(errno));
+		return -1;
+	}
+
+	return 0;
+}
+
 static int notification_handler(str *command)
 {
 	char cmd, *p;
@@ -200,7 +233,7 @@ static int rtpproxy_io_callback(int fd, void *fs, int was_timeout)
 {
 	struct rtpp_notify *notify = (struct rtpp_notify *)fs;
 	char buffer[BUF_LEN];
-	int len, left, offset;
+	int len, left, offset, total;
 	str command;
 	char *p, *start, *sp, *end;
 
@@ -214,59 +247,79 @@ static int rtpproxy_io_callback(int fd, void *fs, int was_timeout)
 		offset = 0;
 	}
 
-	do
-		len = read(fd, buffer + offset, BUF_LEN - offset);
-	while (len == -1 && errno == EINTR);
-
-	if (len < 0) {
-		LM_ERR("reading from socket failed: %s\n",strerror(errno));
-		return -1;
-	}
-	if (len == 0) {
-		LM_DBG("closing rtpproxy notify socket\n");
-		reactor_del_reader(fd, -1, IO_FD_CLOSING);
-		if (notify) {
-			list_del(&notify->list);
-			pkg_free(notify);
+	for (;;) {
+		if (offset == BUF_LEN) {
+			LM_ERR("RTPProxy notification command too large [%.*s]\n",
+					offset, buffer);
+			free_rtpp_notify(fd, notify);
+			return -1;
 		}
-		shutdown(fd, SHUT_RDWR);
-		close(fd);
-		return 0;
-	}
 
-	LM_DBG("Notification(s) received: [%.*s]\n", len, buffer);
-	p = buffer;
-	left = len + offset;
-	end = buffer + left;
+		do
+			len = read(fd, buffer + offset, BUF_LEN - offset);
+		while (len == -1 && errno == EINTR);
 
-	do {
-		start = p;
+		if (len < 0) {
+			if (errno == EAGAIN || errno == EWOULDBLOCK)
+				break;
 
-		sp = q_memchr(p, '\n', left);
-		if (sp == NULL)
-			break;
-		command.s = p;
-		command.len = sp - p;
-		/* skip the command */
-		p = sp + 1;
-		left -= (sp - start) + 1;
-
-		if (notification_handler(&command) < 0)
+			LM_ERR("reading from socket failed: %s\n",strerror(errno));
+			free_rtpp_notify(fd, notify);
 			return -1;
+		}
+		if (len == 0) {
+			if (offset)
+				LM_WARN("dropping partial RTPProxy notification [%.*s]\n",
+						offset, buffer);
+			LM_DBG("closing rtpproxy notify socket\n");
+			free_rtpp_notify(fd, notify);
+			return 0;
+		}
+
+		total = len + offset;
+		LM_DBG("Notification(s) received: [%.*s]\n", total, buffer);
+		p = buffer;
+		left = total;
+		end = buffer + total;
+
+		do {
+			start = p;
+
+			sp = q_memchr(p, '\n', left);
+			if (sp == NULL)
+				break;
+			command.s = p;
+			command.len = sp - p;
+			/* skip the command */
+			p = sp + 1;
+			left -= (sp - start) + 1;
+
+			if (notification_handler(&command) < 0) {
+				LM_ERR("notification_handler failed\n");
+				free_rtpp_notify(fd, notify);
+				return -1;
+			}
 
-		LM_DBG("Left to process: %d\n[%.*s]\n", left, left, p);
+			LM_DBG("Left to process: %d\n[%.*s]\n", left, left, p);
 
-	} while (p < end);
+		} while (p < end);
+
+		offset = end - p;
+		if (offset) {
+			LM_DBG("%d remaining data in buffer!\n", offset);
+			memmove(buffer, p, offset);
+		}
+	}
 
-	if (end - p) {
-		LM_DBG("%d remaining data in buffer!\n", (int)(end - start));
-		if (notify && (notify->remaining = pkg_malloc(end - start)) != NULL) {
-			notify->remaining_len = (int)(end - start);
-			memcpy(notify->remaining, p, notify->remaining_len);
+	if (offset) {
+		if (notify && (notify->remaining = pkg_malloc(offset)) != NULL) {
+			notify->remaining_len = offset;
+			memcpy(notify->remaining, buffer, offset);
 		} else {
-			LM_WARN("dropping remaining data [%.*s]\n", (int)(end - start), start);
+			LM_WARN("dropping remaining data [%.*s]\n", offset, buffer);
 		}
 	}
+
 	return 0;
 }
 
@@ -285,35 +338,43 @@ static int rtpproxy_io_new_callback(int fd, void *fs, int was_timeout)
 		return -1;
 	}
 
-	if (rtpp_notify_socket_un) {
-		LM_DBG("trusting unix socket connection\n");
-		if (reactor_proc_add_fd(fd, rtpproxy_io_callback, NULL)<0) {
-			LM_CRIT("failed to add RTPProxy new connection to reactor\n");
-			return -1;
-		}
-		return 0;
-	}
-	node = rtpproxy_get_node((union sockaddr_union *)&rtpp_info);
-	if (!node) {
-		LM_WARN("connection from unknown RTPProxy node");
-		return -1;
-	}
+	if (set_nonblocking(fd) < 0)
+		goto err;
 
 	notify = pkg_malloc(sizeof *notify);
 	if (!notify) {
 		LM_ERR("could not allocate notify node\n");
-		return -1;
+		goto err;
 	}
 	memset(notify, 0, sizeof *notify);
 	notify->fd = fd;
-	memcpy(&notify->addr, &node->addr, sizeof(union sockaddr_union));
+
+	if (rtpp_notify_socket_un) {
+		LM_DBG("trusting unix socket connection\n");
+	} else {
+		node = rtpproxy_get_node((union sockaddr_union *)&rtpp_info);
+		if (!node) {
+			LM_WARN("connection from unknown RTPProxy node");
+			pkg_free(notify);
+			goto err;
+		}
+		memcpy(&notify->addr, &node->addr, sizeof(union sockaddr_union));
+		list_add(&notify->list, &rtpp_notify_fds);
+		notify->linked = 1;
+	}
+
 	if (reactor_proc_add_fd(fd, rtpproxy_io_callback, notify) < 0) {
 		LM_CRIT("failed to add RTPProxy listen socket to reactor\n");
+		if (notify->linked)
+			list_del(&notify->list);
 		pkg_free(notify);
-		return -1;
+		goto err;
 	}
-	list_add(&notify->list, &rtpp_notify_fds);
 	return 0;
+err:
+	shutdown(fd, SHUT_RDWR);
+	close(fd);
+	return -1;
 }
 
 int init_rtpp_notify(void)
@@ -411,6 +472,8 @@ void notification_listener_process(int rank)
 		return;
 	}
 
+	if (set_nonblocking(socket_fd) < 0)
+		return;
 	if (reactor_proc_add_fd( socket_fd, rtpproxy_io_new_callback, NULL) < 0) {
 		LM_CRIT("failed to add RTPProxy listen socket to reactor\n");
 		return;