2
\$\begingroup\$

Just wrote this small ring buffer system for an embedded device using C++98. Looking for cc, advice and bugs.

Please check it out!

https://github.com/Bambofy/EmbeddedRingBuffer

/* Changelog ------------------------------------------------------------------
 *  06/06/2021     Initial version
 * ---------------------------------------------------------------------------*/

#ifndef RINGBUFFER_RINGBUFFER_H
#define RINGBUFFER_RINGBUFFER_H

#include "Block.h"

/**
 * @brief            A ring buffer is a FIFO structure that can be used to
 *                     spool data between devices.
 *
 *                     There is a Skip() function that allows the client to
 *                     control when the read cursor is changed. This is so the
 *                     client can perform an action after Read() without the
 *                     write cursor overwriting data while the read block is used.
 *
 *                     For e.g with the sequence of events:
 *                         1.    Read(1000, false)
 *                         2.    Busy writing to sd card for 5 seconds
 *                         3.    Skip()
 *
 *                     Because the skip isn't called until the writing
 *                     has finished, another thread can .Append() without
 *                     corrupting the data being written.
 *
 *
 * @attention        The ring buffer can only contain Length-1 number of entries,
 *                   because the last index is reserved for overrun checks.
 *
 * @tparam Length    The length of the backing store array.
 * @tparam T         The type of data stored.
 */
template<unsigned int LENGTH, class T>
class RingBuffer
{
public:
    RingBuffer() : read_position(0), write_position(0)
    {
    }

    ~RingBuffer()
    {
        memset(data, 0, LENGTH);
    }

    /**
     * @brief     Appends a value the end of the
     *            buffer.
     */
    void Append(T value)
    {
        /*
         * If the next position is where the read cursor
         * is then we have a full buffer.
         */
        bool buffer_full;

        buffer_full = ((write_position + 1U) % LENGTH) == read_position;

        if (buffer_full)
        {
            /*
             * Tried to append a value while the buffer is full.
             */
            overrun_flag = true;
        }
        else
        {
            /*
             * Buffer isn't full yet, write to the curr write position
             * and increment it by 1.
             */
            overrun_flag         = false;
            data[write_position] = value;
            write_position       = (write_position + 1U) % LENGTH;
        }
    }

    /**
     * @brief                        Retrieve a continuous block of
     *                               valid buffered data.
     * @param num_reads_requested    How many reads are required.
     * @param skip                   Whether to increment the read position
     *                               automatically, (false for manual skip
     *                               control)
     * @return                       A block of items containing the maximum
     *                               number the buffer can provide at this time.
     */
    Block<T> Read(unsigned int num_reads_requested, bool skip = true)
    {
        bool bridges_zero;
        Block<T> block;

        /*
          * Make sure the number of reads does not bridge the 0 index.
          * This is because we can only provide 1 contiguous block at
          * a time.
          */
        bridges_zero = (read_position > write_position);

        if (bridges_zero)
        {
            unsigned int reads_to_end;
            bool req_surpasses_buffer_end;

            reads_to_end             = LENGTH - read_position;
            req_surpasses_buffer_end = num_reads_requested > reads_to_end;

            if (req_surpasses_buffer_end)
            {
                /*
                 * If the block requested exceeds the buffer end. Then
                 * return a block that reaches the end and no more.
                 */
                block.SetStart(&(data[read_position]));
                block.SetLength(reads_to_end);

                if (skip)
                {
                    read_position = (read_position + reads_to_end) % LENGTH;
                }
            }
            else
            {
                /*
                 * If the block requested does not exceed 0
                 * then return a block that reaches the number of reads required.
                 */
                block.SetStart(&(data[read_position]));
                block.SetLength(num_reads_requested);

                if (skip)
                {
                    read_position =
                            (read_position + num_reads_requested) % LENGTH;
                }
            }
        }
        else
        {
            /*
             * If the block doesn't bridge the zero then
             * return the maximum number of reads to the write
             * cursor.
             */
            unsigned int max_num_reads;
            unsigned int num_reads_to_write_position;

            num_reads_to_write_position = (write_position - read_position);

            if (num_reads_requested > num_reads_to_write_position)
            {
                /*
                 * If the block length requested exceeds the
                 * number of items available, then restrict
                 * the block length to the distance to the write position.
                 */
                max_num_reads = num_reads_to_write_position;
            }
            else
            {
                /*
                 * If the block length requested does not exceed the
                 * number of items available then the entire
                 * block is valid.
                 */
                max_num_reads = num_reads_requested;
            }

            block.SetStart(&(data[read_position]));
            block.SetLength(max_num_reads);

            if (skip)
            {
                read_position = (read_position + max_num_reads) % LENGTH;
            }
        }

        return block;
    }

    /**
     * @brief    Advances the read position.
     *
     */
    void Skip(unsigned int num_reads)
    {
        read_position = (read_position + num_reads) % LENGTH;
    }

    bool Overrun()
    {
        return overrun_flag;
    }

    unsigned int Length()
    {
        return LENGTH;
    }

private:
    unsigned int read_position;
    unsigned int write_position;

    T data[LENGTH];

    bool overrun_flag;
};

#endif

/* Changelog ------------------------------------------------------------------
 *  06/06/2021     Initial version
 * ---------------------------------------------------------------------------*/

#ifndef RINGBUFFER_BLOCK_H
#define RINGBUFFER_BLOCK_H

#include <cstddef>

/**
 * @brief        A block represents a continuous section
 *               of the ring buffer.
 * @tparam T     The type of data stored in the ring buffer.
 */
template<class T>
class Block
{
public:
    Block() : start(nullptr), length(0)
    {
    }

    ~Block()
    {
    }

    /**
     * @brief    Sets the block's starting
     *           position to a point in memory.
     */
    void SetStart(T* start)
    {
        this->start = start;
    }

    /**
     * @brief    Sets the number of items in the
     *           block.
     */
    void SetLength(unsigned int length)
    {
        this->length = length;
    }

    /**
     * @return    The block's starting
     *            point in memory.
     */
    T* Start()
    {
        return this->start;
    }

    /**
     * @return    The number of items in the block.
     */
    unsigned int Length()
    {
        return this->length;
    }

    /**
     * @param index        The index of the item in the block.
     * @return             The item in the block at the index.
     */
    T At(unsigned int index)
    {
        return this->start[index];
    }


private:
    T* start;

    unsigned int length;
};


#endif

Here is a test:

#include <iostream>

#include "RingBuffer.h"

int main()
{
    RingBuffer<100, int> buffer;
    Block<int> block;

    /* Write 100 ints */
    for (int i = 0; i < buffer.Length(); i++)
    {
        buffer.Append(i);
    }

    /* Read a block */
    block = buffer.Read(100);

    /* Print out the block */
    for (int i = 0; i < block.Length(); i++)
    {
        std::cout << block.At(i) << std::endl;
    }

    /* Read another block */
    block = buffer.Read(1000);

    /* Print out the block */
    for (int i = 0; i < block.Length(); i++)
    {
        std::cout << block.At(i) << std::endl;
    }

    return 0;
}
\$\endgroup\$
7
  • \$\begingroup\$ There was a bit of a formatting problem with your code apparently, the current version should be better. Is there a reason your ifndef talks about a RINGBUFFER_RINGBUFFER_H (double ringbuffer) or is this a typo? \$\endgroup\$
    – Mast
    Jun 6 at 11:26
  • \$\begingroup\$ Yes the double ringbuffer include guard is a typo, I had to change everything froms tabs to spaces \$\endgroup\$ Jun 6 at 11:52
  • 1
    \$\begingroup\$ This code is tagged C++98, but uses nullptr. I think you may not be using the version of C++ you think you’re using. (Nobody is still using C++98 anymore these days.) \$\endgroup\$
    – indi
    Jun 6 at 12:16
  • \$\begingroup\$ @indi thanks for the heads up i'll change this to 0 \$\endgroup\$ Jun 6 at 12:35
  • 2
    \$\begingroup\$ The point about your code not being C++98 wasn’t to make your code worse in order to make it 98-compatible, it was to give up on being 98-compatible and upgrade your target version. You mentioned multi-threading: that is literally impossible to do properly in C++98; if you want to properly support multiple threads, you need at least C++11 for its multi-thread-aware memory model (not to mention stuff like std::atomic). \$\endgroup\$
    – indi
    Jun 6 at 16:10
4
\$\begingroup\$
  • A LENGTH confusion.

    In the destructor the memset(data, 0, LENGTH); clears LENGTH bytes, whereas a definition of

    T data[LENGTH]
    

    says there are LENGTH items of sizeof T. A bit more than LENGTH bytes.

    As a side note, the destructor (after fixing the LENGTH issue) works correctly only if T is trivially constructible. BTW, if it is so, do you need it at all?

  • Unrestricted getters and setters defeat the purpose of having start and length private to Block. It is cleaner to define

      Block::Block(T * start, size_t size)
    

    and get rid of setters whatsoever.

  • Every branch of Read does block.SetStart(&(data[read_position]));. Make it clear and lift it out of the if/else cascade. See the next bullet for more.

  • brigdes_zero = (read_position > write_position); does not care of num_reads_requested, and therefore looks suspicious. I understand the underlying motivation, but there is a cleaner way to express it. Consider

      read_hard_end = (read_position > write_position)? LENGTH: write_position;
      max_reads_available = read_hard_end - write_position;
      read_size = std::min(num_read_requested, max_reads_available);
    
      Block<T> block(&data[read_position], read_size);
      return block;
    

    (I intentionally skip the Skip handling).

  • is not . All those this-> only add noise.

\$\endgroup\$
1
  • \$\begingroup\$ Thank you for the cc, i have removed the memset() and researching into your suggestions \$\endgroup\$ Jun 7 at 8:36
3
\$\begingroup\$

This looks quite good, some minor changes could be made though:

Return overrun_flag instead of storing it

Especially on an embedded system, you don't want to store data unnecessarily. Instead of having a member variable overrun_flag and having Append() set it, why not have Append() return a bool indicating whether the attempt to append succeeded or not?

This also forces the caller to immediately check the value, and there is no possibility that another call to Append() might reset overrun_flag to false before it tries to check whether an earlier call to Append() failed.

Pass values as const references where appropriate

If T is an int, passing it by value to a function is fine. But if it becomes a larger struct, maybe one with constructors, this might become inefficient. Consider passing them by const reference instead:

bool Append(const T &value) {
    if ((write_position + 1U) % LENGTH == read_position)
        return false;

    data[write_position] = value;
    write_position       = (write_position + 1U) % LENGTH;
    return true;
}

About Block

Your Block class is a nice way of providing a view of consecutive items in the ringbuffer. However, there are two issues with it.

The most important one is why have a skip parameter? Surely if you immediately update the read_position, another thread might start overwriting the data, before the thread calling Read() had a chance to read the data. I would therefore remove this parameter, and always force the reader to call Skip() manually. Even better, you could make a Block a RAII type, and have it automatically call Skip() when its destructor is called.

But the second problem is that while it looks efficient, it will actually prevent writers from writing to the ring buffer for longer than necessary, because they have to wait for the reader to process the whole block. And if you are going to loop over the block sequentially anyway, then it would actually make more sense to just provide access to the first unread element, like std::queue does. That brings me to:

Consider copying std::queue's API

It's always easier for C++ programmers if the classes they use have a similar interface as those from similar classes in the standard library. It's even better if your own classes are drop-in replacements for STL classes. So in this case, I'd recommend you provide the same member functions as std::queue.

Thread safety

Since you mentioned multiple threads accessing the ringbuffer at the same time, you should definitely think about thread safety. Do you want to allow multiple readers and multiple writers, or just a single reader and/or a single writer? At the moment it is at best only safe for a single reader and a single writer. If that's what is intended, make sure you document that.

However, the question is if this is even safe for a single reader and a single writer accessing the data simultaneously. On 8-bit Atmel MCUs, 16-bit reads and writes are not atomic. You should therefore use some way to synchronize access to the read and write pointers. If you don't, then your code might seem to work as most of the time, the reads and writes won't be done at exactly the same time, but once in a while it might go wrong, and depending on what your embedded device is controlling, the consequences might be grave.

Check if your embedded device has support for atomic reads and writes to unsigned ints, and if not, use a semaphore or mutex primitive to ensure only one thread gets to update the read/write pointers at a time.

\$\endgroup\$
3
  • \$\begingroup\$ Thank you very much for your input! There's a skip parameter on the Read() function because if you dont want to use it in an ISR setting, you dont have to manually skip it. And the block is not ment to be iterated (this is the worst thing that can be done in terms of taking up cpu time), it is meant to be directly transferred using a DMA. The iteration problem is what made me create the skipping feature in the first place, iterating to pull data from the buffer one after another (say 30,000 times) takes too much time thats why you read from it in blocks. \$\endgroup\$ Jun 6 at 12:34
  • \$\begingroup\$ Ah OK, I only saw the Blocks being iterated over in your example, but indeed if you want to do DMA this is a good solution. I would still remove the skip parameter though, and make it more explicit. This will make the code more robust against future changes where you might introduce an ISR that writes to the ringbuffer for example. \$\endgroup\$
    – G. Sliepen
    Jun 6 at 15:35
  • \$\begingroup\$ Ok i'll remove the skip parameter, i agree, it is too hidden. thank you \$\endgroup\$ Jun 6 at 17:58
2
\$\begingroup\$

You don't need version history stored in the file -- that's what version control systems do for you.

~RingBuffer()
    {
        memset(data, 0, LENGTH);
    }

Why zero the memory that you're not going to use again? Is that for debugging? If you do need it, consider std::fill_n instead. But wait a minute... LENGTH is in bytes, but data is an array of T not a byte-buffer. So this is actually wrong, which is one good reason for preferring the C++ algorithms 😀. But does setting objects of type T to 0 even make sense? This is run before the destructors for the individual Ts so overwriting it like this can cause a lot of grief.


LENGTH should be size_t, not unsigned int.


Don't use std::endl (standard Code Review issue!)


write_position = (write_position + 1U) % LENGTH;
That is a very slow way to handle wrap-around. Even on CPUs that support division with a built-in instruction, it is remarkably slow. Worse yet, you're repeating it! Break out a variable, say, next_write_position, to use in all the places that need it.


bool buffer_full;

buffer_full = ((write_position + 1U) % LENGTH) == read_position;

So close... Define variables where you need them, and initialize in the declaration. And use const where you can. So:
const bool buffer_full = ... ;


Use standard names for members that do well-understood things. Using this, it would be a pain to learn that I have to use Length instead of size, Append instead of push_back, etc. Make it fit with what people already know. This is a sequential container that works like a deque, so it should have the same API to the extent possible.


bool Overrun()
    {
        return overrun_flag;
    }

(and others) Access-only member functions should be const.


start(nullptr)
You don't have nullptr in C++98; that was added in C++11. Use the literal constant 0.


Block::Setstart and Setlength:
No... just make those constructor arguments. They are only used to set up the block object to be returned, and always called together. Just record the start/length as local variables in the function and then Block<T>(startpos,length); at the end. Don't define block at the top of the function.

\$\endgroup\$
1
0
\$\begingroup\$

Review addressing the embedded system aspects, assuming the target is Arduino/AVR 328P:

  • Implementing destructors for a class to be used in a bare metal embedded system is very fishy. If you find yourself in a situation where you need to push/pop ring buffers on the stack, you are doing something very wrong. Objects of such classes should always be allocated with static storage duration.

  • RingBuffer<100, int> buffer; is wrong in several ways - it should preferably have static storage duration as mentioned, but more importantly you shouldn't slaughter some 200+ bytes of your stack just like that. Assuming this is for AVR 328P then that's 10% out of your total RAM blown. And maybe some 50% of the stack?

    In addition, there's an old cynical rule saying that we should never allocate any buffers at all on the stack in embedded systems, because buffer overruns and stack overflows are such common bugs.

  • Embedded systems should always use the stdint.h types. This is an 8-bitter so all arithmetic should preferably be done on uint8_t integers whenever possible. You shouldn't use int which is 16 bit and also needlessly signed - signed types are dangerous to use in case they end up in bitwise arithmetic, as is common in embedded systems. Getting rid of int and limiting the max size of the ring buffer to at most 255 will speed up this code considerably.

  • As mentioned in other reviews, any decent ring buffer comes with a "thread-safety" option. In this case interrupt safety, since ring buffers are most commonly used when communicating between for example an UART rx ISR and the UART driver.

    There's a common misconception that the size of variables vs data size of the CPU somehow matters for atomicity. That's plain wrong, C++ code can end up non-atomic even if doing 8 bit access on a 8 bit CPU. You need to use the actual atomic feature (C++11 only?) or inline assembler, or the variables cannot be assumed to be atomic. More info here: https://electronics.stackexchange.com/a/409570/6102.

    The most common way to prevent race conditions in ring buffers is simply to accept a pointer to an interrupt enable register and a mask. So if the UART driver is to utilize the ring buffer, it would pass along a volatile pointer to the UART rx enable/disable register and the ring buffer can simply shut off the interrupt. In the specific case of UART, the data is very slow compared to the program execution, so disabling the interrupt for a few microseconds won't even lead to data loss. Or in case you need tougher real-time than that, you shouldn't be using classes in the first place.

  • Bare metal systems don't return from main(). Who would they return to? You need to have an eternal loop in main(), otherwise you'll just generate pointless bloat return code that chews up stack and flash needlessly. Ideally embedded systems use the implementation-defined form void main (void) but unfortunately there's a misguided PC programmer rule in the C++ standard saying that if a program defines a function main() it must return int. Freestanding systems are not exempt from this rule for reasons unknown. You can avoid this C++ language design flaw by using gcc for AVR and compile for embedded systems with -ffreestanding. (The same option works in C too.)

    To illustrate, here's the code with extra bloat generated by incorrect form int main() with return:

      main:
          push r28
          push r29
          in r28,__SP_L__
          in r29,__SP_H__
          ldi r24,0
          ldi r25,0
          pop r29
          pop r28
          ret
    

    And here's the code generated by correct form void main (void) enabled with -ffreestanding and then an eternal loop:

      main:
          push r28
          push r29
          in r28,__SP_L__
          in r29,__SP_H__
      .L2:
          rjmp .L2
    

    (It's weird that they set the SP from main() but that's another story...)

\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.