coffee/cpp-ipc
1
0
Fork 0
mirror of https://github.com/mutouyun/cpp-ipc.git synced 2026-01-01 03:12:13 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #166 from mutouyun/feature/imp-log

Browse Source 
refactor(log): Replace utility/log with imp/log interface
This commit is contained in:
木头云 2025-12-15 20:18:06 +08:00 committed by GitHub
commit 3269bde1a5
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
21 changed files with 390 additions and 334 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

54
src/libipc/ipc.cpp
View File

@ -19,7 +19,7 @@
#include "libipc/rw_lock.h"
#include "libipc/waiter.h"
#include "libipc/utility/log.h"
#include "libipc/imp/log.h"
#include "libipc/utility/id_pool.h"
#include "libipc/utility/scope_guard.h"
#include "libipc/utility/utility.h"
@ -76,6 +76,7 @@ ipc::buff_t make_cache(T &data, std::size_t size) {
}
acc_t *cc_acc(std::string const &pref) {
LIBIPC_LOG();
static auto *phs = new ipc::unordered_map<std::string, ipc::shm::handle>; // no delete
static std::mutex lock;
std::lock_guard<std::mutex> guard {lock};
@ -84,7 +85,7 @@ acc_t *cc_acc(std::string const &pref) {
std::string shm_name {ipc::make_prefix(pref, "CA_CONN__")};
ipc::shm::handle h;
if (!h.acquire(shm_name.c_str(), sizeof(acc_t))) {
ipc::error("[cc_acc] acquire failed: %s\n", shm_name.c_str());
log.error("[cc_acc] acquire failed: ", shm_name);
return nullptr;
}
it = phs->emplace(pref, std::move(h)).first;
@ -217,10 +218,11 @@ auto& chunk_storages() {
std::mutex lock_;
static bool make_handle(ipc::shm::handle &h, std::string const &shm_name, std::size_t chunk_size) {
LIBIPC_LOG();
if (!h.valid() &&
!h.acquire( shm_name.c_str(),
sizeof(chunk_info_t) + chunk_info_t::chunks_mem_size(chunk_size) )) {
ipc::error("[chunk_storages] chunk_shm.id_info_.acquire failed: chunk_size = %zd\n", chunk_size);
log.error("[chunk_storages] chunk_shm.id_info_.acquire failed: chunk_size = ", chunk_size);
return false;
}
return true;
@ -228,6 +230,7 @@ auto& chunk_storages() {
public:
chunk_info_t *get_info(conn_info_head *inf, std::size_t chunk_size) {
LIBIPC_LOG();
std::string pref {(inf == nullptr) ? std::string{} : inf->prefix_};
std::string shm_name {ipc::make_prefix(pref, "CHUNK_INFO__", chunk_size)};
ipc::shm::handle *h;
@ -240,7 +243,7 @@ auto& chunk_storages() {
}
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);
log.error("[chunk_storages] chunk_shm.id_info_.get failed: chunk_size = ", chunk_size);
return nullptr;
}
return info;
@ -290,8 +293,9 @@ std::pair<ipc::storage_id_t, void*> acquire_storage(conn_info_head *inf, std::si
}
void *find_storage(ipc::storage_id_t id, conn_info_head *inf, std::size_t size) {
LIBIPC_LOG();
if (id < 0) {
ipc::error("[find_storage] id is invalid: id = %ld, size = %zd\n", (long)id, size);
log.error("[find_storage] id is invalid: id = ", (long)id, ", size = ", size);
return nullptr;
}
std::size_t chunk_size = calc_chunk_size(size);
@ -301,8 +305,9 @@ void *find_storage(ipc::storage_id_t id, conn_info_head *inf, std::size_t size)
}
void release_storage(ipc::storage_id_t id, conn_info_head *inf, std::size_t size) {
LIBIPC_LOG();
if (id < 0) {
ipc::error("[release_storage] id is invalid: id = %ld, size = %zd\n", (long)id, size);
log.error("[release_storage] id is invalid: id = ", (long)id, ", size = ", size);
return;
}
std::size_t chunk_size = calc_chunk_size(size);
@ -334,8 +339,9 @@ bool sub_rc(ipc::wr<Rp, Rc, ipc::trans::broadcast>,
template <typename Flag>
void recycle_storage(ipc::storage_id_t id, conn_info_head *inf, std::size_t size, ipc::circ::cc_t curr_conns, ipc::circ::cc_t conn_id) {
LIBIPC_LOG();
if (id < 0) {
ipc::error("[recycle_storage] id is invalid: id = %ld, size = %zd\n", (long)id, size);
log.error("[recycle_storage] id is invalid: id = ", (long)id, ", size = ", size);
return;
}
std::size_t chunk_size = calc_chunk_size(size);
@ -355,11 +361,12 @@ void recycle_storage(ipc::storage_id_t id, conn_info_head *inf, std::size_t size
template <typename MsgT>
bool clear_message(conn_info_head *inf, void* p) {
LIBIPC_LOG();
auto msg = static_cast<MsgT*>(p);
if (msg->storage_) {
std::int32_t r_size = static_cast<std::int32_t>(ipc::data_length) + msg->remain_;
if (r_size <= 0) {
ipc::error("[clear_message] invalid msg size: %d\n", (int)r_size);
log.error("[clear_message] invalid msg size: ", (int)r_size);
return true;
}
release_storage(*reinterpret_cast<ipc::storage_id_t*>(&msg->data_),
@ -518,33 +525,34 @@ static bool wait_for_recv(ipc::handle_t h, std::size_t r_count, std::uint64_t tm
template <typename F>
static bool send(F&& gen_push, ipc::handle_t h, void const * data, std::size_t size) {
LIBIPC_LOG();
if (data == nullptr || size == 0) {
ipc::error("fail: send(%p, %zd)\n", data, size);
log.error("fail: send(", data, ", ", size, ")");
return false;
}
auto que = queue_of(h);
if (que == nullptr) {
ipc::error("fail: send, queue_of(h) == nullptr\n");
log.error("fail: send, queue_of(h) == nullptr");
return false;
}
if (que->elems() == nullptr) {
ipc::error("fail: send, queue_of(h)->elems() == nullptr\n");
log.error("fail: send, queue_of(h)->elems() == nullptr");
return false;
}
if (!que->ready_sending()) {
ipc::error("fail: send, que->ready_sending() == false\n");
log.error("fail: send, que->ready_sending() == false");
return false;
}
ipc::circ::cc_t conns = que->elems()->connections(std::memory_order_relaxed);
if (conns == 0) {
ipc::error("fail: send, there is no receiver on this connection.\n");
log.error("fail: send, there is no receiver on this connection.");
return false;
}
// calc a new message id
conn_info_t *inf = info_of(h);
auto acc = inf->acc();
if (acc == nullptr) {
ipc::error("fail: send, info_of(h)->acc() == nullptr\n");
log.error("fail: send, info_of(h)->acc() == nullptr");
return false;
}
auto msg_id = acc->fetch_add(1, std::memory_order_relaxed);
@ -558,7 +566,7 @@ 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);
//log.debug("fail: shm::handle for big message. msg_id: ", msg_id, ", size: ", size);
}
// push message fragment
std::int32_t offset = 0;
@ -581,14 +589,15 @@ static bool send(F&& gen_push, ipc::handle_t h, void const * data, std::size_t s
}
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) {
LIBIPC_LOG();
return send([tm, &log](auto *info, auto *que, auto msg_id) {
return [tm, &log, info, que, msg_id](std::int32_t remain, void const * data, std::size_t size) {
if (!wait_for(info->wt_waiter_, [&] {
return !que->push(
[](void*) { return true; },
info->cc_id_, msg_id, remain, data, size);
}, tm)) {
ipc::log("force_push: msg_id = %zd, remain = %d, size = %zd\n", msg_id, remain, size);
log.debug("force_push: msg_id = ", msg_id, ", remain = ", remain, ", size = ", size);
if (!que->force_push(
[info](void* p) { return clear_message<typename queue_t::value_t>(info, p); },
info->cc_id_, msg_id, remain, data, size)) {
@ -618,9 +627,10 @@ static bool try_send(ipc::handle_t h, void const * data, std::size_t size, std::
}
static ipc::buff_t recv(ipc::handle_t h, std::uint64_t tm) {
LIBIPC_LOG();
auto que = queue_of(h);
if (que == nullptr) {
ipc::error("fail: recv, queue_of(h) == nullptr\n");
log.error("fail: recv, queue_of(h) == nullptr");
return {};
}
if (!que->connected()) {
@ -648,7 +658,7 @@ static ipc::buff_t recv(ipc::handle_t h, std::uint64_t tm) {
// msg.remain_ may minus & abs(msg.remain_) < data_length
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);
log.error("fail: recv, r_size = ", (int)r_size);
return {};
}
std::size_t msg_size = static_cast<std::size_t>(r_size);
@ -669,7 +679,7 @@ static ipc::buff_t recv(ipc::handle_t h, std::uint64_t tm) {
que->connected_id()
});
if (r_info == nullptr) {
ipc::log("fail: ipc::mem::$new<recycle_t>.\n");
log.error("fail: ipc::mem::$new<recycle_t>.");
return ipc::buff_t{buf, msg_size}; // no recycle
} else {
return ipc::buff_t{buf, msg_size, [](void* p_info, std::size_t size) {
@ -685,7 +695,7 @@ static ipc::buff_t recv(ipc::handle_t h, std::uint64_t tm) {
}, r_info};
}
} else {
ipc::log("fail: shm::handle for large message. msg_id: %zd, buf_id: %zd, size: %zd\n", msg.id_, buf_id, msg_size);
log.error("fail: shm::handle for large message. msg_id: ", msg.id_, ", buf_id: ", buf_id, ", size: ", msg_size);
continue;
}
}

14
src/libipc/platform/linux/condition.h
View File

@ -1,6 +1,6 @@
#pragma once
#include "libipc/utility/log.h"
#include "libipc/imp/log.h"
#include "libipc/mutex.h"
#include "get_wait_time.h"
@ -19,11 +19,12 @@ public:
~condition() = default;
bool wait(ipc::sync::mutex &mtx, std::uint64_t tm) noexcept {
LIBIPC_LOG();
if (!valid()) return false;
if (tm == invalid_value) {
int eno = A0_SYSERR(a0_cnd_wait(native(), static_cast<a0_mtx_t *>(mtx.native())));
if (eno != 0) {
ipc::error("fail condition wait[%d]\n", eno);
log.error("fail condition wait[", eno, "]");
return false;
}
} else {
@ -31,8 +32,7 @@ public:
int eno = A0_SYSERR(a0_cnd_timedwait(native(), static_cast<a0_mtx_t *>(mtx.native()), {ts}));
if (eno != 0) {
if (eno != ETIMEDOUT) {
ipc::error("fail condition timedwait[%d]: tm = %zd, tv_sec = %ld, tv_nsec = %ld\n",
eno, tm, ts.tv_sec, ts.tv_nsec);
log.error("fail condition timedwait[", eno, "]: tm = ", tm, ", tv_sec = ", ts.tv_sec, ", tv_nsec = ", ts.tv_nsec);
}
return false;
}
@ -41,20 +41,22 @@ public:
}
bool notify(ipc::sync::mutex &mtx) noexcept {
LIBIPC_LOG();
if (!valid()) return false;
int eno = A0_SYSERR(a0_cnd_signal(native(), static_cast<a0_mtx_t *>(mtx.native())));
if (eno != 0) {
ipc::error("fail condition notify[%d]\n", eno);
log.error("fail condition notify[", eno, "]");
return false;
}
return true;
}
bool broadcast(ipc::sync::mutex &mtx) noexcept {
LIBIPC_LOG();
if (!valid()) return false;
int eno = A0_SYSERR(a0_cnd_broadcast(native(), static_cast<a0_mtx_t *>(mtx.native())));
if (eno != 0) {
ipc::error("fail condition broadcast[%d]\n", eno);
log.error("fail condition broadcast[", eno, "]");
return false;
}
return true;

13
src/libipc/platform/linux/get_wait_time.h
View File

@ -4,7 +4,7 @@
#include <cinttypes>
#include <system_error>
#include "libipc/utility/log.h"
#include "libipc/imp/log.h"
#include "a0/time.h"
#include "a0/err_macro.h"
@ -14,30 +14,31 @@ namespace linux_ {
namespace detail {
inline bool calc_wait_time(timespec &ts, std::uint64_t tm /*ms*/) noexcept {
LIBIPC_LOG();
std::int64_t add_ns = static_cast<std::int64_t>(tm * 1000000ull);
if (add_ns < 0) {
ipc::error("invalid time = " PRIu64 "\n", tm);
log.error("invalid time = ", tm);
return false;
}
a0_time_mono_t now;
int eno = A0_SYSERR(a0_time_mono_now(&now));
if (eno != 0) {
ipc::error("fail get time[%d]\n", eno);
log.error("fail get time[", eno, "]");
return false;
}
a0_time_mono_t *target = reinterpret_cast<a0_time_mono_t *>(&ts);
if ((eno = A0_SYSERR(a0_time_mono_add(now, add_ns, target))) != 0) {
ipc::error("fail get time[%d]\n", eno);
log.error("fail get time[", eno, "]");
return false;
}
return true;
}
inline timespec make_timespec(std::uint64_t tm /*ms*/) noexcept(false) {
LIBIPC_LOG();
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);
log.error("fail calc_wait_time: tm = ", tm, ", tv_sec = ", ts.tv_sec, ", tv_nsec = ", ts.tv_nsec);
throw std::system_error{static_cast<int>(errno), std::system_category()};
}
return ts;

19
src/libipc/platform/linux/mutex.h
View File

@ -6,7 +6,7 @@
#include <atomic>
#include "libipc/platform/detail.h"
#include "libipc/utility/log.h"
#include "libipc/imp/log.h"
#include "libipc/mem/resource.h"
#include "libipc/shm.h"
@ -23,6 +23,7 @@ namespace sync {
class robust_mutex : public sync::obj_impl<a0_mtx_t> {
public:
bool lock(std::uint64_t tm) noexcept {
LIBIPC_LOG();
if (!valid()) return false;
for (;;) {
auto ts = linux_::detail::make_timespec(tm);
@ -37,24 +38,25 @@ public:
case EOWNERDEAD: {
int eno2 = A0_SYSERR(a0_mtx_consistent(native()));
if (eno2 != 0) {
ipc::error("fail mutex lock[%d] -> consistent[%d]\n", eno, eno2);
log.error("fail mutex lock[", eno, "] -> consistent[", eno2, "]");
return false;
}
int eno3 = A0_SYSERR(a0_mtx_unlock(native()));
if (eno3 != 0) {
ipc::error("fail mutex lock[%d] -> unlock[%d]\n", eno, eno3);
log.error("fail mutex lock[", eno, "] -> unlock[", eno3, "]");
return false;
}
}
break; // loop again
default:
ipc::error("fail mutex lock[%d]\n", eno);
log.error("fail mutex lock[", eno, "]");
return false;
}
}
}
bool try_lock() noexcept(false) {
LIBIPC_LOG();
if (!valid()) return false;
int eno = A0_SYSERR(a0_mtx_timedlock(native(), {linux_::detail::make_timespec(0)}));
switch (eno) {
@ -65,28 +67,29 @@ public:
case EOWNERDEAD: {
int eno2 = A0_SYSERR(a0_mtx_consistent(native()));
if (eno2 != 0) {
ipc::error("fail mutex try_lock[%d] -> consistent[%d]\n", eno, eno2);
log.error("fail mutex try_lock[", eno, "] -> consistent[", eno2, "]");
break;
}
int eno3 = A0_SYSERR(a0_mtx_unlock(native()));
if (eno3 != 0) {
ipc::error("fail mutex try_lock[%d] -> unlock[%d]\n", eno, eno3);
log.error("fail mutex try_lock[", eno, "] -> unlock[", eno3, "]");
break;
}
}
break;
default:
ipc::error("fail mutex try_lock[%d]\n", eno);
log.error("fail mutex try_lock[", eno, "]");
break;
}
throw std::system_error{eno, std::system_category()};
}
bool unlock() noexcept {
LIBIPC_LOG();
if (!valid()) return false;
int eno = A0_SYSERR(a0_mtx_unlock(native()));
if (eno != 0) {
ipc::error("fail mutex unlock[%d]\n", eno);
log.error("fail mutex unlock[", eno, "]");
return false;
}
return true;

5
src/libipc/platform/linux/sync_obj_impl.h
View File

@ -1,6 +1,6 @@
#pragma once
#include "libipc/utility/log.h"
#include "libipc/imp/log.h"
#include "libipc/shm.h"
#include "a0/empty.h"
@ -19,8 +19,9 @@ protected:
sync_t *h_ = nullptr;
sync_t *acquire_handle(char const *name) {
LIBIPC_LOG();
if (!shm_.acquire(name, sizeof(sync_t))) {
ipc::error("[acquire_handle] fail shm.acquire: %s\n", name);
log.error("[acquire_handle] fail shm.acquire: ", name);
return nullptr;
}
return static_cast<sync_t *>(shm_.get());

27
src/libipc/platform/posix/condition.h
View File

@ -5,7 +5,7 @@
#include <pthread.h>
#include "libipc/utility/log.h"
#include "libipc/imp/log.h"
#include "libipc/utility/scope_guard.h"
#include "libipc/mutex.h"
#include "libipc/shm.h"
@ -21,8 +21,9 @@ class condition {
pthread_cond_t *cond_ = nullptr;
pthread_cond_t *acquire_cond(char const *name) {
LIBIPC_LOG();
if (!shm_.acquire(name, sizeof(pthread_cond_t))) {
ipc::error("[acquire_cond] fail shm.acquire: %s\n", name);
log.error("[acquire_cond] fail shm.acquire: ", name);
return nullptr;
}
return static_cast<pthread_cond_t *>(shm_.get());
@ -47,6 +48,7 @@ public:
}
bool open(char const *name) noexcept {
LIBIPC_LOG();
close();
if ((cond_ = acquire_cond(name)) == nullptr) {
return false;
@ -60,17 +62,17 @@ public:
int eno;
pthread_condattr_t cond_attr;
if ((eno = ::pthread_condattr_init(&cond_attr)) != 0) {
ipc::error("fail pthread_condattr_init[%d]\n", eno);
log.error("fail pthread_condattr_init[", eno, "]");
return false;
}
LIBIPC_UNUSED auto guard_cond_attr = guard([&cond_attr] { ::pthread_condattr_destroy(&cond_attr); });
if ((eno = ::pthread_condattr_setpshared(&cond_attr, PTHREAD_PROCESS_SHARED)) != 0) {
ipc::error("fail pthread_condattr_setpshared[%d]\n", eno);
log.error("fail pthread_condattr_setpshared[", 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);
log.error("fail pthread_cond_init[", eno, "]");
return false;
}
finally.dismiss();
@ -78,10 +80,11 @@ public:
}
void close() noexcept {
LIBIPC_LOG();
if ((shm_.ref() <= 1) && cond_ != nullptr) {
int eno;
if ((eno = ::pthread_cond_destroy(cond_)) != 0) {
ipc::error("fail pthread_cond_destroy[%d]\n", eno);
log.error("fail pthread_cond_destroy[", eno, "]");
}
}
shm_.release();
@ -92,7 +95,7 @@ public:
if ((shm_.ref() <= 1) && cond_ != nullptr) {
int eno;
if ((eno = ::pthread_cond_destroy(cond_)) != 0) {
ipc::error("fail pthread_cond_destroy[%d]\n", eno);
log.error("fail pthread_cond_destroy[", eno, "]");
}
}
shm_.clear(); // Make sure the storage is cleaned up.
@ -104,12 +107,13 @@ public:
}
bool wait(ipc::sync::mutex &mtx, std::uint64_t tm) noexcept {
LIBIPC_LOG();
if (!valid()) return false;
switch (tm) {
case invalid_value: {
int eno;
if ((eno = ::pthread_cond_wait(cond_, static_cast<pthread_mutex_t *>(mtx.native()))) != 0) {
ipc::error("fail pthread_cond_wait[%d]\n", eno);
log.error("fail pthread_cond_wait[", eno, "]");
return false;
}
}
@ -119,8 +123,7 @@ public:
int eno;
if ((eno = ::pthread_cond_timedwait(cond_, static_cast<pthread_mutex_t *>(mtx.native()), &ts)) != 0) {
if (eno != ETIMEDOUT) {
ipc::error("fail pthread_cond_timedwait[%d]: tm = %zd, tv_sec = %ld, tv_nsec = %ld\n",
eno, tm, ts.tv_sec, ts.tv_nsec);
log.error("fail pthread_cond_timedwait[", eno, "]: tm = ", tm, ", tv_sec = ", ts.tv_sec, ", tv_nsec = ", ts.tv_nsec);
}
return false;
}
@ -134,7 +137,7 @@ public:
if (!valid()) return false;
int eno;
if ((eno = ::pthread_cond_signal(cond_)) != 0) {
ipc::error("fail pthread_cond_signal[%d]\n", eno);
log.error("fail pthread_cond_signal[", eno, "]");
return false;
}
return true;
@ -144,7 +147,7 @@ public:
if (!valid()) return false;
int eno;
if ((eno = ::pthread_cond_broadcast(cond_)) != 0) {
ipc::error("fail pthread_cond_broadcast[%d]\n", eno);
log.error("fail pthread_cond_broadcast[", eno, "]");
return false;
}
return true;

9
src/libipc/platform/posix/get_wait_time.h
View File

@ -7,17 +7,18 @@
#include <time.h>
#include <errno.h>
#include "libipc/utility/log.h"
#include "libipc/imp/log.h"
namespace ipc {
namespace posix_ {
namespace detail {
inline bool calc_wait_time(timespec &ts, std::uint64_t tm /*ms*/) noexcept {
LIBIPC_LOG();
timeval now;
int eno = ::gettimeofday(&now, NULL);
if (eno != 0) {
ipc::error("fail gettimeofday [%d]\n", eno);
log.error("fail gettimeofday [", eno, "]");
return false;
}
ts.tv_nsec = (now.tv_usec + (tm % 1000) * 1000) * 1000;
@ -27,10 +28,10 @@ inline bool calc_wait_time(timespec &ts, std::uint64_t tm /*ms*/) noexcept {
}
inline timespec make_timespec(std::uint64_t tm /*ms*/) noexcept(false) {
LIBIPC_LOG();
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);
log.error("fail calc_wait_time: tm = ", tm, ", tv_sec = ", ts.tv_sec, ", tv_nsec = ", ts.tv_nsec);
throw std::system_error{static_cast<int>(errno), std::system_category()};
}
return ts;

28
src/libipc/platform/posix/mutex.h
View File

@ -10,7 +10,7 @@
#include <pthread.h>
#include "libipc/platform/detail.h"
#include "libipc/utility/log.h"
#include "libipc/imp/log.h"
#include "libipc/utility/scope_guard.h"
#include "libipc/mem/resource.h"
#include "libipc/shm.h"
@ -127,21 +127,21 @@ public:
int eno;
pthread_mutexattr_t mutex_attr;
if ((eno = ::pthread_mutexattr_init(&mutex_attr)) != 0) {
ipc::error("fail pthread_mutexattr_init[%d]\n", eno);
log.error("fail pthread_mutexattr_init[", eno, "]");
return false;
}
LIBIPC_UNUSED auto guard_mutex_attr = guard([&mutex_attr] { ::pthread_mutexattr_destroy(&mutex_attr); });
if ((eno = ::pthread_mutexattr_setpshared(&mutex_attr, PTHREAD_PROCESS_SHARED)) != 0) {
ipc::error("fail pthread_mutexattr_setpshared[%d]\n", eno);
log.error("fail pthread_mutexattr_setpshared[", eno, "]");
return false;
}
if ((eno = ::pthread_mutexattr_setrobust(&mutex_attr, PTHREAD_MUTEX_ROBUST)) != 0) {
ipc::error("fail pthread_mutexattr_setrobust[%d]\n", eno);
log.error("fail pthread_mutexattr_setrobust[", 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);
log.error("fail pthread_mutex_init[", eno, "]");
return false;
}
finally.dismiss();
@ -149,6 +149,7 @@ public:
}
void close() noexcept {
LIBIPC_LOG();
if ((ref_ != nullptr) && (shm_ != nullptr) && (mutex_ != nullptr)) {
if (shm_->name() != nullptr) {
release_mutex(shm_->name(), [this] {
@ -165,7 +166,7 @@ public:
int eno;
if ((eno = ::pthread_mutex_destroy(mutex_)) != 0) {
ipc::error("fail pthread_mutex_destroy[%d]\n", eno);
log.error("fail pthread_mutex_destroy[", eno, "]");
}
return true;
}
@ -179,6 +180,7 @@ public:
}
void clear() noexcept {
LIBIPC_LOG();
if ((shm_ != nullptr) && (mutex_ != nullptr)) {
if (shm_->name() != nullptr) {
release_mutex(shm_->name(), [this] {
@ -187,7 +189,7 @@ public:
int eno;
if ((eno = ::pthread_mutex_destroy(mutex_)) != 0) {
ipc::error("fail pthread_mutex_destroy[%d]\n", eno);
log.error("fail pthread_mutex_destroy[", eno, "]");
}
shm_->clear();
return true;
@ -206,6 +208,7 @@ public:
}
bool lock(std::uint64_t tm) noexcept {
LIBIPC_LOG();
if (!valid()) return false;
for (;;) {
auto ts = posix_::detail::make_timespec(tm);
@ -222,7 +225,7 @@ public:
// but the previous owner died. We need to make it consistent.
int eno2 = ::pthread_mutex_consistent(mutex_);
if (eno2 != 0) {
ipc::error("fail pthread_mutex_lock[%d], pthread_mutex_consistent[%d]\n", eno, eno2);
log.error("fail pthread_mutex_lock[", eno, "], pthread_mutex_consistent[", eno2, "]");
return false;
}
// After calling pthread_mutex_consistent(), the mutex is now in a
@ -230,7 +233,7 @@ public:
return true;
}
default:
ipc::error("fail pthread_mutex_lock[%d]\n", eno);
log.error("fail pthread_mutex_lock[", eno, "]");
return false;
}
}
@ -250,7 +253,7 @@ public:
// but the previous owner died. We need to make it consistent.
int eno2 = ::pthread_mutex_consistent(mutex_);
if (eno2 != 0) {
ipc::error("fail pthread_mutex_timedlock[%d], pthread_mutex_consistent[%d]\n", eno, eno2);
log.error("fail pthread_mutex_timedlock[", eno, "], pthread_mutex_consistent[", eno2, "]");
throw std::system_error{eno2, std::system_category()};
}
// After calling pthread_mutex_consistent(), the mutex is now in a
@ -258,17 +261,18 @@ public:
return true;
}
default:
ipc::error("fail pthread_mutex_timedlock[%d]\n", eno);
log.error("fail pthread_mutex_timedlock[", eno, "]");
break;
}
throw std::system_error{eno, std::system_category()};
}
bool unlock() noexcept {
LIBIPC_LOG();
if (!valid()) return false;
int eno;
if ((eno = ::pthread_mutex_unlock(mutex_)) != 0) {
ipc::error("fail pthread_mutex_unlock[%d]\n", eno);
log.error("fail pthread_mutex_unlock[", eno, "]");
return false;
}
return true;

24
src/libipc/platform/posix/semaphore_impl.h
View File

@ -7,7 +7,7 @@
#include <semaphore.h>
#include <errno.h>
#include "libipc/utility/log.h"
#include "libipc/imp/log.h"
#include "libipc/shm.h"
#include "get_wait_time.h"
@ -34,9 +34,10 @@ public:
}
bool open(char const *name, std::uint32_t count) noexcept {
LIBIPC_LOG();
close();
if (!shm_.acquire(name, 1)) {
ipc::error("[open_semaphore] fail shm.acquire: %s\n", name);
log.error("[open_semaphore] fail shm.acquire: ", name);
return false;
}
// POSIX semaphore names must start with "/" on some platforms (e.g., FreeBSD)
@ -48,22 +49,23 @@ public:
}
h_ = ::sem_open(sem_name_.c_str(), O_CREAT, 0666, static_cast<unsigned>(count));
if (h_ == SEM_FAILED) {
ipc::error("fail sem_open[%d]: %s\n", errno, sem_name_.c_str());
log.error("fail sem_open[", errno, "]: ", sem_name_);
return false;
}
return true;
}
void close() noexcept {
LIBIPC_LOG();
if (!valid()) return;
if (::sem_close(h_) != 0) {
ipc::error("fail sem_close[%d]: %s\n", errno);
log.error("fail sem_close[", errno, "]");
}
h_ = SEM_FAILED;
if (!sem_name_.empty() && shm_.name() != nullptr) {
if (shm_.release() <= 1) {
if (::sem_unlink(sem_name_.c_str()) != 0) {
ipc::error("fail sem_unlink[%d]: %s, name: %s\n", errno, sem_name_.c_str());
log.error("fail sem_unlink[", errno, "]: ", sem_name_);
}
}
}
@ -71,9 +73,10 @@ public:
}
void clear() noexcept {
LIBIPC_LOG();
if (valid()) {
if (::sem_close(h_) != 0) {
ipc::error("fail sem_close[%d]: %s\n", errno);
log.error("fail sem_close[", errno, "]");
}
h_ = SEM_FAILED;
}
@ -97,18 +100,18 @@ public:
}
bool wait(std::uint64_t tm) noexcept {
LIBIPC_LOG();
if (!valid()) return false;
if (tm == invalid_value) {
if (::sem_wait(h_) != 0) {
ipc::error("fail sem_wait[%d]: %s\n", errno);
log.error("fail sem_wait[", errno, "]");
return false;
}
} else {
auto ts = posix_::detail::make_timespec(tm);
if (::sem_timedwait(h_, &ts) != 0) {
if (errno != ETIMEDOUT) {
ipc::error("fail sem_timedwait[%d]: tm = %zd, tv_sec = %ld, tv_nsec = %ld\n",
errno, tm, ts.tv_sec, ts.tv_nsec);
log.error("fail sem_timedwait[", errno, "]: tm = ", tm, ", tv_sec = ", ts.tv_sec, ", tv_nsec = ", ts.tv_nsec);
}
return false;
}
@ -117,10 +120,11 @@ public:
}
bool post(std::uint32_t count) noexcept {
LIBIPC_LOG();
if (!valid()) return false;
for (std::uint32_t i = 0; i < count; ++i) {
if (::sem_post(h_) != 0) {
ipc::error("fail sem_post[%d]: %s\n", errno);
log.error("fail sem_post[", errno, "]");
return false;
}
}

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

@ -14,7 +14,7 @@
#include "libipc/shm.h"
#include "libipc/def.h"
#include "libipc/utility/log.h"
#include "libipc/imp/log.h"
#include "libipc/mem/resource.h"
#include "libipc/mem/new.h"
@ -45,8 +45,9 @@ namespace ipc {
namespace shm {
id_t acquire(char const * name, std::size_t size, unsigned mode) {
LIBIPC_LOG();
if (!is_valid_string(name)) {
ipc::error("fail acquire: name is empty\n");
log.error("fail acquire: name is empty");
return nullptr;
}
// For portable use, a shared memory object should be identified by name of the form /somename.
@ -79,7 +80,7 @@ id_t acquire(char const * name, std::size_t size, unsigned mode) {
if (fd == -1) {
// only open shm not log error when file not exist
if (open != mode || ENOENT != errno) {
ipc::error("fail shm_open[%d]: %s\n", errno, op_name.c_str());
log.error("fail shm_open[", errno, "]: ", op_name);
}
return nullptr;
}
@ -105,21 +106,23 @@ std::int32_t get_ref(id_t id) {
}
void sub_ref(id_t id) {
LIBIPC_LOG();
if (id == nullptr) {
ipc::error("fail sub_ref: invalid id (null)\n");
log.error("fail sub_ref: invalid id (null)");
return;
}
auto ii = static_cast<id_info_t*>(id);
if (ii->mem_ == nullptr || ii->size_ == 0) {
ipc::error("fail sub_ref: invalid id (mem = %p, size = %zd)\n", ii->mem_, ii->size_);
log.error("fail sub_ref: invalid id (mem = ", ii->mem_, ", size = ", 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) {
LIBIPC_LOG();
if (id == nullptr) {
ipc::error("fail get_mem: invalid id (null)\n");
log.error("fail get_mem: invalid id (null)");
return nullptr;
}
auto ii = static_cast<id_info_t*>(id);
@ -129,31 +132,31 @@ void * get_mem(id_t id, std::size_t * size) {
}
int fd = ii->fd_;
if (fd == -1) {
ipc::error("fail get_mem: invalid id (fd = -1)\n");
log.error("fail get_mem: invalid id (fd = -1)");
return nullptr;
}
if (ii->size_ == 0) {
struct stat st;
if (::fstat(fd, &st) != 0) {
ipc::error("fail fstat[%d]: %s, size = %zd\n", errno, ii->name_.c_str(), ii->size_);
log.error("fail fstat[", errno, "]: ", ii->name_, ", size = ", ii->size_);
return nullptr;
}
ii->size_ = static_cast<std::size_t>(st.st_size);
if ((ii->size_ <= sizeof(info_t)) || (ii->size_ % sizeof(info_t))) {
ipc::error("fail get_mem: %s, invalid size = %zd\n", ii->name_.c_str(), ii->size_);
log.error("fail get_mem: ", ii->name_, ", invalid size = ", ii->size_);
return nullptr;
}
}
else {
ii->size_ = calc_size(ii->size_);
if (::ftruncate(fd, static_cast<off_t>(ii->size_)) != 0) {
ipc::error("fail ftruncate[%d]: %s, size = %zd\n", errno, ii->name_.c_str(), ii->size_);
log.error("fail ftruncate[", errno, "]: ", ii->name_, ", size = ", ii->size_);
return nullptr;
}
}
void* mem = ::mmap(nullptr, ii->size_, PROT_READ | PROT_WRITE, MAP_SHARED, fd, 0);
if (mem == MAP_FAILED) {
ipc::error("fail mmap[%d]: %s, size = %zd\n", errno, ii->name_.c_str(), ii->size_);
log.error("fail mmap[", errno, "]: ", ii->name_, ", size = ", ii->size_);
return nullptr;
}
::close(fd);
@ -165,22 +168,22 @@ void * get_mem(id_t id, std::size_t * size) {
}
std::int32_t release(id_t id) noexcept {
LIBIPC_LOG();
if (id == nullptr) {
ipc::error("fail release: invalid id (null)\n");
log.error("fail release: invalid id (null)");
return -1;
}
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), name = %s\n",
ii->mem_, ii->size_, ii->name_.c_str());
log.error("fail release: invalid id (mem = ", ii->mem_, ", size = ", ii->size_, "), name = ", ii->name_);
}
else if ((ret = acc_of(ii->mem_, ii->size_).fetch_sub(1, std::memory_order_acq_rel)) <= 1) {
::munmap(ii->mem_, ii->size_);
if (!ii->name_.empty()) {
int unlink_ret = ::shm_unlink(ii->name_.c_str());
if (unlink_ret == -1) {
ipc::error("fail shm_unlink[%d]: %s\n", errno, ii->name_.c_str());
log.error("fail shm_unlink[", errno, "]: ", ii->name_);
}
}
}
@ -190,8 +193,9 @@ std::int32_t release(id_t id) noexcept {
}
void remove(id_t id) noexcept {
LIBIPC_LOG();
if (id == nullptr) {
ipc::error("fail remove: invalid id (null)\n");
log.error("fail remove: invalid id (null)");
return;
}
auto ii = static_cast<id_info_t*>(id);
@ -200,14 +204,15 @@ void remove(id_t id) noexcept {
if (!name.empty()) {
int unlink_ret = ::shm_unlink(name.c_str());
if (unlink_ret == -1) {
ipc::error("fail shm_unlink[%d]: %s\n", errno, name.c_str());
log.error("fail shm_unlink[", errno, "]: ", name);
}
}
}
void remove(char const * name) noexcept {
LIBIPC_LOG();
if (!is_valid_string(name)) {
ipc::error("fail remove: name is empty\n");
log.error("fail remove: name is empty");
return;
}
// For portable use, a shared memory object should be identified by name of the form /somename.
@ -219,7 +224,7 @@ void remove(char const * name) noexcept {
}
int unlink_ret = ::shm_unlink(op_name.c_str());
if (unlink_ret == -1) {
ipc::error("fail shm_unlink[%d]: %s\n", errno, op_name.c_str());
log.error("fail shm_unlink[", errno, "]: ", op_name);
}
}

2
src/libipc/platform/win/condition.h
View File

@ -10,7 +10,7 @@
#include <Windows.h>
#endif
#include "libipc/utility/log.h"
#include "libipc/imp/log.h"
#include "libipc/utility/scope_guard.h"
#include "libipc/platform/detail.h"
#include "libipc/mutex.h"

44
src/libipc/platform/win/get_sa.h
View File

@ -5,29 +5,31 @@
namespace ipc {
namespace detail {
inline LPSECURITY_ATTRIBUTES get_sa() {
static struct initiator {
struct sa_initiator {
SECURITY_DESCRIPTOR sd_;
SECURITY_ATTRIBUTES sa_;
bool succ_ = false;
SECURITY_DESCRIPTOR sd_;
SECURITY_ATTRIBUTES sa_;
bool succ_ = false;
initiator() {
if (!::InitializeSecurityDescriptor(&sd_, SECURITY_DESCRIPTOR_REVISION)) {
ipc::error("fail InitializeSecurityDescriptor[%d]\n", static_cast<int>(::GetLastError()));
return;
}
if (!::SetSecurityDescriptorDacl(&sd_, TRUE, NULL, FALSE)) {
ipc::error("fail SetSecurityDescriptorDacl[%d]\n", static_cast<int>(::GetLastError()));
return;
}
sa_.nLength = sizeof(SECURITY_ATTRIBUTES);
sa_.bInheritHandle = FALSE;
sa_.lpSecurityDescriptor = &sd_;
succ_ = true;
template <typename Logger>
sa_initiator(Logger const &log) {
if (!::InitializeSecurityDescriptor(&sd_, SECURITY_DESCRIPTOR_REVISION)) {
log.error("fail InitializeSecurityDescriptor[", static_cast<int>(::GetLastError()), "]");
return;
}
} handle;
if (!::SetSecurityDescriptorDacl(&sd_, TRUE, NULL, FALSE)) {
log.error("fail SetSecurityDescriptorDacl[", static_cast<int>(::GetLastError()), "]");
return;
}
sa_.nLength = sizeof(SECURITY_ATTRIBUTES);
sa_.bInheritHandle = FALSE;
sa_.lpSecurityDescriptor = &sd_;
succ_ = true;
}
};
inline LPSECURITY_ATTRIBUTES get_sa() {
LIBIPC_LOG();
static sa_initiator handle(log);
return handle.succ_ ? &handle.sa_ : nullptr;
}

16
src/libipc/platform/win/mutex.h
View File

@ -9,7 +9,7 @@
#include <Windows.h>
#endif
#include "libipc/utility/log.h"
#include "libipc/imp/log.h"
#include "to_tchar.h"
#include "get_sa.h"
@ -36,10 +36,11 @@ public:
}
bool open(char const *name) noexcept {
LIBIPC_LOG();
close();
h_ = ::CreateMutex(detail::get_sa(), FALSE, detail::to_tchar(name).c_str());
if (h_ == NULL) {
ipc::error("fail CreateMutex[%lu]: %s\n", ::GetLastError(), name);
log.error("fail CreateMutex[", static_cast<unsigned long>(::GetLastError()), "]: ", name);
return false;
}
return true;
@ -59,6 +60,7 @@ public:
}
bool lock(std::uint64_t tm) noexcept {
LIBIPC_LOG();
DWORD ret, ms = (tm == invalid_value) ? INFINITE : static_cast<DWORD>(tm);
for(;;) {
switch ((ret = ::WaitForSingleObject(h_, ms))) {
@ -67,19 +69,20 @@ public:
case WAIT_TIMEOUT:
return false;
case WAIT_ABANDONED:
ipc::log("fail WaitForSingleObject[%lu]: WAIT_ABANDONED, try again.\n", ::GetLastError());
log.warning("fail WaitForSingleObject[", ::GetLastError(), "]: WAIT_ABANDONED, try again.");
if (!unlock()) {
return false;
}
break; // loop again
default:
ipc::error("fail WaitForSingleObject[%lu]: 0x%08X\n", ::GetLastError(), ret);
log.error("fail WaitForSingleObject[", ::GetLastError(), "]: 0x", std::hex, ret, std::dec);
return false;
}
}
}
bool try_lock() noexcept(false) {
LIBIPC_LOG();
DWORD ret = ::WaitForSingleObject(h_, 0);
switch (ret) {
case WAIT_OBJECT_0:
@ -90,14 +93,15 @@ public:
unlock();
LIBIPC_FALLTHROUGH;
default:
ipc::error("fail WaitForSingleObject[%lu]: 0x%08X\n", ::GetLastError(), ret);
log.error("fail WaitForSingleObject[", ::GetLastError(), "]: 0x", std::hex, ret, std::dec);
throw std::system_error{static_cast<int>(ret), std::system_category()};
}
}
bool unlock() noexcept {
LIBIPC_LOG();
if (!::ReleaseMutex(h_)) {
ipc::error("fail ReleaseMutex[%lu]\n", ::GetLastError());
log.error("fail ReleaseMutex[", ::GetLastError(), "]");
return false;
}
return true;

11
src/libipc/platform/win/semaphore.h
View File

@ -8,7 +8,7 @@
#include <Windows.h>
#endif
#include "libipc/utility/log.h"
#include "libipc/imp/log.h"
#include "to_tchar.h"
#include "get_sa.h"
@ -33,12 +33,13 @@ public:
}
bool open(char const *name, std::uint32_t count) noexcept {
LIBIPC_LOG();
close();
h_ = ::CreateSemaphore(detail::get_sa(),
static_cast<LONG>(count), LONG_MAX,
detail::to_tchar(name).c_str());
if (h_ == NULL) {
ipc::error("fail CreateSemaphore[%lu]: %s\n", ::GetLastError(), name);
log.error("fail CreateSemaphore[", static_cast<unsigned long>(::GetLastError()), "]: ", name);
return false;
}
return true;
@ -58,6 +59,7 @@ public:
}
bool wait(std::uint64_t tm) noexcept {
LIBIPC_LOG();
DWORD ret, ms = (tm == invalid_value) ? INFINITE : static_cast<DWORD>(tm);
switch ((ret = ::WaitForSingleObject(h_, ms))) {
case WAIT_OBJECT_0:
@ -66,14 +68,15 @@ public:
return false;
case WAIT_ABANDONED:
default:
ipc::error("fail WaitForSingleObject[%lu]: 0x%08X\n", ::GetLastError(), ret);
log.error("fail WaitForSingleObject[", ::GetLastError(), "]: 0x", std::hex, ret, std::dec);
return false;
}
}
bool post(std::uint32_t count) noexcept {
LIBIPC_LOG();
if (!::ReleaseSemaphore(h_, static_cast<LONG>(count), NULL)) {
ipc::error("fail ReleaseSemaphore[%lu]\n", ::GetLastError());
log.error("fail ReleaseSemaphore[", ::GetLastError(), "]");
return false;
}
return true;

378
src/libipc/platform/win/shm_win.cpp
View File

@ -1,186 +1,192 @@
#if defined(__MINGW32__)
#include <windows.h>
#else
#include <Windows.h>
#endif
#include <atomic>
#include <string>
#include <utility>
#include "libipc/shm.h"
#include "libipc/def.h"
#include "libipc/utility/log.h"
#include "libipc/mem/resource.h"
#include "libipc/mem/new.h"
#include "to_tchar.h"
#include "get_sa.h"
namespace {
struct info_t {
std::atomic<std::int32_t> acc_;
};
struct id_info_t {
HANDLE h_ = NULL;
void* mem_ = nullptr;
std::size_t size_ = 0;
};
constexpr std::size_t calc_size(std::size_t size) {
return ((((size - 1) / alignof(info_t)) + 1) * alignof(info_t)) + sizeof(info_t);
}
inline auto& acc_of(void* mem, std::size_t size) {
return reinterpret_cast<info_t*>(static_cast<ipc::byte_t*>(mem) + size - sizeof(info_t))->acc_;
}
} // internal-linkage
namespace ipc {
namespace shm {
id_t acquire(char const * name, std::size_t size, unsigned mode) {
if (!is_valid_string(name)) {
ipc::error("fail acquire: name is empty\n");
return nullptr;
}
HANDLE h;
auto fmt_name = ipc::detail::to_tchar(name);
// Opens a named file mapping object.
if (mode == open) {
h = ::OpenFileMapping(FILE_MAP_ALL_ACCESS, FALSE, fmt_name.c_str());
if (h == NULL) {
ipc::error("fail OpenFileMapping[%d]: %s\n", static_cast<int>(::GetLastError()), name);
return nullptr;
}
}
// Creates or opens a named file mapping object for a specified file.
else {
std::size_t alloc_size = calc_size(size);
h = ::CreateFileMapping(INVALID_HANDLE_VALUE, detail::get_sa(), PAGE_READWRITE | SEC_COMMIT,
0, static_cast<DWORD>(alloc_size), fmt_name.c_str());
DWORD err = ::GetLastError();
// If the object exists before the function call, the function returns a handle to the existing object
// (with its current size, not the specified size), and GetLastError returns ERROR_ALREADY_EXISTS.
if ((mode == create) && (err == ERROR_ALREADY_EXISTS)) {
if (h != NULL) ::CloseHandle(h);
h = NULL;
}
if (h == NULL) {
ipc::error("fail CreateFileMapping[%d]: %s\n", static_cast<int>(err), name);
return nullptr;
}
}
auto ii = mem::$new<id_info_t>();
ii->h_ = h;
ii->size_ = size;
return ii;
}
std::int32_t get_ref(id_t id) {
if (id == nullptr) {
return 0;
}
auto ii = static_cast<id_info_t*>(id);
if (ii->mem_ == nullptr || ii->size_ == 0) {
return 0;
}
return acc_of(ii->mem_, calc_size(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_info_t*>(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_, calc_size(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");
return nullptr;
}
auto ii = static_cast<id_info_t*>(id);
if (ii->mem_ != nullptr) {
if (size != nullptr) *size = ii->size_;
return ii->mem_;
}
if (ii->h_ == NULL) {
ipc::error("fail to_mem: invalid id (h = null)\n");
return nullptr;
}
LPVOID mem = ::MapViewOfFile(ii->h_, FILE_MAP_ALL_ACCESS, 0, 0, 0);
if (mem == NULL) {
ipc::error("fail MapViewOfFile[%d]\n", static_cast<int>(::GetLastError()));
return nullptr;
}
MEMORY_BASIC_INFORMATION mem_info;
if (::VirtualQuery(mem, &mem_info, sizeof(mem_info)) == 0) {
ipc::error("fail VirtualQuery[%d]\n", static_cast<int>(::GetLastError()));
return nullptr;
}
std::size_t actual_size = static_cast<std::size_t>(mem_info.RegionSize);
if (ii->size_ == 0) {
// Opening existing shared memory
ii->size_ = actual_size - sizeof(info_t);
}
// else: Keep user-requested size (already set in acquire)
ii->mem_ = mem;
if (size != nullptr) *size = ii->size_;
// Initialize or increment reference counter
acc_of(mem, calc_size(ii->size_)).fetch_add(1, std::memory_order_release);
return static_cast<void *>(mem);
}
std::int32_t release(id_t id) noexcept {
if (id == nullptr) {
ipc::error("fail release: invalid id (null)\n");
return -1;
}
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_);
}
else {
ret = acc_of(ii->mem_, calc_size(ii->size_)).fetch_sub(1, std::memory_order_acq_rel);
::UnmapViewOfFile(static_cast<LPCVOID>(ii->mem_));
}
if (ii->h_ == NULL) {
ipc::error("fail release: invalid id (h = null)\n");
}
else ::CloseHandle(ii->h_);
mem::$delete(ii);
return ret;
}
void remove(id_t id) noexcept {
if (id == nullptr) {
ipc::error("fail release: invalid id (null)\n");
return;
}
release(id);
}
void remove(char const * name) noexcept {
if (!is_valid_string(name)) {
ipc::error("fail remove: name is empty\n");
return;
}
// Do Nothing.
}
} // namespace shm
} // namespace ipc
#if defined(__MINGW32__)
#include <windows.h>
#else
#include <Windows.h>
#endif
#include <atomic>
#include <string>
#include <utility>
#include "libipc/shm.h"
#include "libipc/def.h"
#include "libipc/imp/log.h"
#include "libipc/mem/resource.h"
#include "libipc/mem/new.h"
#include "to_tchar.h"
#include "get_sa.h"
namespace {
struct info_t {
std::atomic<std::int32_t> acc_;
};
struct id_info_t {
HANDLE h_ = NULL;
void* mem_ = nullptr;
std::size_t size_ = 0;
};
constexpr std::size_t calc_size(std::size_t size) {
return ((((size - 1) / alignof(info_t)) + 1) * alignof(info_t)) + sizeof(info_t);
}
inline auto& acc_of(void* mem, std::size_t size) {
return reinterpret_cast<info_t*>(static_cast<ipc::byte_t*>(mem) + size - sizeof(info_t))->acc_;
}
} // internal-linkage
namespace ipc {
namespace shm {
id_t acquire(char const * name, std::size_t size, unsigned mode) {
LIBIPC_LOG();
if (!is_valid_string(name)) {
log.error("fail acquire: name is empty");
return nullptr;
}
HANDLE h;
auto fmt_name = ipc::detail::to_tchar(name);
// Opens a named file mapping object.
if (mode == open) {
h = ::OpenFileMapping(FILE_MAP_ALL_ACCESS, FALSE, fmt_name.c_str());
if (h == NULL) {
log.error("fail OpenFileMapping[", static_cast<int>(::GetLastError()), "]: ", name);
return nullptr;
}
}
// Creates or opens a named file mapping object for a specified file.
else {
std::size_t alloc_size = calc_size(size);
h = ::CreateFileMapping(INVALID_HANDLE_VALUE, detail::get_sa(), PAGE_READWRITE | SEC_COMMIT,
0, static_cast<DWORD>(alloc_size), fmt_name.c_str());
DWORD err = ::GetLastError();
// If the object exists before the function call, the function returns a handle to the existing object
// (with its current size, not the specified size), and GetLastError returns ERROR_ALREADY_EXISTS.
if ((mode == create) && (err == ERROR_ALREADY_EXISTS)) {
if (h != NULL) ::CloseHandle(h);
h = NULL;
}
if (h == NULL) {
log.error("fail CreateFileMapping[", static_cast<int>(err), "]: ", name);
return nullptr;
}
}
auto ii = mem::$new<id_info_t>();
ii->h_ = h;
ii->size_ = size;
return ii;
}
std::int32_t get_ref(id_t id) {
if (id == nullptr) {
return 0;
}
auto ii = static_cast<id_info_t*>(id);
if (ii->mem_ == nullptr || ii->size_ == 0) {
return 0;
}
return acc_of(ii->mem_, calc_size(ii->size_)).load(std::memory_order_acquire);
}
void sub_ref(id_t id) {
LIBIPC_LOG();
if (id == nullptr) {
log.error("fail sub_ref: invalid id (null)");
return;
}
auto ii = static_cast<id_info_t*>(id);
if (ii->mem_ == nullptr || ii->size_ == 0) {
log.error("fail sub_ref: invalid id (mem = ", ii->mem_, ", size = ", ii->size_, ")");
return;
}
acc_of(ii->mem_, calc_size(ii->size_)).fetch_sub(1, std::memory_order_acq_rel);
}
void * get_mem(id_t id, std::size_t * size) {
LIBIPC_LOG();
if (id == nullptr) {
log.error("fail get_mem: invalid id (null)");
return nullptr;
}
auto ii = static_cast<id_info_t*>(id);
if (ii->mem_ != nullptr) {
if (size != nullptr) *size = ii->size_;
return ii->mem_;
}
if (ii->h_ == NULL) {
log.error("fail to_mem: invalid id (h = null)");
return nullptr;
}
LPVOID mem = ::MapViewOfFile(ii->h_, FILE_MAP_ALL_ACCESS, 0, 0, 0);
if (mem == NULL) {
log.error("fail MapViewOfFile[", static_cast<int>(::GetLastError()), "]");
return nullptr;
}
MEMORY_BASIC_INFORMATION mem_info;
if (::VirtualQuery(mem, &mem_info, sizeof(mem_info)) == 0) {
log.error("fail VirtualQuery[", static_cast<int>(::GetLastError()), "]");
return nullptr;
}
std::size_t actual_size = static_cast<std::size_t>(mem_info.RegionSize);
if (ii->size_ == 0) {
// Opening existing shared memory
ii->size_ = actual_size - sizeof(info_t);
}
// else: Keep user-requested size (already set in acquire)
ii->mem_ = mem;
if (size != nullptr) *size = ii->size_;
// Initialize or increment reference counter
acc_of(mem, calc_size(ii->size_)).fetch_add(1, std::memory_order_release);
return static_cast<void *>(mem);
}
std::int32_t release(id_t id) noexcept {
LIBIPC_LOG();
if (id == nullptr) {
log.error("fail release: invalid id (null)");
return -1;
}
std::int32_t ret = -1;
auto ii = static_cast<id_info_t*>(id);
if (ii->mem_ == nullptr || ii->size_ == 0) {
log.error("fail release: invalid id (mem = ", ii->mem_, ", size = ", ii->size_, ")");
}
else {
ret = acc_of(ii->mem_, calc_size(ii->size_)).fetch_sub(1, std::memory_order_acq_rel);
::UnmapViewOfFile(static_cast<LPCVOID>(ii->mem_));
}
if (ii->h_ == NULL) {
log.error("fail release: invalid id (h = null)");
}
else ::CloseHandle(ii->h_);
mem::$delete(ii);
return ret;
}
void remove(id_t id) noexcept {
LIBIPC_LOG();
if (id == nullptr) {
log.error("fail release: invalid id (null)");
return;
}
release(id);
}
void remove(char const * name) noexcept {
LIBIPC_LOG();
if (!is_valid_string(name)) {
log.error("fail remove: name is empty");
return;
}
// Do Nothing.
}
} // namespace shm
} // namespace ipc

8
src/libipc/prod_cons.h
View File

@ -10,7 +10,7 @@
#include "libipc/platform/detail.h"
#include "libipc/circ/elem_def.h"
#include "libipc/utility/log.h"
#include "libipc/imp/log.h"
#include "libipc/utility/utility.h"
namespace ipc {
@ -242,6 +242,7 @@ struct prod_cons_impl<wr<relat::single, relat::multi, trans::broadcast>> {
template <typename W, typename F, typename E>
bool force_push(W* wrapper, F&& f, E* elems) {
LIBIPC_LOG();
E* el;
epoch_ += ep_incr;
for (unsigned k = 0;;) {
@ -252,7 +253,7 @@ struct prod_cons_impl<wr<relat::single, relat::multi, trans::broadcast>> {
auto cur_rc = el->rc_.load(std::memory_order_acquire);
circ::cc_t rem_cc = cur_rc & ep_mask;
if (cc & rem_cc) {
ipc::log("force_push: k = %u, cc = %u, rem_cc = %u\n", k, cc, rem_cc);
log.debug("force_push: k = ", k, ", cc = ", cc, ", rem_cc = ", rem_cc);
cc = wrapper->elems()->disconnect_receiver(rem_cc); // disconnect all invalid readers
if (cc == 0) return false; // no reader
}
@ -364,6 +365,7 @@ struct prod_cons_impl<wr<relat::multi, relat::multi, trans::broadcast>> {
template <typename W, typename F, typename E>
bool force_push(W* wrapper, F&& f, E* elems) {
LIBIPC_LOG();
E* el;
circ::u2_t cur_ct;
rc_t epoch = epoch_.fetch_add(ep_incr, std::memory_order_release) + ep_incr;
@ -375,7 +377,7 @@ struct prod_cons_impl<wr<relat::multi, relat::multi, trans::broadcast>> {
auto cur_rc = el->rc_.load(std::memory_order_acquire);
circ::cc_t rem_cc = cur_rc & rc_mask;
if (cc & rem_cc) {
ipc::log("force_push: k = %u, cc = %u, rem_cc = %u\n", k, cc, rem_cc);
log.debug("force_push: k = ", k, ", cc = ", cc, ", rem_cc = ", rem_cc);
cc = wrapper->elems()->disconnect_receiver(rem_cc); // disconnect all invalid readers
if (cc == 0) return false; // no reader
}

7
src/libipc/queue.h
View File

@ -15,7 +15,7 @@
#include "libipc/shm.h"
#include "libipc/rw_lock.h"
#include "libipc/utility/log.h"
#include "libipc/imp/log.h"
#include "libipc/platform/detail.h"
#include "libipc/circ/elem_def.h"
#include "libipc/mem/resource.h"
@ -30,8 +30,9 @@ protected:
template <typename Elems>
Elems* open(char const * name) {
LIBIPC_LOG();
if (!is_valid_string(name)) {
ipc::error("fail open waiter: name is empty!\n");
log.error("fail open waiter: name is empty!");
return nullptr;
}
if (!elems_h_.acquire(name, sizeof(Elems))) {
@ -39,7 +40,7 @@ protected:
}
auto elems = static_cast<Elems*>(elems_h_.get());
if (elems == nullptr) {
ipc::error("fail acquire elems: %s\n", name);
log.error("fail acquire elems: ", name);
return nullptr;
}
elems->init();

7
src/libipc/shm.cpp
View File

@ -5,7 +5,7 @@
#include "libipc/shm.h"
#include "libipc/utility/pimpl.h"
#include "libipc/utility/log.h"
#include "libipc/imp/log.h"
#include "libipc/mem/resource.h"
namespace ipc {
@ -69,12 +69,13 @@ void handle::sub_ref() noexcept {
}
bool handle::acquire(char const * name, std::size_t size, unsigned mode) {
LIBIPC_LOG();
if (!is_valid_string(name)) {
ipc::error("fail acquire: name is empty\n");
log.error("fail acquire: name is empty");
return false;
}
if (size == 0) {
ipc::error("fail acquire: size is 0\n");
log.error("fail acquire: size is 0");
return false;
}
release();

5
src/libipc/sync/condition.cpp
View File

@ -2,7 +2,7 @@
#include "libipc/condition.h"
#include "libipc/utility/pimpl.h"
#include "libipc/utility/log.h"
#include "libipc/imp/log.h"
#include "libipc/mem/resource.h"
#include "libipc/platform/detail.h"
#if defined(LIBIPC_OS_WIN)
@ -50,8 +50,9 @@ bool condition::valid() const noexcept {
}
bool condition::open(char const *name) noexcept {
LIBIPC_LOG();
if (!is_valid_string(name)) {
ipc::error("fail condition open: name is empty\n");
log.error("fail condition open: name is empty");
return false;
}
return impl(p_)->cond_.open(name);

5
src/libipc/sync/mutex.cpp
View File

@ -2,7 +2,7 @@
#include "libipc/mutex.h"
#include "libipc/utility/pimpl.h"
#include "libipc/utility/log.h"
#include "libipc/imp/log.h"
#include "libipc/mem/resource.h"
#include "libipc/platform/detail.h"
#if defined(LIBIPC_OS_WIN)
@ -50,8 +50,9 @@ bool mutex::valid() const noexcept {
}
bool mutex::open(char const *name) noexcept {
LIBIPC_LOG();
if (!is_valid_string(name)) {
ipc::error("fail mutex open: name is empty\n");
log.error("fail mutex open: name is empty");
return false;
}
return impl(p_)->lock_.open(name);

5
src/libipc/sync/semaphore.cpp
View File

@ -2,7 +2,7 @@
#include "libipc/semaphore.h"
#include "libipc/utility/pimpl.h"
#include "libipc/utility/log.h"
#include "libipc/imp/log.h"
#include "libipc/mem/resource.h"
#include "libipc/platform/detail.h"
#if defined(LIBIPC_OS_WIN)
@ -49,8 +49,9 @@ bool semaphore::valid() const noexcept {
}
bool semaphore::open(char const *name, std::uint32_t count) noexcept {
LIBIPC_LOG();
if (!is_valid_string(name)) {
ipc::error("fail semaphore open: name is empty\n");
log.error("fail semaphore open: name is empty");
return false;
}
return impl(p_)->sem_.open(name, count);