tests: lib: mpsc_pbuf:reproduces the deadlock scenario

The test creates two threads: a producer that calls mpsc_pbuf_alloc and
may block inside add_skip_item when entering k_sem_take, and a consumer
that calls mpsc_pbuf_claim when only a small amount of buffer space is
left.

In this condition mpsc_pbuf_claim attempts to reserve space even though
no data is available, preventing the producer from making progress.

Without the fix, the producer thread remains blocked indefinitely in
k_sem_take. With the fix applied, the consumer performs the required
k_sem_give, allowing the producer to resume allocation and the test to
complete.

Signed-off-by: Fei Wang <[email protected]>

Author: Fei Wang

tests/lib/mpsc_pbuf/src/main.c

@@ -1034,6 +1034,140 @@ void start_threads(struct mpsc_pbuf_buffer *buffer)
 	}
 }
 
+static uint32_t buf32_1[128];
+struct test_msg_data {
+	union mpsc_pbuf_generic hdr;
+	uint32_t buf_len; /*contain buf_len and buf*/
+	uint8_t buf[sizeof(buf32_1)];
+};
+struct test_msg_hnd {
+	struct mpsc_pbuf_buffer mpsc_buffer;
+	uint32_t product_max_cnt;
+	uint32_t product_cnt;
+	uint32_t consumer_cnt;
+};
+
+K_THREAD_STACK_DEFINE(t3_stack, 1024);
+static struct k_thread threads_t3;
+static k_tid_t tid3;
+
+static uint32_t test_mpsc_get_used_len(const union mpsc_pbuf_generic *packet)
+{
+	uint32_t size = 0;
+	struct test_msg_data *msg_data = NULL;
+
+	msg_data =  CONTAINER_OF(packet, struct test_msg_data, hdr);
+	size = msg_data->buf_len + sizeof(union mpsc_pbuf_generic);
+	size = ROUND_UP(size, sizeof(uint32_t)); /*return number  of uint32_t */
+	size = size / sizeof(uint32_t);
+	return size;
+}
+
+static void t_data_consumer_entry(void *p0, void *p1, void *p2)
+{
+	uint32_t read_cnt = 0;
+	bool wait = true;
+	const union mpsc_pbuf_generic *msg = NULL;
+	const struct test_msg_data *test_data = NULL;
+	struct test_msg_hnd *test_hnd = (struct test_msg_hnd *)p0;
+
+	while (1) {
+		if (msg == NULL) {
+			msg = mpsc_pbuf_claim(&test_hnd->mpsc_buffer);
+		}
+		test_data = (const struct test_msg_data *)msg;
+		if (test_data == NULL) {
+			continue;
+		}
+		if (test_data && test_hnd->mpsc_buffer.wr_idx == 0) {
+			wait = false;
+		}
+		if (wait == true) {
+			continue;
+		}
+		read_cnt++;
+		mpsc_pbuf_free(&test_hnd->mpsc_buffer, msg);
+		msg = NULL;
+		if (read_cnt == test_hnd->product_max_cnt) {
+			break;
+		}
+
+	}
+	test_hnd->consumer_cnt = read_cnt;
+}
+
+/* test mpsc_pbuf_alloc can get sem_take
+ * one thread product data, one thread consumer data
+ * requirement:
+ *      consumer slow process data
+ * step:
+ *  1:product data len is  0.75 of cfg.size
+ *  2:run product times
+ *
+ */
+ZTEST(log_buffer, test_sema_lock)
+{
+	struct test_msg_hnd  test_hnd = {};
+	struct test_msg_data  test_data;
+	struct test_msg_data  *item = NULL;
+	struct mpsc_pbuf_buffer_config  cfg;
+	uint32_t loop = 0;
+	size_t wlen = 0;
+	bool fist_wait = true;
+
+	if (CONFIG_SYS_CLOCK_TICKS_PER_SEC < 10000) {
+		ztest_test_skip();
+	}
+
+	cfg.buf = buf32_1;
+	cfg.size = ARRAY_SIZE(buf32_1);
+	cfg.get_wlen = test_mpsc_get_used_len;
+
+	mpsc_pbuf_init(&test_hnd.mpsc_buffer, &cfg);
+	test_hnd.product_max_cnt = 2;
+
+	tid3 = k_thread_create(&threads_t3, t3_stack, 1024,
+			t_data_consumer_entry,
+			&test_hnd, NULL, NULL,
+			10, 0, K_NO_WAIT);
+	k_thread_name_set(&threads_t3, "test_mpsc_consumer");
+	for (loop = 0; loop < test_hnd.product_max_cnt; loop++) {
+		if (loop == 0) {
+			test_data.buf_len = sizeof(buf32_1) / 2;
+		} else {
+			test_data.buf_len = sizeof(buf32_1) / 2 + sizeof(buf32_1) / 4;
+		}
+		test_data.buf_len = ROUND_UP(test_data.buf_len, sizeof(uint32_t));
+		memset(test_data.buf, loop + 1, test_data.buf_len);
+
+		wlen = test_data.buf_len + sizeof(test_data.buf_len) +
+			 sizeof(test_data.hdr);
+		wlen = ROUND_UP(wlen, sizeof(uint32_t));
+		wlen = wlen / sizeof(uint32_t);
+
+		if (fist_wait &&
+		   test_data.buf_len == sizeof(buf32_1) / 2 + sizeof(buf32_1) / 4) {
+			PRINT(" mpsc sema wait\n");
+		}
+		item = (struct test_msg_data *)mpsc_pbuf_alloc(&test_hnd.mpsc_buffer,
+								wlen, K_FOREVER);
+		item->hdr.raw = 0;
+		memcpy(item->buf, test_data.buf, test_data.buf_len);
+		item->buf_len = test_data.buf_len + sizeof(test_data.buf_len);
+		mpsc_pbuf_commit(&test_hnd.mpsc_buffer, &item->hdr);
+		if (fist_wait &&
+		    test_data.buf_len == sizeof(buf32_1) / 2 + sizeof(buf32_1) / 4) {
+			PRINT(" mpsc sema wake\n");
+			fist_wait = false;
+		}
+		test_hnd.product_cnt++;
+	}
+	k_thread_join(tid3, K_FOREVER);
+	zassert_equal(test_hnd.product_cnt,
+		      test_hnd.consumer_cnt, "product %d consume %d",
+		      test_hnd.product_cnt, test_hnd.consumer_cnt);
+
+}
 /* Test creates two threads which pends on the buffer until there is a space
  * available. When enough buffers is released threads are woken up and they
  * allocate packets.