net: zperf: Implement asynchronous upload API

Add an API which allows to perform UDP/TCP upload operations
asychronously.

Signed-off-by: Robert Lubos <robert.lubos@nordicsemi.no>

include/zephyr/net/zperf.h

@@ -23,6 +23,12 @@
 extern "C" {
 #endif
 
+enum zperf_status {
+	ZPERF_SESSION_STARTED,
+	ZPERF_SESSION_FINISHED,
+	ZPERF_SESSION_ERROR
+} __packed;
+
 struct zperf_upload_params {
 	struct sockaddr peer_addr;
 	uint32_t duration_ms;
@@ -46,6 +52,17 @@ struct zperf_results {
 	uint32_t nb_packets_errors;
 };
 
+/**
+ * @brief Zperf callback function used for asynchronous operations.
+ *
+ * @param status Session status.
+ * @param result Session results. May be NULL for certain events.
+ * @param user_data A pointer to the user provided data.
+ */
+typedef void (*zperf_callback)(enum zperf_status status,
+			       struct zperf_results *result,
+			       void *user_data);
+
 /**
  * @brief Synchronous UDP upload operation. The function blocks until the upload
  *        is complete.
@@ -70,6 +87,36 @@ int zperf_udp_upload(const struct zperf_upload_params *param,
 int zperf_tcp_upload(const struct zperf_upload_params *param,
 		     struct zperf_results *result);
 
+/**
+ * @brief Asynchronous UDP upload operation.
+ *
+ * @note Only one asynchronous upload can be performed at a time.
+ *
+ * @param param Upload parameters.
+ * @param callback Session results callback.
+ * @param user_data A pointer to the user data to be provided with the callback.
+ *
+ * @return 0 if session was scheduled successfully, a negative error code
+ *         otherwise.
+ */
+int zperf_udp_upload_async(const struct zperf_upload_params *param,
+			   zperf_callback callback, void *user_data);
+
+/**
+ * @brief Asynchronous TCP upload operation.
+ *
+ * @note Only one asynchronous upload can be performed at a time.
+ *
+ * @param param Upload parameters.
+ * @param callback Session results callback.
+ * @param user_data A pointer to the user data to be provided with the callback.
+ *
+ * @return 0 if session was scheduled successfully, a negative error code
+ *         otherwise.
+ */
+int zperf_tcp_upload_async(const struct zperf_upload_params *param,
+			   zperf_callback callback, void *user_data);
+
 #ifdef __cplusplus
 }
 #endif

subsys/net/lib/zperf/zperf_common.c

@@ -7,8 +7,17 @@
 #include <zephyr/logging/log.h>
 #include <zephyr/net/socket.h>
 
+#include "zperf_internal.h"
+
 LOG_MODULE_DECLARE(net_zperf, CONFIG_NET_ZPERF_LOG_LEVEL);
 
+#define ZPERF_WORK_Q_THREAD_PRIORITY K_LOWEST_APPLICATION_THREAD_PRIO
+#define ZPERF_WORK_Q_STACK_SIZE 2048
+
+K_THREAD_STACK_DEFINE(zperf_work_q_stack, ZPERF_WORK_Q_STACK_SIZE);
+
+static struct k_work_q zperf_work_q;
+
 int zperf_prepare_upload_sock(const struct sockaddr *peer_addr, int tos,
 			      int proto)
 {
@@ -87,3 +96,26 @@ uint32_t zperf_packet_duration(uint32_t packet_size, uint32_t rate_in_kbps)
 	return (uint32_t)(((uint64_t)packet_size * 8U * USEC_PER_SEC) /
 			  (rate_in_kbps * 1024U));
 }
+
+void zperf_async_work_submit(struct k_work *work)
+{
+	k_work_submit_to_queue(&zperf_work_q, work);
+}
+
+static int zperf_init(const struct device *unused)
+{
+	ARG_UNUSED(unused);
+
+	k_work_queue_init(&zperf_work_q);
+	k_work_queue_start(&zperf_work_q, zperf_work_q_stack,
+			   K_THREAD_STACK_SIZEOF(zperf_work_q_stack),
+			   ZPERF_WORK_Q_THREAD_PRIORITY, NULL);
+	k_thread_name_set(&zperf_work_q.thread, "zperf_work_q");
+
+	zperf_udp_uploader_init();
+	zperf_tcp_uploader_init();
+
+	return 0;
+}
+
+SYS_INIT(zperf_init, APPLICATION, CONFIG_KERNEL_INIT_PRIORITY_DEFAULT);

subsys/net/lib/zperf/zperf_internal.h

@@ -69,6 +69,13 @@ struct zperf_server_hdr {
 	int32_t jitter2;
 };
 
+struct zperf_async_upload_context {
+	struct k_work work;
+	struct zperf_upload_params param;
+	zperf_callback callback;
+	void *user_data;
+};
+
 static inline uint32_t time_delta(uint32_t ts, uint32_t t)
 {
 	return (t >= ts) ? (t - ts) : (ULONG_MAX - ts + t);
@@ -99,4 +106,8 @@ int zperf_prepare_upload_sock(const struct sockaddr *peer_addr, int tos,
 
 uint32_t zperf_packet_duration(uint32_t packet_size, uint32_t rate_in_kbps);
 
+void zperf_async_work_submit(struct k_work *work);
+void zperf_udp_uploader_init(void);
+void zperf_tcp_uploader_init(void);
+
 #endif /* __ZPERF_INTERNAL_H */

subsys/net/lib/zperf/zperf_tcp_uploader.c

@@ -18,6 +18,8 @@ LOG_MODULE_DECLARE(net_zperf, CONFIG_NET_ZPERF_LOG_LEVEL);
 
 static char sample_packet[PACKET_SIZE_MAX];
 
+static struct zperf_async_upload_context tcp_async_upload_ctx;
+
 static int tcp_upload(int sock,
 		      unsigned int duration_in_ms,
 		      unsigned int packet_size,
@@ -128,3 +130,48 @@ int zperf_tcp_upload(const struct zperf_upload_params *param,
 
 	return ret;
 }
+
+static void tcp_upload_async_work(struct k_work *work)
+{
+	struct zperf_async_upload_context *upload_ctx =
+		CONTAINER_OF(work, struct zperf_async_upload_context, work);
+	struct zperf_results result;
+	int ret;
+
+	upload_ctx->callback(ZPERF_SESSION_STARTED, NULL,
+			     upload_ctx->user_data);
+
+	ret = zperf_tcp_upload(&upload_ctx->param, &result);
+	if (ret < 0) {
+		upload_ctx->callback(ZPERF_SESSION_ERROR, NULL,
+				     upload_ctx->user_data);
+	} else {
+		upload_ctx->callback(ZPERF_SESSION_FINISHED, &result,
+				     upload_ctx->user_data);
+	}
+}
+
+int zperf_tcp_upload_async(const struct zperf_upload_params *param,
+			   zperf_callback callback, void *user_data)
+{
+	if (param == NULL || callback == NULL) {
+		return -EINVAL;
+	}
+
+	if (k_work_is_pending(&tcp_async_upload_ctx.work)) {
+		return -EBUSY;
+	}
+
+	memcpy(&tcp_async_upload_ctx.param, param, sizeof(*param));
+	tcp_async_upload_ctx.callback = callback;
+	tcp_async_upload_ctx.user_data = user_data;
+
+	zperf_async_work_submit(&tcp_async_upload_ctx.work);
+
+	return 0;
+}
+
+void zperf_tcp_uploader_init(void)
+{
+	k_work_init(&tcp_async_upload_ctx.work, tcp_upload_async_work);
+}

subsys/net/lib/zperf/zperf_udp_uploader.c

@@ -18,6 +18,8 @@ static uint8_t sample_packet[sizeof(struct zperf_udp_datagram) +
 			     sizeof(struct zperf_client_hdr_v1) +
 			     PACKET_SIZE_MAX];
 
+static struct zperf_async_upload_context udp_async_upload_ctx;
+
 static inline void zperf_upload_decode_stat(const uint8_t *data,
 					    size_t datalen,
 					    struct zperf_results *results)
@@ -301,3 +303,48 @@ int zperf_udp_upload(const struct zperf_upload_params *param,
 
 	return ret;
 }
+
+static void udp_upload_async_work(struct k_work *work)
+{
+	struct zperf_async_upload_context *upload_ctx =
+		&udp_async_upload_ctx;
+	struct zperf_results result;
+	int ret;
+
+	upload_ctx->callback(ZPERF_SESSION_STARTED, NULL,
+			     upload_ctx->user_data);
+
+	ret = zperf_udp_upload(&upload_ctx->param, &result);
+	if (ret < 0) {
+		upload_ctx->callback(ZPERF_SESSION_ERROR, NULL,
+				     upload_ctx->user_data);
+	} else {
+		upload_ctx->callback(ZPERF_SESSION_FINISHED, &result,
+				     upload_ctx->user_data);
+	}
+}
+
+int zperf_udp_upload_async(const struct zperf_upload_params *param,
+			   zperf_callback callback, void *user_data)
+{
+	if (param == NULL || callback == NULL) {
+		return -EINVAL;
+	}
+
+	if (k_work_is_pending(&udp_async_upload_ctx.work)) {
+		return -EBUSY;
+	}
+
+	memcpy(&udp_async_upload_ctx.param, param, sizeof(*param));
+	udp_async_upload_ctx.callback = callback;
+	udp_async_upload_ctx.user_data = user_data;
+
+	zperf_async_work_submit(&udp_async_upload_ctx.work);
+
+	return 0;
+}
+
+void zperf_udp_uploader_init(void)
+{
+	k_work_init(&udp_async_upload_ctx.work, udp_upload_async_work);
+}