fix some bugs for linux-mutex

Author: mutouyun

include/libipc/shm.h

@@ -15,13 +15,13 @@ enum : unsigned {
     open   = 0x02
 };
 
-IPC_EXPORT id_t   acquire(char const * name, std::size_t size, unsigned mode = create | open);
-IPC_EXPORT void * get_mem(id_t id, std::size_t * size);
-IPC_EXPORT void   release(id_t id);
-IPC_EXPORT void   remove (id_t id);
-IPC_EXPORT void   remove (char const * name);
+IPC_EXPORT id_t         acquire(char const * name, std::size_t size, unsigned mode = create | open);
+IPC_EXPORT void *       get_mem(id_t id, std::size_t * size);
+IPC_EXPORT std::int32_t release(id_t id);
+IPC_EXPORT void         remove (id_t id);
+IPC_EXPORT void         remove (char const * name);
 
-IPC_EXPORT std::uint32_t get_ref(id_t id);
+IPC_EXPORT std::int32_t get_ref(id_t id);
 IPC_EXPORT void sub_ref(id_t id);
 
 class IPC_EXPORT handle {
@@ -39,11 +39,11 @@ class IPC_EXPORT handle {
     std::size_t  size () const noexcept;
     char const * name () const noexcept;
 
-    std::uint32_t ref() const noexcept;
+    std::int32_t ref() const noexcept;
     void sub_ref() noexcept;
 
     bool acquire(char const * name, std::size_t size, unsigned mode = create | open);
-    void release();
+    std::int32_t release();
 
     void* get() const;
 

src/libipc/platform/condition_linux.h

@@ -50,33 +50,34 @@ class condition {
         if ((cond_ = acquire_cond(name)) == nullptr) {
             return false;
         }
-        if (shm_.ref() == 1) {
-            ::pthread_cond_destroy(cond_);
-            auto finally = ipc::guard([this] { close(); }); // close when failed
-            // init condition
-            int eno;
-            pthread_condattr_t cond_attr;
-            if ((eno = ::pthread_condattr_init(&cond_attr)) != 0) {
-                ipc::error("fail pthread_condattr_init[%d]\n", eno);
-                return false;
-            }
-            IPC_UNUSED_ auto guard_cond_attr = unique_ptr(&cond_attr, ::pthread_condattr_destroy);
-            if ((eno = ::pthread_condattr_setpshared(&cond_attr, PTHREAD_PROCESS_SHARED)) != 0) {
-                ipc::error("fail pthread_condattr_setpshared[%d]\n", eno);
-                return false;
-            }
-            *cond_ = PTHREAD_COND_INITIALIZER;
-            if ((eno = ::pthread_cond_init(cond_, &cond_attr)) != 0) {
-                ipc::error("fail pthread_cond_init[%d]\n", eno);
-                return false;
-            }
-            finally.dismiss();
+        if (shm_.ref() > 1) {
+            return valid();
+        }
+        ::pthread_cond_destroy(cond_);
+        auto finally = ipc::guard([this] { close(); }); // close when failed
+        // init condition
+        int eno;
+        pthread_condattr_t cond_attr;
+        if ((eno = ::pthread_condattr_init(&cond_attr)) != 0) {
+            ipc::error("fail pthread_condattr_init[%d]\n", eno);
+            return false;
+        }
+        IPC_UNUSED_ auto guard_cond_attr = unique_ptr(&cond_attr, ::pthread_condattr_destroy);
+        if ((eno = ::pthread_condattr_setpshared(&cond_attr, PTHREAD_PROCESS_SHARED)) != 0) {
+            ipc::error("fail pthread_condattr_setpshared[%d]\n", eno);
+            return false;
+        }
+        *cond_ = PTHREAD_COND_INITIALIZER;
+        if ((eno = ::pthread_cond_init(cond_, &cond_attr)) != 0) {
+            ipc::error("fail pthread_cond_init[%d]\n", eno);
+            return false;
         }
+        finally.dismiss();
         return valid();
     }
 
     void close() noexcept {
-        if (shm_.ref() == 1) {
+        if ((shm_.ref() <= 1) && cond_ != nullptr) {
             int eno;
             if ((eno = ::pthread_cond_destroy(cond_)) != 0) {
                 ipc::error("fail pthread_cond_destroy[%d]\n", eno);
@@ -87,9 +88,8 @@ class condition {
     }
 
     bool wait(ipc::sync::mutex &mtx, std::uint64_t tm) noexcept {
+        if (!valid()) return false;
         switch (tm) {
-        case 0:
-            return true;
         case invalid_value: {
                 int eno;
                 if ((eno = ::pthread_cond_wait(cond_, static_cast<pthread_mutex_t *>(mtx.native()))) != 0) {
@@ -115,6 +115,7 @@ class condition {
     }
 
     bool notify() noexcept {
+        if (!valid()) return false;
         int eno;
         if ((eno = ::pthread_cond_signal(cond_)) != 0) {
             ipc::error("fail pthread_cond_signal[%d]\n", eno);
@@ -124,6 +125,7 @@ class condition {
     }
 
     bool broadcast() noexcept {
+        if (!valid()) return false;
         int eno;
         if ((eno = ::pthread_cond_broadcast(cond_)) != 0) {
             ipc::error("fail pthread_cond_broadcast[%d]\n", eno);

src/libipc/platform/mutex_linux.h

@@ -2,8 +2,10 @@
 
 #include <cstring>
 #include <cassert>
+#include <cstdint>
 #include <system_error>
 #include <mutex>
+#include <atomic>
 
 #include <pthread.h>
 
@@ -19,33 +21,61 @@ namespace detail {
 namespace sync {
 
 class mutex {
-    ipc::shm::handle shm_;
+    ipc::shm::handle *shm_ = nullptr;
+    std::atomic<std::int32_t> *ref_ = nullptr;
     pthread_mutex_t *mutex_ = nullptr;
 
+    struct curr_prog {
+        struct shm_data {
+            ipc::shm::handle shm;
+            std::atomic<std::int32_t> ref;
+
+            struct init {
+                char const *name;
+                std::size_t size;
+            };
+            shm_data(init arg)
+                : shm{arg.name, arg.size}, ref{0} {}
+        };
+        ipc::map<ipc::string, shm_data> mutex_handles;
+        std::mutex lock;
+
+        static curr_prog &get() {
+            static curr_prog info;
+            return info;
+        }
+    };
+
     pthread_mutex_t *acquire_mutex(char const *name) {
-        if (!shm_.acquire(name, sizeof(pthread_mutex_t))) {
-            ipc::error("[acquire_mutex] fail shm.acquire: %s\n", name);
+        if (name == nullptr) {
             return nullptr;
         }
-        return static_cast<pthread_mutex_t *>(shm_.get());
+        auto &info = curr_prog::get();
+        IPC_UNUSED_ std::lock_guard<std::mutex> guard {info.lock};
+        auto it = info.mutex_handles.find(name);
+        if (it == info.mutex_handles.end()) {
+            it = curr_prog::get().mutex_handles.emplace(name, 
+                              curr_prog::shm_data::init{name, sizeof(pthread_mutex_t)}).first;
+        }
+        shm_ = &it->second.shm;
+        ref_ = &it->second.ref;
+        if (shm_ == nullptr) {
+            return nullptr;
+        }
+        return static_cast<pthread_mutex_t *>(shm_->get());
     }
 
-    pthread_mutex_t *get_mutex(char const *name) {
-        if (name == nullptr) {
-            return nullptr;
+    template <typename F>
+    void release_mutex(ipc::string const &name, F &&clear) {
+        if (name.empty()) return;
+        IPC_UNUSED_ std::lock_guard<std::mutex> guard {curr_prog::get().lock};
+        auto it = curr_prog::get().mutex_handles.find(name);
+        if (it == curr_prog::get().mutex_handles.end()) {
+            return;
         }
-        static ipc::map<ipc::string, pthread_mutex_t *> mutex_handles;
-        static std::mutex lock;
-        IPC_UNUSED_ std::lock_guard<std::mutex> guard {lock};
-        auto it = mutex_handles.find(name);
-        if (it == mutex_handles.end()) {
-            auto ptr = acquire_mutex(name);
-            if (ptr != nullptr) {
-                mutex_handles.emplace(name, ptr);
-            }
-            return ptr;
+        if (clear()) {
+            curr_prog::get().mutex_handles.erase(it);
         }
-        return it->second;
     }
 
 public:
@@ -62,56 +92,69 @@ class mutex {
 
     bool valid() const noexcept {
         static const char tmp[sizeof(pthread_mutex_t)] {};
-        return (mutex_ != nullptr)
+        return (shm_ != nullptr) && (ref_ != nullptr) && (mutex_ != nullptr)
             && (std::memcmp(tmp, mutex_, sizeof(pthread_mutex_t)) != 0);
     }
 
     bool open(char const *name) noexcept {
         close();
-        if ((mutex_ = get_mutex(name)) == nullptr) {
+        if ((mutex_ = acquire_mutex(name)) == nullptr) {
             return false;
         }
-        if (shm_.ref() == 1) {
-            ::pthread_mutex_destroy(mutex_);
-            auto finally = ipc::guard([this] { close(); }); // close when failed
-            // init mutex
-            int eno;
-            pthread_mutexattr_t mutex_attr;
-            if ((eno = ::pthread_mutexattr_init(&mutex_attr)) != 0) {
-                ipc::error("fail pthread_mutexattr_init[%d]\n", eno);
-                return false;
-            }
-            IPC_UNUSED_ auto guard_mutex_attr = unique_ptr(&mutex_attr, ::pthread_mutexattr_destroy);
-            if ((eno = ::pthread_mutexattr_setpshared(&mutex_attr, PTHREAD_PROCESS_SHARED)) != 0) {
-                ipc::error("fail pthread_mutexattr_setpshared[%d]\n", eno);
-                return false;
-            }
-            if ((eno = ::pthread_mutexattr_setrobust(&mutex_attr, PTHREAD_MUTEX_ROBUST)) != 0) {
-                ipc::error("fail pthread_mutexattr_setrobust[%d]\n", eno);
-                return false;
-            }
-            *mutex_ = PTHREAD_MUTEX_INITIALIZER;
-            if ((eno = ::pthread_mutex_init(mutex_, &mutex_attr)) != 0) {
-                ipc::error("fail pthread_mutex_init[%d]\n", eno);
-                return false;
-            }
-            finally.dismiss();
+        auto self_ref = ref_->fetch_add(1, std::memory_order_relaxed);
+        if (shm_->ref() > 1 || self_ref > 0) {
+            return valid();
+        }
+        ::pthread_mutex_destroy(mutex_);
+        auto finally = ipc::guard([this] { close(); }); // close when failed
+        // init mutex
+        int eno;
+        pthread_mutexattr_t mutex_attr;
+        if ((eno = ::pthread_mutexattr_init(&mutex_attr)) != 0) {
+            ipc::error("fail pthread_mutexattr_init[%d]\n", eno);
+            return false;
+        }
+        IPC_UNUSED_ auto guard_mutex_attr = unique_ptr(&mutex_attr, ::pthread_mutexattr_destroy);
+        if ((eno = ::pthread_mutexattr_setpshared(&mutex_attr, PTHREAD_PROCESS_SHARED)) != 0) {
+            ipc::error("fail pthread_mutexattr_setpshared[%d]\n", eno);
+            return false;
+        }
+        if ((eno = ::pthread_mutexattr_setrobust(&mutex_attr, PTHREAD_MUTEX_ROBUST)) != 0) {
+            ipc::error("fail pthread_mutexattr_setrobust[%d]\n", eno);
+            return false;
         }
+        *mutex_ = PTHREAD_MUTEX_INITIALIZER;
+        if ((eno = ::pthread_mutex_init(mutex_, &mutex_attr)) != 0) {
+            ipc::error("fail pthread_mutex_init[%d]\n", eno);
+            return false;
+        }
+        finally.dismiss();
         return valid();
     }
 
     void close() noexcept {
-        if (shm_.ref() == 1) {
-            int eno;
-            if ((eno = ::pthread_mutex_destroy(mutex_)) != 0) {
-                ipc::error("fail pthread_mutex_destroy[%d]\n", eno);
-            }
+        if ((ref_ != nullptr) && (shm_ != nullptr) && (mutex_ != nullptr)) {
+            if (shm_->name() != nullptr) {
+                release_mutex(shm_->name(), [this] {
+                    auto self_ref = ref_->fetch_sub(1, std::memory_order_relaxed);
+                    if ((shm_->ref() <= 1) && (self_ref <= 1)) {
+                        int eno;
+                        if ((eno = ::pthread_mutex_destroy(mutex_)) != 0) {
+                            ipc::error("fail pthread_mutex_destroy[%d]\n", eno);
+                        }
+                        return true;
+                    }
+                    return false;
+                });
+            } else shm_->release();
         }
-        shm_.release();
+        shm_   = nullptr;
+        ref_   = nullptr;
         mutex_ = nullptr;
     }
 
     bool lock(std::uint64_t tm) noexcept {
+        if (!valid()) return false;
         for (;;) {
             auto ts = detail::make_timespec(tm);
             int eno = (tm == invalid_value) 
@@ -123,8 +166,8 @@ class mutex {
             case ETIMEDOUT:
                 return false;
             case EOWNERDEAD: {
-                    if (shm_.ref() > 1) {
-                        shm_.sub_ref();
+                    if (shm_->ref() > 1) {
+                        shm_->sub_ref();
                     }
                     int eno2 = ::pthread_mutex_consistent(mutex_);
                     if (eno2 != 0) {
@@ -146,6 +189,7 @@ class mutex {
     }
 
     bool try_lock() noexcept(false) {
+        if (!valid()) return false;
         auto ts = detail::make_timespec(0);
         int eno = ::pthread_mutex_timedlock(mutex_, &ts);
         switch (eno) {
@@ -154,8 +198,8 @@ class mutex {
         case ETIMEDOUT:
             return false;
         case EOWNERDEAD: {
-                if (shm_.ref() > 1) {
-                    shm_.sub_ref();
+                if (shm_->ref() > 1) {
+                    shm_->sub_ref();
                 }
                 int eno2 = ::pthread_mutex_consistent(mutex_);
                 if (eno2 != 0) {
@@ -177,6 +221,7 @@ class mutex {
     }
 
     bool unlock() noexcept {
+        if (!valid()) return false;
         int eno;
         if ((eno = ::pthread_mutex_unlock(mutex_)) != 0) {
             ipc::error("fail pthread_mutex_unlock[%d]\n", eno);

src/libipc/platform/semaphore_linux.h

@@ -50,13 +50,15 @@ class semaphore {
         if (::sem_close(h_) != 0) {
             ipc::error("fail sem_close[%d]: %s\n", errno);
         }
-        if (shm_.ref() == 1) {
-            if (::sem_unlink(shm_.name()) != 0) {
-                ipc::error("fail sem_unlink[%d]: %s\n", errno);
+        h_ = SEM_FAILED;
+        if (shm_.name() != nullptr) {
+            std::string name = shm_.name();
+            if (shm_.release() <= 1) {
+                if (::sem_unlink(name.c_str()) != 0) {
+                    ipc::error("fail sem_unlink[%d]: %s, name: %s\n", errno, name.c_str());
+                }
             }
         }
-        shm_.release();
-        h_ = SEM_FAILED;
     }
 
     bool wait(std::uint64_t tm) noexcept {