/***************************************************************************/ /* Copyright (c) 2026 Eclipse ThreadX contributors */ /* */ /* This program and the accompanying materials are made available under */ /* the terms of the MIT License which is available at */ /* https://opensource.org/licenses/MIT. */ /* */ /* AI Disclosure: This file was largely AI-generated by Claude Code */ /* (Opus 5). The AI-generated portions may be considered public domain */ /* (CC0-1.0) and not subject to the project's licence. The human */ /* contributor has reviewed and verified that the code is correct. */ /* */ /* SPDX-License-Identifier: MIT and CC0-1.0 */ /***************************************************************************/ /* This test covers the message buffer mq_send() used to leak when the send failed, reported as issue #568. mq_send() allocates a private buffer from the queue's own byte pool, copies the caller's message into it and passes the address through the queue. The receiver owns the buffer from then on and releases it. When the send itself fails the message never reaches a receiver, so mq_send() has to release the buffer before returning or it is lost. The failure is provoked the way an application would meet it: the queue is filled, a sender blocks on it, and its wait is aborted. The queue survives, so any buffer that is not released stays lost. Aborting repeatedly and watching the queue's byte pool separates the two outcomes: with the buffer released the pool returns to its original size, and without it the pool shrinks by a message every time. */ #include #include "tx_api.h" #include "pthread.h" /* The queue's pool is sized (mq_maxmsg + 1) * (mq_msgsize + 11), with a floor of 100 bytes, so it holds barely more than a full queue. Keep the messages small and the queue shallow: that floor then leaves room for a few leaked buffers to accumulate and be measured, instead of exhausting the pool on the first one and hanging the sender inside posix_internal_error(). */ #define TEST_MESSAGE_SIZE 8 #define TEST_QUEUE_DEPTH 1 #define TEST_ABORT_COUNT 2 static ULONG posix_region[192 * 1024 / sizeof(ULONG)]; static pthread_t sender; static pthread_attr_t sender_attr; static pthread_t controller; static pthread_attr_t controller_attr; static mqd_t queue_des; static struct mq_attr queue_attr; /* Written by the sender, read by the controller. */ static TX_THREAD * volatile sender_thread_ptr; static volatile INT send_failures; static volatile INT stop_sending; extern void test_control_return(UINT status); static VOID *sender_entry(VOID *input); static VOID *controller_entry(VOID *input); /* Report how many bytes the queue's message pool has free. */ static ULONG pool_available(void) { ULONG available; available = 0; tx_byte_pool_info_get(&(queue_des -> f_data -> vq_message_area), TX_NULL, &available, TX_NULL, TX_NULL, TX_NULL, TX_NULL); return (available); } /* Define what the initial system looks like. */ #ifdef CTEST void test_application_define(void *first_unused_memory) #else void posix_mq_send_abort_application_define(void *first_unused_memory) #endif { struct sched_param param; VOID *storage_ptr; (void) first_unused_memory; storage_ptr = posix_initialize(posix_region); queue_attr.mq_maxmsg = TEST_QUEUE_DEPTH; queue_attr.mq_msgsize = TEST_MESSAGE_SIZE; queue_des = mq_open("mq_abort", O_CREAT | O_RDWR, 0, &queue_attr); if (queue_des == (mqd_t) ERROR) { printf("Running POSIX Message Queue Send Abort Test....................... ERROR #1\n"); test_control_return(1); } /* The controller drives the test and runs above the sender, so it keeps control while the sender is being unblocked. */ pthread_attr_init(&controller_attr); param.sched_priority = 15; pthread_attr_setschedparam(&controller_attr, ¶m); pthread_attr_setstackaddr(&controller_attr, storage_ptr); storage_ptr = (ULONG *) storage_ptr + (4096 / sizeof(ULONG)); pthread_attr_init(&sender_attr); param.sched_priority = 10; pthread_attr_setschedparam(&sender_attr, ¶m); pthread_attr_setstackaddr(&sender_attr, storage_ptr); if (pthread_create(&controller, &controller_attr, controller_entry, NULL) != OK) { printf("Running POSIX Message Queue Send Abort Test....................... ERROR #2\n"); test_control_return(1); } if (pthread_create(&sender, &sender_attr, sender_entry, NULL) != OK) { printf("Running POSIX Message Queue Send Abort Test....................... ERROR #3\n"); test_control_return(1); } } /* Sends until told to stop. Every send blocks, because the controller keeps the queue full, and every one is released by an aborted wait. */ static VOID *sender_entry(VOID *input) { CHAR message[TEST_MESSAGE_SIZE]; INT i; (void) input; for (i = 0; i < TEST_MESSAGE_SIZE; i++) { message[i] = (CHAR) i; } /* Publish our own thread pointer so the controller can abort our wait without casting a pthread ID back to a ThreadX thread. */ sender_thread_ptr = tx_thread_identify(); while (stop_sending == 0) { if (mq_send(queue_des, message, TEST_MESSAGE_SIZE, 1) != OK) { send_failures++; } } return (NULL); } static VOID *controller_entry(VOID *input) { CHAR message[TEST_MESSAGE_SIZE]; ULONG baseline; ULONG after; ULONG lost; INT i; (void) input; printf("Running POSIX Message Queue Send Abort Test....................... "); for (i = 0; i < TEST_MESSAGE_SIZE; i++) { message[i] = (CHAR) i; } /* Fill the queue so that any further send has to wait. */ if (mq_send(queue_des, message, TEST_MESSAGE_SIZE, 1) != OK) { printf("ERROR #4\n"); test_control_return(1); } /* Let the sender reach its first blocking send, then take the baseline. The sender is holding one allocation at this point, which is the state every later measurement is compared against. */ tx_thread_sleep(2); if (sender_thread_ptr == TX_NULL) { printf("ERROR #5\n"); test_control_return(1); } baseline = pool_available(); for (i = 0; i < TEST_ABORT_COUNT; i++) { /* Release the sender from its wait. Its send fails, and the buffer it allocated has to come back to the pool. */ if (tx_thread_wait_abort(sender_thread_ptr) != TX_SUCCESS) { printf("ERROR #6\n"); test_control_return(1); } /* Give the sender time to return from mq_send() and block again. */ tx_thread_sleep(2); } if (send_failures != TEST_ABORT_COUNT) { printf("ERROR #7\n"); test_control_return(1); } /* The sender is blocked again holding exactly one allocation, the same as at the baseline, so the pool should read the same. */ after = pool_available(); if (after > baseline) { printf("ERROR #8\n"); test_control_return(1); } lost = baseline - after; /* Anything approaching a message means buffers are not coming back. A leak would cost TEST_ABORT_COUNT messages, so this separates the two cases comfortably without depending on byte pool overhead. */ if (lost >= (ULONG) TEST_MESSAGE_SIZE) { printf("ERROR #9\n"); test_control_return(1); } /* Shut the sender down and confirm the queue still works. */ stop_sending = 1; tx_thread_wait_abort(sender_thread_ptr); tx_thread_sleep(2); if (mq_receive(queue_des, message, TEST_MESSAGE_SIZE, TX_NULL) != (ssize_t) TEST_MESSAGE_SIZE) { printf("ERROR #10\n"); test_control_return(1); } printf("SUCCESS!\n"); test_control_return(0); return (NULL); }