Skip to content

Commit

Permalink
reconnect cannot reconnect when you are out of authority
Browse files Browse the repository at this point in the history
  • Loading branch information
mutouyun committed Oct 21, 2023
1 parent fb52c81 commit 76d9b41
Show file tree
Hide file tree
Showing 4 changed files with 78 additions and 36 deletions.
17 changes: 15 additions & 2 deletions demo/win_service/client/main.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -11,14 +11,27 @@ int _tmain (int argc, TCHAR *argv[]) {
ipc::channel ipc_r{ipc::prefix{"Global\\"}, "service ipc r", ipc::receiver};
ipc::channel ipc_w{ipc::prefix{"Global\\"}, "service ipc w", ipc::sender};
while (1) {
if (!ipc_r.reconnect(ipc::receiver)) {
Sleep(1000);
continue;
}
auto msg = ipc_r.recv();
if (msg.empty()) {
_tprintf(_T("My Sample Client: message recv error\n"));
return -1;
ipc_r.disconnect();
continue;
}
printf("My Sample Client: message recv: [%s]\n", (char const *)msg.data());
while (!ipc_w.send("Copy.")) {
for (;;) {
if (!ipc_w.reconnect(ipc::sender)) {
Sleep(1000);
continue;
}
if (ipc_w.send("Copy.")) {
break;
}
_tprintf(_T("My Sample Client: message send error\n"));
ipc_w.disconnect();
Sleep(1000);
}
_tprintf(_T("My Sample Client: message send [Copy]\n"));
Expand Down
72 changes: 45 additions & 27 deletions src/libipc/ipc.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -112,16 +112,27 @@ struct conn_info_head {
ipc::shm::handle acc_h_;

conn_info_head(char const * prefix, char const * name)
: prefix_ {ipc::make_string(prefix)}
, name_ {ipc::make_string(name)}
, cc_id_ {}
, cc_waiter_{ipc::make_prefix(prefix_, {"CC_CONN__", name_}).c_str()}
, wt_waiter_{ipc::make_prefix(prefix_, {"WT_CONN__", name_}).c_str()}
, rd_waiter_{ipc::make_prefix(prefix_, {"RD_CONN__", name_}).c_str()}
, acc_h_ {ipc::make_prefix(prefix_, {"AC_CONN__", name_}).c_str(), sizeof(acc_t)} {
: prefix_{ipc::make_string(prefix)}
, name_ {ipc::make_string(name)}
, cc_id_ {} {}

void init() {
if (!cc_waiter_.valid()) cc_waiter_.open(ipc::make_prefix(prefix_, {"CC_CONN__", name_}).c_str());
if (!wt_waiter_.valid()) wt_waiter_.open(ipc::make_prefix(prefix_, {"WT_CONN__", name_}).c_str());
if (!rd_waiter_.valid()) rd_waiter_.open(ipc::make_prefix(prefix_, {"RD_CONN__", name_}).c_str());
if (!acc_h_.valid()) acc_h_.acquire(ipc::make_prefix(prefix_, {"AC_CONN__", name_}).c_str(), sizeof(acc_t));
if (cc_id_ != 0) {
return;
}
acc_t *pacc = cc_acc(prefix_);
if (pacc != nullptr) {
cc_id_ = pacc->fetch_add(1, std::memory_order_relaxed);
if (pacc == nullptr) {
// Failed to obtain the global accumulator.
return;
}
cc_id_ = pacc->fetch_add(1, std::memory_order_relaxed) + 1;
if (cc_id_ == 0) {
// The identity cannot be 0.
cc_id_ = pacc->fetch_add(1, std::memory_order_relaxed) + 1;
}
}

Expand Down Expand Up @@ -362,12 +373,18 @@ struct queue_generator {
queue_t que_;

conn_info_t(char const * pref, char const * name)
: conn_info_head{pref, name}
, que_{ipc::make_prefix(prefix_, {
"QU_CONN__",
ipc::to_string(DataSize), "__",
ipc::to_string(AlignSize), "__",
name}).c_str()} {}
: conn_info_head{pref, name} { init(); }

void init() {
conn_info_head::init();
if (!que_.valid()) {
que_.open(ipc::make_prefix(prefix_, {
"QU_CONN__",
ipc::to_string(DataSize), "__",
ipc::to_string(AlignSize), "__",
this->name_}).c_str());
}
}

void disconnect_receiver() {
bool dis = que_.disconnect();
Expand Down Expand Up @@ -397,6 +414,18 @@ constexpr static queue_t* queue_of(ipc::handle_t h) noexcept {

/* API implementations */

static bool connect(ipc::handle_t * ph, ipc::prefix pref, char const * name, bool start_to_recv) {
assert(ph != nullptr);
if (*ph == nullptr) {
*ph = ipc::mem::alloc<conn_info_t>(pref.str, name);
}
return reconnect(ph, start_to_recv);
}

static bool connect(ipc::handle_t * ph, char const * name, bool start_to_recv) {
return connect(ph, {nullptr}, name, start_to_recv);
}

static void disconnect(ipc::handle_t h) {
auto que = queue_of(h);
if (que == nullptr) {
Expand All @@ -414,6 +443,7 @@ static bool reconnect(ipc::handle_t * ph, bool start_to_recv) {
if (que == nullptr) {
return false;
}
info_of(*ph)->init();
if (start_to_recv) {
que->shut_sending();
if (que->connect()) { // wouldn't connect twice
Expand All @@ -429,18 +459,6 @@ static bool reconnect(ipc::handle_t * ph, bool start_to_recv) {
return que->ready_sending();
}

static bool connect(ipc::handle_t * ph, ipc::prefix pref, char const * name, bool start_to_recv) {
assert(ph != nullptr);
if (*ph == nullptr) {
*ph = ipc::mem::alloc<conn_info_t>(pref.str, name);
}
return reconnect(ph, start_to_recv);
}

static bool connect(ipc::handle_t * ph, char const * name, bool start_to_recv) {
return connect(ph, {nullptr}, name, start_to_recv);
}

static void destroy(ipc::handle_t h) {
disconnect(h);
ipc::mem::free(info_of(h));
Expand Down
17 changes: 11 additions & 6 deletions src/libipc/platform/win/shm_win.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -37,21 +37,26 @@ id_t acquire(char const * name, std::size_t size, unsigned mode) {
// 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 {
h = ::CreateFileMapping(INVALID_HANDLE_VALUE, detail::get_sa(), PAGE_READWRITE | SEC_COMMIT,
0, static_cast<DWORD>(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) && (::GetLastError() == ERROR_ALREADY_EXISTS)) {
::CloseHandle(h);
if ((mode == create) && (err == ERROR_ALREADY_EXISTS)) {
if (h != NULL) ::CloseHandle(h);
h = NULL;
}
}
if (h == NULL) {
ipc::error("fail CreateFileMapping/OpenFileMapping[%d]: %s\n", static_cast<int>(::GetLastError()), name);
return nullptr;
if (h == NULL) {
ipc::error("fail CreateFileMapping[%d]: %s\n", static_cast<int>(err), name);
return nullptr;
}
}
auto ii = mem::alloc<id_info_t>();
ii->h_ = h;
Expand Down
8 changes: 7 additions & 1 deletion src/libipc/queue.h
Original file line number Diff line number Diff line change
Expand Up @@ -104,7 +104,7 @@ class queue_base : public queue_conn {

explicit queue_base(char const * name)
: queue_base{} {
elems_ = open<elems_t>(name);
elems_ = queue_conn::template open<elems_t>(name);
}

explicit queue_base(elems_t * elems) noexcept
Expand All @@ -117,6 +117,12 @@ class queue_base : public queue_conn {
base_t::close();
}

bool open(char const * name) noexcept {
base_t::close();
elems_ = queue_conn::template open<elems_t>(name);
return elems_ != nullptr;
}

elems_t * elems() noexcept { return elems_; }
elems_t const * elems() const noexcept { return elems_; }

Expand Down

0 comments on commit 76d9b41

Please sign in to comment.