coffee/cpp-ipc
1
0
Fork 0
mirror of https://github.com/mutouyun/cpp-ipc.git synced 2025-12-06 16:56:45 +08:00
Code Issues Packages Projects Releases Wiki Activity

Fixed memory access exception in multithreading.

Browse Source
This commit is contained in:
mutouyun 2023-09-24 18:02:37 +08:00
parent f52b125c3f
commit 8c02f37ff9
5 changed files with 40 additions and 22 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
3rdparty/gtest/src/gtest-death-test.cc vendored
View File

@ -1296,8 +1296,8 @@ static void StackLowerThanAddress(const void* ptr, bool* result) {
GTEST_ATTRIBUTE_NO_SANITIZE_ADDRESS_
GTEST_ATTRIBUTE_NO_SANITIZE_HWADDRESS_
static bool StackGrowsDown() {
int dummy;
bool result;
int dummy {};
bool result {};
StackLowerThanAddress(&dummy, &result);
return result;
}

30
src/libipc/ipc.cpp
View File

@ -182,6 +182,7 @@ struct chunk_info_t {
auto& chunk_storages() {
class chunk_handle_t {
ipc::unordered_map<ipc::string, ipc::shm::handle> handles_;
std::mutex lock_;
static bool make_handle(ipc::shm::handle &h, ipc::string const &shm_name, std::size_t chunk_size) {
if (!h.valid() &&
@ -197,11 +198,15 @@ auto& chunk_storages() {
chunk_info_t *get_info(conn_info_head *inf, std::size_t chunk_size) {
ipc::string pref {(inf == nullptr) ? ipc::string{} : inf->prefix_};
ipc::string shm_name {ipc::make_prefix(pref, {"CHUNK_INFO__", ipc::to_string(chunk_size)})};
ipc::shm::handle &h = handles_[pref];
if (!make_handle(h, shm_name, chunk_size)) {
return nullptr;
ipc::shm::handle *h;
{
std::lock_guard<std::mutex> guard {lock_};
h = &(handles_[pref]);
if (!make_handle(*h, shm_name, chunk_size)) {
return nullptr;
}
}
auto *info = static_cast<chunk_info_t*>(h.get());
auto *info = static_cast<chunk_info_t*>(h->get());
if (info == nullptr) {
ipc::error("[chunk_storages] chunk_shm.id_info_.get failed: chunk_size = %zd\n", chunk_size);
return nullptr;
@ -209,7 +214,9 @@ auto& chunk_storages() {
return info;
}
};
static ipc::map<std::size_t, chunk_handle_t> chunk_hs;
using deleter_t = void (*)(chunk_handle_t*);
using chunk_handle_ptr_t = std::unique_ptr<chunk_handle_t, deleter_t>;
static ipc::map<std::size_t, chunk_handle_ptr_t> chunk_hs;
return chunk_hs;
}
@ -220,13 +227,17 @@ chunk_info_t *chunk_storage_info(conn_info_head *inf, std::size_t chunk_size) {
static ipc::rw_lock lock;
IPC_UNUSED_ std::shared_lock<ipc::rw_lock> guard {lock};
if ((it = storages.find(chunk_size)) == storages.end()) {
using chunk_handle_t = std::decay_t<decltype(storages)>::value_type::second_type;
using chunk_handle_ptr_t = std::decay_t<decltype(storages)>::value_type::second_type;
using chunk_handle_t = chunk_handle_ptr_t::element_type;
guard.unlock();
IPC_UNUSED_ std::lock_guard<ipc::rw_lock> guard {lock};
it = storages.emplace(chunk_size, chunk_handle_t{}).first;
it = storages.emplace(chunk_size, chunk_handle_ptr_t{
ipc::mem::alloc<chunk_handle_t>(), [](chunk_handle_t *p) {
ipc::mem::destruct(p);
}}).first;
}
}
return it->second.get_info(inf, chunk_size);
return it->second->get_info(inf, chunk_size);
}
std::pair<ipc::storage_id_t, void*> acquire_storage(conn_info_head *inf, std::size_t size, ipc::circ::cc_t conns) {
@ -356,8 +367,7 @@ struct queue_generator {
"QU_CONN__",
ipc::to_string(DataSize), "__",
ipc::to_string(AlignSize), "__",
name}).c_str()} {
}
name}).c_str()} {}
void disconnect_receiver() {
bool dis = que_.disconnect();

16
src/libipc/platform/posix/shm_posix.cpp
View File

@ -49,6 +49,9 @@ id_t acquire(char const * name, std::size_t size, unsigned mode) {
ipc::error("fail acquire: name is empty\n");
return nullptr;
}
// For portable use, a shared memory object should be identified by name of the form /somename.
// see: https://man7.org/linux/man-pages/man3/shm_open.3.html
ipc::string op_name = ipc::string{"/"} + name;
// Open the object for read-write access.
int flag = O_RDWR;
switch (mode) {
@ -65,17 +68,17 @@ id_t acquire(char const * name, std::size_t size, unsigned mode) {
flag |= O_CREAT;
break;
}
int fd = ::shm_open(name, flag, S_IRUSR | S_IWUSR |
S_IRGRP | S_IWGRP |
S_IROTH | S_IWOTH);
int fd = ::shm_open(op_name.c_str(), flag, S_IRUSR | S_IWUSR |
S_IRGRP | S_IWGRP |
S_IROTH | S_IWOTH);
if (fd == -1) {
ipc::error("fail shm_open[%d]: %s\n", errno, name);
ipc::error("fail shm_open[%d]: %s\n", errno, op_name.c_str());
return nullptr;
}
auto ii = mem::alloc<id_info_t>();
ii->fd_ = fd;
ii->size_ = size;
ii->name_ = name;
ii->name_ = std::move(op_name);
return ii;
}
@ -158,7 +161,8 @@ std::int32_t release(id_t id) {
std::int32_t ret = -1;
auto ii = static_cast<id_info_t*>(id);
if (ii->mem_ == nullptr || ii->size_ == 0) {
ipc::error("fail release: invalid id (mem = %p, size = %zd)\n", ii->mem_, ii->size_);
ipc::error("fail release: invalid id (mem = %p, size = %zd), name = %s\n",
ii->mem_, ii->size_, ii->name_.c_str());
}
else if ((ret = acc_of(ii->mem_, ii->size_).fetch_sub(1, std::memory_order_acq_rel)) <= 1) {
::munmap(ii->mem_, ii->size_);

10
test/test_ipc.cpp
View File

@ -151,11 +151,15 @@ void test_sr(char const * name, int s_cnt, int r_cnt) {
} // internal-linkage
TEST(IPC, basic) {
TEST(IPC, basic_ssu) {
test_basic<relat::single, relat::single, trans::unicast >("ssu");
//test_basic<relat::single, relat::multi , trans::unicast >("smu");
//test_basic<relat::multi , relat::multi , trans::unicast >("mmu");
}
TEST(IPC, basic_smb) {
test_basic<relat::single, relat::multi , trans::broadcast>("smb");
}
TEST(IPC, basic_mmb) {
test_basic<relat::multi , relat::multi , trans::broadcast>("mmb");
}

2
test/test_sync.cpp
View File

@ -131,7 +131,7 @@ TEST(Sync, Condition) {
{
std::lock_guard<ipc::sync::mutex> guard {lock};
while (que.empty()) {
EXPECT_TRUE(cond.wait(lock, 1000));
ASSERT_TRUE(cond.wait(lock, 1000));
}
val = que.front();
que.pop_front();