From 62d9c36fcf7b0081db077ff6cb27bfff56685b89 Mon Sep 17 00:00:00 2001 From: Jitpanu Maneeratpongsuk Date: Thu, 19 Dec 2024 15:09:19 +0000 Subject: [PATCH 01/10] feat: Add basic tcp connection to socket node type Signed-off-by: Jitpanu Maneeratpongsuk --- include/villas/nodes/socket.hpp | 1 + include/villas/socket_addr.hpp | 2 +- lib/nodes/socket.cpp | 87 +++++++++++++++++++++++++++++++-- lib/socket_addr.cpp | 6 +++ 4 files changed, 90 insertions(+), 6 deletions(-) diff --git a/include/villas/nodes/socket.hpp b/include/villas/nodes/socket.hpp index 8882a607d..671d508f4 100644 --- a/include/villas/nodes/socket.hpp +++ b/include/villas/nodes/socket.hpp @@ -22,6 +22,7 @@ class NodeCompat; struct Socket { int sd; // The socket descriptor + int clt_sd; // TCP client socket descriptor int verify_source; // Verify the source address of incoming packets against socket::remote. enum SocketLayer diff --git a/include/villas/socket_addr.hpp b/include/villas/socket_addr.hpp index d3947a171..34dbe9bc9 100644 --- a/include/villas/socket_addr.hpp +++ b/include/villas/socket_addr.hpp @@ -35,7 +35,7 @@ union sockaddr_union { namespace villas { namespace node { -enum class SocketLayer { ETH, IP, UDP, UNIX }; +enum class SocketLayer { ETH, IP, UDP, UNIX, TCP_CLIENT, TCP_SERVER}; /* Generate printable socket address depending on the address family * diff --git a/lib/nodes/socket.cpp b/lib/nodes/socket.cpp index e7e56ed08..784441b92 100644 --- a/lib/nodes/socket.cpp +++ b/lib/nodes/socket.cpp @@ -28,6 +28,9 @@ #include #endif // WITH_NETEM +#define MAX_CONNECTION_RETRIES 5 +#define RETRIES_DELAY 2 + using namespace villas; using namespace villas::utils; using namespace villas::node; @@ -97,6 +100,11 @@ char *villas::node::socket_print(NodeCompat *n) { case SocketLayer::UNIX: layer = "unix"; break; + + case SocketLayer::TCP_SERVER: + case SocketLayer::TCP_CLIENT: + layer = "tcp"; + break; } char *local = socket_print_addr((struct sockaddr *)&s->in.saddr); @@ -195,6 +203,11 @@ int villas::node::socket_start(NodeCompat *n) { s->sd = socket(s->in.saddr.sa.sa_family, SOCK_DGRAM, 0); break; + case SocketLayer::TCP_SERVER: + case SocketLayer::TCP_CLIENT: + s->sd = socket(s->in.saddr.sa.sa_family, SOCK_STREAM, 0); + break; + default: throw RuntimeError("Invalid socket type!"); } @@ -233,9 +246,43 @@ int villas::node::socket_start(NodeCompat *n) { addrlen = sizeof(s->in.saddr); } - ret = bind(s->sd, (struct sockaddr *)&s->in.saddr, addrlen); - if (ret < 0) - throw SystemError("Failed to bind socket"); + if (s->layer == SocketLayer::TCP_CLIENT) { + //Attempt to connect to TCP server + int retries = 0; + while (retries < MAX_CONNECTION_RETRIES) { + n->logger->info("Attempting({}) to connect to server..", retries + 1); + ret = connect(s->sd, (struct sockaddr *)&s->out.saddr, addrlen); + if (ret == 0) { + break; + } else { + retries++; + if (retries < MAX_CONNECTION_RETRIES) { + sleep(RETRIES_DELAY); + } + } + } + + } else { + ret = bind(s->sd, (struct sockaddr *)&s->in.saddr, addrlen); + } + + if (ret < 0) { + if (s->layer == SocketLayer::TCP_CLIENT) { + throw SystemError("Failed to connect to TCP server"); + } else { + throw SystemError("Failed to bind socket"); + } + } + + //TCP Server listen for client connection + if (s->layer == SocketLayer::TCP_SERVER) { + listen(s->sd, 5); + //Accept client connection and get client socket descriptor + s->clt_sd = accept(s->sd, nullptr, nullptr); + if (s->clt_sd < 0) { + throw SystemError("Failed to accept connection"); + } + } if (s->multicast.enabled) { ret = setsockopt(s->sd, IPPROTO_IP, IP_MULTICAST_LOOP, &s->multicast.loop, @@ -258,6 +305,8 @@ int villas::node::socket_start(NodeCompat *n) { int prio; switch (s->layer) { case SocketLayer::UDP: + case SocketLayer::TCP_SERVER: + case SocketLayer::TCP_CLIENT: case SocketLayer::IP: prio = IPTOS_LOWDELAY; if (setsockopt(s->sd, IPPROTO_IP, IP_TOS, &prio, sizeof(prio))) @@ -316,7 +365,12 @@ int villas::node::socket_stop(NodeCompat *n) { } if (s->sd >= 0) { + //Close client socket descriptor + if (s->layer == SocketLayer::TCP_SERVER) + close(s->clt_sd); + ret = close(s->sd); + if (ret) return ret; } @@ -340,7 +394,17 @@ int villas::node::socket_read(NodeCompat *n, struct Sample *const smps[], socklen_t srclen = sizeof(src); // Receive next sample - bytes = recvfrom(s->sd, s->in.buf, s->in.buflen, 0, &src.sa, &srclen); + + if (s->layer == SocketLayer::TCP_CLIENT) { + //Receive data from server + bytes = recv(s->sd, s->in.buf, s->in.buflen, 0); + } else if (s->layer == SocketLayer::TCP_SERVER) { + //Receive data from client + bytes = recv(s->clt_sd, s->in.buf, s->in.buflen, 0); + } else { + bytes = recvfrom(s->sd, s->in.buf, s->in.buflen, 0, &src.sa, &srclen); + } + if (bytes < 0) { if (errno == EINTR) return -1; @@ -445,8 +509,17 @@ int villas::node::socket_write(NodeCompat *n, struct Sample *const smps[], } retry2: - bytes = sendto(s->sd, s->out.buf, wbytes, 0, (struct sockaddr *)&s->out.saddr, + if (s->layer == SocketLayer::TCP_CLIENT) { + //Send data to TCP server + bytes = send(s->sd, s->out.buf, wbytes, 0); + } else if (s->layer == SocketLayer::TCP_SERVER) { + //Send data to TCP client + bytes = send(s->clt_sd, s->out.buf, wbytes, 0); + } else { + bytes = sendto(s->sd, s->out.buf, wbytes, 0, (struct sockaddr *)&s->out.saddr, addrlen); + } + if (bytes < 0) { if ((errno == EPERM) || (errno == ENOENT && s->layer == SocketLayer::UNIX)) n->logger->warn("Failed sendto(): {}", strerror(errno)); @@ -505,6 +578,10 @@ int villas::node::socket_parse(NodeCompat *n, json_t *json) { s->layer = SocketLayer::UDP; else if (!strcmp(layer, "unix") || !strcmp(layer, "local")) s->layer = SocketLayer::UNIX; + else if (!strcmp(layer, "tcp_client")) + s->layer = SocketLayer::TCP_CLIENT; + else if (!strcmp(layer, "tcp_server")) + s->layer = SocketLayer::TCP_SERVER; else throw SystemError("Invalid layer '{}'", layer); } diff --git a/lib/socket_addr.cpp b/lib/socket_addr.cpp index a8e4e5f4c..2da3bbc73 100644 --- a/lib/socket_addr.cpp +++ b/lib/socket_addr.cpp @@ -156,6 +156,12 @@ int villas::node::socket_parse_address(const char *addr, struct sockaddr *saddr, hint.ai_protocol = IPPROTO_UDP; break; + case SocketLayer::TCP_CLIENT: + case SocketLayer::TCP_SERVER: + hint.ai_socktype = SOCK_STREAM; + hint.ai_protocol = IPPROTO_TCP; + break; + default: throw RuntimeError("Invalid address type"); } From ac0a7a444ca1448bbec1c47675c707a52a96077a Mon Sep 17 00:00:00 2001 From: Jitpanu Maneeratpongsuk Date: Tue, 7 Jan 2025 19:27:13 +0000 Subject: [PATCH 02/10] fix: Comment format and naming Signed-off-by: Jitpanu Maneeratpongsuk --- lib/nodes/socket.cpp | 27 +++++++++++++++------------ 1 file changed, 15 insertions(+), 12 deletions(-) diff --git a/lib/nodes/socket.cpp b/lib/nodes/socket.cpp index 784441b92..47983df17 100644 --- a/lib/nodes/socket.cpp +++ b/lib/nodes/socket.cpp @@ -102,8 +102,11 @@ char *villas::node::socket_print(NodeCompat *n) { break; case SocketLayer::TCP_SERVER: + layer = "tcp-server"; + break; + case SocketLayer::TCP_CLIENT: - layer = "tcp"; + layer = "tcp-client"; break; } @@ -247,10 +250,10 @@ int villas::node::socket_start(NodeCompat *n) { } if (s->layer == SocketLayer::TCP_CLIENT) { - //Attempt to connect to TCP server + // Attempt to connect to TCP server. int retries = 0; while (retries < MAX_CONNECTION_RETRIES) { - n->logger->info("Attempting({}) to connect to server..", retries + 1); + n->logger->info("Attempting to connect to server: attempt={}...", retries + 1); ret = connect(s->sd, (struct sockaddr *)&s->out.saddr, addrlen); if (ret == 0) { break; @@ -274,10 +277,10 @@ int villas::node::socket_start(NodeCompat *n) { } } - //TCP Server listen for client connection + // TCP Server listen for client connection. if (s->layer == SocketLayer::TCP_SERVER) { listen(s->sd, 5); - //Accept client connection and get client socket descriptor + // Accept client connection and get client socket descriptor. s->clt_sd = accept(s->sd, nullptr, nullptr); if (s->clt_sd < 0) { throw SystemError("Failed to accept connection"); @@ -365,7 +368,7 @@ int villas::node::socket_stop(NodeCompat *n) { } if (s->sd >= 0) { - //Close client socket descriptor + // Close client socket descriptor. if (s->layer == SocketLayer::TCP_SERVER) close(s->clt_sd); @@ -396,10 +399,10 @@ int villas::node::socket_read(NodeCompat *n, struct Sample *const smps[], // Receive next sample if (s->layer == SocketLayer::TCP_CLIENT) { - //Receive data from server + // Receive data from server. bytes = recv(s->sd, s->in.buf, s->in.buflen, 0); } else if (s->layer == SocketLayer::TCP_SERVER) { - //Receive data from client + // Receive data from client. bytes = recv(s->clt_sd, s->in.buf, s->in.buflen, 0); } else { bytes = recvfrom(s->sd, s->in.buf, s->in.buflen, 0, &src.sa, &srclen); @@ -510,10 +513,10 @@ int villas::node::socket_write(NodeCompat *n, struct Sample *const smps[], retry2: if (s->layer == SocketLayer::TCP_CLIENT) { - //Send data to TCP server + // Send data to TCP server. bytes = send(s->sd, s->out.buf, wbytes, 0); } else if (s->layer == SocketLayer::TCP_SERVER) { - //Send data to TCP client + // Send data to TCP client. bytes = send(s->clt_sd, s->out.buf, wbytes, 0); } else { bytes = sendto(s->sd, s->out.buf, wbytes, 0, (struct sockaddr *)&s->out.saddr, @@ -578,9 +581,9 @@ int villas::node::socket_parse(NodeCompat *n, json_t *json) { s->layer = SocketLayer::UDP; else if (!strcmp(layer, "unix") || !strcmp(layer, "local")) s->layer = SocketLayer::UNIX; - else if (!strcmp(layer, "tcp_client")) + else if (!strcmp(layer, "tcp-client")) s->layer = SocketLayer::TCP_CLIENT; - else if (!strcmp(layer, "tcp_server")) + else if (!strcmp(layer, "tcp-server")) s->layer = SocketLayer::TCP_SERVER; else throw SystemError("Invalid layer '{}'", layer); From 467b0512d3691a1cab88e054d0a72ed22bcc7809 Mon Sep 17 00:00:00 2001 From: Jitpanu Maneeratpongsuk Date: Thu, 9 Jan 2025 15:10:25 +0000 Subject: [PATCH 03/10] fix: Add reconnect attempt after the connection is broken. Move the establishment of tcp connection out of socket_start so that it will not block other node. Signed-off-by: Jitpanu Maneeratpongsuk --- include/villas/nodes/socket.hpp | 3 ++ lib/nodes/socket.cpp | 85 +++++++++++++++++++-------------- 2 files changed, 51 insertions(+), 37 deletions(-) diff --git a/include/villas/nodes/socket.hpp b/include/villas/nodes/socket.hpp index 671d508f4..1c78de648 100644 --- a/include/villas/nodes/socket.hpp +++ b/include/villas/nodes/socket.hpp @@ -24,6 +24,7 @@ struct Socket { int sd; // The socket descriptor int clt_sd; // TCP client socket descriptor int verify_source; // Verify the source address of incoming packets against socket::remote. + bool tcp_connect = false; // TCP connection status bit enum SocketLayer layer; // The OSI / IP layer which should be used for this socket @@ -69,6 +70,8 @@ int socket_read(NodeCompat *n, struct Sample *const smps[], unsigned cnt); int socket_parse(NodeCompat *n, json_t *json); +void socket_tcp_connection(NodeCompat *n, Socket *s); + char *socket_print(NodeCompat *n); } // namespace node diff --git a/lib/nodes/socket.cpp b/lib/nodes/socket.cpp index 47983df17..6050c1208 100644 --- a/lib/nodes/socket.cpp +++ b/lib/nodes/socket.cpp @@ -28,7 +28,7 @@ #include #endif // WITH_NETEM -#define MAX_CONNECTION_RETRIES 5 +#define MAX_CONNECTION_RETRIES 40 #define RETRIES_DELAY 2 using namespace villas; @@ -249,43 +249,11 @@ int villas::node::socket_start(NodeCompat *n) { addrlen = sizeof(s->in.saddr); } - if (s->layer == SocketLayer::TCP_CLIENT) { - // Attempt to connect to TCP server. - int retries = 0; - while (retries < MAX_CONNECTION_RETRIES) { - n->logger->info("Attempting to connect to server: attempt={}...", retries + 1); - ret = connect(s->sd, (struct sockaddr *)&s->out.saddr, addrlen); - if (ret == 0) { - break; - } else { - retries++; - if (retries < MAX_CONNECTION_RETRIES) { - sleep(RETRIES_DELAY); - } - } - } - - } else { + if (s->layer != SocketLayer::TCP_CLIENT) ret = bind(s->sd, (struct sockaddr *)&s->in.saddr, addrlen); - } - if (ret < 0) { - if (s->layer == SocketLayer::TCP_CLIENT) { - throw SystemError("Failed to connect to TCP server"); - } else { - throw SystemError("Failed to bind socket"); - } - } - - // TCP Server listen for client connection. - if (s->layer == SocketLayer::TCP_SERVER) { - listen(s->sd, 5); - // Accept client connection and get client socket descriptor. - s->clt_sd = accept(s->sd, nullptr, nullptr); - if (s->clt_sd < 0) { - throw SystemError("Failed to accept connection"); - } - } + if (ret < 0) + throw SystemError("Failed to bind socket"); if (s->multicast.enabled) { ret = setsockopt(s->sd, IPPROTO_IP, IP_MULTICAST_LOOP, &s->multicast.loop, @@ -400,9 +368,15 @@ int villas::node::socket_read(NodeCompat *n, struct Sample *const smps[], if (s->layer == SocketLayer::TCP_CLIENT) { // Receive data from server. + if (!s->tcp_connect) + villas::node::socket_tcp_connection(n, s); + bytes = recv(s->sd, s->in.buf, s->in.buflen, 0); } else if (s->layer == SocketLayer::TCP_SERVER) { // Receive data from client. + if (!s->tcp_connect) + villas::node::socket_tcp_connection(n, s); + bytes = recv(s->clt_sd, s->in.buf, s->in.buflen, 0); } else { bytes = recvfrom(s->sd, s->in.buf, s->in.buflen, 0, &src.sa, &srclen); @@ -413,8 +387,12 @@ int villas::node::socket_read(NodeCompat *n, struct Sample *const smps[], return -1; throw SystemError("Failed recvfrom()"); - } else if (bytes == 0) + } else if (bytes == 0) { + if (s->layer == SocketLayer::TCP_CLIENT || s->layer == SocketLayer::TCP_SERVER) + s->tcp_connect = false; + return 0; + } ptr = s->in.buf; @@ -456,6 +434,39 @@ int villas::node::socket_read(NodeCompat *n, struct Sample *const smps[], return ret; } +void villas::node::socket_tcp_connection(NodeCompat *n, Socket *s) { + int ret; + if (s->layer == SocketLayer::TCP_CLIENT) { + close(s->sd); + s->sd = socket(s->in.saddr.sa.sa_family, SOCK_STREAM, 0); + // Attemp to connect to TCP server. + int retries = 0; + while (retries < MAX_CONNECTION_RETRIES) { + n->logger->info("Attempting to connect to server: attempt={}...", retries + 1); + ret = connect(s->sd, (struct sockaddr *)&s->out.saddr, sizeof(s->in.saddr)); + if (ret == 0) { + s->tcp_connect = true; + break; + } else { + retries++; + if (retries < MAX_CONNECTION_RETRIES) { + sleep(RETRIES_DELAY); + } + } + } + if (ret < 0) + throw SystemError("Failed to conenct to TCP server"); + } else if (s->layer == SocketLayer::TCP_SERVER) { + listen(s->sd, 5); + // Accept client connection and get client socket descriptor. + s->clt_sd = accept(s->sd, nullptr, nullptr); + if (s->clt_sd < 0) { + throw SystemError("Failed to accept connection"); + } + s->tcp_connect = true; + } +} + int villas::node::socket_write(NodeCompat *n, struct Sample *const smps[], unsigned cnt) { auto *s = n->getData(); From b0d2b888e05b157a1511397705f48ec6f43c0da9 Mon Sep 17 00:00:00 2001 From: Jitpanu Maneeratpongsuk Date: Fri, 10 Jan 2025 15:24:38 +0000 Subject: [PATCH 04/10] feat: Add TCP to the OpenAPI spec of socket node type Signed-off-by: Jitpanu Maneeratpongsuk --- doc/openapi/components/schemas/config/nodes/socket.yaml | 2 ++ 1 file changed, 2 insertions(+) diff --git a/doc/openapi/components/schemas/config/nodes/socket.yaml b/doc/openapi/components/schemas/config/nodes/socket.yaml index 6330d3d89..9e69e86c8 100644 --- a/doc/openapi/components/schemas/config/nodes/socket.yaml +++ b/doc/openapi/components/schemas/config/nodes/socket.yaml @@ -14,6 +14,8 @@ allOf: - udp - ip - eth + - tcp-client + - tcp-server default: udp description: | Select the network layer which should be used for the socket. Please note that `eth` can only be used locally in a LAN as it contains no routing information for the internet. From ad8199e473ed680c7c9795a03bbe4a74f4b510e4 Mon Sep 17 00:00:00 2001 From: Jitpanu Maneeratpongsuk Date: Wed, 15 Jan 2025 13:13:07 +0000 Subject: [PATCH 05/10] feat: Extended integration test for socket TCP connection Signed-off-by: Jitpanu Maneeratpongsuk --- tests/integration/node-loopback-socket.sh | 82 +++++++++++++++++++++++ 1 file changed, 82 insertions(+) diff --git a/tests/integration/node-loopback-socket.sh b/tests/integration/node-loopback-socket.sh index 5e96cde5e..1d79c4e94 100755 --- a/tests/integration/node-loopback-socket.sh +++ b/tests/integration/node-loopback-socket.sh @@ -87,3 +87,85 @@ wait %% # Send / Receive data to node VILLAS_LOG_PREFIX="[compare] " \ villas compare input.dat output.dat + +sleep 1 + +cat > config.json < Date: Tue, 21 Jan 2025 16:12:49 +0000 Subject: [PATCH 06/10] fix: Uninitialize variable Signed-off-by: Jitpanu Maneeratpongsuk --- lib/nodes/socket.cpp | 2 ++ 1 file changed, 2 insertions(+) diff --git a/lib/nodes/socket.cpp b/lib/nodes/socket.cpp index 6050c1208..3018dd523 100644 --- a/lib/nodes/socket.cpp +++ b/lib/nodes/socket.cpp @@ -251,6 +251,8 @@ int villas::node::socket_start(NodeCompat *n) { if (s->layer != SocketLayer::TCP_CLIENT) ret = bind(s->sd, (struct sockaddr *)&s->in.saddr, addrlen); + else + ret = 0; if (ret < 0) throw SystemError("Failed to bind socket"); From da049577b08ff24086268f647a4ee8e8bb1e48d6 Mon Sep 17 00:00:00 2001 From: Jitpanu Maneeratpongsuk Date: Wed, 29 Jan 2025 12:07:42 +0000 Subject: [PATCH 07/10] fix: Add error checking fix: Naming and comment fix: Remove socket_tcp_connection from header file Signed-off-by: Jitpanu Maneeratpongsuk --- include/villas/nodes/socket.hpp | 4 +- lib/nodes/socket.cpp | 92 +++++++++++++++++++-------------- 2 files changed, 54 insertions(+), 42 deletions(-) diff --git a/include/villas/nodes/socket.hpp b/include/villas/nodes/socket.hpp index 1c78de648..b84de9909 100644 --- a/include/villas/nodes/socket.hpp +++ b/include/villas/nodes/socket.hpp @@ -24,7 +24,7 @@ struct Socket { int sd; // The socket descriptor int clt_sd; // TCP client socket descriptor int verify_source; // Verify the source address of incoming packets against socket::remote. - bool tcp_connect = false; // TCP connection status bit + bool tcp_connected = false; // TCP connection status bit enum SocketLayer layer; // The OSI / IP layer which should be used for this socket @@ -70,8 +70,6 @@ int socket_read(NodeCompat *n, struct Sample *const smps[], unsigned cnt); int socket_parse(NodeCompat *n, json_t *json); -void socket_tcp_connection(NodeCompat *n, Socket *s); - char *socket_print(NodeCompat *n); } // namespace node diff --git a/lib/nodes/socket.cpp b/lib/nodes/socket.cpp index 3018dd523..69ab742c6 100644 --- a/lib/nodes/socket.cpp +++ b/lib/nodes/socket.cpp @@ -339,11 +339,17 @@ int villas::node::socket_stop(NodeCompat *n) { if (s->sd >= 0) { // Close client socket descriptor. - if (s->layer == SocketLayer::TCP_SERVER) + if (s->layer == SocketLayer::TCP_SERVER) { close(s->clt_sd); + if (ret) + throw SystemError("Failed to close TCP client socket descriptor"); + } ret = close(s->sd); + //Reset socket descriptor + s->sd = -1; + if (ret) return ret; } @@ -354,6 +360,47 @@ int villas::node::socket_stop(NodeCompat *n) { return 0; } +static void socket_tcp_connection(NodeCompat *n, Socket *s) { + int ret; + if (s->layer == SocketLayer::TCP_CLIENT) { + if (s->sd >= 0) { + ret = close(s->sd); + if (ret < 0) + throw SystemError("Failed to close socket descriptor"); + } + s->sd = socket(s->in.saddr.sa.sa_family, SOCK_STREAM, 0); + if (s->sd < 0) + throw SystemError("Failed to create socket"); + // Attempt to connect to TCP server. + int retries = 0; + while (retries < MAX_CONNECTION_RETRIES) { + n->logger->info("Attempting to connect to TCP server: attempt={}...", retries + 1); + ret = connect(s->sd, (struct sockaddr *)&s->out.saddr, sizeof(s->in.saddr)); + if (ret == 0) { + s->tcp_connected = true; + break; + } else { + retries++; + if (retries < MAX_CONNECTION_RETRIES) { + sleep(RETRIES_DELAY); + } + } + } + if (ret < 0) + throw SystemError("Failed to conenct to TCP server"); + } else if (s->layer == SocketLayer::TCP_SERVER) { + ret = listen(s->sd, 5); + if (ret < 0) + throw SystemError("Failed to listen for TCP client connection"); + // Accept client connection and get client socket descriptor. + s->clt_sd = accept(s->sd, nullptr, nullptr); + if (s->clt_sd < 0) { + throw SystemError("Failed to accept TCP client connection"); + } + s->tcp_connected = true; + } +} + int villas::node::socket_read(NodeCompat *n, struct Sample *const smps[], unsigned cnt) { int ret; @@ -370,14 +417,14 @@ int villas::node::socket_read(NodeCompat *n, struct Sample *const smps[], if (s->layer == SocketLayer::TCP_CLIENT) { // Receive data from server. - if (!s->tcp_connect) - villas::node::socket_tcp_connection(n, s); + if (!s->tcp_connected) + socket_tcp_connection(n, s); bytes = recv(s->sd, s->in.buf, s->in.buflen, 0); } else if (s->layer == SocketLayer::TCP_SERVER) { // Receive data from client. - if (!s->tcp_connect) - villas::node::socket_tcp_connection(n, s); + if (!s->tcp_connected) + socket_tcp_connection(n, s); bytes = recv(s->clt_sd, s->in.buf, s->in.buflen, 0); } else { @@ -391,7 +438,7 @@ int villas::node::socket_read(NodeCompat *n, struct Sample *const smps[], throw SystemError("Failed recvfrom()"); } else if (bytes == 0) { if (s->layer == SocketLayer::TCP_CLIENT || s->layer == SocketLayer::TCP_SERVER) - s->tcp_connect = false; + s->tcp_connected = false; return 0; } @@ -436,39 +483,6 @@ int villas::node::socket_read(NodeCompat *n, struct Sample *const smps[], return ret; } -void villas::node::socket_tcp_connection(NodeCompat *n, Socket *s) { - int ret; - if (s->layer == SocketLayer::TCP_CLIENT) { - close(s->sd); - s->sd = socket(s->in.saddr.sa.sa_family, SOCK_STREAM, 0); - // Attemp to connect to TCP server. - int retries = 0; - while (retries < MAX_CONNECTION_RETRIES) { - n->logger->info("Attempting to connect to server: attempt={}...", retries + 1); - ret = connect(s->sd, (struct sockaddr *)&s->out.saddr, sizeof(s->in.saddr)); - if (ret == 0) { - s->tcp_connect = true; - break; - } else { - retries++; - if (retries < MAX_CONNECTION_RETRIES) { - sleep(RETRIES_DELAY); - } - } - } - if (ret < 0) - throw SystemError("Failed to conenct to TCP server"); - } else if (s->layer == SocketLayer::TCP_SERVER) { - listen(s->sd, 5); - // Accept client connection and get client socket descriptor. - s->clt_sd = accept(s->sd, nullptr, nullptr); - if (s->clt_sd < 0) { - throw SystemError("Failed to accept connection"); - } - s->tcp_connect = true; - } -} - int villas::node::socket_write(NodeCompat *n, struct Sample *const smps[], unsigned cnt) { auto *s = n->getData(); From 1053d9e72233186c8d998b5452f7a2cf7238166b Mon Sep 17 00:00:00 2001 From: Jitpanu Maneeratpongsuk Date: Wed, 29 Jan 2025 12:39:09 +0000 Subject: [PATCH 08/10] fix: Remove sleep from socket integration test Signed-off-by: Jitpanu Maneeratpongsuk --- tests/integration/node-loopback-socket.sh | 2 -- 1 file changed, 2 deletions(-) diff --git a/tests/integration/node-loopback-socket.sh b/tests/integration/node-loopback-socket.sh index 1d79c4e94..af4df8e7d 100755 --- a/tests/integration/node-loopback-socket.sh +++ b/tests/integration/node-loopback-socket.sh @@ -88,8 +88,6 @@ wait %% VILLAS_LOG_PREFIX="[compare] " \ villas compare input.dat output.dat -sleep 1 - cat > config.json < Date: Thu, 30 Jan 2025 12:28:09 +0000 Subject: [PATCH 09/10] fix: Comment format Signed-off-by: Jitpanu Maneeratpongsuk --- lib/nodes/socket.cpp | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/lib/nodes/socket.cpp b/lib/nodes/socket.cpp index 69ab742c6..09c356a83 100644 --- a/lib/nodes/socket.cpp +++ b/lib/nodes/socket.cpp @@ -347,7 +347,7 @@ int villas::node::socket_stop(NodeCompat *n) { ret = close(s->sd); - //Reset socket descriptor + // Reset socket descriptor. s->sd = -1; if (ret) From 8a1ac3fc99dd1180a4320f773ef8a1688aa009b5 Mon Sep 17 00:00:00 2001 From: Jitpanu Maneeratpongsuk Date: Thu, 30 Jan 2025 15:07:12 +0000 Subject: [PATCH 10/10] fix: Broken error checking Signed-off-by: Jitpanu Maneeratpongsuk --- lib/nodes/socket.cpp | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/lib/nodes/socket.cpp b/lib/nodes/socket.cpp index 09c356a83..092959ceb 100644 --- a/lib/nodes/socket.cpp +++ b/lib/nodes/socket.cpp @@ -340,7 +340,7 @@ int villas::node::socket_stop(NodeCompat *n) { if (s->sd >= 0) { // Close client socket descriptor. if (s->layer == SocketLayer::TCP_SERVER) { - close(s->clt_sd); + ret = close(s->clt_sd); if (ret) throw SystemError("Failed to close TCP client socket descriptor"); }