README.md

Boost shared memory lockfree circular buffer queue for custom struct

References

• Boost shared memory lockfree circular buffer queue
  • Cosmetic modifications
• Boost shared memory and synchronized queue

CMakeLists.txt

cmake_minimum_required(VERSION 3.0.0)
project(circular-buffer-queue VERSION 0.1.0)

set(Boost_USE_MULTITHREADED OFF)
add_definitions(-DBOOST_ALL_NO_LIB)
find_package( Boost 1.54 COMPONENTS system thread REQUIRED )
find_package( OpenCV REQUIRED )

include_directories(
    ${Boost_INCLUDE_DIRS}
    ${OpenCV_INCLUDE_DIRS})

set(LINK_LIBRARIES
    rt
    pthread
    ${Boost_LIBRARIES}
    ${OpenCV_LIBS})

add_executable(circular-buffer-queue-consumer consumer.cpp)
target_link_libraries(circular-buffer-queue-consumer ${LINK_LIBRARIES})

add_executable(circular-buffer-queue-producer producer.cpp)
target_link_libraries(circular-buffer-queue-producer ${LINK_LIBRARIES})

consumer.cpp

#include "shared_memory_struct_ring_buffer.h"

#include <iostream>

#include <boost/thread/thread.hpp>
#include <boost/random.hpp>

int main() {
    boost::mt19937 generator;
    boost::uniform_int<> range( 10, 30 );
    boost::variate_generator< boost::mt19937, boost::uniform_int<> > nextRandomValue(generator, range);

    bip::managed_shared_memory segment(bip::open_or_create, "MySharedMemory", 65536);

    shm::ring_buffer *queue = segment.find_or_construct<shm::ring_buffer>("queue")();

    while (true) {
        gps_position pos;
        if (queue->pop(pos))
            std::cout << "Consumer #" << getpid() << ": " << pos.degrees << "° " << pos.minutes << "' " <<  pos.seconds << '"' << std::endl;
        
        boost::this_thread::sleep(boost::posix_time::milliseconds(nextRandomValue()));
    }
}

producer.cpp

#include "shared_memory_struct_ring_buffer.h"

#include <unistd.h>

#include <boost/thread/thread.hpp>
#include <boost/random.hpp>

int main() {
    boost::mt19937 generator;
    boost::uniform_int<> range( 50, 100 );
    boost::variate_generator< boost::mt19937, boost::uniform_int<> > nextRandomValue(generator, range);

    bip::managed_shared_memory segment(bip::open_or_create, "MySharedMemory", 65536);
    
    // Ringbuffer fully constructed in shared memory. The struct elements are
    // also allocated from the same shared memory segment. This vector can be
    // safely accessed from other processes.
    shm::ring_buffer *queue = segment.find_or_construct<shm::ring_buffer>("queue")();

    while (true) {
        gps_position pos(nextRandomValue(), nextRandomValue(), nextRandomValue());
        queue->push(pos);

        std::cout << "Producer #" << getpid() << ": " << pos.degrees << "° " << pos.minutes << "' " <<  pos.seconds << '"' << std::endl;
        40° 26′ 46″
        
        boost::this_thread::sleep(boost::posix_time::milliseconds(nextRandomValue()));
    }
}

shared_memory_struct_ring_buffer.h

#ifndef SHARED_MEMORY_STRUCT_RING_BUFFER_H 
#define SHARED_MEMORY_STRUCT_RING_BUFFER_H 

#include <boost/lockfree/spsc_queue.hpp> // ring buffer

#include <boost/interprocess/managed_shared_memory.hpp>
#include <boost/interprocess/allocators/allocator.hpp>

struct gps_position {
    int degrees;
    int minutes;
    float seconds;

    gps_position(int d=0, int m=0, float s=0) : degrees(d), minutes(m), seconds(s) {}
};

namespace bip = boost::interprocess;
namespace shm
{
    typedef bip::allocator<gps_position, bip::managed_shared_memory::segment_manager> struct_alloc;

    typedef boost::lockfree::spsc_queue<
        gps_position, 
        boost::lockfree::capacity<200>,
        struct_alloc
    > ring_buffer;
}

#endif // SHARED_MEMORY_STRUCT_RING_BUFFER_H