kernel: workq: introduce work timeout:

Introduce work timeout, which is an optional workqueue configuration
which enables monitoring for work items which take longer than
expected. This could be due to long running or deadlocked handlers.

Signed-off-by: Bjarki Arge Andreasen <[email protected]>

Author: bjarki-andreasen

include/zephyr/kernel.h

@@ -3663,6 +3663,19 @@ int k_work_queue_unplug(struct k_work_q *queue);
  */
 int k_work_queue_stop(struct k_work_q *queue, k_timeout_t timeout);
 
+/** @brief Check if a work queue is blocked.
+ *
+ * Checks if a work queue is blocked by a work item. The work queue is considered
+ * blocked if a work item takes longer to execute than the work_timeout provided
+ * to k_work_queue_start in the k_work_queue_config structure.
+ *
+ * @param queue Pointer to the queue structure.
+ *
+ * @retval true if the work queue is blocked
+ * @retval false if the work queue is not blocked
+ */
+bool k_work_queue_is_blocked(struct k_work_q *queue);
+
 /** @brief Initialize a delayable work structure.
  *
  * This must be invoked before scheduling a delayable work structure for the
@@ -3974,6 +3987,8 @@ enum {
 	K_WORK_QUEUE_PLUGGED = BIT(K_WORK_QUEUE_PLUGGED_BIT),
 	K_WORK_QUEUE_STOP_BIT = 4,
 	K_WORK_QUEUE_STOP = BIT(K_WORK_QUEUE_STOP_BIT),
+	K_WORK_QUEUE_BLOCKED_BIT = 5,
+	K_WORK_QUEUE_BLOCKED = BIT(K_WORK_QUEUE_BLOCKED_BIT),
 
 	/* Static work queue flags */
 	K_WORK_QUEUE_NO_YIELD_BIT = 8,
@@ -4168,6 +4183,14 @@ struct k_work_queue_config {
 	 * essential thread.
 	 */
 	bool essential;
+
+	/** Controls whether work queue monitors work timeouts.
+	 *
+	 * If set to a positive value, the work queue will monitor the
+	 * duration of each work item, and warn the user if the work
+	 * item handler takes longer than work_timeout to execute.
+	 */
+	uint32_t work_timeout_ms;
 };
 
 /** @brief A structure used to hold work until it can be processed. */
@@ -4190,6 +4213,15 @@ struct k_work_q {
 
 	/* Flags describing queue state. */
 	uint32_t flags;
+
+	/* Timeout which invokes sentinal */
+	struct _timeout workto;
+
+	/* Work monitored by sentinal */
+	struct k_work *work;
+
+	/* Maximum work duration */
+	k_timeout_t work_timeout;
 };
 
 /* Provide the implementation for inline functions declared above */

kernel/Kconfig

@@ -600,6 +600,13 @@ config SYSTEM_WORKQUEUE_NO_YIELD
 	  cooperative and a sequence of work items is expected to complete
 	  without yielding.
 
+config SYSTEM_WORKQUEUE_WORK_TIMEOUT_MS
+	int "Select system work queue work timeout in milliseconds"
+	default 10000 if DEBUG
+	default 0
+	help
+	  Set to 0 to disable work timeout for system workqueue.
+
 endmenu
 
 menu "Barrier Operations"

kernel/system_work_q.c

@@ -25,6 +25,7 @@ static int k_sys_work_q_init(void)
 		.name = "sysworkq",
 		.no_yield = IS_ENABLED(CONFIG_SYSTEM_WORKQUEUE_NO_YIELD),
 		.essential = true,
+		.work_timeout_ms = CONFIG_SYSTEM_WORKQUEUE_WORK_TIMEOUT_MS,
 	};
 
 	k_work_queue_start(&k_sys_work_q,

kernel/work.c

@@ -17,6 +17,9 @@
 #include <errno.h>
 #include <ksched.h>
 #include <zephyr/sys/printk.h>
+#include <zephyr/logging/log.h>
+
+LOG_MODULE_DECLARE(os, CONFIG_KERNEL_LOG_LEVEL);
 
 static inline void flag_clear(uint32_t *flagp,
 			      uint32_t bit)
@@ -599,6 +602,53 @@ bool k_work_cancel_sync(struct k_work *work,
 	return pending;
 }
 
+static void workto_handler(struct _timeout *to)
+{
+	struct k_work_q *queue = CONTAINER_OF(to, struct k_work_q, workto);
+	k_spinlock_key_t key;
+	const char *name;
+	struct k_work *work;
+	k_work_handler_t handler;
+
+	key = k_spin_lock(&lock);
+
+	flag_set(&queue->flags, K_WORK_QUEUE_BLOCKED_BIT);
+
+	name = k_thread_name_get(&queue->thread);
+	work = queue->work;
+	handler = work->handler;
+
+	if (name != NULL) {
+		LOG_WRN("queue %s blocked by work %p with handler %p", name, work, handler);
+	} else {
+		LOG_WRN("queue %p blocked by work %p with handler %p", queue, work, handler);
+	}
+
+	k_spin_unlock(&lock, key);
+}
+
+static void work_timeout_start_locked(struct k_work_q *queue, struct k_work *work)
+{
+	if (K_TIMEOUT_EQ(queue->work_timeout, K_FOREVER)) {
+		return;
+	}
+
+	queue->work = work;
+	z_add_timeout(&queue->workto, workto_handler, queue->work_timeout);
+}
+
+static void work_timeout_stop_locked(struct k_work_q *queue)
+{
+	if (K_TIMEOUT_EQ(queue->work_timeout, K_FOREVER)) {
+		return;
+	}
+
+	z_abort_timeout(&queue->workto);
+	if (flag_test_and_clear(&queue->flags, K_WORK_QUEUE_BLOCKED_BIT)) {
+		LOG_INF("queue %p unblocked", queue);
+	}
+}
+
 /* Loop executed by a work queue thread.
  *
  * @param workq_ptr pointer to the work queue structure
@@ -678,6 +728,8 @@ static void work_queue_main(void *workq_ptr, void *p2, void *p3)
 			continue;
 		}
 
+		work_timeout_start_locked(queue, work);
+
 		k_spin_unlock(&lock, key);
 
 		__ASSERT_NO_MSG(handler != NULL);
@@ -690,6 +742,8 @@ static void work_queue_main(void *workq_ptr, void *p2, void *p3)
 		 */
 		key = k_spin_lock(&lock);
 
+		work_timeout_stop_locked(queue);
+
 		flag_clear(&work->flags, K_WORK_RUNNING_BIT);
 		if (flag_test(&work->flags, K_WORK_FLUSHING_BIT)) {
 			finalize_flush_locked(work);
@@ -761,6 +815,12 @@ void k_work_queue_start(struct k_work_q *queue,
 		queue->thread.base.user_options |= K_ESSENTIAL;
 	}
 
+	if ((cfg != NULL) && (cfg->work_timeout_ms)) {
+		queue->work_timeout = K_MSEC(cfg->work_timeout_ms);
+	} else {
+		queue->work_timeout = K_FOREVER;
+	}
+
 	k_thread_start(&queue->thread);
 
 	SYS_PORT_TRACING_OBJ_FUNC_EXIT(k_work_queue, start, queue);
@@ -853,6 +913,11 @@ int k_work_queue_stop(struct k_work_q *queue, k_timeout_t timeout)
 	return 0;
 }
 
+bool k_work_queue_is_blocked(struct k_work_q *queue)
+{
+	return flag_test(&queue->flags, K_WORK_QUEUE_BLOCKED_BIT);
+}
+
 #ifdef CONFIG_SYS_CLOCK_EXISTS
 
 /* Timeout handler for delayable work.