From e20dd7d5e3553ffafa1ad6a2fb39bf9c9aaf9385 Mon Sep 17 00:00:00 2001 From: mutouyun Date: Sun, 20 Jun 2021 15:23:44 +0800 Subject: [PATCH] add sync::mutex for windows/linux --- .gitignore | 1 + include/libipc/def.h | 7 +- include/libipc/ipc.h | 26 ++-- include/libipc/mutex.h | 38 ++++++ include/libipc/shm.h | 13 +- include/libipc/waiter.h | 4 +- src/ipc.cpp | 18 +-- src/libipc/platform/detail.h | 12 ++ src/libipc/platform/get_wait_time.h | 39 ++++++ src/libipc/platform/mutex_linux.h | 167 +++++++++++++++++++++++ src/libipc/platform/mutex_win.h | 100 ++++++++++++++ src/libipc/platform/shm_linux.cpp | 29 +++- src/libipc/platform/shm_win.cpp | 8 ++ src/libipc/platform/waiter_linux.h | 17 ++- src/libipc/platform/waiter_win.h | 10 +- src/libipc/platform/waiter_wrapper.h | 19 ++- src/libipc/waiter_helper.h | 2 +- src/mutex.cpp | 70 ++++++++++ src/shm.cpp | 14 +- src/waiter.cpp | 4 +- test/CMakeLists.txt | 3 +- test/{test_pthread.cpp => test_sync.cpp} | 132 +++++++++++------- 22 files changed, 622 insertions(+), 111 deletions(-) create mode 100644 include/libipc/mutex.h create mode 100644 src/libipc/platform/get_wait_time.h create mode 100644 src/libipc/platform/mutex_linux.h create mode 100644 src/libipc/platform/mutex_win.h create mode 100644 src/mutex.cpp rename test/{test_pthread.cpp => test_sync.cpp} (59%) mode change 100755 => 100644 diff --git a/.gitignore b/.gitignore index 5ea6006..c83c97d 100644 --- a/.gitignore +++ b/.gitignore @@ -44,3 +44,4 @@ CMakeLists.txt.user* # My output files build +.vscode \ No newline at end of file diff --git a/include/libipc/def.h b/include/libipc/def.h index 9be5206..d02ff43 100755 --- a/include/libipc/def.h +++ b/include/libipc/def.h @@ -25,13 +25,16 @@ using uint_t = typename uint::type; // constants +enum : std::uint32_t { + invalid_value = (std::numeric_limits::max)(), + default_timeout = 100, // ms +}; + enum : std::size_t { - invalid_value = (std::numeric_limits::max)(), data_length = 64, large_msg_limit = data_length, large_msg_align = 512, large_msg_cache = 32, - default_timeout = 100 // ms }; enum class relat { // multiplicity of the relationship diff --git a/include/libipc/ipc.h b/include/libipc/ipc.h index f6380ae..64b262c 100755 --- a/include/libipc/ipc.h +++ b/include/libipc/ipc.h @@ -27,12 +27,12 @@ struct IPC_EXPORT chan_impl { static char const * name(ipc::handle_t h); static std::size_t recv_count(ipc::handle_t h); - static bool wait_for_recv(ipc::handle_t h, std::size_t r_count, std::size_t tm); + static bool wait_for_recv(ipc::handle_t h, std::size_t r_count, std::uint64_t tm); - static bool send(ipc::handle_t h, void const * data, std::size_t size, std::size_t tm); - static buff_t recv(ipc::handle_t h, std::size_t tm); + static bool send(ipc::handle_t h, void const * data, std::size_t size, std::uint64_t tm); + static buff_t recv(ipc::handle_t h, std::uint64_t tm); - static bool try_send(ipc::handle_t h, void const * data, std::size_t size, std::size_t tm); + static bool try_send(ipc::handle_t h, void const * data, std::size_t size, std::uint64_t tm); static buff_t try_recv(ipc::handle_t h); }; @@ -120,41 +120,41 @@ public: return detail_t::recv_count(h_); } - bool wait_for_recv(std::size_t r_count, std::size_t tm = invalid_value) const { + bool wait_for_recv(std::size_t r_count, std::uint64_t tm = invalid_value) const { return detail_t::wait_for_recv(h_, r_count, tm); } - static bool wait_for_recv(char const * name, std::size_t r_count, std::size_t tm = invalid_value) { + static bool wait_for_recv(char const * name, std::size_t r_count, std::uint64_t tm = invalid_value) { return chan_wrapper(name).wait_for_recv(r_count, tm); } /** * If timeout, this function would call 'force_push' to send the data forcibly. */ - bool send(void const * data, std::size_t size, std::size_t tm = default_timeout) { + bool send(void const * data, std::size_t size, std::uint64_t tm = default_timeout) { return detail_t::send(h_, data, size, tm); } - bool send(buff_t const & buff, std::size_t tm = default_timeout) { + bool send(buff_t const & buff, std::uint64_t tm = default_timeout) { return this->send(buff.data(), buff.size(), tm); } - bool send(std::string const & str, std::size_t tm = default_timeout) { + bool send(std::string const & str, std::uint64_t tm = default_timeout) { return this->send(str.c_str(), str.size() + 1, tm); } /** * If timeout, this function would just return false. */ - bool try_send(void const * data, std::size_t size, std::size_t tm = default_timeout) { + bool try_send(void const * data, std::size_t size, std::uint64_t tm = default_timeout) { return detail_t::try_send(h_, data, size, tm); } - bool try_send(buff_t const & buff, std::size_t tm = default_timeout) { + bool try_send(buff_t const & buff, std::uint64_t tm = default_timeout) { return this->try_send(buff.data(), buff.size(), tm); } - bool try_send(std::string const & str, std::size_t tm = default_timeout) { + bool try_send(std::string const & str, std::uint64_t tm = default_timeout) { return this->try_send(str.c_str(), str.size() + 1, tm); } - buff_t recv(std::size_t tm = invalid_value) { + buff_t recv(std::uint64_t tm = invalid_value) { return detail_t::recv(h_, tm); } diff --git a/include/libipc/mutex.h b/include/libipc/mutex.h new file mode 100644 index 0000000..23f7b53 --- /dev/null +++ b/include/libipc/mutex.h @@ -0,0 +1,38 @@ +#pragma once + +#include // std::uint64_t +#include + +#include "libipc/export.h" +#include "libipc/def.h" + +namespace ipc { +namespace sync { + +class IPC_EXPORT mutex { + mutex(mutex const &) = delete; + mutex &operator=(mutex const &) = delete; + +public: + mutex(); + explicit mutex(char const *name); + ~mutex(); + + void const *native() const noexcept; + void *native() noexcept; + + bool valid() const noexcept; + + bool open(char const *name) noexcept; + void close() noexcept; + bool lock(std::uint64_t tm = ipc::invalid_value) noexcept; + bool try_lock() noexcept(false); // std::system_error + bool unlock() noexcept; + +private: + class mutex_; + mutex_* p_; +}; + +} // namespace sync +} // namespace ipc diff --git a/include/libipc/shm.h b/include/libipc/shm.h index 91a68e4..395b6c8 100755 --- a/include/libipc/shm.h +++ b/include/libipc/shm.h @@ -1,6 +1,7 @@ #pragma once #include +#include #include "libipc/export.h" @@ -20,6 +21,9 @@ IPC_EXPORT void 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 void sub_ref(id_t id); + class IPC_EXPORT handle { public: handle(); @@ -31,9 +35,12 @@ public: void swap(handle& rhs); handle& operator=(handle rhs); - bool valid() const; - std::size_t size () const; - char const * name () const; + bool valid() const noexcept; + std::size_t size () const noexcept; + char const * name () const noexcept; + + std::uint32_t ref() const noexcept; + void sub_ref() noexcept; bool acquire(char const * name, std::size_t size, unsigned mode = create | open); void release(); diff --git a/include/libipc/waiter.h b/include/libipc/waiter.h index a4c3c09..37197e0 100755 --- a/include/libipc/waiter.h +++ b/include/libipc/waiter.h @@ -54,7 +54,7 @@ public: bool open (char const * name, long count = 0); void close(); - bool wait(std::size_t tm = invalid_value); + bool wait(std::uint64_t tm = invalid_value); bool post(long count = 1); private: @@ -81,7 +81,7 @@ public: bool open (char const * name); void close(); - bool wait(mutex&, std::size_t tm = invalid_value); + bool wait(mutex&, std::uint64_t tm = invalid_value); bool notify(); bool broadcast(); diff --git a/src/ipc.cpp b/src/ipc.cpp index 12a629b..bfa1376 100755 --- a/src/ipc.cpp +++ b/src/ipc.cpp @@ -291,7 +291,7 @@ struct conn_info_head { }; template -bool wait_for(W& waiter, F&& pred, std::size_t tm) { +bool wait_for(W& waiter, F&& pred, std::uint64_t tm) { if (tm == 0) return !pred(); for (unsigned k = 0; pred();) { bool loop = true, ret = true; @@ -403,7 +403,7 @@ static std::size_t recv_count(ipc::handle_t h) noexcept { return que->conn_count(); } -static bool wait_for_recv(ipc::handle_t h, std::size_t r_count, std::size_t tm) { +static bool wait_for_recv(ipc::handle_t h, std::size_t r_count, std::uint64_t tm) { auto que = queue_of(h); if (que == nullptr) { return false; @@ -475,7 +475,7 @@ static bool send(F&& gen_push, ipc::handle_t h, void const * data, std::size_t s return true; } -static bool send(ipc::handle_t h, void const * data, std::size_t size, std::size_t tm) { +static bool send(ipc::handle_t h, void const * data, std::size_t size, std::uint64_t tm) { return send([tm](auto info, auto que, auto msg_id) { return [tm, info, que, msg_id](std::int32_t remain, void const * data, std::size_t size) { if (!wait_for(info->wt_waiter_, [&] { @@ -500,7 +500,7 @@ static bool send(ipc::handle_t h, void const * data, std::size_t size, std::size }, h, data, size); } -static bool try_send(ipc::handle_t h, void const * data, std::size_t size, std::size_t tm) { +static bool try_send(ipc::handle_t h, void const * data, std::size_t size, std::uint64_t tm) { return send([tm](auto info, auto que, auto msg_id) { return [tm, info, que, msg_id](std::int32_t remain, void const * data, std::size_t size) { if (!wait_for(info->wt_waiter_, [&] { @@ -514,7 +514,7 @@ static bool try_send(ipc::handle_t h, void const * data, std::size_t size, std:: }, h, data, size); } -static ipc::buff_t recv(ipc::handle_t h, std::size_t tm) { +static ipc::buff_t recv(ipc::handle_t h, std::uint64_t tm) { auto que = queue_of(h); if (que == nullptr) { ipc::error("fail: recv, queue_of(h) == nullptr\n"); @@ -630,22 +630,22 @@ std::size_t chan_impl::recv_count(ipc::handle_t h) { } template -bool chan_impl::wait_for_recv(ipc::handle_t h, std::size_t r_count, std::size_t tm) { +bool chan_impl::wait_for_recv(ipc::handle_t h, std::size_t r_count, std::uint64_t tm) { return detail_impl>::wait_for_recv(h, r_count, tm); } template -bool chan_impl::send(ipc::handle_t h, void const * data, std::size_t size, std::size_t tm) { +bool chan_impl::send(ipc::handle_t h, void const * data, std::size_t size, std::uint64_t tm) { return detail_impl>::send(h, data, size, tm); } template -buff_t chan_impl::recv(ipc::handle_t h, std::size_t tm) { +buff_t chan_impl::recv(ipc::handle_t h, std::uint64_t tm) { return detail_impl>::recv(h, tm); } template -bool chan_impl::try_send(ipc::handle_t h, void const * data, std::size_t size, std::size_t tm) { +bool chan_impl::try_send(ipc::handle_t h, void const * data, std::size_t size, std::uint64_t tm) { return detail_impl>::try_send(h, data, size, tm); } diff --git a/src/libipc/platform/detail.h b/src/libipc/platform/detail.h index fbf539a..97bbd12 100755 --- a/src/libipc/platform/detail.h +++ b/src/libipc/platform/detail.h @@ -22,6 +22,18 @@ # error "IPC_CONSTEXPR_ has been defined." #endif +// detect platform + +#if defined(WIN64) || defined(_WIN64) || defined(__WIN64__) || \ + defined(WIN32) || defined(_WIN32) || defined(__WIN32__) || defined(__NT__) || \ + defined(WINCE) || defined(_WIN32_WCE) +# define IPC_OS_WINDOWS_ +#endif/*WIN*/ + +#if defined(__linux__) || defined(__linux) +# define IPC_OS_LINUX_ +#endif/*linux*/ + #if __cplusplus >= 201703L #define IPC_UNUSED_ [[maybe_unused]] diff --git a/src/libipc/platform/get_wait_time.h b/src/libipc/platform/get_wait_time.h new file mode 100644 index 0000000..785cd75 --- /dev/null +++ b/src/libipc/platform/get_wait_time.h @@ -0,0 +1,39 @@ +#pragma once + +#include +#include + +#include +#include +#include + +#include "libipc/utility/log.h" + +namespace ipc { +namespace detail { + +inline bool calc_wait_time(timespec &ts, std::uint64_t tm /*ms*/) noexcept { + timeval now; + int eno = ::gettimeofday(&now, NULL); + if (eno != 0) { + ipc::error("fail gettimeofday [%d]\n", eno); + return false; + } + ts.tv_nsec = (now.tv_usec + (tm % 1000) * 1000) * 1000; + ts.tv_sec = now.tv_sec + (tm / 1000) + (ts.tv_nsec / 1000000000l); + ts.tv_nsec %= 1000000000l; + return true; +} + +inline timespec make_timespec(std::uint64_t tm /*ms*/) noexcept(false) { + timespec ts {}; + if (!calc_wait_time(ts, tm)) { + ipc::error("fail calc_wait_time: tm = %zd, tv_sec = %ld, tv_nsec = %ld\n", + tm, ts.tv_sec, ts.tv_nsec); + throw std::system_error{static_cast(errno), std::system_category()}; + } + return ts; +} + +} // namespace detail +} // namespace ipc diff --git a/src/libipc/platform/mutex_linux.h b/src/libipc/platform/mutex_linux.h new file mode 100644 index 0000000..370a297 --- /dev/null +++ b/src/libipc/platform/mutex_linux.h @@ -0,0 +1,167 @@ +#pragma once + +#include +#include +#include + +#include + +#include "libipc/platform/get_wait_time.h" +#include "libipc/platform/detail.h" +#include "libipc/utility/log.h" +#include "libipc/utility/scope_guard.h" +#include "libipc/shm.h" + +namespace ipc { +namespace detail { +namespace sync { + +class mutex { + ipc::shm::handle shm_; + pthread_mutex_t *mutex_ = nullptr; + +public: + mutex() noexcept = default; + explicit mutex(char const *name) noexcept { + open(name); + } + + ~mutex() noexcept = default; + + pthread_mutex_t const *native() const noexcept { + return mutex_; + } + + pthread_mutex_t *native() noexcept { + return mutex_; + } + + bool valid() const noexcept { + static const tmp[sizeof pthread_mutex_t] {}; + return shm_.valid() + && (mutex_ != nullptr) + && (std::memcmp(tmp, mutex_, sizeof pthread_mutex_t) != 0); + } + + bool open(char const *name) noexcept { + close(); + if (!shm_.acquire(name, sizeof pthread_mutex_t)) { + ipc::error("fail shm.acquire: %s\n", name); + return false; + } + mutex_ = static_cast(shm_.get()); + assert(mutex_ != nullptr); + if ((shm_.ref() == 1) && valid()/*it means mutex has been inited*/) { + ::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); + } + } + shm_.release(); + mutex_ = nullptr; + } + + bool lock(std::uint64_t tm) noexcept { + for (;;) { + int eno = (tm == invalid_value) + ? ::pthread_mutex_lock(mutex_) + : ::pthread_mutex_timedlock(mutex_, detail::make_timespec(tm)); + switch (eno) { + case 0: + return true; + case ETIMEDOUT: + return false; + case EOWNERDEAD: + if (shm_.ref() > 1) { + shm_.sub_ref(); + } + int eno2 = ::pthread_mutex_consistent(mutex_); + if (eno2 != 0) { + ipc::error("fail pthread_mutex_lock[%d], pthread_mutex_consistent[%d]\n", eno, eno2); + return false; + } + int eno3 = ::pthread_mutex_unlock(mutex_); + if (eno3 != 0) { + ipc::error("fail pthread_mutex_lock[%d], pthread_mutex_unlock[%d]\n", eno, eno3); + return false; + } + break; // loop again + default: + ipc::error("fail pthread_mutex_lock[%d]\n", eno); + return false; + } + } + } + + bool try_lock() noexcept(false) { + int eno = ::pthread_mutex_timedlock(mutex_, detail::make_timespec(0)); + switch (eno) { + case 0: + return true; + case ETIMEDOUT: + return false; + case EOWNERDEAD: + if (shm_.ref() > 1) { + shm_.sub_ref(); + } + int eno2 = ::pthread_mutex_consistent(mutex_); + if (eno2 != 0) { + ipc::error("fail pthread_mutex_timedlock[%d], pthread_mutex_consistent[%d]\n", eno, eno2); + break; + } + int eno3 = ::pthread_mutex_unlock(mutex_); + if (eno3 != 0) { + ipc::error("fail pthread_mutex_timedlock[%d], pthread_mutex_unlock[%d]\n", eno, eno3); + break; + } + break; + default: + ipc::error("fail pthread_mutex_timedlock[%d]\n", eno); + break; + } + throw std::system_error{eno, std::system_category()}; + } + + bool unlock() noexcept { + int eno; + if ((eno = ::pthread_mutex_unlock(mutex_)) != 0) { + ipc::error("fail pthread_mutex_unlock[%d]\n", eno); + return false; + } + return true; + } +}; + +} // namespace sync +} // namespace detail +} // namespace ipc diff --git a/src/libipc/platform/mutex_win.h b/src/libipc/platform/mutex_win.h new file mode 100644 index 0000000..cac5d82 --- /dev/null +++ b/src/libipc/platform/mutex_win.h @@ -0,0 +1,100 @@ +#pragma once + +#include +#include + +#include + +#include "libipc/utility/log.h" + +#include "libipc/platform/to_tchar.h" +#include "libipc/platform/get_sa.h" + +namespace ipc { +namespace detail { +namespace sync { + +class mutex { + HANDLE h_ = NULL; + +public: + mutex() noexcept = default; + explicit mutex(char const *name) noexcept { + open(name); + } + + ~mutex() noexcept = default; + + HANDLE native() const noexcept { + return h_; + } + + bool valid() const noexcept { + return h_ != NULL; + } + + bool open(char const *name) noexcept { + close(); + h_ = ::CreateMutex(detail::get_sa(), FALSE, ipc::detail::to_tchar(name).c_str()); + if (h_ == NULL) { + ipc::error("fail CreateMutex[%lu]: %s\n", ::GetLastError(), name); + return false; + } + return true; + } + + void close() noexcept { + if (!valid()) return; + ::CloseHandle(h_); + h_ = NULL; + } + + bool lock(std::uint64_t tm) noexcept { + DWORD ret, ms = (tm == invalid_value) ? INFINITE : static_cast(tm); + for(;;) { + switch ((ret = ::WaitForSingleObject(h_, ms))) { + case WAIT_OBJECT_0: + return true; + case WAIT_TIMEOUT: + return false; + case WAIT_ABANDONED: + ipc::log("fail WaitForSingleObject[%lu]: WAIT_ABANDONED, try again.\n", ::GetLastError()); + if (!unlock()) { + return false; + } + break; // loop again + default: + ipc::error("fail WaitForSingleObject[%lu]: 0x%08X\n", ::GetLastError(), ret); + return false; + } + } + } + + bool try_lock() noexcept(false) { + DWORD ret = ::WaitForSingleObject(h_, 0); + switch (ret) { + case WAIT_OBJECT_0: + return true; + case WAIT_TIMEOUT: + return false; + case WAIT_ABANDONED: + unlock(); + IPC_FALLTHROUGH_; + default: + ipc::error("fail WaitForSingleObject[%lu]: 0x%08X\n", ::GetLastError(), ret); + throw std::system_error{static_cast(ret), std::system_category()}; + } + } + + bool unlock() noexcept { + if (!::ReleaseMutex(h_)) { + ipc::error("fail ReleaseMutex[%lu]\n", ::GetLastError()); + return false; + } + return true; + } +}; + +} // namespace sync +} // namespace detail +} // namespace ipc diff --git a/src/libipc/platform/shm_linux.cpp b/src/libipc/platform/shm_linux.cpp index 9b523d6..f3396b7 100755 --- a/src/libipc/platform/shm_linux.cpp +++ b/src/libipc/platform/shm_linux.cpp @@ -22,7 +22,7 @@ namespace { struct info_t { - std::atomic_size_t acc_; + std::atomic acc_; }; struct id_info_t { @@ -81,6 +81,31 @@ id_t acquire(char const * name, std::size_t size, unsigned mode) { return ii; } +std::uint32_t get_ref(id_t id) { + if (id == nullptr) { + ipc::error("fail get_ref: invalid id (null)\n"); + return 0; + } + auto ii = static_cast(id); + if (ii->mem_ == nullptr || ii->size_ == 0) { + return 0; + } + return acc_of(mem, ii->size_).load(std::memory_order_acquire); +} + +void sub_ref(id_t id) { + if (id == nullptr) { + ipc::error("fail sub_ref: invalid id (null)\n"); + return; + } + auto ii = static_cast(id); + if (ii->mem_ == nullptr || ii->size_ == 0) { + ipc::error("fail sub_ref: invalid id (mem = %p, size = %zd)\n", ii->mem_, ii->size_); + return; + } + acc_of(ii->mem_, ii->size_).fetch_sub(1, std::memory_order_acq_rel); +} + void * get_mem(id_t id, std::size_t * size) { if (id == nullptr) { ipc::error("fail get_mem: invalid id (null)\n"); @@ -137,7 +162,7 @@ void release(id_t id) { if (ii->mem_ == nullptr || ii->size_ == 0) { ipc::error("fail release: invalid id (mem = %p, size = %zd)\n", ii->mem_, ii->size_); } - else if (acc_of(ii->mem_, ii->size_).fetch_sub(1, std::memory_order_acquire) == 1) { + else if (acc_of(ii->mem_, ii->size_).fetch_sub(1, std::memory_order_acq_rel) == 1) { ::munmap(ii->mem_, ii->size_); if (!ii->name_.empty()) { ::shm_unlink(ii->name_.c_str()); diff --git a/src/libipc/platform/shm_win.cpp b/src/libipc/platform/shm_win.cpp index 389372d..858a1e5 100755 --- a/src/libipc/platform/shm_win.cpp +++ b/src/libipc/platform/shm_win.cpp @@ -58,6 +58,14 @@ id_t acquire(char const * name, std::size_t size, unsigned mode) { return ii; } +std::uint32_t get_ref(id_t) { + return 0; +} + +void sub_ref(id_t) { + // Do Nothing. +} + void * get_mem(id_t id, std::size_t * size) { if (id == nullptr) { ipc::error("fail get_mem: invalid id (null)\n"); diff --git a/src/libipc/platform/waiter_linux.h b/src/libipc/platform/waiter_linux.h index eb9a3b1..006ba6a 100755 --- a/src/libipc/platform/waiter_linux.h +++ b/src/libipc/platform/waiter_linux.h @@ -22,7 +22,7 @@ namespace ipc { namespace detail { -inline static bool calc_wait_time(timespec& ts, std::size_t tm /*ms*/) { +inline static bool calc_wait_time(timespec& ts, std::uint64_t tm /*ms*/) { timeval now; int eno = ::gettimeofday(&now, NULL); if (eno != 0) { @@ -89,8 +89,7 @@ public: return true; case EOWNERDEAD: if (::pthread_mutex_consistent(&mutex_) == 0) { - ::pthread_mutex_unlock(&mutex_); - break; + return true; } IPC_FALLTHROUGH_; case ENOTRECOVERABLE: @@ -138,7 +137,7 @@ public: IPC_PTHREAD_FUNC_(pthread_cond_destroy, &cond_); } - bool wait(mutex& mtx, std::size_t tm = invalid_value) { + bool wait(mutex& mtx, std::uint64_t tm = invalid_value) { switch (tm) { case 0: return true; @@ -221,7 +220,7 @@ public: return true; } - static bool wait(handle_t h, std::size_t tm = invalid_value) { + static bool wait(handle_t h, std::uint64_t tm = invalid_value) { if (h == invalid()) return false; switch (tm) { case 0: @@ -289,7 +288,7 @@ private: return ipc::detail::unique_lock(me_->lock_); } - bool sema_wait(std::size_t tm) { + bool sema_wait(std::uint64_t tm) { return sem_helper::wait(std::get<1>(h_), tm); } @@ -297,7 +296,7 @@ private: return sem_helper::post(std::get<1>(h_), count); } - bool handshake_wait(std::size_t tm) { + bool handshake_wait(std::uint64_t tm) { return sem_helper::wait(std::get<2>(h_), tm); } @@ -339,7 +338,7 @@ public: } template - bool wait_if(handle_t const & h, wait_flags * flags, F&& pred, std::size_t tm = invalid_value) { + bool wait_if(handle_t const & h, wait_flags * flags, F&& pred, std::uint64_t tm = invalid_value) { assert(flags != nullptr); contrl ctrl { this, flags, h }; @@ -400,7 +399,7 @@ public: } template - bool wait_if(handle_t h, waiter_helper::wait_flags * flags, F && pred, std::size_t tm = invalid_value) { + bool wait_if(handle_t h, waiter_helper::wait_flags * flags, F && pred, std::uint64_t tm = invalid_value) { if (h == invalid()) return false; return helper_.wait_if(h, flags, std::forward(pred), tm); } diff --git a/src/libipc/platform/waiter_win.h b/src/libipc/platform/waiter_win.h index 4f3d080..9f5f888 100755 --- a/src/libipc/platform/waiter_win.h +++ b/src/libipc/platform/waiter_win.h @@ -40,7 +40,7 @@ public: ::CloseHandle(h_); } - bool wait(std::size_t tm = invalid_value) { + bool wait(std::uint64_t tm = invalid_value) { DWORD ret, ms = (tm == invalid_value) ? INFINITE : static_cast(tm); switch ((ret = ::WaitForSingleObject(h_, ms))) { case WAIT_OBJECT_0: @@ -102,7 +102,7 @@ class condition { return ipc::detail::unique_lock(me_->lock_); } - bool sema_wait(std::size_t tm) { + bool sema_wait(std::uint64_t tm) { return me_->sema_.wait(tm); } @@ -110,7 +110,7 @@ class condition { return me_->sema_.post(count); } - bool handshake_wait(std::size_t tm) { + bool handshake_wait(std::uint64_t tm) { return me_->handshake_.wait(tm); } @@ -151,7 +151,7 @@ public: } template - bool wait_if(Mutex & mtx, wait_flags * flags, F && pred, std::size_t tm = invalid_value) { + bool wait_if(Mutex & mtx, wait_flags * flags, F && pred, std::uint64_t tm = invalid_value) { assert(flags != nullptr); contrl ctrl { this, flags }; return waiter_helper::wait_if(ctrl, mtx, std::forward(pred), tm); @@ -201,7 +201,7 @@ public: } template - bool wait_if(handle_t& h, waiter_helper::wait_flags * flags, F&& pred, std::size_t tm = invalid_value) { + bool wait_if(handle_t& h, waiter_helper::wait_flags * flags, F&& pred, std::uint64_t tm = invalid_value) { if (h == invalid()) return false; class non_mutex { diff --git a/src/libipc/platform/waiter_wrapper.h b/src/libipc/platform/waiter_wrapper.h index 553d2e1..18c06e0 100755 --- a/src/libipc/platform/waiter_wrapper.h +++ b/src/libipc/platform/waiter_wrapper.h @@ -8,9 +8,7 @@ #include "libipc/memory/resource.h" #include "libipc/platform/detail.h" -#if defined(WIN64) || defined(_WIN64) || defined(__WIN64__) || \ - defined(WIN32) || defined(_WIN32) || defined(__WIN32__) || defined(__NT__) || \ - defined(WINCE) || defined(_WIN32_WCE) +#if defined(IPC_OS_WINDOWS_) #include "libipc/platform/waiter_win.h" @@ -52,7 +50,7 @@ public: cnt_h_.release(); } - bool wait(mutex_impl& mtx, std::size_t tm = invalid_value) { + bool wait(mutex_impl& mtx, std::uint64_t tm = invalid_value) { return base_t::wait_if(mtx, &flags_, [] { return true; }, tm); } }; @@ -60,7 +58,7 @@ public: } // namespace detail } // namespace ipc -#else /*!WIN*/ +#elif defined(IPC_OS_LINUX_) #include "libipc/platform/waiter_linux.h" @@ -123,7 +121,7 @@ public: class condition_impl : public object_impl { public: - bool wait(mutex_impl& mtx, std::size_t tm = invalid_value) { + bool wait(mutex_impl& mtx, std::uint64_t tm = invalid_value) { return object().wait(mtx.object(), tm); } @@ -168,7 +166,7 @@ public: opened_.release(); } - bool wait(std::size_t tm = invalid_value) { + bool wait(std::uint64_t tm = invalid_value) { return sem_helper::wait(h_, tm); } @@ -179,8 +177,9 @@ public: } // namespace detail } // namespace ipc - -#endif/*!WIN*/ +#else/*linux*/ +# error "Unsupported platform." +#endif namespace ipc { namespace detail { @@ -235,7 +234,7 @@ public: } template - bool wait_if(F && pred, std::size_t tm = invalid_value) { + bool wait_if(F && pred, std::uint64_t tm = invalid_value) { if (!valid()) return false; return w_->wait_if(h_, &flags_, std::forward(pred), tm); } diff --git a/src/libipc/waiter_helper.h b/src/libipc/waiter_helper.h index a32035b..7bc6a08 100644 --- a/src/libipc/waiter_helper.h +++ b/src/libipc/waiter_helper.h @@ -24,7 +24,7 @@ struct waiter_helper { }; template - static bool wait_if(Ctrl & ctrl, Mutex & mtx, F && pred, std::size_t tm) { + static bool wait_if(Ctrl & ctrl, Mutex & mtx, F && pred, std::uint64_t tm) { auto & flags = ctrl.flags(); if (flags.is_closed_.load(std::memory_order_acquire)) { return false; diff --git a/src/mutex.cpp b/src/mutex.cpp new file mode 100644 index 0000000..813e334 --- /dev/null +++ b/src/mutex.cpp @@ -0,0 +1,70 @@ + +#include "libipc/mutex.h" + +#include "libipc/utility/pimpl.h" +#include "libipc/memory/resource.h" +#include "libipc/platform/detail.h" +#if defined(IPC_OS_WINDOWS_) +#include "libipc/platform/mutex_win.h" +#elif defined(IPC_OS_LINUX_) +#include "libipc/platform/mutex_linux.h" +#else/*linux*/ +# error "Unsupported platform." +#endif + +namespace ipc { +namespace sync { + +class mutex::mutex_ : public ipc::pimpl { +public: + ipc::detail::sync::mutex lock_; +}; + +mutex::mutex() + : p_(p_->make()) { +} + +mutex::mutex(char const * name) + : mutex() { + open(name); +} + +mutex::~mutex() { + close(); + p_->clear(); +} + +void const *mutex::native() const noexcept { + return impl(p_)->lock_.native(); +} + +void *mutex::native() noexcept { + return impl(p_)->lock_.native(); +} + +bool mutex::valid() const noexcept { + return impl(p_)->lock_.valid(); +} + +bool mutex::open(char const *name) noexcept { + return impl(p_)->lock_.open(name); +} + +void mutex::close() noexcept { + impl(p_)->lock_.close(); +} + +bool mutex::lock(std::uint64_t tm) noexcept { + return impl(p_)->lock_.lock(tm); +} + +bool mutex::try_lock() noexcept(false) { + return impl(p_)->lock_.try_lock(); +} + +bool mutex::unlock() noexcept { + return impl(p_)->lock_.unlock(); +} + +} // namespace sync +} // namespace ipc diff --git a/src/shm.cpp b/src/shm.cpp index a2eed93..24cb377 100755 --- a/src/shm.cpp +++ b/src/shm.cpp @@ -47,18 +47,26 @@ handle& handle::operator=(handle rhs) { return *this; } -bool handle::valid() const { +bool handle::valid() const noexcept { return impl(p_)->m_ != nullptr; } -std::size_t handle::size() const { +std::size_t handle::size() const noexcept { return impl(p_)->s_; } -char const * handle::name() const { +char const * handle::name() const noexcept { return impl(p_)->n_.c_str(); } +std::uint32_t handle::ref() const noexcept { + return shm::get_ref(impl(p_)->id_); +} + +void handle::sub_ref() noexcept { + shm::sub_ref(impl(p_)->id_); +} + bool handle::acquire(char const * name, std::size_t size, unsigned mode) { release(); impl(p_)->id_ = shm::acquire((impl(p_)->n_ = name).c_str(), size, mode); diff --git a/src/waiter.cpp b/src/waiter.cpp index 24a1382..2ac32c0 100755 --- a/src/waiter.cpp +++ b/src/waiter.cpp @@ -44,7 +44,7 @@ bool mutex::unlock() { #include "libipc/waiter_template.inc" -bool semaphore::wait(std::size_t tm) { +bool semaphore::wait(std::uint64_t tm) { return impl(p_)->h_.wait(tm); } @@ -62,7 +62,7 @@ bool semaphore::post(long count) { #include "libipc/waiter_template.inc" -bool condition::wait(mutex& mtx, std::size_t tm) { +bool condition::wait(mutex& mtx, std::uint64_t tm) { return impl(p_)->h_.wait(impl(mtx.p_)->h_, tm); } diff --git a/test/CMakeLists.txt b/test/CMakeLists.txt index 07b0a34..9398aa3 100755 --- a/test/CMakeLists.txt +++ b/test/CMakeLists.txt @@ -17,7 +17,8 @@ include_directories( file(GLOB SRC_FILES ${LIBIPC_PROJECT_DIR}/test/*.cpp - ${LIBIPC_PROJECT_DIR}/test/profiler/*.cpp) + # ${LIBIPC_PROJECT_DIR}/test/profiler/*.cpp + ) file(GLOB HEAD_FILES ${LIBIPC_PROJECT_DIR}/test/*.h) add_executable(${PROJECT_NAME} ${SRC_FILES} ${HEAD_FILES}) diff --git a/test/test_pthread.cpp b/test/test_sync.cpp old mode 100755 new mode 100644 similarity index 59% rename from test/test_pthread.cpp rename to test/test_sync.cpp index 36d0acf..d2f78bf --- a/test/test_pthread.cpp +++ b/test/test_sync.cpp @@ -1,49 +1,83 @@ - -#include -#include - -#include "test.h" - -#if defined(__linux__) || defined(__linux) -#include -#include - -TEST(PThread, Robust) { - pthread_mutexattr_t ma; - pthread_mutexattr_init(&ma); - pthread_mutexattr_setrobust(&ma, PTHREAD_MUTEX_ROBUST); - pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER; - pthread_mutex_init(&mutex, &ma); - - std::thread{[&mutex] { - pthread_mutex_lock(&mutex); - // pthread_mutex_unlock(&mutex); - }}.join(); - - struct timespec tout; - clock_gettime(CLOCK_REALTIME, &tout); - int r = pthread_mutex_timedlock(&mutex, &tout); - EXPECT_EQ(r, EOWNERDEAD); - - pthread_mutex_consistent(&mutex); - pthread_mutex_unlock(&mutex); - pthread_mutex_destroy(&mutex); -} -#elif defined(WIN64) || defined(_WIN64) || defined(__WIN64__) || \ - defined(WIN32) || defined(_WIN32) || defined(__WIN32__) || defined(__NT__) -#include -#include - -TEST(PThread, Robust) { - HANDLE lock = CreateMutex(NULL, FALSE, _T("test-robust")); - std::thread{[] { - HANDLE lock = CreateMutex(NULL, FALSE, _T("test-robust")); - WaitForSingleObject(lock, 0); - }}.join(); - - DWORD r = WaitForSingleObject(lock, 0); - EXPECT_EQ(r, WAIT_ABANDONED); - - CloseHandle(lock); -} -#endif // !__linux__ \ No newline at end of file + +#include +#include +#include +#include + +#include "test.h" + +#if defined(__linux__) || defined(__linux) +#include +#include + +TEST(PThread, Robust) { + pthread_mutexattr_t ma; + pthread_mutexattr_init(&ma); + pthread_mutexattr_setrobust(&ma, PTHREAD_MUTEX_ROBUST); + pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER; + pthread_mutex_init(&mutex, &ma); + + std::thread{[&mutex] { + pthread_mutex_lock(&mutex); + // pthread_mutex_unlock(&mutex); + }}.join(); + + struct timespec tout; + clock_gettime(CLOCK_REALTIME, &tout); + int r = pthread_mutex_timedlock(&mutex, &tout); + EXPECT_EQ(r, EOWNERDEAD); + + pthread_mutex_consistent(&mutex); + pthread_mutex_unlock(&mutex); + pthread_mutex_destroy(&mutex); +} +#elif defined(WIN64) || defined(_WIN64) || defined(__WIN64__) || \ + defined(WIN32) || defined(_WIN32) || defined(__WIN32__) || defined(__NT__) +#include +#include + +TEST(PThread, Robust) { + HANDLE lock = CreateMutex(NULL, FALSE, _T("test-robust")); + std::thread{[] { + HANDLE lock = CreateMutex(NULL, FALSE, _T("test-robust")); + WaitForSingleObject(lock, 0); + }}.join(); + + DWORD r = WaitForSingleObject(lock, 0); + EXPECT_EQ(r, WAIT_ABANDONED); + + CloseHandle(lock); +} +#endif // !__linux__ + +#include "libipc/mutex.h" + +TEST(Sync, Mutex) { + ipc::sync::mutex lock; + EXPECT_TRUE(lock.open("test-mutex-robust")); + + std::thread{[] { + ipc::sync::mutex lock{"test-mutex-robust"}; + EXPECT_TRUE(lock.valid()); + EXPECT_TRUE(lock.lock()); + }}.join(); + + EXPECT_THROW(lock.try_lock(), std::system_error); + + int i = 0; + EXPECT_TRUE(lock.lock()); + i = 100; + auto t2 = std::thread{[&i] { + ipc::sync::mutex lock{"test-mutex-robust"}; + EXPECT_TRUE(lock.valid()); + EXPECT_FALSE(lock.try_lock()); + EXPECT_TRUE(lock.lock()); + i += i; + EXPECT_TRUE(lock.unlock()); + }}; + std::this_thread::sleep_for(std::chrono::seconds(1)); + EXPECT_EQ(i, 100); + EXPECT_TRUE(lock.unlock()); + t2.join(); + EXPECT_EQ(i, 200); +} \ No newline at end of file