cv::Mat shared memory serializer deserializer into lockfree ring buffer

README.md

cv::Mat shared memory serializer deserializer into lockfree ring buffer

References

• Boost shared memory lockfree circular buffer queue
  • Cosmetic modifications
• Boost shared memory and synchronized queue
• Boost shared memory lockfree circular buffer queue for custom struct
• Runtime queue size definiton
• How to construct boost spsc_queue with runtime size parameter to exchange cv::Mat objects between two processes using shared memory?

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_ring_buffer_base.h"

#include <iostream>

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

#include <opencv2/opencv.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);

    auto& queue = *segment.find_or_construct<shm::BufferQueue>("queue")();

    while (true) {
        shm::Buffer v(segment.get_segment_manager());
        if (queue.pop(v)) {
            size_t offset = 0;
            shm::MatHeader& header = *((shm::MatHeader*) v.data()); offset += shm::MatHeader::size();

            cv::Mat mat(header.rows, header.cols, header.elemType, v.data() + offset, header.step);
            
            std::cout << mat.size() << " sum: " << cv::sum(mat)[0] << std::endl;
        }

        boost::this_thread::sleep(boost::posix_time::milliseconds(nextRandomValue()));
    }
}

producer.cpp

#include "shared_memory_ring_buffer_base.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::shared_memory_object::remove("MySharedMemory");

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

    auto& queue = *segment.find_or_construct<shm::BufferQueue>("queue")();
    
    while (true) {
        cv::Mat mat(20, 30, CV_8UC1, cvScalar(0));
        mat.data[10] = 1;
        mat.data[100] = 2;
        mat.data[200] = 3;

        shm::MatHeader header(mat.cols, mat.rows, mat.step[0], mat.elemSize(), mat.type());

        auto buf = shm::Buffer(header.totalSize(), segment.get_segment_manager());

        size_t offset = 0;
        memcpy(buf.data(), (uchar*) &header, shm::MatHeader::size()); offset += shm::MatHeader::size();
        memcpy(buf.data() + offset, mat.data, header.dataSize());
        queue.push(buf);

        std::cout << mat.size() << " sum: " << cv::sum(mat)[0] << std::endl;

        boost::this_thread::sleep(boost::posix_time::milliseconds(nextRandomValue()));
    }
}

shared_memory_ring_buffer_base.h

#ifndef SHARED_MEMORY_RING_BUFFER_BASE_H 
#define SHARED_MEMORY_RING_BUFFER_BASE_H 


#include <vector>

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

#include <opencv2/opencv.hpp>

namespace bip = boost::interprocess;
namespace bc = boost::container;
namespace shm
{
    using Mem = bip::managed_shared_memory;
    using Segment = Mem::segment_manager;

    template <typename T> using Alloc = bip::allocator<T, Segment>;
    template <typename T, int cap> using Queue = boost::lockfree::spsc_queue<T, boost::lockfree::capacity<cap>  >;
    using Buffer = std::vector<uchar, Alloc<uchar> >;
    using BufferQueue = Queue<Buffer, 1024>;

    typedef struct MatHeader_ {
        int cols;
        int rows;
        size_t step;
        size_t elemSize;
        size_t elemType;

        MatHeader_(int cols, int rows, size_t step, size_t elemSize, size_t elemType): 
            cols(cols), rows(rows), step(step), elemSize(elemSize), elemType(elemType)
        {}

        static size_t size() {
            return sizeof(MatHeader_);
        }

        size_t dataSize() {
            return rows * step;
        }

        size_t totalSize() {
            return dataSize() + size();
        }
    } MatHeader;
}

#endif // SHARED_MEMORY_RING_BUFFER_BASE_H