Circular buffer using pointers in C

Question

I have a queue structure, that I attempted to implement using a circular buffer, which I am using in a networking application. I am looking for some guidance and feedback. First, let me present the relevant code.

typedef struct nwk_packet_type
{
    uint8_t dest_address[NRF24_ADDR_LEN];
    uint8_t data[32];
    uint8_t data_len;
}nwk_packet_t;

/* The circular fifo on which outgoing packets are stored */
nwk_packet_t nwk_send_queue[NWK_QUEUE_SIZE];
nwk_packet_t* send_queue_in; /* pointer to queue head */
nwk_packet_t* send_queue_out; /* pointer to queue tail */

static nwk_packet_t* nwk_tx_pkt_allocate(void)
{
    /* Make sure the send queue is not full */
    if(send_queue_in == (send_queue_out - 1 + NWK_QUEUE_SIZE) % NWK_QUEUE_SIZE)
        return 0;

    /* return pointer to the next add and increment the tracker */
    return send_queue_in++;//TODO: it's not just ++, it has to be modular by packet size
}

/* External facing function for application layer to send network data */
// simply adds the packet to the network queue if there is space
// returns an appropriate error code if anything goes wrong
uint8_t nwk_send(uint8_t* address, uint8_t* data, uint8_t len)
{
    /* First check all the parameters */
    if(!address)
        return NWK_BAD_ADDRESS;
    if(!data)
        return NWK_BAD_DATA_PTR;
    if(!len || len > 32)
        return NWK_BAD_DATA_LEN;

    //TODO: PROBABLY NEED TO START BLOCKING HERE
    /* Allocate the packet on the queue */
    nwk_packet_t* packet;
    if(!( packet = nwk_tx_pkt_allocate() ))
        return NWK_QUEUE_FULL;

    /* Build the packet */
    memcpy(packet->dest_address, address, NRF24_ADDR_LEN);
    memcpy(packet->data, data, len);
    packet->data_len = len;
    //TODO: PROBABLY SAFE TO STOP BLOCKING HERE

    return NWK_SUCCESS;
}

/* Only called during NWK_IDLE, pushes the next item on the send queue out to the chip's "MAC" layer over SPI */
void nwk_transmit_pkt(void)
{
    nwk_packet_t tx_pkt = nwk_send_queue[send_queue_out];
    nrf24_send(tx_pkt->data, tx_pkt->data_len);
}

/* The callback for transceiver interrupt when a sent packet is either completed or ran out of retries */
void nwk_tx_result_cb(bool completed)
{
    if( (completed) && (nwk_tx_state == NWK_SENDING))
        send_queue_out++;//TODO: it's not just ++, it has to be modular by packet size with in the buffer

}

Ok now for a quick explanation and then my questions. So the basic idea is that I've got this queue for data which is being sent onto the network. The function nwk_send() can be called from anywhere in application code, which by the wall will be a small pre-emptive task based operating system (FreeRTOS) and thus can happen from lots of places in the code and be interrupted by the OS tick interrupt.

Now since that function is modifying the pointers into the global queue, I know it needs to be blocking when it is doing that. Am I correct in my comments on the code about where I should be blocking (ie disabling interrupts)? Also would be smarter to make a mutex using a global boolean variable or something rather than just disabling interrupts?

Also, I think there's a second place I should be blocking when things are being taken off the queue, but I'm not sure where that is exactly. Is it in nwk_transmit_pkt() where I'm actually copying the data off the queue and into a local ram variable?

Final question, how do I achieve the modulus operation on my pointers within the arrays? I feel like it should look something like:

send_queue_in = ((send_queue_in + 1) % (NWK_QUEUE_SIZE*sizeof(nwk_packet_t))) + nwk_send_queue;

Any feedback is greatly appreciated, thank you.

Answer 1

Without an overall driver framework to develop with, and when communicating with interrupt-state on a uC, you need to be very careful.

You cannot use OS synchro primitives to communicate to interrupt state. Attmpting to do so will certainly crash your OS because interrupt-handlers cannot block.

Copying the actual bulk data should be avoided.

On an 8-bit uC, I suggest queueing an index onto a buffer array pool, where the number of buffers is <256. That means that only one byte needs to be queued up and so, with an appropriate queue class that stores the value before updating internal byte-size indexes, it is possible to safely communicate buffers into a tx handler without excessive interrupt-disabling.

Access to the pool array should be thread-safe and 'insertion/deletion' should be quick - I have 'succ/pred' byte-fields in each buffer struct, so forming a double-linked list, access protected by a mutex. As well as I/O, I use this pool of buffers for all inter-thread comms.

For tx, get a buffer struct from teh pool, fill with data, push the index onto a tx queue, disable interrupts for only long enough to determine whether the tx interrupt needs 'primimg'. If priming is required, shove in a FIFO-full of data before re-enabling interrupts.

When the tx interrupt-handler has sent the buffer, it can push the 'used' index back onto a 'scavenge' queue and signal a semaphore to make a handler thread run. This thread can then take the entry from the scavenge queue and return it to the pool.

This scheme only works if interrupt-handlers do not re-enable higher-priority interrupts using the same buffering scheme.

Answer 2

About locking it will be best to use some existing mutex primitive from the OS you use. I am not familiar with FreeRTOS but it should have builtin primitives for locking between interrupt and user context.

For circular buffer you may use these:

check for empty queue

send_queue_in == send_queue_out

check for full queue

(send_queue_in + 1) % NWK_QUEUE_SIZE == send_queue_out

push element [pseudo code]

if (queue is full)
    return error;
queue[send_queue_in] = new element;
send_queue_in = (send_queue_in + 1) % NWK_QUEUE_SIZE;

pop element [pseudo code]

if (queue is empty)
   return error;
element = queue[send_queue_out];
send_queue_out = (send_queue_out + 1) % NWK_QUEUE_SIZE;

It looks that you copy and do not just reference the packet data before sending. This means that you can hold the lock until the copy is done.