Boost shared memory lockfree circular buffer queue

consumer.cpp

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

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

namespace bip = boost::interprocess;
namespace shm
{
    typedef bip::allocator<char, bip::managed_shared_memory::segment_manager> char_alloc;
    typedef bip::basic_string<char, std::char_traits<char>, char_alloc >      shared_string;

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

#include <iostream>

int main()
{
    // create segment and corresponding allocator
    bip::managed_shared_memory segment(bip::open_or_create, "MySharedMemory", 65536);
    shm::char_alloc char_alloc(segment.get_segment_manager());

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

    while (true)
    {
        shm::shared_string v(char_alloc);
        if (queue->pop(v))
            std::cout << "Processed: '" << v << "'\n";
    }
}

Makefile

all:consumer producer

CPPFLAGS+=-std=c++03 -Wall -pedantic
CPPFLAGS+=-g -O0
CPPFLAGS+=-isystem ~/custom/boost/

LDFLAGS+=-L ~/custom/boost/stage/lib/ -Wl,-rpath,/home/sehe/custom/boost/stage/lib
LDFLAGS+=-lboost_system -lrt -lpthread

%:%.cpp
	$(CXX) $(CPPFLAGS) $^ -o $@ $(LDFLAGS)

producer.cpp

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

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

namespace bip = boost::interprocess;
namespace shm
{
    typedef bip::allocator<char, bip::managed_shared_memory::segment_manager> char_alloc;
    typedef bip::basic_string<char, std::char_traits<char>, char_alloc >      shared_string;

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

#include <unistd.h>

int main()
{
    // create segment and corresponding allocator
    bip::managed_shared_memory segment(bip::open_or_create, "MySharedMemory", 65536);
    shm::char_alloc char_alloc(segment.get_segment_manager());

    // Ringbuffer fully constructed in shared memory. The element strings 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")();

    const char* messages[] = { "hello world", "the answer is 42", "where is your towel", 0 };

    for (const char** msg_it = messages; *msg_it; ++msg_it)
    {
        sleep(1);
        queue->push(shm::shared_string(*msg_it, char_alloc));
    }
}

I'm trying to compile your snippet but unfortunately it doesnt work. Compiling with g++-4.6 and boost 1.54.

The error is:

g++ -std=c++03 -Wall -pedantic -g -O0 -isystem /opt/boost/include consumer.cpp -o consumer -L /opt/boost/lib/ -Wl,-rpath,/opt/boost/lib -lboost_system -lrt -lpthread
In file included from /opt/boost/include/boost/interprocess/containers/string.hpp:19:0,
from consumer.cpp:5:
/opt/boost/include/boost/container/string.hpp: In constructor ‘boost::container::container_detail::basic_string_base::members_holder::members_holder() [with Allocator = boost::interprocess::allocator<char, boost::interprocess::segment_manager<char, boost::interprocess::rbtree_best_fitboost::interprocess::mutex_family, boost::interprocess::iset_index> >]’:
/opt/boost/include/boost/container/string.hpp:104:18: instantiated from ‘boost::container::container_detail::basic_string_base::basic_string_base() [with Allocator = boost::interprocess::allocator<char, boost::interprocess::segment_manager<char, boost::interprocess::rbtree_best_fitboost::interprocess::mutex_family, boost::interprocess::iset_index> >]’
/opt/boost/include/boost/container/string.hpp:590:16: instantiated from ‘boost::container::basic_string<CharT, Traits, Allocator>::basic_string() [with CharT = char, Traits = std::char_traits, Allocator = boost::interprocess::allocator<char, boost::interprocess::segment_manager<char, boost::interprocess::rbtree_best_fitboost::interprocess::mutex_family, boost::interprocess::iset_index> >]’
/opt/boost/include/boost/array.hpp:60:11: instantiated from ‘boost::lockfree::spsc_queue<T, A0, A1>::spsc_queue() [with T = boost::container::basic_string<char, std::char_traits, boost::interprocess::allocator<char, boost::interprocess::segment_manager<char, boost::interprocess::rbtree_best_fitboost::interprocess::mutex_family, boost::interprocess::iset_index> > >, A0 = boost::lockfree::capacity<200ul>, A1 = boost::parameter::void_]’
/opt/boost/include/boost/interprocess/detail/named_proxy.hpp:155:10: instantiated from ‘void boost::interprocess::ipcdetail::Ctor0Arg::construct_n(void_, std::size_t, std::size_t&) [with T = boost::lockfree::spsc_queue<boost::container::basic_string<char, std::char_traits, boost::interprocess::allocator<char, boost::interprocess::segment_manager<char, boost::interprocess::rbtree_best_fitboost::interprocess::mutex_family, boost::interprocess::iset_index> > >, boost::lockfree::capacity<200ul> >, std::size_t = long unsigned int]’
consumer.cpp:35:1: instantiated from here
/opt/boost/include/boost/container/string.hpp:218:22: error: no matching function for call to ‘boost::interprocess::allocator<char, boost::interprocess::segment_manager<char, boost::interprocess::rbtree_best_fitboost::interprocess::mutex_family, boost::interprocess::iset_index> >::allocator()’
/opt/boost/include/boost/container/string.hpp:218:22: note: candidates are:
/opt/boost/include/boost/interprocess/allocators/allocator.hpp:140:4: note: template boost::interprocess::allocator::allocator(const boost::interprocess::allocator<T2, SegmentManager>&)
/opt/boost/include/boost/interprocess/allocators/allocator.hpp:134:4: note: boost::interprocess::allocator<T, SegmentManager>::allocator(const boost::interprocess::allocator<T, SegmentManager>&) [with T = char, SegmentManager = boost::interprocess::segment_manager<char, boost::interprocess::rbtree_best_fitboost::interprocess::mutex_family, boost::interprocess::iset_index>, boost::interprocess::allocator<T, SegmentManager> = boost::interprocess::allocator<char, boost::interprocess::segment_manager<char, boost::interprocess::rbtree_best_fitboost::interprocess::mutex_family, boost::interprocess::iset_index> >]
/opt/boost/include/boost/interprocess/allocators/allocator.hpp:134:4: note: candidate expects 1 argument, 0 provided
/opt/boost/include/boost/interprocess/allocators/allocator.hpp:129:4: note: boost::interprocess::allocator<T, SegmentManager>::allocator(boost::interprocess::allocator<T, SegmentManager>::segment_manager_) [with T = char, SegmentManager = boost::interprocess::segment_manager<char, boost::interprocess::rbtree_best_fitboost::interprocess::mutex_family, boost::interprocess::iset_index>, boost::interprocess::allocator<T, SegmentManager>::segment_manager = boost::interprocess::segment_manager<char, boost::interprocess::rbtree_best_fitboost::interprocess::mutex_family, boost::interprocess::iset_index>]
/opt/boost/include/boost/interprocess/allocators/allocator.hpp:129:4: note: candidate expects 1 argument, 0 provided

Which boost version did you use to compile this example?

I was able to build it.
boost: 1.56.0
c++: c++ (Gentoo 4.9.3 p1.5, pie-0.6.4) 4.9.3

I didn't use the Makefile. I just created a shell script:

#!/bin/sh

c++ -std=c++14 -lboost_system -lpthread -lrt consumer.cpp -o consumer
c++ -std=c++14 -lboost_system -lpthread -lrt producer.cpp -o producer

And thank you sehe for the quick easy example.

Hi sehe,

Thanks for the example.

I am looking for a float array example instead of string array. Can you give an example how to code that? Basically what is the equivalent of basic_string for float array?

Thanks,
Narsi

With CMake:

cmake_minimum_required (VERSION 3.6.2 FATAL_ERROR)

find_package(Boost 1.64.0 REQUIRED COMPONENTS system)
include_directories(${Boost_INCLUDE_DIRS})

add_executable (consumer consumer.cpp)
target_link_libraries(consumer ${Boost_LIBRARIES})

add_executable (producer producer.cpp)
target_link_libraries(producer ${Boost_LIBRARIES})

Adding the updated CMakeLists.txt

cmake_minimum_required (VERSION 3.6.2 FATAL_ERROR)

SET (CMAKE_VERBOSE_MAKEFILE 0)

set(THREADS_PREFER_PTHREAD_FLAG ON)
find_package(Threads REQUIRED)

find_package(Boost 1.64.0 REQUIRED COMPONENTS system)
include_directories(${Boost_INCLUDE_DIRS})

add_executable (consumer consumer.cpp)
target_link_libraries(consumer LINK_PUBLIC 
	${Boost_LIBRARIES}
	rt
	Threads::Threads
)

add_executable (producer producer.cpp)
target_link_libraries(producer LINK_PUBLIC 
	${Boost_LIBRARIES}
	rt
	Threads::Threads
)