提升进程间:在循环中分配共享内存

Boost Interprocess: Allocating Shared Memory within a loop

本文关键字：分配 共享 内存 循环 进程      更新时间：2023-10-16

我正在使用boosts进程间库在多个程序之间共享内存。 3 或 4 个其他程序将在 cust_order 中读取和写入共享内存内容。对内存空间的访问需要序列化。

在下面的示例程序中，引擎循环访问risk_queue的内容，如果已填充，则获取第一个cust_order编号并定位该顺序进行处理。

  • 在每个程序使用的 objects.h 头文件中：

    struct cust_order {
int ID;
char CLID[128];
int CUST_ID;
char ORDER_STATUS;
};
cust_order o;
boost::interprocess::managed_shared_memory
cust_order_segment(boost::interprocess::open_or_create, "cust_order",
65536 * 100);
typedef int cust_order_KeyType;
typedef cust_order cust_order_MappedType;
typedef std::pair<const int, cust_order> cust_order_ValueType;
typedef boost::interprocess::allocator<
cust_order_ValueType,
boost::interprocess::managed_shared_memory::segment_manager>
cust_order_ShmemAllocator;
typedef boost::interprocess::map<cust_order_KeyType, cust_order_MappedType,
std::less<cust_order_KeyType>,
cust_order_ShmemAllocator>
cust_order_MySHMMap;
cust_order_MySHMMap::iterator cust_order_iter;
boost::interprocess::managed_shared_memory
risk_queue_segment(boost::interprocess::open_or_create, "risk_queue",
65536 * 100);
typedef int risk_queue_KeyType;
typedef int risk_queue_MappedType;
typedef std::pair<const int, int> risk_queue_ValueType;
typedef boost::interprocess::allocator<
risk_queue_ValueType,
boost::interprocess::managed_shared_memory::segment_manager>
risk_queue_ShmemAllocator;
typedef boost::interprocess::map<risk_queue_KeyType, risk_queue_MappedType,
std::less<risk_queue_KeyType>,
risk_queue_ShmemAllocator>
risk_queue_MySHMMap;
risk_queue_MySHMMap::iterator risk_queue_iter;

  • 在引擎中.cpp：

    int main() {
risk_queue_ShmemAllocator risk_queue_alloc_inst(
risk_queue_segment.get_segment_manager());
cust_order_ShmemAllocator cust_order_alloc_inst(
cust_order_segment.get_segment_manager());
for (; 0 < 1;) {
boost::interprocess::offset_ptr<risk_queue_MySHMMap> risk_queue_m_pmap =
risk_queue_segment.find<risk_queue_MySHMMap>("risk_queue").first;
boost::interprocess::offset_ptr<cust_order_MySHMMap> cust_order_m_pmap =
cust_order_segment.find<cust_order_MySHMMap>("cust_order").first;
risk_queue_iter = risk_queue_m_pmap->begin();
if (risk_queue_iter != risk_queue_m_pmap->end()) {
ordid = risk_queue_iter->second;
cust_order_m_pmap->find(ordid)->second = o;
o.ORDER_STATUS = '0';
o = cust_order_m_pmap->find(ordid)->second;
risk_queue_m_pmap->erase(ordid);
}
};
return 0;
}

这是我在程序完美运行几秒钟后遇到的错误：

engine: /usr/include/boost/intrusive/bstree.hpp:1331: boost::intrusive::bstree_impl<ValueTraits, VoidOrKeyOfValue, VoidOrKeyComp, SizeType, ConstantTimeSize, AlgoType, HeaderHolder>::iterator boost::intrusive::bstree_impl<ValueTraits, VoidOrKeyOfValue, VoidOrKeyComp, SizeType, ConstantTimeSize, AlgoType, HeaderHolder>::insert_unique_commit(boost::intrusive::bstree_impl<ValueTraits, VoidOrKeyOfValue, VoidOrKeyComp, SizeType, ConstantTimeSize, AlgoType, HeaderHolder>::reference, const insert_commit_data&) [with ValueTraits = boost::intrusive::bhtraits<boost::container::container_detail::tree_node<std::pair<const int, event>, boost::interprocess::offset_ptr<void>, (boost::container::tree_type_enum)0, true>, boost::intrusive::rbtree_node_traits<boost::interprocess::offset_ptr<void>, true>, (boost::intrusive::link_mode_type)0, boost::intrusive::dft_tag, 3>; VoidOrKeyOfValue = void; VoidOrKeyComp = boost::container::value_to_node_compare<boost::container::container_detail::tree_node<std::pair<const int, event>, boost::interprocess::offset_ptr<void>, (boost::container::tree_type_enum)0, true>, boost::intrusive::tree_value_compare<boost::interprocess::offset_ptr<std::pair<const int, event>, long int, long unsigned int, 0>, std::less<int>, boost::container::container_detail::select1st<int>, false> >; SizeType = long unsigned int; bool ConstantTimeSize = true; boost::intrusive::algo_types AlgoType = (boost::intrusive::algo_types)5; HeaderHolder = void; boost::intrusive::bstree_impl<ValueTraits, VoidOrKeyOfValue, VoidOrKeyComp, SizeType, ConstantTimeSize, AlgoType, HeaderHolder>::iterator = boost::intrusive::tree_iterator<boost::intrusive::bhtraits<boost::container::container_detail::tree_node<std::pair<const int, event>, boost::interprocess::offset_ptr<void>, (boost::container::tree_type_enum)0, true>, boost::intrusive::rbtree_node_traits<boost::interprocess::offset_ptr<void>, true>, (boost::intrusive::link_mode_type)0, boost::intrusive::dft_tag, 3>, false>; boost::intrusive::bstree_impl<ValueTraits, VoidOrKeyOfValue, VoidOrKeyComp, SizeType, ConstantTimeSize, AlgoType, HeaderHolder>::reference = boost::container::container_detail::tree_node<std::pair<const int, event>, boost::interprocess::offset_ptr<void>, (boost::container::tree_type_enum)0, true>&; boost::intrusive::bstree_impl<ValueTraits, VoidOrKeyOfValue, VoidOrKeyComp, SizeType, ConstantTimeSize, AlgoType, HeaderHolder>::insert_commit_data = boost::intrusive::insert_commit_data_t<boost::interprocess::offset_ptr<boost::intrusive::compact_rbtree_node<boost::interprocess::offset_ptr<void> >, long int, long unsigned int, 0> >]: Assertion `( p == this->end() || !this->comp()(*p, value) )' failed.
Aborted (core dumped)

请您帮助我了解错误。让我知道，如果有更好的方法来做我需要做的事情。

但是要完成这项工作，我必须在循环中分配内存

我不明白为什么会这样。您的共享内存大小容量限制为 64kb(65536 字节)，因此这很容易：预先分配一次。

我的应用程序是面向低延迟的，我想知道是否有一种方法可以在内存中读取更新的对象，而无需在每次循环迭代中不断重新分配？

是的。索引到映射到进程地址空间的共享内存区域，而无需分配。

下面是我必须放置在循环中的代码，以便从访问相同内存空间的其他程序接收更新。

您显示的代码不会分配任何内容。它在托管段中定位已分配(和构造)的对象。并返回指向它的指针。

如果我将其放在循环之外，当其他程序在共享内存空间中操作或添加数据时，我将不会收到更新

啊！你没有告诉我们X_MySHMMap是什么样的，但我可以开始猜测它可能包含类似std::array<char, N>的东西或你循环的另一个容器。

如果从多个进程对该对象进行不同步访问，则会进入数据争用。编译器有一个内存模型(参见同一页)，允许它对内存操作重新排序并消除加载/存储周期。例如，如果你写

static bool ready = false;
int main() {
while (!ready) {
std::cout << "waiting...n";
}
}

编译器将此视为有保证的无限循环。想象ready在共享内存中，编译器根本没有区别。您必须添加同步(通过使用互斥、互斥+条件变量、原子学或在简单情况下，您可以将对象标记为volatile)。

我的猜测是，当您每次在循环中find对象/again/时，您会意外地在进程之间同步(因为段管理器使用互斥来同步对段元数据的访问)，这就是使编译器发出新读取的原因。

怎么办？

您没有提供足够的代码来实际为您显示修复程序。我想我可以说managed_shared_memory可能把事情复杂化了。您可以使用一个简单的shared_memory_object(它没有段管理器的开销)，并确保让编译器知道您的数据是从另一个"线程"¹可变/并发访问的。这样可以防止它优化所需的负载。

简单示例：

住在科利鲁²

#include <boost/interprocess/shared_memory_object.hpp>
#include <boost/interprocess/mapped_region.hpp>
#include <array>
#include <atomic>
using SharedBuffer = std::array<std::atomic_char, 65536>;
static_assert(std::is_pod<SharedBuffer>{}, "assumed POD"); // Warning: trait is not correct on some versions of MSVC
namespace bip = boost::interprocess;
int main() {
bip::shared_memory_object smo(bip::open_or_create, "yoho", bip::mode_t::read_write);
smo.truncate(sizeof(SharedBuffer));
bip::mapped_region region(smo, bip::mode_t::read_write, 0, 65536);
auto& buffer = *reinterpret_cast<SharedBuffer*>(region.get_address());
// do something with that buffer
std::fill(buffer.begin(), buffer.end(), 0); // start with all zeroes
// loop over it or something
}

¹(线程在另一个进程中没有区别) ² Coliru 不允许共享内存访问

相关文章：