base.hpp

#include <boost/interprocess/sync/interprocess_mutex.hpp>
#include <boost/interprocess/containers/vector.hpp>
#include <boost/interprocess/allocators/allocator.hpp>

namespace ipc = boost::interprocess;

struct Data {
  bool end;
  ipc::interprocess_mutex mutex;
};

using ShmemAllocator = ipc::allocator<uint8_t, ipc::managed_shared_memory::segment_manager>;
using ShmemVector = std::vector<uint8_t, ShmemAllocator>;

build_and_run.sh`

#!/usr/bin/env bash

set -euxo pipefail

docker build -f Dockerfile.producer -t shmem_producer:latest . &
docker build -f Dockerfile.consumer -t shmem_consumer:latest . &
wait

docker run --name shmem_producer --ipc=shareable --rm shmem_producer &
sleep 5

docker run --ipc=container:shmem_producer --rm shmem_consumer &

Dockerfile.consumer

FROM alpine:latest

RUN apk add build-base
RUN apk add clang
RUN apk add boost-dev

COPY ./base.hpp .
COPY ./shmem_profile_consumer.cpp .
RUN clang++ -O3 -std=c++17 -o /usr/bin/consumer shmem_profile_consumer.cpp -lrt -I. -pthread

CMD [ "/usr/bin/consumer" ]

Dockerfile.producer

FROM alpine:latest

RUN apk add build-base
RUN apk add clang
RUN apk add boost-dev

COPY ./base.hpp .
COPY ./shmem_profile_producer.cpp .
RUN clang++ -O3 -std=c++17 -o /usr/bin/producer shmem_profile_producer.cpp -lrt -I. -pthread

CMD [ "/usr/bin/producer" ]

shmem_profile_consumer.cpp

#include <numeric>
#include <iostream>
#include <chrono>
#include <cstdlib>

#include <boost/interprocess/mapped_region.hpp>
#include <boost/interprocess/shared_memory_object.hpp>
#include <boost/interprocess/sync/scoped_lock.hpp>
#include <boost/interprocess/managed_shared_memory.hpp>

#include "base.hpp"

namespace ipc = boost::interprocess;

int main(int argc, const char* argv[]) {
  size_t nruns = 10000;
  std::cout << "nruns: " << nruns << '\n';

  using unit = std::chrono::microseconds;
  static const char* unit_name = "us";

  std::vector<unit> points;
  points.reserve(nruns);

  {
    ipc::shared_memory_object shm_obj(ipc::open_only, "shared_memory_mutex", ipc::read_write);
    ipc::mapped_region region(shm_obj, ipc::read_write);

    void* addr = region.get_address();

    Data* data = static_cast<Data*>(addr);
    if (data == nullptr) {
      std::cerr << "shared memory region is nullptr" << std::endl;
      return 1;
    }

    ipc::managed_shared_memory segment(ipc::open_only, "shared_memory");

    ipc::scoped_lock<ipc::interprocess_mutex> lock(data->mutex);
    auto vec = segment.find<ShmemVector>("vec").first;
    for (size_t run = 0; run < nruns; ++run) {
      const size_t num_bytes = vec->size();
      std::vector<uint8_t> out(num_bytes, 0);

      auto start = std::chrono::steady_clock::now();
      // std::copy(vec->cbegin(), vec->cend(), out.begin());
      std::memcpy(out.data(), vec->data(), vec->size());
      auto end = std::chrono::steady_clock::now();

      points.push_back(std::chrono::duration_cast<unit>(end - start));
      std::cout << run << std::endl;
    }
    segment.destroy<ShmemVector>("vec");
    data->end = true;
  }

  using limits_ssize_t = std::numeric_limits<ssize_t>;
  auto avg = std::accumulate(points.cbegin(), points.cend(), unit(0)) / nruns;
  auto min = std::accumulate(points.cbegin(), points.cend(), unit(limits_ssize_t::max()), [](auto a, auto b){return std::min(a, b);});
  auto max = std::accumulate(points.cbegin(), points.cend(), unit(limits_ssize_t::min()), [](auto a, auto b){return std::max(a, b);});
  std::cout << "avg: " << avg.count() << " " << unit_name << '\n'
            << "min: " << min.count() << " " << unit_name << '\n'
            << "max: " << max.count() << " " << unit_name << std::endl;

  return 0;
}

shmem_profile_producer.cpp

#include <algorithm>
#include <random>
#include <iostream>
#include <chrono>
#include <cstdlib>
#include <thread>

#include <boost/interprocess/mapped_region.hpp>
#include <boost/interprocess/shared_memory_object.hpp>
#include <boost/interprocess/sync/scoped_lock.hpp>
#include <boost/interprocess/managed_shared_memory.hpp>
#include <boost/interprocess/containers/vector.hpp>
#include <boost/interprocess/allocators/allocator.hpp>

#include "base.hpp"

namespace ipc = boost::interprocess;

int main(int argc, const char* argv[]) {
  std::random_device rd;
  std::mt19937 rng(rd());
  std::uniform_int_distribution<uint8_t> dist(0, 254);

  static constexpr size_t bytes = 2448 * 1378 * 3 / 2;
  std::cout << "bytes: " << bytes << '\n';

  struct shm_remove {
    shm_remove() {
      ipc::shared_memory_object::remove("shared_memory");
      ipc::shared_memory_object::remove("shared_memory_mutex");
    }
    ~shm_remove() {
      ipc::shared_memory_object::remove("shared_memory");
      ipc::shared_memory_object::remove("shared_memory_mutex");
    }
  } remover;

  //Obtain the page size of the system
  std::size_t page_size = ipc::mapped_region::get_page_size();
  std::cout << "page size " << page_size << std::endl;

  ipc::managed_shared_memory segment(ipc::create_only, "shared_memory", page_size * 2450);
  const ShmemAllocator alloc_inst(segment.get_segment_manager());

  auto vec = segment.construct<ShmemVector>("vec")(alloc_inst);
  vec->resize(2448*1378*3/2);
  std::fill(vec->begin(), vec->end(), 0);

  ipc::shared_memory_object shm_obj(ipc::create_only, "shared_memory_mutex", ipc::read_write);
  shm_obj.truncate(sizeof(Data));
  ipc::mapped_region region(shm_obj, ipc::read_write);
  Data *data = new (region.get_address()) Data;

  size_t i = 0;
  {
    ipc::scoped_lock<ipc::interprocess_mutex> lock(data->mutex);
    std::generate(vec->begin(), vec->end(), [&dist, &rng]() { return dist(rng); });
  }
  while (!data->end) {
    std::this_thread::sleep_for(std::chrono::milliseconds(33));
    std::cout << "copied data: step " << i << std::endl;
    ++i;
  }
  return 0;
}