net: zperf: Add -w option to delay upload jobs

Use -w option to delay the startup of the upload job.
Then when ready, do "zperf jobs start" to launch all upload
sessions at the same time.

Signed-off-by: Jukka Rissanen <[email protected]>

Author: jukkar

doc/connectivity/networking/api/zperf.rst

@@ -104,7 +104,12 @@ If :kconfig:option:`CONFIG_ZPERF_SESSION_PER_THREAD` option is set, then
 multiple upload sessions can be done at the same time if user supplies ``-a``
 option when starting the upload. Each session will have their own work queue
 to run the test. The session test results can be viewed also after the tests
-have finished.
+have finished. The sessions can be started with ``-w`` option which then
+lets the worker threads to wait a start signal so that all the threads can
+be started at the same time. This will prevent the case where the zperf shell
+cannot run because it is running in lower priority than the already started
+session thread. If you have only one upload session, then the ``-w`` is not
+really needed.
 
 Following zperf shell commands are available for session management:
 
@@ -115,6 +120,7 @@ Following zperf shell commands are available for session management:
    "``jobs``", "Show currently active or finished sessions"
    "``jobs all``", "Show statistics of finished sessions"
    "``jobs clear``", "Clear finished session statistics"
+   "``jobs start``", "Start all the waiting sessions"
 
 Example:
 
@@ -172,3 +178,61 @@ Example:
    uart:~$ zperf jobs
    No active upload sessions
    No finished sessions found
+
+The ``-w`` option can be used like this to delay the startup of the jobs.
+
+.. code-block:: console
+
+   uart:~$ zperf tcp upload -a -t 6 -w 192.0.2.2 5001 10 1K
+   Remote port is 5001
+   Connecting to 192.0.2.2
+   Duration:       10.00 s
+   Packet size:    1000 bytes
+   Rate:           10 kbps
+   Waiting "zperf jobs start" command.
+   [01:06:51.392,288] <inf> net_zperf: [0] TCP waiting for start
+
+   uart:~$ zperf udp upload -a -t 6 -w 192.0.2.2 5001 10 1K 10M
+   Remote port is 5001
+   Connecting to 192.0.2.2
+   Duration:       10.00 s
+   Packet size:    1000 bytes
+   Rate:           10000 kbps
+   Waiting "zperf jobs start" command.
+   Rate:           10.00 Mbps
+   Packet duration 781 us
+   [01:06:58.064,552] <inf> net_zperf: [0] UDP waiting for start
+
+   uart:~$ zperf jobs start
+   -
+   Upload completed!
+   -
+   Upload completed!
+
+   # Note that the output may be garbled as two threads printed
+   # output at the same time. Just print out the fresh listing
+   # like this.
+
+   uart:~$ zperf jobs all
+   -
+   Upload completed!
+   Statistics:             server  (client)
+   Duration:               9.99 s  (10.00 s)
+   Num packets:            11429   (11429)
+   Num packets out order:  0
+   Num packets lost:       0
+   Jitter:                 164 us
+   Rate:                   9.14 Mbps       (9.14 Mbps)
+   Thread priority:        6
+   Protocol:               UDP
+   Session id:             0
+   -
+   Upload completed!
+   Duration:               10.00 s
+   Num packets:            15487
+   Num errors:             0 (retry or fail)
+   Rate:                   12.38 Mbps
+   Thread priority:        6
+   Protocol:               TCP
+   Session id:             0
+   Total 2 sessions done

include/zephyr/net/zperf.h

@@ -47,6 +47,7 @@ struct zperf_upload_params {
 		int priority;
 #ifdef CONFIG_ZPERF_SESSION_PER_THREAD
 		int thread_priority;
+		bool wait_for_start;
 #endif
 		uint32_t report_interval_ms;
 	} options;

subsys/net/lib/zperf/Kconfig

@@ -24,6 +24,7 @@ config NET_ZPERF_LEGACY_HEADER_COMPAT
 
 config ZPERF_SESSION_PER_THREAD
 	bool "Run each session in a separate thread"
+	select EVENTS
 	help
 	  Each session is started in its own thread. This means
 	  that the system will use more memory because multiple

subsys/net/lib/zperf/zperf_common.c

@@ -47,11 +47,7 @@ struct sockaddr_in *zperf_get_sin(void)
 	      K_LOWEST_APPLICATION_THREAD_PRIO)
 
 #if defined(CONFIG_ZPERF_SESSION_PER_THREAD)
-struct zperf_work {
-	struct k_work_q *queue;
-	struct z_thread_stack_element *stack;
-	size_t stack_size;
-};
+static K_EVENT_DEFINE(start_event);
 
 #define CREATE_WORK_Q(i, _)					       \
 	static struct k_work_q zperf_work_q_##i;		       \
@@ -80,7 +76,7 @@ static struct zperf_work zperf_work_q[] = {
 	LISTIFY(MAX_SESSION_COUNT, SET_WORK_Q, (,), _)
 };
 
-struct k_work_q *get_queue(enum session_proto proto, int session_id)
+struct zperf_work *get_queue(enum session_proto proto, int session_id)
 {
 	if (session_id < 0 || session_id >= CONFIG_NET_ZPERF_MAX_SESSIONS) {
 		return NULL;
@@ -95,9 +91,14 @@ struct k_work_q *get_queue(enum session_proto proto, int session_id)
 		proto * SESSION_INDEX + session_id,
 		session_id);
 
-	return zperf_work_q[proto * SESSION_INDEX + session_id].queue;
+
+	return &zperf_work_q[proto * SESSION_INDEX + session_id];
 }
 
+void start_jobs(void)
+{
+	k_event_set(&start_event, START_EVENT);
+}
 #else /* CONFIG_ZPERF_SESSION_PER_THREAD */
 
 K_THREAD_STACK_DEFINE(zperf_work_q_stack, CONFIG_ZPERF_WORK_Q_STACK_SIZE);
@@ -298,6 +299,8 @@ static int zperf_init(void)
 			.no_yield = false,
 		};
 
+		zperf_work_q[i].start_event = &start_event;
+
 #define MAX_NAME_LEN sizeof("zperf_work_q[xxx]")
 		char name[MAX_NAME_LEN];
 

subsys/net/lib/zperf/zperf_internal.h

@@ -116,7 +116,17 @@ int zperf_get_ipv4_addr(char *host, struct in_addr *addr);
 struct sockaddr_in *zperf_get_sin(void);
 
 extern void connect_ap(char *ssid);
-extern struct k_work_q *get_queue(enum session_proto proto, int session_id);
+
+struct zperf_work {
+	struct k_work_q *queue;
+	struct z_thread_stack_element *stack;
+	struct k_event *start_event;
+	size_t stack_size;
+};
+
+#define START_EVENT 0x0001
+extern void start_jobs(void);
+extern struct zperf_work *get_queue(enum session_proto proto, int session_id);
 
 int zperf_prepare_upload_sock(const struct sockaddr *peer_addr, uint8_t tos,
 			      int priority, int tcp_nodelay, int proto);

subsys/net/lib/zperf/zperf_session.h

@@ -54,7 +54,9 @@ struct session {
 #ifdef CONFIG_ZPERF_SESSION_PER_THREAD
 	struct zperf_results result;
 	struct zperf_async_upload_context async_upload_ctx;
+	struct zperf_work *zperf;
 	bool in_progress; /* is this session finished or not */
+	bool wait_for_start; /* wait until the user starts the sessions */
 #endif /* CONFIG_ZPERF_SESSION_PER_THREAD */
 };
 

subsys/net/lib/zperf/zperf_shell.c

@@ -806,7 +806,14 @@ static int execute_upload(const struct shell *sh,
 		      param->packet_size);
 	shell_fprintf(sh, SHELL_NORMAL, "Rate:\t\t%u kbps\n",
 		      param->rate_kbps);
-	shell_fprintf(sh, SHELL_NORMAL, "Starting...\n");
+
+	if (IS_ENABLED(CONFIG_ZPERF_SESSION_PER_THREAD) &&
+	    COND_CODE_1(CONFIG_ZPERF_SESSION_PER_THREAD,
+			(param->options.wait_for_start), (0))) {
+		shell_fprintf(sh, SHELL_NORMAL, "Waiting \"zperf jobs start\" command.\n");
+	} else {
+		shell_fprintf(sh, SHELL_NORMAL, "Starting...\n");
+	}
 
 	if (IS_ENABLED(CONFIG_NET_IPV6) && param->peer_addr.sa_family == AF_INET6) {
 		struct sockaddr_in6 *ipv6 =
@@ -1000,6 +1007,11 @@ static int shell_cmd_upload(const struct shell *sh, size_t argc,
 			opt_cnt += 2;
 			async = true;
 			break;
+
+		case 'w':
+			param.options.wait_for_start = true;
+			opt_cnt += 1;
+			break;
 #endif /* CONFIG_ZPERF_SESSION_PER_THREAD */
 
 #ifdef CONFIG_NET_CONTEXT_PRIORITY
@@ -1247,6 +1259,11 @@ static int shell_cmd_upload2(const struct shell *sh, size_t argc,
 			opt_cnt += 2;
 			async = true;
 			break;
+
+		case 'w':
+			param.options.wait_for_start = true;
+			opt_cnt += 1;
+			break;
 #endif /* CONFIG_ZPERF_SESSION_PER_THREAD */
 
 #ifdef CONFIG_NET_CONTEXT_PRIORITY
@@ -1784,6 +1801,22 @@ static int cmd_jobs_clear(const struct shell *sh, size_t argc, char *argv[])
 	return 0;
 }
 
+static int cmd_jobs_start(const struct shell *sh, size_t argc, char *argv[])
+{
+#ifdef CONFIG_ZPERF_SESSION_PER_THREAD
+	ARG_UNUSED(argc);
+	ARG_UNUSED(argv);
+	ARG_UNUSED(sh);
+
+	start_jobs();
+#else
+	shell_fprintf(sh, SHELL_INFO,
+		      "Zephyr has not been built with %s support.\n",
+		      "CONFIG_ZPERF_SESSION_PER_THREAD");
+#endif
+	return 0;
+}
+
 void zperf_shell_init(void)
 {
 	int ret;
@@ -1862,6 +1895,7 @@ SHELL_STATIC_SUBCMD_SET_CREATE(zperf_cmd_tcp,
 		  "-n: Disable Nagle's algorithm\n"
 #ifdef CONFIG_ZPERF_SESSION_PER_THREAD
 		  "-t: Specify custom thread priority\n"
+		  "-w: Wait for start signal before starting the tests\n"
 #endif /* CONFIG_ZPERF_SESSION_PER_THREAD */
 #ifdef CONFIG_NET_CONTEXT_PRIORITY
 		  "-p: Specify custom packet priority\n"
@@ -1885,6 +1919,7 @@ SHELL_STATIC_SUBCMD_SET_CREATE(zperf_cmd_tcp,
 		  "-n: Disable Nagle's algorithm\n"
 #ifdef CONFIG_ZPERF_SESSION_PER_THREAD
 		  "-t: Specify custom thread priority\n"
+		  "-w: Wait for start signal before starting the tests\n"
 #endif /* CONFIG_ZPERF_SESSION_PER_THREAD */
 #ifdef CONFIG_NET_CONTEXT_PRIORITY
 		  "-p: Specify custom packet priority\n"
@@ -1937,6 +1972,7 @@ SHELL_STATIC_SUBCMD_SET_CREATE(zperf_cmd_udp,
 		  "-a: Asynchronous call (shell will not block for the upload)\n"
 #ifdef CONFIG_ZPERF_SESSION_PER_THREAD
 		  "-t: Specify custom thread priority\n"
+		  "-w: Wait for start signal before starting the tests\n"
 #endif /* CONFIG_ZPERF_SESSION_PER_THREAD */
 #ifdef CONFIG_NET_CONTEXT_PRIORITY
 		  "-p: Specify custom packet priority\n"
@@ -1961,6 +1997,7 @@ SHELL_STATIC_SUBCMD_SET_CREATE(zperf_cmd_udp,
 		  "-a: Asynchronous call (shell will not block for the upload)\n"
 #ifdef CONFIG_ZPERF_SESSION_PER_THREAD
 		  "-t: Specify custom thread priority\n"
+		  "-w: Wait for start signal before starting the tests\n"
 #endif /* CONFIG_ZPERF_SESSION_PER_THREAD */
 #ifdef CONFIG_NET_CONTEXT_PRIORITY
 		  "-p: Specify custom packet priority\n"
@@ -1994,6 +2031,7 @@ SHELL_STATIC_SUBCMD_SET_CREATE(zperf_cmd_udp,
 SHELL_STATIC_SUBCMD_SET_CREATE(zperf_cmd_jobs,
 	SHELL_CMD(all, NULL, "Show all statistics", cmd_jobs_all),
 	SHELL_CMD(clear, NULL, "Clear all statistics", cmd_jobs_clear),
+	SHELL_CMD(start, NULL, "Start waiting jobs", cmd_jobs_start),
 );
 
 SHELL_STATIC_SUBCMD_SET_CREATE(zperf_commands,

subsys/net/lib/zperf/zperf_tcp_uploader.c

@@ -160,6 +160,15 @@ static void tcp_upload_async_work(struct k_work *work)
 	ses = CONTAINER_OF(work, struct session, async_upload_ctx.work);
 	upload_ctx = &ses->async_upload_ctx;
 
+	if (ses->wait_for_start) {
+		NET_INFO("[%d] %s waiting for start", ses->id, "TCP");
+
+		/* Wait for the start event to be set */
+		k_event_wait(ses->zperf->start_event, START_EVENT, true, K_FOREVER);
+
+		NET_INFO("[%d] %s starting", ses->id, "TCP");
+	}
+
 	NET_DBG("[%d] thread %p priority %d name %s", ses->id, k_current_get(),
 		k_thread_priority_get(k_current_get()),
 		k_thread_name_get(k_current_get()));
@@ -250,6 +259,7 @@ int zperf_tcp_upload_async(const struct zperf_upload_params *param,
 
 #ifdef CONFIG_ZPERF_SESSION_PER_THREAD
 	struct k_work_q *queue;
+	struct zperf_work *zperf;
 	struct session *ses;
 	k_tid_t tid;
 
@@ -270,7 +280,9 @@ int zperf_tcp_upload_async(const struct zperf_upload_params *param,
 	ses->async_upload_ctx.callback = callback;
 	ses->async_upload_ctx.user_data = user_data;
 
-	queue = get_queue(SESSION_TCP, ses->id);
+	zperf = get_queue(SESSION_TCP, ses->id);
+
+	queue = zperf->queue;
 	if (queue == NULL) {
 		NET_ERR("Cannot get a work queue!");
 		return -ENOENT;
@@ -282,6 +294,8 @@ int zperf_tcp_upload_async(const struct zperf_upload_params *param,
 	k_work_init(&ses->async_upload_ctx.work, tcp_upload_async_work);
 
 	ses->start_time = k_uptime_ticks();
+	ses->zperf = zperf;
+	ses->wait_for_start = param->options.wait_for_start;
 
 	zperf_async_work_submit(SESSION_TCP, ses->id, &ses->async_upload_ctx.work);
 

subsys/net/lib/zperf/zperf_udp_uploader.c

@@ -346,6 +346,15 @@ static void udp_upload_async_work(struct k_work *work)
 	ses = CONTAINER_OF(work, struct session, async_upload_ctx.work);
 	upload_ctx = &ses->async_upload_ctx;
 
+	if (ses->wait_for_start) {
+		NET_INFO("[%d] %s waiting for start", ses->id, "UDP");
+
+		/* Wait for the start event to be set */
+		k_event_wait(ses->zperf->start_event, START_EVENT, true, K_FOREVER);
+
+		NET_INFO("[%d] %s starting", ses->id, "UDP");
+	}
+
 	NET_DBG("[%d] thread %p priority %d name %s", ses->id, k_current_get(),
 		k_thread_priority_get(k_current_get()),
 		k_thread_name_get(k_current_get()));
@@ -383,6 +392,7 @@ int zperf_udp_upload_async(const struct zperf_upload_params *param,
 
 #ifdef CONFIG_ZPERF_SESSION_PER_THREAD
 	struct k_work_q *queue;
+	struct zperf_work *zperf;
 	struct session *ses;
 	k_tid_t tid;
 
@@ -403,7 +413,9 @@ int zperf_udp_upload_async(const struct zperf_upload_params *param,
 	ses->async_upload_ctx.callback = callback;
 	ses->async_upload_ctx.user_data = user_data;
 
-	queue = get_queue(SESSION_UDP, ses->id);
+	zperf = get_queue(SESSION_UDP, ses->id);
+
+	queue = zperf->queue;
 	if (queue == NULL) {
 		NET_ERR("Cannot get a work queue!");
 		return -ENOENT;
@@ -415,6 +427,8 @@ int zperf_udp_upload_async(const struct zperf_upload_params *param,
 	k_work_init(&ses->async_upload_ctx.work, udp_upload_async_work);
 
 	ses->start_time = k_uptime_ticks();
+	ses->zperf = zperf;
+	ses->wait_for_start = param->options.wait_for_start;
 
 	zperf_async_work_submit(SESSION_UDP, ses->id, &ses->async_upload_ctx.work);