path: root/test/posix/regression/posix_mq_send_abort_test.c

/***************************************************************************/
/* 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 <stdio.h>
#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, &param);
    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, &param);
    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);
}