meshx_txcm.c

Go to the documentation of this file.

#include "meshx.h"
#include "sys/queue.h"
#include "string.h"
#include "meshx_txcm.h"
 
#if CONFIG_TXCM_ENABLE
#define MESHX_TXCM_INIT_MAGIC       0x4455
#define MESHX_TXCM_TASK_STACK_SIZE  2048
#define MESHX_TXCM_TASK_PRIO        5
#define MESHX_TXCM_SIG_Q_LEN        10
#define MESHX_TXCM_SIG_Q_DEPTH      sizeof(meshx_txcm_request_t)

typedef meshx_err_t (*meshx_txcm_sig_proc_t)(meshx_txcm_request_t *request);

typedef struct meshx_txcm_tx_q
{
    uint16_t dest_addr;                                 
    uint16_t retry_cnt;                                 
    uint16_t msg_param_len;                             
    meshx_txcm_msg_type_t msg_type;                     
    meshx_txcm_msg_state_t msg_state;                   
    meshx_txcm_fn_model_send_t send_fn;                 
    uint8_t msg_param[MESHX_TXCM_MSG_PARAM_MAX_LEN];    
}meshx_txcm_tx_q_t;

typedef struct {
    int16_t head;                                      
    int16_t tail;                                      
    uint16_t count;                                    
    meshx_txcm_tx_q_t q_param[MESHX_TXCM_TX_Q_LEN];    
} meshx_tx_queue_t;
 
/****************************************************************************
 * Private functions Ring buffer
 ****************************************************************************/
static meshx_err_t meshx_tx_queue_is_full(const meshx_tx_queue_t *q);
static meshx_err_t meshx_tx_queue_is_empty(const meshx_tx_queue_t *q);
static meshx_err_t meshx_tx_queue_dequeue(meshx_tx_queue_t *q, meshx_txcm_tx_q_t *item);
static meshx_err_t meshx_tx_queue_peek(const meshx_tx_queue_t *q, meshx_txcm_tx_q_t *item);
static meshx_err_t meshx_tx_queue_enqueue(meshx_tx_queue_t *q, const meshx_txcm_tx_q_t *item);
static meshx_err_t meshx_tx_queue_enqueue_front(meshx_tx_queue_t *q, const meshx_txcm_tx_q_t *item);
 
/****************************************************************************
 * Private functions Tx Control
 ****************************************************************************/
static void meshx_txcm_task_handler(const dev_struct_t *args);
static meshx_err_t meshx_txcm_sig_resend(meshx_txcm_request_t *request);
static meshx_err_t meshx_txcm_sig_enq_send(meshx_txcm_request_t *request);
static meshx_err_t meshx_txcm_sig_ack(const meshx_txcm_request_t *request);
static meshx_err_t meshx_txcm_sig_direct_send(meshx_txcm_request_t *request);

static struct{
    uint16_t init_magic;
    meshx_task_t txcm_task;
    meshx_msg_q_t txcm_sig_queue;
    meshx_tx_queue_t txcm_tx_queue;
} g_txcm = {
    .init_magic = 0,
    .txcm_tx_queue = {
        .head = 0,
        .tail = 0,
        .count = 0,
    },
    .txcm_sig_queue = {
        .max_msg_length = MESHX_TXCM_SIG_Q_LEN,
        .max_msg_depth  = MESHX_TXCM_SIG_Q_DEPTH,
    },
    .txcm_task = {
        .task_name  = "meshx_txcm_task",
        .priority   = MESHX_TXCM_TASK_PRIO,
        .stack_size = MESHX_TXCM_TASK_STACK_SIZE,
        .task_cb    = (meshx_task_cb_t) &meshx_txcm_task_handler,
    }
};

static meshx_txcm_sig_proc_t g_sig_proc_table[MESHX_TXCM_SIG_MAX] =
{
    [MESHX_TXCM_SIG_ACK]         = (meshx_txcm_sig_proc_t)&meshx_txcm_sig_ack,
    [MESHX_TXCM_SIG_RESEND]      = (meshx_txcm_sig_proc_t)&meshx_txcm_sig_resend,
    [MESHX_TXCM_SIG_ENQ_SEND]    = (meshx_txcm_sig_proc_t)&meshx_txcm_sig_enq_send,
    [MESHX_TXCM_SIG_DIRECT_SEND] = (meshx_txcm_sig_proc_t)&meshx_txcm_sig_direct_send,
};

static meshx_err_t meshx_tx_queue_is_full(const meshx_tx_queue_t *q)
{
    return q->count == MESHX_TXCM_TX_Q_LEN ? MESHX_SUCCESS : MESHX_NO_MEM;
}

static meshx_err_t meshx_tx_queue_is_empty(const meshx_tx_queue_t *q)
{
    return q->count == 0 ? MESHX_SUCCESS : MESHX_INVALID_STATE;
}

static meshx_err_t meshx_tx_queue_enqueue(meshx_tx_queue_t *q, const meshx_txcm_tx_q_t *item)
{
    meshx_err_t err = MESHX_SUCCESS;
    err = meshx_tx_queue_is_full(q);
    if (err == MESHX_SUCCESS)
    {
        return MESHX_NO_MEM;
    }
 
    // Copy the entire item structure into the q_param.
    memcpy(q->q_param + q->tail, item, sizeof(meshx_txcm_tx_q_t));
 
    q->tail = (q->tail + 1) % MESHX_TXCM_TX_Q_LEN;
 
    // Increment count.
    q->count++;
 
    return MESHX_SUCCESS;
}

static meshx_err_t meshx_tx_queue_enqueue_front(meshx_tx_queue_t *q, const meshx_txcm_tx_q_t *item)
{
    meshx_err_t err = MESHX_SUCCESS;
    err = meshx_tx_queue_is_full(q);
    if (err == MESHX_SUCCESS)
    {
        return MESHX_NO_MEM;
    }
    // Increment tail and wrap around if necessary.
    q->head = (q->head - 1 + MESHX_TXCM_TX_Q_LEN) % MESHX_TXCM_TX_Q_LEN;
 
    // Copy the entire item structure into the q_param.
    memcpy(q->q_param + q->head, item, sizeof(meshx_txcm_tx_q_t));
 
    // Increment count.
    q->count++;
 
    return MESHX_SUCCESS;
}
static meshx_err_t meshx_tx_queue_peek(const meshx_tx_queue_t *q, meshx_txcm_tx_q_t *item)
{
    meshx_err_t err = MESHX_SUCCESS;
    err = meshx_tx_queue_is_empty(q);
    if (err == MESHX_SUCCESS)
    {
        return MESHX_INVALID_STATE;
    }
 
    // Copy the entire item structure from the q_param.
    memcpy(item, q->q_param + q->head, sizeof(meshx_txcm_tx_q_t));
 
    return MESHX_SUCCESS;
}

static meshx_err_t meshx_tx_queue_search(
    const meshx_tx_queue_t *q,
    const uint8_t *param,
    uint16_t dest_addr)
{
    meshx_err_t err = MESHX_SUCCESS;
    err = meshx_tx_queue_is_empty(q);
    if (err == MESHX_SUCCESS)
    {
        return MESHX_INVALID_STATE;
    }
 
    int16_t head = q->head;
    int16_t tail = q->tail;
 
    while(head != tail)
    {
        // Move tail back
        tail = (tail - 1 + MESHX_TXCM_TX_Q_LEN) % MESHX_TXCM_TX_Q_LEN;
        // compare param with q_param[tail]
        if (strcmp((const char*)param, (const char*)q->q_param[tail].msg_param) == 0
            && q->q_param[tail].dest_addr == dest_addr)
        {
            MESHX_LOGD(MODULE_ID_TXCM, "Found param in queue");
            return MESHX_SUCCESS;
        }
    }
 
    return MESHX_NOT_FOUND;
}
static meshx_err_t meshx_tx_queue_dequeue(meshx_tx_queue_t *q, meshx_txcm_tx_q_t *item)
{
    meshx_err_t err = meshx_tx_queue_peek(q, item);
    if (err != MESHX_SUCCESS)
    {
        return err;
    }
 
    // Move head forward
    q->head = (q->head + 1) % MESHX_TXCM_TX_Q_LEN;
 
    // Decrement count
    q->count--;
 
    return MESHX_SUCCESS;
}

static meshx_err_t meshx_txcm_msg_q_front_try_send(bool resend)
{
    meshx_err_t err = MESHX_SUCCESS;
    meshx_txcm_tx_q_t front_tx =
    {
        .msg_state = MESHX_TXCM_MSG_STATE_MAX
    };
 
    MESHX_LOGD(MODULE_ID_TXCM, "TXCM_Q Stat: %x|%x|%x", g_txcm.txcm_tx_queue.head, g_txcm.txcm_tx_queue.tail, g_txcm.txcm_tx_queue.count);
    if (resend == false && (meshx_tx_queue_peek(&g_txcm.txcm_tx_queue, &front_tx) != MESHX_SUCCESS))
    {
        return MESHX_SUCCESS;
    }
 
    if (!(resend || front_tx.msg_state == MESHX_TXCM_MSG_STATE_NEW))
    {
        return MESHX_SUCCESS;
    }
    if(resend)
    {
        MESHX_LOGD(MODULE_ID_TXCM, "Try to send message from Tx Control Tx Queue resend|state:%d|%d", resend,front_tx.msg_state);
    }
    err = meshx_tx_queue_dequeue(&g_txcm.txcm_tx_queue, &front_tx);
    if (err)
    {
        MESHX_LOGE(MODULE_ID_TXCM, "Failed to receive message from Tx Control Tx Queue: %p", (void *)err);
        return err;
    }
 
    if ((int16_t)front_tx.retry_cnt-- <= 0)
    {
        front_tx.msg_state = MESHX_TXCM_MSG_STATE_NACK;
        return MESHX_TIMEOUT;
    }
 
    front_tx.msg_state = MESHX_TXCM_MSG_STATE_SENDING;
    if (front_tx.send_fn != NULL)
    {
        err = front_tx.send_fn(front_tx.msg_param, front_tx.msg_param_len);
        if(err != MESHX_SUCCESS)
        {
            front_tx.msg_state = MESHX_TXCM_MSG_STATE_NACK;
            return err;
        }
    }
 
    if (front_tx.msg_type == MESHX_TXCM_MSG_TYPE_ACKED)
    {
        /* Requeue to TX Front as we are waiting for Ack */
        front_tx.msg_state = MESHX_TXCM_MSG_STATE_WAITING_ACK;
        err = meshx_tx_queue_enqueue_front(&g_txcm.txcm_tx_queue, &front_tx);
        if (err)
        {
            MESHX_LOGE(MODULE_ID_TXCM, "Failed to send message to Tx Control Tx Queue: %p", (void *)err);
            return err;
        }
    }
    else
    {
        /* Un-Acked msg need not to be retained */
        MESHX_DO_NOTHING;
    }
    return err;
}

static meshx_err_t meshx_txcm_proccess_request_msg(
    meshx_txcm_request_t *request,
    meshx_txcm_msg_type_t msg_type)
{
    meshx_err_t err = MESHX_SUCCESS;
    static meshx_txcm_tx_q_t new_tx;
 
    if (   request == NULL
        || request->send_fn == NULL
        || msg_type >= MESHX_TXCM_MSG_TYPE_MAX
        || request->msg_param_len >= MESHX_TXCM_MSG_PARAM_MAX_LEN
    )
    {
        return MESHX_INVALID_ARG;
    }
 
    MESHX_LOGD(MODULE_ID_TXCM, "Processing a new request");
 
    err = meshx_tx_queue_search(&g_txcm.txcm_tx_queue, request->msg_param, request->dest_addr);
    if (err == MESHX_SUCCESS)
    {
        MESHX_LOGD(MODULE_ID_TXCM, "Message already in queue");
        return MESHX_SUCCESS;
    }
 
    memset(&new_tx, 0, sizeof(meshx_txcm_tx_q_t));
 
    new_tx.msg_type      = msg_type;
    new_tx.send_fn       = request->send_fn;
    new_tx.dest_addr     = request->dest_addr;
    new_tx.msg_param_len = request->msg_param_len;
    new_tx.msg_state     = MESHX_TXCM_MSG_STATE_NEW;
    new_tx.retry_cnt     = MESHX_TXCM_MSG_RETRY_MAX;
 
    memcpy(new_tx.msg_param, request->msg_param, request->msg_param_len);
 
    err = meshx_tx_queue_enqueue(&g_txcm.txcm_tx_queue, &new_tx);
    if (err)
    {
        MESHX_LOGE(MODULE_ID_TXCM, "Failed to send message to Tx Control Tx Queue: %p", (void *)err);
        return err;
    }
 
    err = meshx_txcm_msg_q_front_try_send(false);
    if(err)
    {
        MESHX_LOGE(MODULE_ID_TXCM, "Failed to process front of Tx Control Tx Queue: %p", (void *)err);
    }
    return err;
}

static meshx_err_t meshx_txcm_sig_enq_send(meshx_txcm_request_t *request)
{
    MESHX_LOGD(MODULE_ID_TXCM, "Enqueuing a new request");
    return meshx_txcm_proccess_request_msg(request, MESHX_TXCM_MSG_TYPE_ACKED);
}

static meshx_err_t meshx_txcm_sig_direct_send(meshx_txcm_request_t *request)
{
    MESHX_LOGD(MODULE_ID_TXCM, "Processing a new direct request");
    /* The same path can take care as the msg type  */
    return meshx_txcm_proccess_request_msg(request, MESHX_TXCM_MSG_TYPE_UNACKED);
}

static meshx_err_t meshx_txcm_sig_resend(meshx_txcm_request_t *request)
{
    MESHX_UNUSED(request);
    meshx_err_t err = MESHX_SUCCESS;
 
    MESHX_LOGD(MODULE_ID_TXCM, "Processing a retry");
    err = meshx_txcm_msg_q_front_try_send(true);
    if(err == MESHX_TIMEOUT)
    {
        MESHX_LOGD(MODULE_ID_TXCM, "Timeout");
        err = control_task_msg_publish(
            CONTROL_TASK_MSG_CODE_TXCM,
            CONTROL_TASK_MSG_EVT_TXCM_MSG_TIMEOUT,
            request->msg_param,
            request->msg_param_len
        );
        if(err)
        {
            MESHX_LOGE(MODULE_ID_TXCM, "Failed to process front of Tx Control Tx Queue: %p", (void *)err);
        }
        err = meshx_txcm_msg_q_front_try_send(false);
    }
    meshx_rtos_free(&request->msg_param);
    return err;
}

static meshx_err_t meshx_txcm_sig_ack(const meshx_txcm_request_t *request)
{
    MESHX_UNUSED(request);
    meshx_err_t err = MESHX_SUCCESS;
 
    MESHX_LOGD(MODULE_ID_TXCM, "Processing an ack");
    meshx_txcm_tx_q_t front_tx;
    if(meshx_tx_queue_peek(&g_txcm.txcm_tx_queue, &front_tx) == MESHX_SUCCESS)
    {
        if(front_tx.dest_addr == request->dest_addr)
        {
            err = meshx_tx_queue_dequeue(&g_txcm.txcm_tx_queue, &front_tx);
            if (err)
            {
                MESHX_LOGE(MODULE_ID_TXCM, "Failed to receive message from Tx Control Tx Queue: %p", (void *)err);
                return err;
            }
            MESHX_LOGD(MODULE_ID_TXCM, "Received message from Tx Control Tx Queue");
            front_tx.msg_state = MESHX_TXCM_MSG_STATE_ACK;
        }
        else
        {
            MESHX_LOGW(MODULE_ID_TXCM, "Async status received from address 0x%X", front_tx.dest_addr);
            MESHX_DO_NOTHING;
        }
    }
    err = meshx_rtos_free(request->msg_param);
    if (err)
    {
        MESHX_LOGE(MODULE_ID_TXCM, "RTOS Free failed: %p", (void *)err);
    }
    err = meshx_txcm_msg_q_front_try_send(false);
    return err;
}

static void meshx_txcm_task_handler(const dev_struct_t *args)
{
    MESHX_UNUSED(args);
    MESHX_LOGI(MODULE_ID_TXCM, "MeshX Tx Control Task started");
    meshx_err_t err = MESHX_SUCCESS;
 
    while(true)
    {
        meshx_txcm_request_t request;
        /* Wait for a signal from the signal queue */
        if(meshx_msg_q_recv(&g_txcm.txcm_sig_queue, &request, UINT32_MAX) != MESHX_SUCCESS)
        {
            MESHX_LOGD(MODULE_ID_TXCM, "Failed to receive signal from Tx Control Signal Queue");
            continue;
        }
        MESHX_LOGD(MODULE_ID_TXCM, "Processing sig: %d", request.request_type);
 
        /* Process the signal based on the request type */
        err = g_sig_proc_table[request.request_type](&request);
        if(err)
        {
            MESHX_LOGE(MODULE_ID_TXCM, "Failed to process the request (%p): %p",(void*)request.request_type, (void *)err);
        }
    }
}

meshx_err_t meshx_txcm_init(dev_struct_t *pdev)
{
    MESHX_LOGI(MODULE_ID_TXCM, "Initializing MeshX Tx Control Module");
    if(g_txcm.init_magic == MESHX_TXCM_INIT_MAGIC)
    {
        return MESHX_SUCCESS;
    }
    meshx_err_t err = MESHX_SUCCESS;
    g_txcm.init_magic = MESHX_TXCM_INIT_MAGIC;
    g_txcm.txcm_task.arg = pdev;
 
    /* Create Control Signal Queue */
    err = meshx_msg_q_create(&g_txcm.txcm_sig_queue);
    if(err)
    {
        MESHX_LOGE(MODULE_ID_TXCM, "Failed to create Tx Control Signal Queue: %p", (void *)err);
        return err;
    }
 
    err = meshx_task_create(&g_txcm.txcm_task);
    if(err)
    {
        MESHX_LOGE(MODULE_ID_TXCM, "Failed to create Tx Control task: %p", (void *)err);
        return err;
    }
 
    return MESHX_SUCCESS;
}

meshx_err_t meshx_txcm_request_send(
    meshx_txcm_sig_t request_type,
    uint16_t dest_addr,
    meshx_cptr_t msg_param,
    uint16_t msg_param_len,
    meshx_txcm_fn_model_send_t send_fn
)
{
    meshx_err_t err = MESHX_SUCCESS;
    static meshx_txcm_request_t new_req;
    memset(&new_req, 0, sizeof(meshx_txcm_request_t));
 
    /* Prepare request message */
    new_req.dest_addr     = dest_addr;
    new_req.send_fn       = send_fn;
    new_req.request_type  = request_type;
    new_req.msg_param_len = msg_param_len;
 
    if(msg_param_len != 0 && msg_param != NULL)
    {
        /* Retain message context */
        err = meshx_rtos_malloc(&new_req.msg_param, msg_param_len);
        if (err)
        {
            MESHX_LOGE(MODULE_ID_TXCM, "Malloc Failure: %p", (void *)err);
            return err;
        }
        memcpy(new_req.msg_param, msg_param, msg_param_len);
    }
    else
    {
        new_req.msg_param = NULL;
    }
    err = meshx_msg_q_send(&g_txcm.txcm_sig_queue, &new_req, sizeof(meshx_txcm_request_t), 0);
    if (err)
    {
        MESHX_LOGE(MODULE_ID_TXCM, "TXCM Signal failed: %p", (void *)err);
        meshx_rtos_free(&new_req.msg_param);
        return err;
    }
    return err;
}

meshx_err_t meshx_txcm_event_cb_reg(meshx_txcm_cb_t event_cb)
{
    return control_task_msg_subscribe(
        CONTROL_TASK_MSG_CODE_TXCM,
        CONTROL_TASK_MSG_EVT_TXCM_MSG_TIMEOUT,
        event_cb);
}
#endif /* CONFIG_TXCM_ENABLE */