fix some bugs, adjust the test cases

Author: mutouyun

include/libipc/def.h

@@ -29,7 +29,7 @@ enum : std::size_t {
     invalid_value   = (std::numeric_limits<std::size_t>::max)(),
     data_length     = 64,
     large_msg_limit = data_length,
-    large_msg_align = 512,
+    large_msg_align = 1024,
     large_msg_cache = 32,
     default_timeout = 100 // ms
 };

src/ipc.cpp

@@ -50,15 +50,16 @@ struct msg_t : msg_t<0, AlignSize> {
     std::aligned_storage_t<DataSize, AlignSize> data_ {};
 
     msg_t() = default;
-    msg_t(msg_id_t c, msg_id_t i, std::int32_t r, void const * d, std::size_t s)
-        : msg_t<0, AlignSize> { c, i, r, (d == nullptr) || (s == 0) } {
+    msg_t(msg_id_t conn, msg_id_t id, std::int32_t remain, void const * data, std::size_t size)
+        : msg_t<0, AlignSize> {conn, id, remain, (data == nullptr) || (size == 0)} {
         if (this->storage_) {
-            if (d != nullptr) {
+            if (data != nullptr) {
                 // copy storage-id
-                *reinterpret_cast<std::size_t*>(&data_) = *static_cast<std::size_t const *>(d);
+                *reinterpret_cast<ipc::storage_id_t*>(&data_) =
+                     *static_cast<ipc::storage_id_t const *>(data);
             }
         }
-        else std::memcpy(&data_, d, s);
+        else std::memcpy(&data_, data, size);
     }
 };
 
@@ -94,17 +95,13 @@ struct chunk_info_t {
     ipc::id_pool<> pool_;
     ipc::spin_lock lock_;
 
-    IPC_CONSTEXPR_ static std::size_t chunks_elem_size(std::size_t chunk_size) noexcept {
-        return ipc::make_align(alignof(std::max_align_t), chunk_size);
-    }
-
     IPC_CONSTEXPR_ static std::size_t chunks_mem_size(std::size_t chunk_size) noexcept {
-        return ipc::id_pool<>::max_count * chunks_elem_size(chunk_size);
+        return ipc::id_pool<>::max_count * chunk_size;
     }
 
-    ipc::byte_t *at(std::size_t chunk_size, std::size_t id) noexcept {
-        if (id == ipc::invalid_value) return nullptr;
-        return reinterpret_cast<ipc::byte_t *>(this + 1) + (chunks_elem_size(chunk_size) * id);
+    ipc::byte_t *at(std::size_t chunk_size, ipc::storage_id_t id) noexcept {
+        if (id < 0) return nullptr;
+        return reinterpret_cast<ipc::byte_t *>(this + 1) + (chunk_size * id);
     }
 };
 
@@ -128,29 +125,22 @@ auto& chunk_storages() {
             return info;
         }
     };
-    static ipc::unordered_map<std::size_t, chunk_t> chunk_s;
+    thread_local ipc::unordered_map<std::size_t, chunk_t> chunk_s;
     return chunk_s;
 }
 
-auto& chunk_lock() {
-    static ipc::spin_lock chunk_l;
-    return chunk_l;
+IPC_CONSTEXPR_ std::size_t calc_chunk_size(std::size_t size) noexcept {
+    return ipc::make_align(alignof(std::max_align_t), 
+                          (((size - 1) / ipc::large_msg_align) + 1) * ipc::large_msg_align);
 }
 
-constexpr std::size_t calc_chunk_size(std::size_t size) noexcept {
-    return ( ((size - 1) / ipc::large_msg_align) + 1 ) * ipc::large_msg_align;
+chunk_info_t *chunk_storage_info(std::size_t chunk_size) {
+    return chunk_storages()[chunk_size].get_info(chunk_size);
 }
 
-auto& chunk_storage(std::size_t chunk_size) {
-    IPC_UNUSED_ auto guard = ipc::detail::unique_lock(chunk_lock());
-    return chunk_storages()[chunk_size];
-}
-
-std::pair<std::size_t, void*> apply_storage(std::size_t size) {
+std::pair<ipc::storage_id_t, void*> apply_storage(std::size_t size) {
     std::size_t chunk_size = calc_chunk_size(size);
-    auto &      chunk_shm  = chunk_storage(chunk_size);
-
-    auto info = chunk_shm.get_info(chunk_size);
+    auto info = chunk_storage_info(chunk_size);
     if (info == nullptr) return {};
 
     info->lock_.lock();
@@ -162,27 +152,25 @@ std::pair<std::size_t, void*> apply_storage(std::size_t size) {
     return { id, info->at(chunk_size, id) };
 }
 
-void *find_storage(std::size_t id, std::size_t size) {
-    if (id == ipc::invalid_value) {
+void *find_storage(ipc::storage_id_t id, std::size_t size) {
+    if (id < 0) {
         ipc::error("[find_storage] id is invalid: id = %ld, size = %zd\n", (long)id, size);
         return nullptr;
     }
     std::size_t chunk_size = calc_chunk_size(size);
-    auto &      chunk_shm  = chunk_storage(chunk_size);
-    auto info = chunk_shm.get_info(chunk_size);
+    auto info = chunk_storage_info(chunk_size);
     if (info == nullptr) return nullptr;
     return info->at(chunk_size, id);
 }
 
-void clear_storage(std::size_t id, std::size_t size) {
-    if (id == ipc::invalid_value) {
+void release_storage(ipc::storage_id_t id, std::size_t size) {
+    if (id < 0) {
         ipc::error("[clear_storage] id is invalid: id = %ld, size = %zd\n", (long)id, size);
         return;
     }
 
     std::size_t chunk_size = calc_chunk_size(size);
-    auto &      chunk_shm  = chunk_storage(chunk_size);
-    auto info = chunk_shm.get_info(chunk_size);
+    auto info = chunk_storage_info(chunk_size);
     if (info == nullptr) return;
 
     info->lock_.lock();
@@ -194,9 +182,14 @@ template <typename MsgT>
 bool recycle_message(void* p) {
     auto msg = static_cast<MsgT*>(p);
     if (msg->storage_) {
-        clear_storage(
-            *reinterpret_cast<std::size_t*>(&msg->data_),
-            static_cast<std::int32_t>(ipc::data_length) + msg->remain_);
+        std::int32_t r_size = static_cast<std::int32_t>(ipc::data_length) + msg->remain_;
+        if (r_size <= 0) {
+            ipc::error("[recycle_message] invalid msg size: %d\n", (int)r_size);
+            return true;
+        }
+        release_storage(
+            *reinterpret_cast<ipc::storage_id_t*>(&msg->data_),
+            static_cast<std::size_t>(r_size));
     }
     return true;
 }
@@ -396,11 +389,11 @@ static bool send(F&& gen_push, ipc::handle_t h, void const * data, std::size_t s
                             static_cast<std::int32_t>(ipc::data_length), &(dat.first), 0);
         }
         // try using message fragment
-        // ipc::log("fail: shm::handle for big message. msg_id: %zd, size: %zd\n", msg_id, size);
+        //ipc::log("fail: shm::handle for big message. msg_id: %zd, size: %zd\n", msg_id, size);
     }
     // push message fragment
     std::int32_t offset = 0;
-    for (int i = 0; i < static_cast<int>(size / ipc::data_length); ++i, offset += ipc::data_length) {
+    for (std::int32_t i = 0; i < static_cast<std::int32_t>(size / ipc::data_length); ++i, offset += ipc::data_length) {
         if (!try_push(static_cast<std::int32_t>(size) - offset - static_cast<std::int32_t>(ipc::data_length),
                       static_cast<ipc::byte_t const *>(data) + offset, ipc::data_length)) {
             return false;
@@ -466,7 +459,7 @@ static ipc::buff_t recv(ipc::handle_t h, std::size_t tm) {
         return {};
     }
     auto& rc = info_of(h)->recv_cache();
-    while (1) {
+    for (;;) {
         // pop a new message
         typename queue_t::value_t msg;
         if (!wait_for(info_of(h)->rd_waiter_, [que, &msg] { return !que->pop(msg); }, tm)) {
@@ -478,20 +471,28 @@ static ipc::buff_t recv(ipc::handle_t h, std::size_t tm) {
             continue; // ignore message to self
         }
         // msg.remain_ may minus & abs(msg.remain_) < data_length
-        std::size_t remain = static_cast<std::int32_t>(ipc::data_length) + msg.remain_;
+        std::int32_t r_size = static_cast<std::int32_t>(ipc::data_length) + msg.remain_;
+        if (r_size <= 0) {
+            ipc::error("fail: recv, r_size = %d\n", (int)r_size);
+            return {};
+        }
+        std::size_t msg_size = static_cast<std::size_t>(r_size);
         // find cache with msg.id_
         auto cac_it = rc.find(msg.id_);
         if (cac_it == rc.end()) {
-            if (remain <= ipc::data_length) {
-                return make_cache(msg.data_, remain);
+            if (msg_size <= ipc::data_length) {
+                return make_cache(msg.data_, msg_size);
             }
             if (msg.storage_) {
                 std::size_t buf_id = *reinterpret_cast<std::size_t*>(&msg.data_);
-                void      * buf    = find_storage(buf_id, remain);
+                void      * buf    = find_storage(buf_id, msg_size);
                 if (buf != nullptr) {
-                    return ipc::buff_t{buf, remain};
+                    return ipc::buff_t{buf, msg_size};
+                }
+                else {
+                    ipc::log("fail: shm::handle for big message. msg_id: %zd, buf_id: %zd, size: %zd\n", msg.id_, buf_id, msg_size);
+                    continue;
                 }
-                else ipc::log("fail: shm::handle for big message. msg_id: %zd, buf_id: %zd, size: %zd\n", msg.id_, buf_id, remain);
             }
             // gc
             if (rc.size() > 1024) {
@@ -505,14 +506,14 @@ static ipc::buff_t recv(ipc::handle_t h, std::size_t tm) {
                 for (auto id : need_del) rc.erase(id);
             }
             // cache the first message fragment
-            rc.emplace(msg.id_, cache_t { ipc::data_length, make_cache(msg.data_, remain) });
+            rc.emplace(msg.id_, cache_t { ipc::data_length, make_cache(msg.data_, msg_size) });
         }
         // has cached before this message
         else {
             auto& cac = cac_it->second;
             // this is the last message fragment
             if (msg.remain_ <= 0) {
-                cac.append(&(msg.data_), remain);
+                cac.append(&(msg.data_), msg_size);
                 // finish this message, erase it from cache
                 auto buff = std::move(cac.buff_);
                 rc.erase(cac_it);

src/libipc/memory/resource.h

@@ -5,6 +5,7 @@
 #include <utility>
 #include <functional>
 #include <unordered_map>
+#include <map>
 #include <string>
 #include <cstdio>
 
@@ -49,6 +50,11 @@ using unordered_map = std::unordered_map<
     Key, T, std::hash<Key>, std::equal_to<Key>, ipc::mem::allocator<std::pair<const Key, T>>
 >;
 
+template <typename Key, typename T>
+using map = std::map<
+    Key, T, std::less<Key>, ipc::mem::allocator<std::pair<const Key, T>>
+>;
+
 template <typename Char>
 using basic_string = std::basic_string<
     Char, std::char_traits<Char>, ipc::mem::allocator<Char>

src/libipc/utility/id_pool.h

@@ -2,21 +2,23 @@
 
 #include <type_traits>  // std::aligned_storage_t
 #include <cstring>      // std::memcmp
+#include <cstdint>
 
 #include "libipc/def.h"
-
 #include "libipc/platform/detail.h"
 
 namespace ipc {
 
+using storage_id_t = std::int32_t;
+
 template <std::size_t DataSize, std::size_t AlignSize>
 struct id_type;
 
 template <std::size_t AlignSize>
 struct id_type<0, AlignSize> {
     uint_t<8> id_;
 
-    id_type& operator=(std::size_t val) {
+    id_type& operator=(storage_id_t val) {
         id_ = static_cast<uint_t<8>>(val);
         return (*this);
     }
@@ -57,7 +59,7 @@ class id_pool {
     }
 
     void init() {
-        for (std::size_t i = 0; i < max_count;) {
+        for (storage_id_t i = 0; i < max_count;) {
             i = next_[i] = (i + 1);
         }
     }
@@ -71,31 +73,31 @@ class id_pool {
         return cursor_ == max_count;
     }
 
-    std::size_t acquire() {
-        if (empty()) return invalid_value;
-        std::size_t id = cursor_;
+    storage_id_t acquire() {
+        if (empty()) return -1;
+        storage_id_t id = cursor_;
         cursor_ = next_[id]; // point to next
         return id;
     }
 
-    bool release(std::size_t id) {
-        if (id == invalid_value) return false;
+    bool release(storage_id_t id) {
+        if (id < 0) return false;
         next_[id] = cursor_;
         cursor_ = static_cast<uint_t<8>>(id); // put it back
         return true;
     }
 
-    void       * at(std::size_t id)       { return &(next_[id].data_); }
-    void const * at(std::size_t id) const { return &(next_[id].data_); }
+    void       * at(storage_id_t id)       { return &(next_[id].data_); }
+    void const * at(storage_id_t id) const { return &(next_[id].data_); }
 };
 
 template <typename T>
 class obj_pool : public id_pool<sizeof(T), alignof(T)> {
     using base_t = id_pool<sizeof(T), alignof(T)>;
 
 public:
-    T       * at(std::size_t id)       { return reinterpret_cast<T       *>(base_t::at(id)); }
-    T const * at(std::size_t id) const { return reinterpret_cast<T const *>(base_t::at(id)); }
+    T       * at(storage_id_t id)       { return reinterpret_cast<T       *>(base_t::at(id)); }
+    T const * at(storage_id_t id) const { return reinterpret_cast<T const *>(base_t::at(id)); }
 };
 
 } // namespace ipc