net: zperf: allow TCP receiver to handle multiple connections

Currently the zperf_tcp_receiver can handle only one TCP connection
each time, modify the code to poll and handle multiple connections.

Take the occasion to unify the bind and listen part of the code
between ipv4 and ipv6 part using a structure introduced to handle
the multiple connections.

Now in case the zsock_recv fails, we can't stop every connection
and fail through the error label, so just print the error message
and report the failure through the callback.

Signed-off-by: Luca Fancellu <[email protected]>

Author: luca-fancellu

subsys/net/lib/zperf/zperf_tcp_receiver.c

@@ -32,10 +32,8 @@ LOG_MODULE_DECLARE(net_zperf, CONFIG_NET_ZPERF_LOG_LEVEL);
 #define TCP_RECEIVER_STACK_SIZE 2048
 
 #define SOCK_ID_IPV4_LISTEN 0
-#define SOCK_ID_IPV4_DATA 1
-#define SOCK_ID_IPV6_LISTEN 2
-#define SOCK_ID_IPV6_DATA 3
-#define SOCK_ID_MAX 4
+#define SOCK_ID_IPV6_LISTEN 1
+#define SOCK_ID_MAX         (CONFIG_NET_ZPERF_MAX_SESSIONS + 2)
 
 #define TCP_RECEIVER_BUF_SIZE 1500
 #define POLL_TIMEOUT_MS 100
@@ -100,10 +98,50 @@ static void tcp_received(const struct sockaddr *addr, size_t datalen)
 	}
 }
 
+static int tcp_bind_listen_connection(struct zsock_pollfd *pollfd,
+				      struct sockaddr *address)
+{
+	uint16_t port;
+	int ret;
+
+	if (address->sa_family == AF_INET) {
+		port = ntohs(net_sin(address)->sin_port);
+	} else {
+		port = ntohs(net_sin6(address)->sin6_port);
+	}
+
+	ret = zsock_bind(pollfd->fd, address, sizeof(*address));
+	if (ret < 0) {
+		NET_ERR("Cannot bind IPv%d TCP port %d (%d)",
+			(address->sa_family == AF_INET ? 4 : 6), port, errno);
+		goto out;
+	}
+
+	ret = zsock_listen(pollfd->fd, 1);
+	if (ret < 0) {
+		NET_ERR("Cannot listen IPv%d TCP (%d)",
+			(address->sa_family == AF_INET ? 4 : 6), errno);
+		goto out;
+	}
+
+	pollfd->events = ZSOCK_POLLIN;
+
+out:
+	return ret;
+}
+
+static void tcp_session_error_report(void)
+{
+	if (tcp_session_cb != NULL) {
+		tcp_session_cb(ZPERF_SESSION_ERROR, NULL, tcp_user_data);
+	}
+}
+
 static void tcp_server_session(void)
 {
 	static uint8_t buf[TCP_RECEIVER_BUF_SIZE];
-	struct zsock_pollfd fds[SOCK_ID_MAX] = { 0 };
+	static struct zsock_pollfd fds[SOCK_ID_MAX];
+	static struct sockaddr sock_addr[SOCK_ID_MAX];
 	int ret;
 
 	for (int i = 0; i < ARRAY_SIZE(fds); i++) {
@@ -134,7 +172,7 @@ static void tcp_server_session(void)
 			const struct in_addr *addr =
 				zperf_get_default_if_in4_addr();
 			if (!addr) {
-				NET_ERR("Unable to get IPv4 by default\n");
+				NET_ERR("Unable to get IPv4 by default");
 				goto error;
 			}
 			memcpy(&in4_addr->sin_addr, addr,
@@ -146,22 +184,15 @@ static void tcp_server_session(void)
 		NET_INFO("Binding to %s",
 			 net_sprint_ipv4_addr(&in4_addr->sin_addr));
 
-		ret = zsock_bind(fds[SOCK_ID_IPV4_LISTEN].fd,
-				 (struct sockaddr *)in4_addr,
-				 sizeof(struct sockaddr_in));
-		if (ret < 0) {
-			NET_ERR("Cannot bind IPv4 TCP port %d (%d)",
-				ntohs(in4_addr->sin_port), errno);
-			goto error;
-		}
+		memcpy(net_sin(&sock_addr[SOCK_ID_IPV4_LISTEN]), in4_addr,
+		       sizeof(struct sockaddr_in));
 
-		ret = zsock_listen(fds[SOCK_ID_IPV4_LISTEN].fd, 1);
+		ret = tcp_bind_listen_connection(
+				&fds[SOCK_ID_IPV4_LISTEN],
+				&sock_addr[SOCK_ID_IPV4_LISTEN]);
 		if (ret < 0) {
-			NET_ERR("Cannot listen IPv4 TCP (%d)", errno);
 			goto error;
 		}
-
-		fds[SOCK_ID_IPV4_LISTEN].events = ZSOCK_POLLIN;
 	}
 
 	if (IS_ENABLED(CONFIG_NET_IPV6)) {
@@ -189,7 +220,7 @@ static void tcp_server_session(void)
 			const struct in6_addr *addr =
 				zperf_get_default_if_in6_addr();
 			if (!addr) {
-				NET_ERR("Unable to get IPv6 by default\n");
+				NET_ERR("Unable to get IPv6 by default");
 				goto error;
 			}
 			memcpy(&in6_addr->sin6_addr, addr,
@@ -201,29 +232,20 @@ static void tcp_server_session(void)
 		NET_INFO("Binding to %s",
 			 net_sprint_ipv6_addr(&in6_addr->sin6_addr));
 
-		ret = zsock_bind(fds[SOCK_ID_IPV6_LISTEN].fd,
-				 (struct sockaddr *)in6_addr,
-				 sizeof(struct sockaddr_in6));
-		if (ret < 0) {
-			NET_ERR("Cannot bind IPv6 TCP port %d (%d)",
-				ntohs(in6_addr->sin6_port), errno);
-			goto error;
-		}
+		memcpy(net_sin6(&sock_addr[SOCK_ID_IPV6_LISTEN]), in6_addr,
+		       sizeof(struct sockaddr_in6));
 
-		ret = zsock_listen(fds[SOCK_ID_IPV6_LISTEN].fd, 1);
+		ret = tcp_bind_listen_connection(
+				&fds[SOCK_ID_IPV6_LISTEN],
+				&sock_addr[SOCK_ID_IPV6_LISTEN]);
 		if (ret < 0) {
-			NET_ERR("Cannot listen IPv6 TCP (%d)", errno);
 			goto error;
 		}
-
-		fds[SOCK_ID_IPV6_LISTEN].events = ZSOCK_POLLIN;
 	}
 
 	NET_INFO("Listening on port %d", tcp_server_port);
 
 	while (true) {
-		struct sockaddr addr_ipv4, addr_ipv6;
-
 		ret = zsock_poll(fds, ARRAY_SIZE(fds), POLL_TIMEOUT_MS);
 		if (ret < 0) {
 			NET_ERR("TCP receiver poll error (%d)", errno);
@@ -239,84 +261,84 @@ static void tcp_server_session(void)
 		}
 
 		for (int i = 0; i < ARRAY_SIZE(fds); i++) {
-			struct sockaddr *addr = &addr_ipv6;
-			socklen_t addrlen = sizeof(struct sockaddr);
-
 			if ((fds[i].revents & ZSOCK_POLLERR) ||
 			    (fds[i].revents & ZSOCK_POLLNVAL)) {
 				NET_ERR("TCP receiver IPv%d socket error",
-					(i <= SOCK_ID_IPV4_DATA) ? 4 : 6);
+					(sock_addr[i].sa_family == AF_INET
+						? 4 : 6));
 				goto error;
 			}
 
 			if (!(fds[i].revents & ZSOCK_POLLIN)) {
 				continue;
 			}
 
-			switch (i) {
-			case SOCK_ID_IPV4_LISTEN:
-				addr = &addr_ipv4;
-				__fallthrough;
-			case SOCK_ID_IPV6_LISTEN:{
-				int sock = zsock_accept(fds[i].fd, addr, &addrlen);
+			if ((i >= SOCK_ID_IPV4_LISTEN) && (i <= SOCK_ID_IPV6_LISTEN)) {
+				int j = SOCK_ID_IPV6_LISTEN + 1;
+				struct sockaddr addr_incoming_conn;
+				socklen_t addrlen = sizeof(struct sockaddr);
+				int sock = zsock_accept(fds[i].fd,
+							&addr_incoming_conn,
+							&addrlen);
 
 				if (sock < 0) {
 					NET_ERR("TCP receiver IPv%d accept error",
-						(i <= SOCK_ID_IPV4_DATA) ? 4 : 6);
+						(sock_addr[i].sa_family == AF_INET
+							? 4 : 6));
 					goto error;
 				}
 
-				if (i == SOCK_ID_IPV4_LISTEN &&
-				    fds[SOCK_ID_IPV4_DATA].fd < 0) {
-					fds[SOCK_ID_IPV4_DATA].fd = sock;
-					fds[SOCK_ID_IPV4_DATA].events = ZSOCK_POLLIN;
-				} else if (i == SOCK_ID_IPV6_LISTEN &&
-					   fds[SOCK_ID_IPV6_DATA].fd < 0) {
-					fds[SOCK_ID_IPV6_DATA].fd = sock;
-					fds[SOCK_ID_IPV6_DATA].events = ZSOCK_POLLIN;
-				} else {
+				for (; j < SOCK_ID_MAX; j++) {
+					if (fds[j].fd < 0) {
+						break;
+					}
+				}
+
+				if (j == SOCK_ID_MAX) {
 					/* Too many connections. */
+					NET_ERR("Dropping TCP connection, reached maximum limit.");
 					zsock_close(sock);
-					break;
+				} else {
+					fds[j].fd = sock;
+					fds[j].events = ZSOCK_POLLIN;
+					memcpy(&sock_addr[j],
+					       &addr_incoming_conn,
+					       addrlen);
 				}
-
-				break;
-			}
-
-			case SOCK_ID_IPV4_DATA:
-				addr = &addr_ipv4;
-				__fallthrough;
-			case SOCK_ID_IPV6_DATA:
+			} else if ((i > SOCK_ID_IPV6_LISTEN) && (i < SOCK_ID_MAX)) {
 				ret = zsock_recv(fds[i].fd, buf, sizeof(buf), 0);
 				if (ret < 0) {
 					NET_ERR("recv failed on IPv%d socket (%d)",
-						(i <= SOCK_ID_IPV4_DATA) ? 4 : 6,
+						(sock_addr[i].sa_family == AF_INET
+							? 4 : 6),
 						errno);
-					goto error;
+					tcp_session_error_report();
+					/* This will close the zperf session */
+					ret = 0;
 				}
 
-				tcp_received(addr, ret);
+				tcp_received(&sock_addr[i], ret);
 
 				if (ret == 0) {
 					zsock_close(fds[i].fd);
 					fds[i].fd = -1;
+					memset(&sock_addr[i], 0,
+					sizeof(struct sockaddr));
 				}
-
-				break;
+			} else {
+				goto error;
 			}
 		}
 	}
 
 error:
-	if (tcp_session_cb != NULL) {
-		tcp_session_cb(ZPERF_SESSION_ERROR, NULL,
-			       tcp_user_data);
-	}
+	tcp_session_error_report();
 
 cleanup:
 	for (int i = 0; i < ARRAY_SIZE(fds); i++) {
 		if (fds[i].fd >= 0) {
 			zsock_close(fds[i].fd);
+			memset(&sock_addr[i], 0, sizeof(struct sockaddr));
 		}
 	}
 }