diff mbox series

[bpf-next,v2,2/2] selftests/bpf: Export send_recv_data helper

Message ID a8153ab2b82c8cd57aca2c6d44d5d327e8c7be92.1712547287.git.tanggeliang@kylinos.cn (mailing list archive)
State Changes Requested
Delegated to: BPF
Headers show
Series export send_byte and send_recv_data | expand

Checks

Context Check Description
bpf/vmtest-bpf-next-PR success PR summary
netdev/series_format success Posting correctly formatted
netdev/tree_selection success Clearly marked for bpf-next
netdev/ynl success Generated files up to date; no warnings/errors; no diff in generated;
netdev/fixes_present success Fixes tag not required for -next series
netdev/header_inline success No static functions without inline keyword in header files
netdev/build_32bit success Errors and warnings before: 8 this patch: 8
netdev/build_tools success Errors and warnings before: 0 this patch: 0
netdev/cc_maintainers success CCed 17 of 17 maintainers
netdev/build_clang success Errors and warnings before: 8 this patch: 8
netdev/verify_signedoff success Signed-off-by tag matches author and committer
netdev/deprecated_api success None detected
netdev/check_selftest success No net selftest shell script
netdev/verify_fixes success No Fixes tag
netdev/build_allmodconfig_warn success Errors and warnings before: 8 this patch: 8
netdev/checkpatch success total: 0 errors, 0 warnings, 0 checks, 197 lines checked
netdev/build_clang_rust success No Rust files in patch. Skipping build
netdev/kdoc success Errors and warnings before: 2 this patch: 2
netdev/source_inline success Was 0 now: 0
bpf/vmtest-bpf-next-VM_Test-0 success Logs for Lint
bpf/vmtest-bpf-next-VM_Test-1 success Logs for ShellCheck
bpf/vmtest-bpf-next-VM_Test-2 success Logs for Unittests
bpf/vmtest-bpf-next-VM_Test-3 success Logs for Validate matrix.py
bpf/vmtest-bpf-next-VM_Test-5 success Logs for aarch64-gcc / build-release
bpf/vmtest-bpf-next-VM_Test-4 success Logs for aarch64-gcc / build / build for aarch64 with gcc
bpf/vmtest-bpf-next-VM_Test-10 success Logs for aarch64-gcc / veristat
bpf/vmtest-bpf-next-VM_Test-12 success Logs for s390x-gcc / build-release
bpf/vmtest-bpf-next-VM_Test-9 success Logs for aarch64-gcc / test (test_verifier, false, 360) / test_verifier on aarch64 with gcc
bpf/vmtest-bpf-next-VM_Test-11 success Logs for s390x-gcc / build / build for s390x with gcc
bpf/vmtest-bpf-next-VM_Test-16 success Logs for s390x-gcc / test (test_verifier, false, 360) / test_verifier on s390x with gcc
bpf/vmtest-bpf-next-VM_Test-17 success Logs for s390x-gcc / veristat
bpf/vmtest-bpf-next-VM_Test-19 success Logs for x86_64-gcc / build / build for x86_64 with gcc
bpf/vmtest-bpf-next-VM_Test-18 success Logs for set-matrix
bpf/vmtest-bpf-next-VM_Test-20 success Logs for x86_64-gcc / build-release
bpf/vmtest-bpf-next-VM_Test-28 success Logs for x86_64-llvm-17 / build / build for x86_64 with llvm-17
bpf/vmtest-bpf-next-VM_Test-29 success Logs for x86_64-llvm-17 / build-release / build for x86_64 with llvm-17 and -O2 optimization
bpf/vmtest-bpf-next-VM_Test-30 success Logs for x86_64-llvm-17 / test (test_maps, false, 360) / test_maps on x86_64 with llvm-17
bpf/vmtest-bpf-next-VM_Test-33 success Logs for x86_64-llvm-17 / test (test_verifier, false, 360) / test_verifier on x86_64 with llvm-17
bpf/vmtest-bpf-next-VM_Test-34 success Logs for x86_64-llvm-17 / veristat
bpf/vmtest-bpf-next-VM_Test-35 success Logs for x86_64-llvm-18 / build / build for x86_64 with llvm-18
bpf/vmtest-bpf-next-VM_Test-36 success Logs for x86_64-llvm-18 / build-release / build for x86_64 with llvm-18 and -O2 optimization
bpf/vmtest-bpf-next-VM_Test-42 success Logs for x86_64-llvm-18 / veristat
bpf/vmtest-bpf-next-VM_Test-6 success Logs for aarch64-gcc / test (test_maps, false, 360) / test_maps on aarch64 with gcc
bpf/vmtest-bpf-next-VM_Test-8 success Logs for aarch64-gcc / test (test_progs_no_alu32, false, 360) / test_progs_no_alu32 on aarch64 with gcc
bpf/vmtest-bpf-next-VM_Test-7 success Logs for aarch64-gcc / test (test_progs, false, 360) / test_progs on aarch64 with gcc
bpf/vmtest-bpf-next-VM_Test-21 success Logs for x86_64-gcc / test (test_maps, false, 360) / test_maps on x86_64 with gcc
bpf/vmtest-bpf-next-VM_Test-24 success Logs for x86_64-gcc / test (test_progs_no_alu32_parallel, true, 30) / test_progs_no_alu32_parallel on x86_64 with gcc
bpf/vmtest-bpf-next-VM_Test-25 success Logs for x86_64-gcc / test (test_progs_parallel, true, 30) / test_progs_parallel on x86_64 with gcc
bpf/vmtest-bpf-next-VM_Test-26 success Logs for x86_64-gcc / test (test_verifier, false, 360) / test_verifier on x86_64 with gcc
bpf/vmtest-bpf-next-VM_Test-27 success Logs for x86_64-gcc / veristat / veristat on x86_64 with gcc
bpf/vmtest-bpf-next-VM_Test-31 success Logs for x86_64-llvm-17 / test (test_progs, false, 360) / test_progs on x86_64 with llvm-17
bpf/vmtest-bpf-next-VM_Test-32 success Logs for x86_64-llvm-17 / test (test_progs_no_alu32, false, 360) / test_progs_no_alu32 on x86_64 with llvm-17
bpf/vmtest-bpf-next-VM_Test-37 success Logs for x86_64-llvm-18 / test (test_maps, false, 360) / test_maps on x86_64 with llvm-18
bpf/vmtest-bpf-next-VM_Test-41 success Logs for x86_64-llvm-18 / test (test_verifier, false, 360) / test_verifier on x86_64 with llvm-18
bpf/vmtest-bpf-next-VM_Test-13 success Logs for s390x-gcc / test (test_maps, false, 360) / test_maps on s390x with gcc
bpf/vmtest-bpf-next-VM_Test-22 success Logs for x86_64-gcc / test (test_progs, false, 360) / test_progs on x86_64 with gcc
bpf/vmtest-bpf-next-VM_Test-23 success Logs for x86_64-gcc / test (test_progs_no_alu32, false, 360) / test_progs_no_alu32 on x86_64 with gcc
bpf/vmtest-bpf-next-VM_Test-38 success Logs for x86_64-llvm-18 / test (test_progs, false, 360) / test_progs on x86_64 with llvm-18
bpf/vmtest-bpf-next-VM_Test-39 success Logs for x86_64-llvm-18 / test (test_progs_cpuv4, false, 360) / test_progs_cpuv4 on x86_64 with llvm-18
bpf/vmtest-bpf-next-VM_Test-40 success Logs for x86_64-llvm-18 / test (test_progs_no_alu32, false, 360) / test_progs_no_alu32 on x86_64 with llvm-18
bpf/vmtest-bpf-next-VM_Test-15 success Logs for s390x-gcc / test (test_progs_no_alu32, false, 360) / test_progs_no_alu32 on s390x with gcc
bpf/vmtest-bpf-next-VM_Test-14 success Logs for s390x-gcc / test (test_progs, false, 360) / test_progs on s390x with gcc

Commit Message

Geliang Tang April 8, 2024, 3:45 a.m. UTC 
From: Geliang Tang <tanggeliang@kylinos.cn>

This patch extracts the code to send and receive data into a new
helper named send_recv_data() in network_helpers.c and export it
in network_helpers.h.

This helper will be used for MPTCP BPF selftests.

Signed-off-by: Geliang Tang <tanggeliang@kylinos.cn>
---
 tools/testing/selftests/bpf/network_helpers.c | 85 +++++++++++++++++++
 tools/testing/selftests/bpf/network_helpers.h |  1 +
 .../selftests/bpf/prog_tests/bpf_tcp_ca.c     | 81 +-----------------
 3 files changed, 87 insertions(+), 80 deletions(-)

Comments

Geliang Tang April 9, 2024, 3:51 a.m. UTC | #1 
Hi Martin,

On Mon, 2024-04-08 at 11:45 +0800, Geliang Tang wrote:
> From: Geliang Tang <tanggeliang@kylinos.cn>
> 
> This patch extracts the code to send and receive data into a new
> helper named send_recv_data() in network_helpers.c and export it
> in network_helpers.h.
> 
> This helper will be used for MPTCP BPF selftests.
> 
> Signed-off-by: Geliang Tang <tanggeliang@kylinos.cn>
> ---
>  tools/testing/selftests/bpf/network_helpers.c | 85
> +++++++++++++++++++
>  tools/testing/selftests/bpf/network_helpers.h |  1 +
>  .../selftests/bpf/prog_tests/bpf_tcp_ca.c     | 81 +----------------
> -
>  3 files changed, 87 insertions(+), 80 deletions(-)
> 
> diff --git a/tools/testing/selftests/bpf/network_helpers.c
> b/tools/testing/selftests/bpf/network_helpers.c
> index 04175e16195a..e17d19f88a36 100644
> --- a/tools/testing/selftests/bpf/network_helpers.c
> +++ b/tools/testing/selftests/bpf/network_helpers.c
> @@ -545,3 +545,88 @@ int set_hw_ring_size(char *ifname, struct
> ethtool_ringparam *ring_param)
>  	close(sockfd);
>  	return 0;
>  }
> +
> +struct send_recv_arg {
> +	int		fd;
> +	uint32_t	bytes;
> +	int		stop;
> +};
> +
> +static void *send_recv_server(void *arg)
> +{
> +	struct send_recv_arg *a = (struct send_recv_arg *)arg;
> +	ssize_t nr_sent = 0, bytes = 0;
> +	char batch[1500];
> +	int err = 0, fd;
> +
> +	fd = accept(a->fd, NULL, NULL);
> +	while (fd == -1) {
> +		if (errno == EINTR)
> +			continue;
> +		err = -errno;
> +		goto done;
> +	}
> +
> +	if (settimeo(fd, 0)) {
> +		err = -errno;
> +		goto done;
> +	}
> +
> +	while (bytes < a->bytes && !READ_ONCE(a->stop)) {
> +		nr_sent = send(fd, &batch,
> +			       MIN(a->bytes - bytes, sizeof(batch)),
> 0);
> +		if (nr_sent == -1 && errno == EINTR)
> +			continue;
> +		if (nr_sent == -1) {
> +			err = -errno;
> +			break;
> +		}
> +		bytes += nr_sent;
> +	}
> +
> +	ASSERT_EQ(bytes, a->bytes, "send");
> +
> +done:
> +	if (fd >= 0)
> +		close(fd);
> +	if (err) {
> +		WRITE_ONCE(a->stop, 1);
> +		return ERR_PTR(err);
> +	}
> +	return NULL;
> +}
> +
> +void send_recv_data(int lfd, int fd, uint32_t total_bytes)
> +{
> +	ssize_t nr_recv = 0, bytes = 0;
> +	struct send_recv_arg arg = {
> +		.fd	= lfd,
> +		.bytes	= total_bytes,
> +		.stop	= 0,
> +	};
> +	pthread_t srv_thread;
> +	void *thread_ret;
> +	char batch[1500];
> +	int err;
> +
> +	err = pthread_create(&srv_thread, NULL, send_recv_server,
> (void *)&arg);
> +	if (!ASSERT_OK(err, "pthread_create"))
> +		return;
> +
> +	/* recv total_bytes */
> +	while (bytes < total_bytes && !READ_ONCE(arg.stop)) {
> +		nr_recv = recv(fd, &batch,
> +			       MIN(total_bytes - bytes,
> sizeof(batch)), 0);
> +		if (nr_recv == -1 && errno == EINTR)
> +			continue;
> +		if (nr_recv == -1)
> +			break;
> +		bytes += nr_recv;
> +	}
> +
> +	ASSERT_EQ(bytes, total_bytes, "recv");

I think we should avoid using ASSERT_* in network_helpers.c, but I'm
not sure. What do you think?

Thanks,
-Geliang

> +
> +	WRITE_ONCE(arg.stop, 1);
> +	pthread_join(srv_thread, &thread_ret);
> +	ASSERT_OK(IS_ERR(thread_ret), "thread_ret");
> +}
> diff --git a/tools/testing/selftests/bpf/network_helpers.h
> b/tools/testing/selftests/bpf/network_helpers.h
> index 6457445cc6e2..5172f0b7bf6e 100644
> --- a/tools/testing/selftests/bpf/network_helpers.h
> +++ b/tools/testing/selftests/bpf/network_helpers.h
> @@ -76,6 +76,7 @@ struct nstoken;
>   */
>  struct nstoken *open_netns(const char *name);
>  void close_netns(struct nstoken *token);
> +void send_recv_data(int lfd, int fd, uint32_t total_bytes);
>  
>  static __u16 csum_fold(__u32 csum)
>  {
> diff --git a/tools/testing/selftests/bpf/prog_tests/bpf_tcp_ca.c
> b/tools/testing/selftests/bpf/prog_tests/bpf_tcp_ca.c
> index 64f172f02a9a..3f822100c2b3 100644
> --- a/tools/testing/selftests/bpf/prog_tests/bpf_tcp_ca.c
> +++ b/tools/testing/selftests/bpf/prog_tests/bpf_tcp_ca.c
> @@ -33,75 +33,15 @@ static int settcpca(int fd, const char *tcp_ca)
>  	return 0;
>  }
>  
> -struct send_recv_arg {
> -	int		fd;
> -	uint32_t	bytes;
> -	int		stop;
> -};
> -
> -static void *server(void *arg)
> -{
> -	struct send_recv_arg *a = (struct send_recv_arg *)arg;
> -	ssize_t nr_sent = 0, bytes = 0;
> -	char batch[1500];
> -	int err = 0, fd;
> -
> -	fd = accept(a->fd, NULL, NULL);
> -	while (fd == -1) {
> -		if (errno == EINTR)
> -			continue;
> -		err = -errno;
> -		goto done;
> -	}
> -
> -	if (settimeo(fd, 0)) {
> -		err = -errno;
> -		goto done;
> -	}
> -
> -	while (bytes < a->bytes && !READ_ONCE(a->stop)) {
> -		nr_sent = send(fd, &batch,
> -			       MIN(a->bytes - bytes, sizeof(batch)),
> 0);
> -		if (nr_sent == -1 && errno == EINTR)
> -			continue;
> -		if (nr_sent == -1) {
> -			err = -errno;
> -			break;
> -		}
> -		bytes += nr_sent;
> -	}
> -
> -	ASSERT_EQ(bytes, a->bytes, "send");
> -
> -done:
> -	if (fd >= 0)
> -		close(fd);
> -	if (err) {
> -		WRITE_ONCE(a->stop, 1);
> -		return ERR_PTR(err);
> -	}
> -	return NULL;
> -}
> -
>  static void do_test(const char *tcp_ca, const struct bpf_map
> *sk_stg_map)
>  {
> -	ssize_t nr_recv = 0, bytes = 0;
> -	struct send_recv_arg arg = {
> -		.bytes	= total_bytes,
> -		.stop	= 0,
> -	};
>  	int lfd = -1, fd = -1;
> -	pthread_t srv_thread;
> -	void *thread_ret;
> -	char batch[1500];
>  	int err;
>  
>  	lfd = start_server(AF_INET6, SOCK_STREAM, NULL, 0, 0);
>  	if (!ASSERT_NEQ(lfd, -1, "socket"))
>  		return;
>  
> -	arg.fd = lfd;
> -
>  	fd = socket(AF_INET6, SOCK_STREAM, 0);
>  	if (!ASSERT_NEQ(fd, -1, "socket")) {
>  		close(lfd);
> @@ -133,26 +73,7 @@ static void do_test(const char *tcp_ca, const
> struct bpf_map *sk_stg_map)
>  			goto done;
>  	}
>  
> -	err = pthread_create(&srv_thread, NULL, server, (void
> *)&arg);
> -	if (!ASSERT_OK(err, "pthread_create"))
> -		goto done;
> -
> -	/* recv total_bytes */
> -	while (bytes < total_bytes && !READ_ONCE(arg.stop)) {
> -		nr_recv = recv(fd, &batch,
> -			       MIN(total_bytes - bytes,
> sizeof(batch)), 0);
> -		if (nr_recv == -1 && errno == EINTR)
> -			continue;
> -		if (nr_recv == -1)
> -			break;
> -		bytes += nr_recv;
> -	}
> -
> -	ASSERT_EQ(bytes, total_bytes, "recv");
> -
> -	WRITE_ONCE(arg.stop, 1);
> -	pthread_join(srv_thread, &thread_ret);
> -	ASSERT_OK(IS_ERR(thread_ret), "thread_ret");
> +	send_recv_data(lfd, fd, total_bytes);
>  
>  done:
>  	close(lfd);
Martin KaFai Lau April 9, 2024, 4:52 a.m. UTC | #2 
On 4/8/24 8:51 PM, Geliang Tang wrote:
>> +static void *send_recv_server(void *arg)
>> +{
>> +	struct send_recv_arg *a = (struct send_recv_arg *)arg;
>> +	ssize_t nr_sent = 0, bytes = 0;
>> +	char batch[1500];
>> +	int err = 0, fd;
>> +
>> +	fd = accept(a->fd, NULL, NULL);
>> +	while (fd == -1) {
>> +		if (errno == EINTR)
>> +			continue;
>> +		err = -errno;
>> +		goto done;
>> +	}
>> +
>> +	if (settimeo(fd, 0)) {
>> +		err = -errno;
>> +		goto done;
>> +	}
>> +
>> +	while (bytes < a->bytes && !READ_ONCE(a->stop)) {
>> +		nr_sent = send(fd, &batch,
>> +			       MIN(a->bytes - bytes, sizeof(batch)),
>> 0);
>> +		if (nr_sent == -1 && errno == EINTR)
>> +			continue;
>> +		if (nr_sent == -1) {
>> +			err = -errno;
>> +			break;
>> +		}
>> +		bytes += nr_sent;
>> +	}
>> +
>> +	ASSERT_EQ(bytes, a->bytes, "send");
>> +
>> +done:
>> +	if (fd >= 0)
>> +		close(fd);
>> +	if (err) {
>> +		WRITE_ONCE(a->stop, 1);
>> +		return ERR_PTR(err);
>> +	}
>> +	return NULL;
>> +}
>> +
>> +void send_recv_data(int lfd, int fd, uint32_t total_bytes)
>> +{
>> +	ssize_t nr_recv = 0, bytes = 0;
>> +	struct send_recv_arg arg = {
>> +		.fd	= lfd,
>> +		.bytes	= total_bytes,
>> +		.stop	= 0,
>> +	};
>> +	pthread_t srv_thread;
>> +	void *thread_ret;
>> +	char batch[1500];
>> +	int err;
>> +
>> +	err = pthread_create(&srv_thread, NULL, send_recv_server,
>> (void *)&arg);
>> +	if (!ASSERT_OK(err, "pthread_create"))
>> +		return;
>> +
>> +	/* recv total_bytes */
>> +	while (bytes < total_bytes && !READ_ONCE(arg.stop)) {
>> +		nr_recv = recv(fd, &batch,
>> +			       MIN(total_bytes - bytes,
>> sizeof(batch)), 0);
>> +		if (nr_recv == -1 && errno == EINTR)
>> +			continue;
>> +		if (nr_recv == -1)
>> +			break;
>> +		bytes += nr_recv;
>> +	}
>> +
>> +	ASSERT_EQ(bytes, total_bytes, "recv");
> I think we should avoid using ASSERT_* in network_helpers.c, but I'm
> not sure. What do you think?

There is log_err which is used by other helpers in network_helpers.c. May be use 
log_err instead and return int instead of void here. The caller can decide if it 
expects error or not and uses ASSERT accordingly.

pw-bot: cr
diff mbox series

Patch

diff --git a/tools/testing/selftests/bpf/network_helpers.c b/tools/testing/selftests/bpf/network_helpers.c
index 04175e16195a..e17d19f88a36 100644
--- a/tools/testing/selftests/bpf/network_helpers.c
+++ b/tools/testing/selftests/bpf/network_helpers.c
@@ -545,3 +545,88 @@  int set_hw_ring_size(char *ifname, struct ethtool_ringparam *ring_param)
 	close(sockfd);
 	return 0;
 }
+
+struct send_recv_arg {
+	int		fd;
+	uint32_t	bytes;
+	int		stop;
+};
+
+static void *send_recv_server(void *arg)
+{
+	struct send_recv_arg *a = (struct send_recv_arg *)arg;
+	ssize_t nr_sent = 0, bytes = 0;
+	char batch[1500];
+	int err = 0, fd;
+
+	fd = accept(a->fd, NULL, NULL);
+	while (fd == -1) {
+		if (errno == EINTR)
+			continue;
+		err = -errno;
+		goto done;
+	}
+
+	if (settimeo(fd, 0)) {
+		err = -errno;
+		goto done;
+	}
+
+	while (bytes < a->bytes && !READ_ONCE(a->stop)) {
+		nr_sent = send(fd, &batch,
+			       MIN(a->bytes - bytes, sizeof(batch)), 0);
+		if (nr_sent == -1 && errno == EINTR)
+			continue;
+		if (nr_sent == -1) {
+			err = -errno;
+			break;
+		}
+		bytes += nr_sent;
+	}
+
+	ASSERT_EQ(bytes, a->bytes, "send");
+
+done:
+	if (fd >= 0)
+		close(fd);
+	if (err) {
+		WRITE_ONCE(a->stop, 1);
+		return ERR_PTR(err);
+	}
+	return NULL;
+}
+
+void send_recv_data(int lfd, int fd, uint32_t total_bytes)
+{
+	ssize_t nr_recv = 0, bytes = 0;
+	struct send_recv_arg arg = {
+		.fd	= lfd,
+		.bytes	= total_bytes,
+		.stop	= 0,
+	};
+	pthread_t srv_thread;
+	void *thread_ret;
+	char batch[1500];
+	int err;
+
+	err = pthread_create(&srv_thread, NULL, send_recv_server, (void *)&arg);
+	if (!ASSERT_OK(err, "pthread_create"))
+		return;
+
+	/* recv total_bytes */
+	while (bytes < total_bytes && !READ_ONCE(arg.stop)) {
+		nr_recv = recv(fd, &batch,
+			       MIN(total_bytes - bytes, sizeof(batch)), 0);
+		if (nr_recv == -1 && errno == EINTR)
+			continue;
+		if (nr_recv == -1)
+			break;
+		bytes += nr_recv;
+	}
+
+	ASSERT_EQ(bytes, total_bytes, "recv");
+
+	WRITE_ONCE(arg.stop, 1);
+	pthread_join(srv_thread, &thread_ret);
+	ASSERT_OK(IS_ERR(thread_ret), "thread_ret");
+}
diff --git a/tools/testing/selftests/bpf/network_helpers.h b/tools/testing/selftests/bpf/network_helpers.h
index 6457445cc6e2..5172f0b7bf6e 100644
--- a/tools/testing/selftests/bpf/network_helpers.h
+++ b/tools/testing/selftests/bpf/network_helpers.h
@@ -76,6 +76,7 @@  struct nstoken;
  */
 struct nstoken *open_netns(const char *name);
 void close_netns(struct nstoken *token);
+void send_recv_data(int lfd, int fd, uint32_t total_bytes);
 
 static __u16 csum_fold(__u32 csum)
 {
diff --git a/tools/testing/selftests/bpf/prog_tests/bpf_tcp_ca.c b/tools/testing/selftests/bpf/prog_tests/bpf_tcp_ca.c
index 64f172f02a9a..3f822100c2b3 100644
--- a/tools/testing/selftests/bpf/prog_tests/bpf_tcp_ca.c
+++ b/tools/testing/selftests/bpf/prog_tests/bpf_tcp_ca.c
@@ -33,75 +33,15 @@  static int settcpca(int fd, const char *tcp_ca)
 	return 0;
 }
 
-struct send_recv_arg {
-	int		fd;
-	uint32_t	bytes;
-	int		stop;
-};
-
-static void *server(void *arg)
-{
-	struct send_recv_arg *a = (struct send_recv_arg *)arg;
-	ssize_t nr_sent = 0, bytes = 0;
-	char batch[1500];
-	int err = 0, fd;
-
-	fd = accept(a->fd, NULL, NULL);
-	while (fd == -1) {
-		if (errno == EINTR)
-			continue;
-		err = -errno;
-		goto done;
-	}
-
-	if (settimeo(fd, 0)) {
-		err = -errno;
-		goto done;
-	}
-
-	while (bytes < a->bytes && !READ_ONCE(a->stop)) {
-		nr_sent = send(fd, &batch,
-			       MIN(a->bytes - bytes, sizeof(batch)), 0);
-		if (nr_sent == -1 && errno == EINTR)
-			continue;
-		if (nr_sent == -1) {
-			err = -errno;
-			break;
-		}
-		bytes += nr_sent;
-	}
-
-	ASSERT_EQ(bytes, a->bytes, "send");
-
-done:
-	if (fd >= 0)
-		close(fd);
-	if (err) {
-		WRITE_ONCE(a->stop, 1);
-		return ERR_PTR(err);
-	}
-	return NULL;
-}
-
 static void do_test(const char *tcp_ca, const struct bpf_map *sk_stg_map)
 {
-	ssize_t nr_recv = 0, bytes = 0;
-	struct send_recv_arg arg = {
-		.bytes	= total_bytes,
-		.stop	= 0,
-	};
 	int lfd = -1, fd = -1;
-	pthread_t srv_thread;
-	void *thread_ret;
-	char batch[1500];
 	int err;
 
 	lfd = start_server(AF_INET6, SOCK_STREAM, NULL, 0, 0);
 	if (!ASSERT_NEQ(lfd, -1, "socket"))
 		return;
 
-	arg.fd = lfd;
-
 	fd = socket(AF_INET6, SOCK_STREAM, 0);
 	if (!ASSERT_NEQ(fd, -1, "socket")) {
 		close(lfd);
@@ -133,26 +73,7 @@  static void do_test(const char *tcp_ca, const struct bpf_map *sk_stg_map)
 			goto done;
 	}
 
-	err = pthread_create(&srv_thread, NULL, server, (void *)&arg);
-	if (!ASSERT_OK(err, "pthread_create"))
-		goto done;
-
-	/* recv total_bytes */
-	while (bytes < total_bytes && !READ_ONCE(arg.stop)) {
-		nr_recv = recv(fd, &batch,
-			       MIN(total_bytes - bytes, sizeof(batch)), 0);
-		if (nr_recv == -1 && errno == EINTR)
-			continue;
-		if (nr_recv == -1)
-			break;
-		bytes += nr_recv;
-	}
-
-	ASSERT_EQ(bytes, total_bytes, "recv");
-
-	WRITE_ONCE(arg.stop, 1);
-	pthread_join(srv_thread, &thread_ret);
-	ASSERT_OK(IS_ERR(thread_ret), "thread_ret");
+	send_recv_data(lfd, fd, total_bytes);
 
 done:
 	close(lfd);