From df3fdbe5ec3444203a5d46204daa1e803e3e7248 Mon Sep 17 00:00:00 2001 From: Lokesh Muthuraj Date: Mon, 1 Jun 2026 09:16:35 +0000 Subject: [PATCH 1/3] tcp stream and system counter - resource leak fixes --- src/const.h | 2 +- src/main.c | 22 +++++++++++++++++++++- src/ntttcp.c | 14 ++++++++++++-- src/ntttcp.h | 2 +- src/oscounter.c | 4 +++- src/tcpstream.c | 35 +++++++++++++++++++++++++++-------- 6 files changed, 65 insertions(+), 14 deletions(-) diff --git a/src/const.h b/src/const.h index 29f5116..175463b 100644 --- a/src/const.h +++ b/src/const.h @@ -5,7 +5,7 @@ // ---------------------------------------------------------------------------------- #define TOOL_NAME "NTTTCP for Linux" -#define TOOL_VERSION "1.4.3" +#define TOOL_VERSION "1.4.7" #define AUTHOR_NAME "Shihua (Simon) Xiao, sixiao@microsoft.com" #define TCP SOCK_STREAM diff --git a/src/main.c b/src/main.c index 3550e41..90c232c 100644 --- a/src/main.c +++ b/src/main.c @@ -39,10 +39,14 @@ int run_ntttcp_sender(struct ntttcp_test_endpoint *tep) reply_received = query_receiver_busy_state(tep->synch_socket); if (reply_received == -1) { PRINT_ERR("sender: failed to query receiver state"); + if (test->no_synch == false) + close(tep->synch_socket); return ERROR_GENERAL; } if (reply_received == 1) { PRINT_ERR("sender: receiver is busy with another test"); + if (test->no_synch == false) + close(tep->synch_socket); return ERROR_GENERAL; } @@ -50,6 +54,8 @@ int run_ntttcp_sender(struct ntttcp_test_endpoint *tep) test->warmup + test->duration + test->cooldown); if (reply_received == -1) { PRINT_ERR("sender: failed to negotiate test cycle time with receiver"); + if (test->no_synch == false) + close(tep->synch_socket); return ERROR_GENERAL; } if (reply_received != test->duration) { @@ -145,10 +151,14 @@ int run_ntttcp_sender(struct ntttcp_test_endpoint *tep) tep->test->last_client ? (int)'L' : (int)'R'); if (reply_received == -1) { PRINT_ERR("sender: failed to sync with receiver to start test"); + if (test->no_synch == false) + close(tep->synch_socket); return ERROR_GENERAL; } if (reply_received == 0) { PRINT_ERR("sender: receiver refuse to start test right now"); + if (test->no_synch == false) + close(tep->synch_socket); return ERROR_GENERAL; } @@ -162,6 +172,8 @@ int run_ntttcp_sender(struct ntttcp_test_endpoint *tep) if (tep->negotiated_test_cycle_time == 0) { sleep(UINT_MAX); /* either sleep has elapsed, or sleep was interrupted by a signal */ + if (test->no_synch == false) + close(tep->synch_socket); return err_code; } @@ -333,7 +345,15 @@ int main(int argc, char **argv) exit(-1); } - default_ntttcp_test(test); + // Handle error return from default_ntttcp_test + err_code = default_ntttcp_test(test); + if (err_code != NO_ERROR) { + PRINT_ERR("main: error when initializing default test parameters"); + free(test->bind_address); + free(test); + exit(err_code); + } + err_code = parse_arguments(test, argc, argv); if (err_code != NO_ERROR) { PRINT_ERR("main: error when parsing args"); diff --git a/src/ntttcp.c b/src/ntttcp.c index 432b30d..b527c9e 100644 --- a/src/ntttcp.c +++ b/src/ntttcp.c @@ -18,7 +18,7 @@ struct ntttcp_test *new_ntttcp_test() return test; } -void default_ntttcp_test(struct ntttcp_test *test) +int default_ntttcp_test(struct ntttcp_test *test) { test->server_role = false; test->client_role = false; @@ -29,7 +29,6 @@ void default_ntttcp_test(struct ntttcp_test *test) test->use_client_address = false; test->exit_after_done = true; test->mapping = "16,*,*"; - test->bind_address = "0.0.0.0"; test->client_address = "0.0.0.0"; test->cpu_affinity = -1; /* no hard cpu affinity */ test->server_ports = DEFAULT_NUM_SERVER_PORTS; //default:16 */ @@ -59,6 +58,15 @@ void default_ntttcp_test(struct ntttcp_test *test) test->json_log_filename = DEFAULT_JSON_LOG_FILE_NAME; /* "ntttcp-for-linux-log.json" */ test->quiet = false; test->verbose = false; + + /* Allocate bind_address last to ensure all other fields are initialized if allocation fails */ + test->bind_address = strdup("0.0.0.0"); + if (!test->bind_address) { + PRINT_ERR("failed to allocate memory for bind_address in defaults"); + return ERROR_MEMORY_ALLOC; + } + + return NO_ERROR; } bool is_running_tty(void) @@ -228,6 +236,8 @@ void free_ntttcp_test_endpoint_and_test(struct ntttcp_test_endpoint *e) for (i = 0; i < total_threads; i++) free(e->results->threads[i]); + + free(e->results->threads); free(e->results->init_cpu_usage); free(e->results->init_cpu_ps); free(e->results->init_tcp_retrans); diff --git a/src/ntttcp.h b/src/ntttcp.h index 63ba083..076d281 100644 --- a/src/ntttcp.h +++ b/src/ntttcp.h @@ -143,7 +143,7 @@ struct ntttcp_stream_server{ }; struct ntttcp_test *new_ntttcp_test(); -void default_ntttcp_test(struct ntttcp_test *test); +int default_ntttcp_test(struct ntttcp_test *test); struct ntttcp_test_endpoint *new_ntttcp_test_endpoint(struct ntttcp_test *test, int endpoint_role); void set_ntttcp_test_endpoint_test_continuous(struct ntttcp_test_endpoint* e); diff --git a/src/oscounter.c b/src/oscounter.c index 2a8f900..2777166 100644 --- a/src/oscounter.c +++ b/src/oscounter.c @@ -87,8 +87,10 @@ uint64_t get_interrupts_from_proc_by_dev(char *dev_name) PRINT_ERR("Cannot open /proc/interrupts"); return 0; } - if (!strcmp(dev_name, "")) + if (!strcmp(dev_name, "")) { + fclose(file); return 0; + } /* the max number of chars in each line = 64 + 12 chars/cpu * 1024 cpus + 128 = 12,480 */ char buffer[12480]; diff --git a/src/tcpstream.c b/src/tcpstream.c index ff1cf76..9cf1086 100644 --- a/src/tcpstream.c +++ b/src/tcpstream.c @@ -326,6 +326,7 @@ int ntttcp_server_listen(struct ntttcp_stream_server *ss) ASPRINTF(&port_str, "%d", ss->server_port); if (getaddrinfo(ss->bind_address, port_str, &hints, &serv_info) != 0) { PRINT_ERR("cannot get address info for receiver"); + free(port_str); return -1; } free(port_str); @@ -349,17 +350,17 @@ int ntttcp_server_listen(struct ntttcp_stream_server *ss) if (setsockopt(sockfd, SOL_SOCKET, SO_REUSEADDR, (char *)&opt, sizeof(opt)) < 0) { ASPRINTF(&log, "cannot set socket options: %d", sockfd); PRINT_ERR_FREE(log); + close(sockfd); freeaddrinfo(serv_info); free(local_addr_str); - close(sockfd); return -1; } if (set_socket_non_blocking(sockfd) == -1) { ASPRINTF(&log, "cannot set socket as non-blocking: %d", sockfd); PRINT_ERR_FREE(log); + close(sockfd); freeaddrinfo(serv_info); free(local_addr_str); - close(sockfd); return -1; } @@ -369,9 +370,18 @@ int ntttcp_server_listen(struct ntttcp_stream_server *ss) local_addr_str = retrive_ip_address_str((struct sockaddr_storage *)p->ai_addr, local_addr_str, ip_addr_max_size), sockfd, i); - if (i == -1) /* append more info to log */ - ASPRINTF(&log, "%s. errcode = %d", log, errno); - PRINT_DBG_FREE(log); + if (i == -1 && log != NULL) { /* append more info to log */ + char *old_log = log; + ASPRINTF(&log, "%s. errcode = %d", old_log, errno); + if (log != NULL) { + free(old_log); + } else { + log = old_log; /* restore original message if second ASPRINTF failed */ + } + } + if (log != NULL) + PRINT_DBG_FREE(log); + close(sockfd); continue; } else { break; /* connected */ @@ -382,7 +392,6 @@ int ntttcp_server_listen(struct ntttcp_stream_server *ss) if (p == NULL) { ASPRINTF(&log, "cannot bind the socket on address: %s", ss->bind_address); PRINT_ERR_FREE(log); - close(sockfd); return -1; } @@ -495,11 +504,15 @@ int ntttcp_server_epoll(struct ntttcp_stream_server *ss) if (set_socket_non_blocking(newfd) == -1) { ASPRINTF(&log, "cannot set the new socket as non-blocking: %d error [%s]", newfd, strerror(errno)); PRINT_DBG_FREE(log); + close(newfd); + continue; } if (ss->tcp_nodelay && set_socket_tcp_nodelay(newfd) == -1) { ASPRINTF(&log, "cannot set the TCP_NODELAY for socket [%d] error [%s]", newfd, strerror(errno)); PRINT_DBG_FREE(log); + close(newfd); + continue; } local_addr_size = sizeof(local_addr); @@ -522,9 +535,11 @@ int ntttcp_server_epoll(struct ntttcp_stream_server *ss) event.data.fd = newfd; event.events = EPOLLIN; - if (epoll_ctl(efd, EPOLL_CTL_ADD, newfd, &event) != 0) + if (epoll_ctl(efd, EPOLL_CTL_ADD, newfd, &event) != 0) { PRINT_ERR("epoll_ctl failed"); - + close(newfd); + continue; + } /* if there is no synch thread, if any new connection coming, indicates ss started */ if (ss->no_synch) turn_on_light(); @@ -635,11 +650,15 @@ int ntttcp_server_select(struct ntttcp_stream_server *ss) if (set_socket_non_blocking(newfd) == -1) { ASPRINTF(&log, "cannot set the new socket as non-blocking: %d error [%s]", newfd, strerror(errno)); PRINT_DBG_FREE(log); + close(newfd); + continue; } if (ss->tcp_nodelay && set_socket_tcp_nodelay(newfd) == -1) { ASPRINTF(&log, "cannot set the TCP_NODELAY for socket [%d] error [%s]", newfd, strerror(errno)); PRINT_DBG_FREE(log); + close(newfd); + continue; } FD_SET(newfd, &ss->read_set); /* add the new one to read_set */ From 09c8f6f0fce8c0fb02e76c6e1ccef72d88a1c848 Mon Sep 17 00:00:00 2001 From: Lokesh Muthuraj Date: Wed, 15 Jul 2026 03:58:18 +0000 Subject: [PATCH 2/3] Addressing review comments --- src/tcpstream.c | 70 +++++++++++++++++++++++++++++-------------------- 1 file changed, 41 insertions(+), 29 deletions(-) diff --git a/src/tcpstream.c b/src/tcpstream.c index 9cf1086..37fa255 100644 --- a/src/tcpstream.c +++ b/src/tcpstream.c @@ -100,6 +100,7 @@ void *run_ntttcp_sender_tcp_stream(void *ptr) ASPRINTF(&port_str, "%d", sc->server_port); if (getaddrinfo(sc->bind_address, port_str, &hints, &remote_serv_info) != 0) { PRINT_ERR("cannot get address info for receiver"); + free(port_str); return 0; } free(port_str); @@ -168,33 +169,35 @@ void *run_ntttcp_sender_tcp_stream(void *ptr) if (sc->client_port != 0) { /* 2. bind this socket fd to a local fixed TCP port */ client_port = sc->client_port + i; - } - - /* update client port information */ - ntttcp_update_client_port_info(&client_addr, client_port); - - ret = ntttcp_bind_socket(sockfd, &client_addr); - if (ret != 0) { - ASPRINTF(&log, "failed to do tcp bind : socket domain [%d] client_port [%d] errno [%d]", - sc->domain, client_port, errno); - PRINT_INFO_FREE(log); - close(sockfd); - sockfds[i] = -1; - continue; } - /* perform SO_BINDTODEVICE operation for a socket */ - if (sc->use_client_address) { - ret = ntttcp_bind_to_device(sockfd, sc, if_name); - if (ret != NO_ERROR) { - ASPRINTF(&log, "failed to do tcp bind to device : socket domain [%d] client_port [%d] errno [%d] ifname [%s]", - sc->domain, client_port, errno, if_name); - PRINT_INFO_FREE(log); - close(sockfd); - sockfds[i] = -1; - continue; - } - } + /* update client port information */ + ntttcp_update_client_port_info(&client_addr, client_port); + + ret = ntttcp_bind_socket(sockfd, &client_addr); + if (ret != 0) { + int bind_errno = errno; + ASPRINTF(&log, "failed to do tcp bind : socket domain [%d] client_port [%d] errno [%d]", + sc->domain, client_port, bind_errno); + PRINT_INFO_FREE(log); + close(sockfd); + sockfds[i] = -1; + continue; + } + + /* perform SO_BINDTODEVICE operation for a socket */ + if (sc->use_client_address) { + ret = ntttcp_bind_to_device(sockfd, sc, if_name); + if (ret != NO_ERROR) { + int bind_dev_errno = errno; + ASPRINTF(&log, "failed to do tcp bind to device : socket domain [%d] client_port [%d] errno [%d] ifname [%s]", + sc->domain, client_port, bind_dev_errno, if_name); + PRINT_INFO_FREE(log); + close(sockfd); + sockfds[i] = -1; + continue; + } + } /* 3. connect to receiver */ remote_addr_str = retrive_ip_address_str((struct sockaddr_storage *)p->ai_addr, remote_addr_str, ip_addr_max_size); @@ -365,6 +368,7 @@ int ntttcp_server_listen(struct ntttcp_stream_server *ss) } if ((i = bind(sockfd, p->ai_addr, p->ai_addrlen)) < 0) { + int bind_errno = errno; /* capture errno immediately after bind failure */ ASPRINTF(&log, "failed to bind the socket to local address: %s on socket: %d. return = %d", local_addr_str = retrive_ip_address_str((struct sockaddr_storage *)p->ai_addr, local_addr_str, ip_addr_max_size), @@ -372,7 +376,7 @@ int ntttcp_server_listen(struct ntttcp_stream_server *ss) if (i == -1 && log != NULL) { /* append more info to log */ char *old_log = log; - ASPRINTF(&log, "%s. errcode = %d", old_log, errno); + ASPRINTF(&log, "%s. errcode = %d", old_log, bind_errno); if (log != NULL) { free(old_log); } else { @@ -464,6 +468,13 @@ int ntttcp_server_epoll(struct ntttcp_stream_server *ss) /* Buffer where events are returned */ events = calloc(MAX_EPOLL_EVENTS, sizeof event); + if (events == NULL) { + PRINT_ERR("cannot allocate memory for epoll events"); + free(buffer); + free(ip_address_str); + close(efd); + return ERROR_MEMORY_ALLOC; + } while (1) { if (ss->endpoint->receiver_exit_after_done && @@ -495,7 +506,8 @@ int ntttcp_server_epoll(struct ntttcp_stream_server *ss) /* We have processed all incoming connections. */ break; } else { - ASPRINTF(&log, "error to accept new connections. errno = %d", errno) + int accept_errno = errno; + ASPRINTF(&log, "error to accept new connections. errno = %d", accept_errno) PRINT_ERR_FREE(log); break; } @@ -557,10 +569,10 @@ int ntttcp_server_epoll(struct ntttcp_stream_server *ss) if (nbytes <= 0) { if (errno != EAGAIN) { if (nbytes == 0) { - ASPRINTF(&log, "socket closed: %d", i); + ASPRINTF(&log, "socket closed: %d", current_fd); PRINT_DBG_FREE(log); } else { - ASPRINTF(&log, "error: cannot read data from socket: %d", i); + ASPRINTF(&log, "error: cannot read data from socket: %d", current_fd); PRINT_INFO_FREE(log); err_code = ERROR_NETWORK_READ; /* need to continue ss and check other socket, so don't end the ss */ From e5c24c7965191349e955f207e51b1e316a219c08 Mon Sep 17 00:00:00 2001 From: Lokesh Muthuraj Date: Mon, 20 Jul 2026 03:03:34 +0000 Subject: [PATCH 3/3] Addressing review comments in tcpstream.c file --- src/tcpstream.c | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/src/tcpstream.c b/src/tcpstream.c index 37fa255..74ffb00 100644 --- a/src/tcpstream.c +++ b/src/tcpstream.c @@ -439,12 +439,14 @@ int ntttcp_server_epoll(struct ntttcp_stream_server *ss) if ((buffer = (char *)malloc(ss->recv_buf_size)) == (char *)NULL) { PRINT_ERR("cannot allocate memory for receive buffer"); + close(ss->listener); return ERROR_MEMORY_ALLOC; } ip_addr_max_size = (ss->domain == AF_INET ? INET_ADDRSTRLEN : INET6_ADDRSTRLEN); if ((ip_address_str = (char *)malloc(ip_addr_max_size)) == (char *)NULL) { PRINT_ERR("cannot allocate memory for ip address of peer"); free(buffer); + close(ss->listener); return ERROR_MEMORY_ALLOC; } @@ -453,6 +455,7 @@ int ntttcp_server_epoll(struct ntttcp_stream_server *ss) PRINT_ERR("epoll_create1 failed"); free(buffer); free(ip_address_str); + close(ss->listener); return ERROR_EPOLL; } @@ -463,6 +466,7 @@ int ntttcp_server_epoll(struct ntttcp_stream_server *ss) free(buffer); free(ip_address_str); close(efd); + close(ss->listener); return ERROR_EPOLL; } @@ -473,6 +477,7 @@ int ntttcp_server_epoll(struct ntttcp_stream_server *ss) free(buffer); free(ip_address_str); close(efd); + close(ss->listener); return ERROR_MEMORY_ALLOC; } @@ -617,12 +622,14 @@ int ntttcp_server_select(struct ntttcp_stream_server *ss) if ((buffer = (char *)malloc(ss->recv_buf_size)) == (char *)NULL) { PRINT_ERR("cannot allocate memory for receive buffer"); + close(ss->listener); return ERROR_MEMORY_ALLOC; } ip_addr_max_size = (ss->domain == AF_INET ? INET_ADDRSTRLEN : INET6_ADDRSTRLEN); if ((ip_address_str = (char *)malloc(ip_addr_max_size)) == (char *)NULL) { PRINT_ERR("cannot allocate memory for ip address of peer"); free(buffer); + close(ss->listener); return ERROR_MEMORY_ALLOC; }