Fixed memory access exception in multithreading.

Author: mutouyun

3rdparty/gtest/src/gtest-death-test.cc

@@ -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;
 }

src/libipc/ipc.cpp

@@ -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,19 +198,25 @@ 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;
             }
             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();

src/libipc/platform/posix/shm_posix.cpp

@@ -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_);

test/test_ipc.cpp

@@ -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");
 }
 

test/test_sync.cpp

@@ -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();