Circular queue implementation using two regions

Question

Inspired by the "bip buffer", I coded a fixed-size circular queue that uses two regions (two pairs of begin/end pointers) to account for the wrap-around.

The main goal of this design was to simplify the queue operations implementation, and in the case of cqueue_size, cqueue_empty and cqueue_full it happened. However, cqueue_pop and cqueue_push are messy, and I am concerned that a bug may be hiding in those functions.

EDIT: I noticed that my post has no clear question. So here it is: can you find any bugs in the code?

#pragma once

#include <stdbool.h>
#include <stddef.h>
#include <string.h>

// WARN: these macros are meant for internal use exclusively!
#define BUFFER_SIZE         8
#define BUFFER_BEGIN(pcq)   ((pcq)->buffer)
#define BUFFER_END(pcq)     ((pcq)->buffer + BUFFER_SIZE)
#define REGIONA_SIZE(pcq)   ((pcq)->ra_end - (pcq)->ra_begin)
#define REGIONB_SIZE(pcq)   ((pcq)->rb_end - (pcq)->rb_begin)

#define REGIONB_ACTIVATE(pcq)                           \
if ((pcq)->rb_begin == NULL && (pcq)->rb_end == NULL) { \
    (pcq)->rb_begin     = BUFFER_BEGIN(pcq);            \
    (pcq)->rb_end       = BUFFER_BEGIN(pcq);            \
} else (void)0

#define REGIONB_DEACTIVATE(pcq) if (true) { \
    (pcq)->rb_begin     = NULL;             \
    (pcq)->rb_end       = NULL;             \
} else (void)0

///
/// @brief Circular queue.
/// @details Circular queue with fixed capacity, with design inspired by the "bip buffer" (bipartite buffer).
///     This implementation attempts to simplify the code of the circular queue operations by
///     abstracting away the wrap-around as two regions in the internal buffer: RegionA and RegionB.
///     RegionA can begin anywhere in the InternalBuffer but only end at the InternalBuffer's end.
///     RegionB can only begin at the beginning of the InternalBuffer.
///     RegionB can only exist if RegionA has reached the end of the InternalBuffer (i.e. RegionA is full).
///     RegionB becomes the new RegionA when the old RegionA becomes empty.
/// @remarks At the cost of keeping two extra pointers for RegionB, the pseudocode is simplified thusly:
/// @remarks CQUEUE_EMPTY:  RegionA.begin == RegionA.end
/// @remarks CQUEUE_FULL:   RegionA.begin == RegionB.end || RegionA.size == InternalBuffer.size
/// @remarks CQUEUE_SIZE:   (RegionA.end - RegionA.begin) + (RegionB.end - RegionB.begin)
///
typedef struct
{
    char    buffer[BUFFER_SIZE]; ///< The circular queue's byte buffer.
    char   *ra_begin;   ///< RegionA beginning.
    char   *ra_end;     ///< RegionA end.
    char   *rb_begin;   ///< RegionB beginning.
    char   *rb_end;     ///< RegionB end.
} cqueue_t;

// circular queue operations
static inline void cqueue_reset(cqueue_t *cq);
static inline char cqueue_pop(cqueue_t *cq);
static inline void cqueue_push(cqueue_t *cq, char c);
static inline size_t cqueue_size(const cqueue_t *cq);
static inline bool cqueue_empty(const cqueue_t *cq);
static inline bool cqueue_full(const cqueue_t *cq);

///
/// @brief Resets the circular queue `cq` to a default empty state.
/// @note The circular queue's internal buffer is cleared to 0.
/// @param [out] cq     Pointer to the circular queue to be reset.
///
static inline void cqueue_reset(cqueue_t *cq)
{
    memset(BUFFER_BEGIN(cq), 0, BUFFER_SIZE);
    cq->ra_begin = BUFFER_BEGIN(cq);
    cq->ra_end   = BUFFER_BEGIN(cq);
    REGIONB_DEACTIVATE(cq); // RegionB doesn't exist unless RegionA is full
}

///
/// @brief Pops the next byte from the circular queue `cq`.
/// @warning If the circular queue is empty, a spurious value of 0 will be "popped".
/// @param [in,out] cq      Pointer to the circular queue to be changed.
/// @returns The value of the current top byte.
/// @retval 0 If the circular queue is empty.
///
static inline char cqueue_pop(cqueue_t *cq)
{
    if (REGIONA_SIZE(cq) != 0) // does RegionA have some data?
    {
        // retrieve value of popped byte
        const char r = *(cq->ra_begin)++;

        if (REGIONA_SIZE(cq) == 0 && REGIONB_SIZE(cq) != 0) // is RegionA empty while RegionB isn't?
        {
            // RegionB becomes RegionA
            cq->ra_begin = cq->rb_begin;
            cq->ra_end   = cq->rb_end;
            REGIONB_DEACTIVATE(cq);
        }

        return r;
    }
    else // the circular queue is empty
        return 0;
}

///
/// @brief Pushes the new byte `c` to the circular queue `cq`.
/// @warning If the circular queue is full, the new byte will overwrite the top byte.
/// @param [in,out] cq      Pointer to the circular queue to be changed.
/// @param c                Byte to be pushed.
///
static inline void cqueue_push(cqueue_t *cq, char c)
{
    if (cqueue_full(cq)) // is the circular queue full?
    {
        // push to RegionB and overwrite the current top byte in RegionA
        REGIONB_ACTIVATE(cq);
        *(cq->rb_end)++ = c;
        ++cq->ra_begin;

        if (cq->ra_begin == BUFFER_END(cq)) // is RegionA empty?
        {
            // RegionB becomes RegionA
            cq->ra_begin = cq->rb_begin;
            REGIONB_DEACTIVATE(cq);
        }
    }
    else
    if (cq->ra_end == BUFFER_END(cq)) // is RegionA full?
    {
        // push to RegionB
        REGIONB_ACTIVATE(cq);
        *(cq->rb_end)++ = c;
    }
    else // RegionA is not full
    {
        // push to RegionA
        *(cq->ra_end)++ = c;
    }
}

///
/// @brief Returns the size of the circular queue `cq`.
/// @param [in] cq      Pointer to the circular queue to be analyzed.
/// @returns The size in bytes of the circular queue.
///
static inline size_t cqueue_size(const cqueue_t *cq)
{
    return REGIONA_SIZE(cq) + REGIONB_SIZE(cq);
}

///
/// @brief Returns whether or not the circular queue `cq` is empty.
/// @param [in] cq      Pointer to the circular queue to be analyzed.
/// @retval true If the circular queue is empty.
/// @retval false If the circular queue is not empty.
///
static inline bool cqueue_empty(const cqueue_t *cq)
{
    // NOTE: if RegionA doesn't exist, neither does RegionB;
    //  this is why it's enough to check only the size of RegionA
    //  to determine if the circular queue is empty
    return REGIONA_SIZE(cq) == 0;
}

///
/// @brief Returns whether or not the circular queue `cq` is full.
/// @param [in] cq      Pointer to the circular queue to be analyzed.
/// @retval true If the circular queue is full.
/// @retval false If the circular queue is not full.
///
static inline bool cqueue_full(const cqueue_t *cq)
{
    return cq->ra_begin == cq->rb_end || REGIONA_SIZE(cq) == BUFFER_SIZE;
}

Answer 1

#define REGIONB_SIZE(pcq)   ((pcq)->rb_end - (pcq)->rb_begin)

This will likely work as your program is written now, but it is not robust in the face of possible changes. In particular, if you changed how you handled an inactive region B. It would be safer to check if region B is active (missing macro) and explicitly return 0 when it is not. It is also vulnerable to direct manipulation of rb_begin and rb_end. Since this is C not C++, you have no encapsulation protection.

If you do keep this the way it is now, you should comment it so that someone modifying this code in the future knows what you have done. Essentially you have optimized this based on the fact that an inactive region B will have both start and end set to NULL and NULL - NULL == 0.

#define REGIONB_ACTIVATE(pcq)                           \
if ((pcq)->rb_begin == NULL && (pcq)->rb_end == NULL) { \

Again, I think that this should be written:

#define IS_REGIONB_ACTIVE(pcq) ((pcq)->rb_begin == NULL && (pcq)->rb_end == NULL)

#define REGIONB_ACTIVATE(pcq)                           \
if (IS_REGIONB_ACTIVE(pcq)) {                           \

This makes it clearer why REGIONB_DEACTIVATE sets both to NULL and allows the logic to be reused elsewhere.

    if (REGIONA_SIZE(cq) == 0 && REGIONB_SIZE(cq) != 0) // is RegionA empty while RegionB isn't?
    {
        // RegionB becomes RegionA
        cq->ra_begin = cq->rb_begin;
        cq->ra_end   = cq->rb_end;
        REGIONB_DEACTIVATE(cq);
    }

I'm not sure that REGIONB_SIZE(cq) != 0) is what you want here. First, the size should never be less than zero. If it is, then there is a bug. You tell it to continue in the face of this bug. This moves the bug from region B to region A. However, you can essentially eliminate that copy with no cost (no extra checks or comparisons):

    if ( REGIONA_SIZE(cq) == 0 )
    {
        if ( REGIONB_SIZE(cq) > 0 ) // is RegionA empty while RegionB isn't?
        {
            // RegionB becomes RegionA
            cq->ra_begin = cq->rb_begin;
            cq->ra_end   = cq->rb_end;
        }
        else
        {
            cq->ra_begin = BUFFER_BEGIN(cq);
            cq->ra_end   = cq->ra_begin;
        }

        REGIONB_DEACTIVATE(cq);
    }

Now we know for sure that region B is deactivated if region A is empty. If region B was in a bad state, we dumped it rather than copy to A. It's possible that we may deactivate an already inactive B, but this avoids the situation where we have an active but empty region B. Note that an active but empty B is risky in the face of future changes. For example, what if you could remove arbitrary elements and not just pop and push elements? Then B might not start at the beginning of the buffer, so you could end up with a B that was inside A.

This also sets A to start at the beginning of the buffer. You don't need to do this, but it can help reduce how often you create a region B.

/// @warning If the circular queue is full, the new byte will overwrite the top byte.

Perhaps this is the correct behavior for your application, but it seems odd. This would seem better as an error case for most applications.

char   *rb_begin;   ///< RegionB beginning.

Why do you need this? The way that your application works, region B always starts at the beginning of the buffer. There is never a time when you change rb_begin other than to set it either to NULL or the beginning of the buffer. Why not get rid of it entirely? Then you'd just have to maintain rb_end. That would also eliminate an entire class of problems that could arise because someone modified rb_begin outside your methods. Note that you can read from BUFFER_BEGIN(cq) where you currently read from rb_begin. You can drop the two assignments:

#define REGIONB_ACTIVATE(pcq)                           \
if ((pcq)->rb_end == NULL) {                            \
    (pcq)->rb_end       = BUFFER_BEGIN(pcq);            \
} else (void)0

#define REGIONB_DEACTIVATE(pcq) (pcq)->rb_end = NULL

This actually makes REGIONB_DEACTIVATE quite a bit simpler.

static inline size_t cqueue_size(const cqueue_t *cq)
{
    return REGIONA_SIZE(cq) + REGIONB_SIZE(cq);
}

Note that the two size macros return ptrdiff_t while the function returns a size_t. I don't know that there will ever be a bug as a result, but you should be aware of the implicit cast. If nothing else, note that you are casting a signed value to unsigned.

I'm not crazy about the use of pointers here. It's not wrong. It's just that it leads to odd things like BUFFER_END(cq) being outside the buffer. It also limits you in some places:

if (cq->ra_end == BUFFER_END(cq)) // is RegionA full?

I would prefer to write that with an inequality. The reason being that an equality like this is fragile in the face of other changes. Note that if cq->ra_end > BUFFER_END(cq) somehow becomes true, your code will happily keep pushing into unknown space until it crashes. By contrast, if you were using indexes to track these, you could write it with an inequality that would halt even when in an inconsistent state.

Strictly speaking, these are not bugs. This is a place where a bug could be created by interactions with other code (not included in this program now). Such code could be external to the code here, or it could be new functionality that someone adds to this library. I mention them because it may be worthwhile to write your code in such a way that it is more robust in the face of future code.

return cq->ra_begin == cq->rb_end || REGIONA_SIZE(cq) == BUFFER_SIZE;

The latter portion should be an inequality:

return cq->ra_begin == cq->rb_end || REGIONA_SIZE(cq) >= BUFFER_SIZE;

It should never happen, but if it does, at least you'll stop putting things in A.

    *(cq->rb_end)++ = c;

What happens if cq->rb_end > BUFFER_END(cq)? Note that if cq->ra_begin is in an inconsistent state, you can add to rb_end indefinitely. The assumption is that you will never increment cq->rb_end that far because then cq->ra_begin would hit BUFFER_END(cq) and turn off B. What happens if you (or someone else) change that behavior? Will you remember that you have to add additional checks for cq->rb_end?