[PATCH bpf-next v2 0/2] export send_byte and send_recv_data

[PATCH bpf-next v2 0/2] export send_byte and send_recv_data

[Geliang Tang]

From: Geliang Tang <tanggeliang@kylinos.cn>

Address Martin's comments for v1 (thanks.)
 - drop patch 1, "export send_byte helper".
 - drop "WRITE_ONCE(arg.stop, 0)".
 - rebased.

send_recv_data will be re-used in MPTCP bpf tests, but not included
in this set because it depends on other patches that have not been
in the bpf-next yet. It will be sent as another set soon.

Geliang Tang (2):
  selftests/bpf: Add struct send_recv_arg
  selftests/bpf: Export send_recv_data helper

 tools/testing/selftests/bpf/network_helpers.c | 85 +++++++++++++++++++
 tools/testing/selftests/bpf/network_helpers.h |  1 +
 .../selftests/bpf/prog_tests/bpf_tcp_ca.c     | 71 +---------------
 3 files changed, 87 insertions(+), 70 deletions(-)

-- 
2.40.1

[PATCH bpf-next v2 1/2] selftests/bpf: Add struct send_recv_arg

[Geliang Tang]

From: Geliang Tang <tanggeliang@kylinos.cn>

Avoid setting total_bytes and stop as global variables, this patch adds
a new struct named send_recv_arg to pass arguments between threads. Put
these two variables together with fd into this struct and pass it to
server thread, so that server thread can access these two variables without
setting them as global ones.

Signed-off-by: Geliang Tang <tanggeliang@kylinos.cn>
---
 .../selftests/bpf/prog_tests/bpf_tcp_ca.c     | 34 ++++++++++++-------
 1 file changed, 22 insertions(+), 12 deletions(-)

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 077b107130f6..64f172f02a9a 100644
--- a/tools/testing/selftests/bpf/prog_tests/bpf_tcp_ca.c
+++ b/tools/testing/selftests/bpf/prog_tests/bpf_tcp_ca.c
@@ -21,7 +21,6 @@
 
 static const unsigned int total_bytes = 10 * 1024 * 1024;
 static int expected_stg = 0xeB9F;
-static int stop;
 
 static int settcpca(int fd, const char *tcp_ca)
 {
@@ -34,13 +33,20 @@ 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)
 {
-	int lfd = (int)(long)arg, err = 0, fd;
+	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(lfd, NULL, NULL);
+	fd = accept(a->fd, NULL, NULL);
 	while (fd == -1) {
 		if (errno == EINTR)
 			continue;
@@ -53,9 +59,9 @@ static void *server(void *arg)
 		goto done;
 	}
 
-	while (bytes < total_bytes && !READ_ONCE(stop)) {
+	while (bytes < a->bytes && !READ_ONCE(a->stop)) {
 		nr_sent = send(fd, &batch,
-			       MIN(total_bytes - bytes, sizeof(batch)), 0);
+			       MIN(a->bytes - bytes, sizeof(batch)), 0);
 		if (nr_sent == -1 && errno == EINTR)
 			continue;
 		if (nr_sent == -1) {
@@ -65,13 +71,13 @@ static void *server(void *arg)
 		bytes += nr_sent;
 	}
 
-	ASSERT_EQ(bytes, total_bytes, "send");
+	ASSERT_EQ(bytes, a->bytes, "send");
 
 done:
 	if (fd >= 0)
 		close(fd);
 	if (err) {
-		WRITE_ONCE(stop, 1);
+		WRITE_ONCE(a->stop, 1);
 		return ERR_PTR(err);
 	}
 	return NULL;
@@ -80,18 +86,22 @@ static void *server(void *arg)
 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;
 
-	WRITE_ONCE(stop, 0);
-
 	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);
@@ -123,12 +133,12 @@ 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 *)(long)lfd);
+	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(stop)) {
+	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)
@@ -140,7 +150,7 @@ static void do_test(const char *tcp_ca, const struct bpf_map *sk_stg_map)
 
 	ASSERT_EQ(bytes, total_bytes, "recv");
 
-	WRITE_ONCE(stop, 1);
+	WRITE_ONCE(arg.stop, 1);
 	pthread_join(srv_thread, &thread_ret);
 	ASSERT_OK(IS_ERR(thread_ret), "thread_ret");
 
-- 
2.40.1

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

[Geliang Tang]

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");
+
+	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);
-- 
2.40.1

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

[Geliang Tang]

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);

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

[Martin KaFai Lau]

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