TripleBuffer.h

Go to the documentation of this file.

// Modified by Raphael Grimm, 2017
// changed interface / renamed functions
//
// Modified by Adrian Gierakowski, 2014
//
// added (const T&) constructor to enable custom buffer initialisation
// added getReadRef and getWriteRef to enable by reference access to
// objects contained in the buffer
//
// minore code cleanup
//
//
// ORIGINAL COPYRIGHT NOTICE
//==============================================================================
// Name        : TripleBuffer.hxx
// Author      : André Pacheco Neves
// Version     : 1.0 (27/01/13)
// Copyright   : Copyright (c) 2013, André Pacheco Neves
//               All rights reserved.
 
/*
Redistribution and use in source and binary forms, with or without
modification, are permitted provided that the following conditions are met:
- Redistributions of source code must retain the above copyright
notice, this list of conditions and the following disclaimer.
- Redistributions in binary form must reproduce the above copyright
notice, this list of conditions and the following disclaimer in the
documentation and/or other materials provided with the distribution.
- Neither the name of the <organization> nor the
names of its contributors may be used to endorse or promote products
derived from this software without specific prior written permission.
THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
DISCLAIMED. IN NO EVENT SHALL <COPYRIGHT HOLDER> BE LIABLE FOR ANY
DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
(INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
*/
 
// Description :
//    Template class for a TripleBuffer as a concurrency mechanism,
//    using atomic operations
// Credits :
//    http://remis-thoughts.blogspot.pt/2012/01/triple-buffering-as-concurrency_30.html
//==============================================================================
 
#pragma once
 
#include <atomic>
#include <iostream>
#include <type_traits>
 
#include <SimoxUtility/threads/system_thread_id.h>
 
namespace armarx
{
    /**
     * @brief A simple triple buffer for lockfree comunication between a single writer and a single reader.
     *
     * If more than one reader/writer is required, you have to synchronize the access.
     * Writers must not have: writes to the same data cell (except it is an atomic et al) and no writes are allowed during swapping of the buffer.
     * Readers must not read when the buffer is swapped.
     * If a writer only writes parts of the data structure, data loss could occure! (use a WriteBufferedTripleBuffer instead)
     *
     */
    template <typename T>
    class TripleBuffer
    {
    public:
#define DEBUG_MODE_PRINT                                                                           \
    if (debugMode.load())                                                                          \
    std::cout << '[' << simox::system_thread_id() << "] [" << this << "] "
 
        TripleBuffer(T&& initR, T&& initH, T&& initW)
        {
            buffer[getReadBufferIndex()] = std::move(initR);
            buffer[getHiddenBufferIndex()] = std::move(initH);
            buffer[getWriteBufferIndex()] = std::move(initW);
        }
 
        template <class... Ts>
        TripleBuffer(const Ts&... ts) : buffer{T(ts...), T(ts...), T(ts...)}
        {
        }
 
        /// @return the write buffer
        T&
        getWriteBuffer()
        {
            DEBUG_MODE_PRINT << "access write buffer\n";
            return buffer[getWriteBufferIndex()];
        }
 
        /// @return the write buffer
        const T&
        getWriteBuffer() const
        {
            DEBUG_MODE_PRINT << "access write buffer\n";
            return buffer[getWriteBufferIndex()];
        }
 
        /// @return the current read buffer
        const T&
        getReadBuffer() const
        {
            DEBUG_MODE_PRINT << "access read buffer\n";
            return buffer[getReadBufferIndex()];
        }
 
        /// @return the most up to date read buffer
        const T&
        getUpToDateReadBuffer() const
        {
            updateReadBuffer();
            return getReadBuffer();
        }
 
        /// @return the current read buffer (sometimes required for more complex initialization)
        T&
        _getNonConstReadBuffer()
        {
            DEBUG_MODE_PRINT << "access non const read buffer\n";
            return buffer[getReadBufferIndex()];
        }
 
        /// @return the current read buffer (sometimes required for more complex initialization)
        T&
        _getNonConstHiddenBuffer()
        {
            DEBUG_MODE_PRINT << "access non const hidden buffer\n";
            return buffer[getHiddenBufferIndex()];
        }
 
        /**
         * @brief Swaps in the hidden buffer if it contains new data.
         * @return True if new data is available
         */
        bool
        updateReadBuffer() const
        {
            uint_fast8_t flagsNow(flags.load(std::memory_order_consume));
 
            if (!hasNewWrite(flagsNow))
            {
                DEBUG_MODE_PRINT << "update read buffer. no update available\n";
                return false;
            }
 
            while (!flags.compare_exchange_weak(flagsNow,
                                                flagSwapReadWithHidden(flagsNow),
                                                std::memory_order_release,
                                                std::memory_order_consume))
                ;
            DEBUG_MODE_PRINT << "update read buffer. got update\n";
            return true;
        }
 
        void
        commitWrite()
        {
            DEBUG_MODE_PRINT << "commit write buffer.\n";
            swapWriteAndHiddenBuffer();
        }
 
        template <class U = T>
        typename std::enable_if<std::is_copy_constructible<U>::value>::type
        reinitAllBuffers(const T& init)
        {
            buffer[0] = init;
            buffer[1] = init;
            buffer[2] = init;
        }
 
        void
        reinitAllBuffers(T&& writeBuff, T&& hiddenBuff, T&& readBuff)
        {
            buffer[getWriteBufferIndex()] = std::move(writeBuff);
            buffer[getHiddenBufferIndex()] = std::move(hiddenBuff);
            buffer[getReadBufferIndex()] = std::move(readBuff);
        }
 
        void
        setDebugMode(bool mode)
        {
            std::cout << '[' << simox::system_thread_id() << "] [0x" << this
                      << "] set debug mode to " << mode << '\n';
            debugMode = mode;
        }
 
#undef DEBUG_MODE_PRINT
    protected:
        // non-copyable behavior
        TripleBuffer(const TripleBuffer&) = delete;
        TripleBuffer& operator=(const TripleBuffer&) = delete;
 
        // /////////////////////////////////////////////////////////////////////////////// //
        // get indices
        uint_fast8_t
        getReadBufferIndex() const
        {
            return (flags.load(std::memory_order_consume) & readBufferIndexMask) >>
                   readBufferIndexShift;
        }
 
        uint_fast8_t
        getWriteBufferIndex() const
        {
            return (flags.load(std::memory_order_consume) & writeBufferIndexMask) >>
                   writeBufferIndexShift;
        }
 
        uint_fast8_t
        getHiddenBufferIndex() const
        {
            return (flags.load(std::memory_order_consume) & hiddenBufferIndexMask) >>
                   hiddenBufferIndexShift;
        }
 
        // /////////////////////////////////////////////////////////////////////////////// //
        //buffer operations
        /// @brief Swap the write buffer with the hidden buffer
        void
        swapWriteAndHiddenBuffer()
        {
            uint_fast8_t flagsNow(flags.load(std::memory_order_consume));
            while (!flags.compare_exchange_weak(flagsNow,
                                                flagSwapWriteWithHidden(flagsNow),
                                                std::memory_order_release,
                                                std::memory_order_consume))
                ;
        }
 
        // void swapReadAndHiddenBuffer(); is done by updateReadBuffer()
 
        // /////////////////////////////////////////////////////////////////////////////// //
        //flag opertions
        // check if the newWrite bit is 1
        bool
        hasNewWrite(uint_fast8_t flags) const
        {
            return flags & dirtyBitMask;
        }
 
        /// swap read and hidden indexes
        /**
         * @brief swap read and hidden indexes of given flags (set dirty to 0)
         * @param flags the current flags
         * @return the new flags
         */
        static uint_fast8_t
        flagSwapReadWithHidden(uint_fast8_t flags)
        {
            return 0 //set dirty bit to 0
                   | ((flags & hiddenBufferIndexMask) >>
                      hiddenToReadShift) // hidden index now is read   index
                   | ((flags & readBufferIndexMask)
                      << hiddenToReadShift) // read   index now is hidden index
                   | (flags & writeBufferIndexMask); // keep write index
        }
 
        /**
         * @brief swap write and hidden indexes of given flags (set dirty to 1)
         * @param flags the current flags
         * @return the new flags
         */
        static uint_fast8_t
        flagSwapWriteWithHidden(uint_fast8_t flags)
        {
            return dirtyBitMask //set dirty bit to 1
                   | ((flags & hiddenBufferIndexMask)
                      << hiddenToWriteShift) // hidden index now is write  index
                   | ((flags & writeBufferIndexMask) >>
                      hiddenToWriteShift) // write  index now is hidden index
                   | (flags & readBufferIndexMask); // keep read index
        }
 
        // /////////////////////////////////////////////////////////////////////////////// //
        // constants
        static const uint_fast8_t readBufferIndexShift = 0;
        static const uint_fast8_t hiddenBufferIndexShift = 2;
        static const uint_fast8_t writeBufferIndexShift = 4;
        static const uint_fast8_t dirtyBitShift = 7;
 
        static const uint_fast8_t hiddenToReadShift = hiddenBufferIndexShift - readBufferIndexShift;
        static const uint_fast8_t hiddenToWriteShift =
            writeBufferIndexShift - hiddenBufferIndexShift;
 
        static const uint_fast8_t readBufferIndexMask = 3 << readBufferIndexShift; //0b00000011;
        static const uint_fast8_t hiddenBufferIndexMask = 3 << hiddenBufferIndexShift; //0b00001100;
        static const uint_fast8_t writeBufferIndexMask = 3 << writeBufferIndexShift; //0b00110000;
        static const uint_fast8_t dirtyBitMask = 1 << dirtyBitShift; //0b10000000;
        // initially dirty = 0, write 0, hidden = 1, read = 2
        static const uint_fast8_t initialFlags = 2 + (1 << 2); //0b00000110;
 
        // /////////////////////////////////////////////////////////////////////////////// //
        // data
        mutable std::atomic_uint_fast8_t flags{initialFlags};
        T buffer[3];
        std::atomic_bool debugMode{false};
    };
 
    /**
     * @see TripleBuffer
     * @brief Same as the TripleBuffer, but partial writes of the data structure are ok.
     * The write operation may be a bit slower and memory consumption may be 1/3 higher.
     */
    template <typename T>
    class WriteBufferedTripleBuffer
    {
    public:
        WriteBufferedTripleBuffer() : writeBuffer(T())
        {
        }
 
        WriteBufferedTripleBuffer(const T& init) : tripleBuffer{init}, writeBuffer(init)
        {
        }
 
        /// @return the write buffer
        T&
        getWriteBuffer()
        {
            return writeBuffer;
        }
 
        /// @return the write buffer
        const T&
        getWriteBuffer() const
        {
            return writeBuffer;
        }
 
        /// @return the current read buffer
        const T&
        getReadBuffer() const
        {
            return tripleBuffer.getReadBuffer();
        }
 
        /// @return the most up to date read buffer
        const T&
        getUpToDateReadBuffer() const
        {
            return tripleBuffer.getUpToDateReadBuffer();
        }
 
        /// @return the current read buffer (sometimes required for more complex initialization)
        T&
        _getNonConstReadBuffer()
        {
            return tripleBuffer._getNonConstReadBuffer();
        }
 
        /// @return the current read buffer (sometimes required for more complex initialization)
        T&
        _getNonConstHiddenBuffer()
        {
            return tripleBuffer._getNonConstHiddenBuffer();
        }
 
        /// @return the current read buffer (sometimes required for more complex initialization)
        T&
        _getNonConstHiddenWriteBuffer()
        {
            return tripleBuffer.getReadBuffer();
        }
 
        /**
         * @brief Swaps in the hidden buffer if it contains new data.
         * @return True if new data is available
         */
        bool
        updateReadBuffer() const
        {
            return tripleBuffer.updateReadBuffer();
        }
 
        void
        commitWrite()
        {
            tripleBuffer.getWriteBuffer() = writeBuffer;
            tripleBuffer.commitWrite();
        }
 
        void
        reinitAllBuffers(const T& init)
        {
            writeBuffer = init;
            tripleBuffer.reinitAllBuffers(init);
        }
 
        void
        setDebugMode(bool mode)
        {
            tripleBuffer.setDebugMode(mode);
        }
 
    private:
        TripleBuffer<T> tripleBuffer;
        T writeBuffer;
    };
} // namespace armarx