From 305dbeb746e4496ec5ea9cc98685639c70fbfaf7 Mon Sep 17 00:00:00 2001 From: xiehan <52160700+Barenboim@users.noreply.github.com> Date: Wed, 1 Jan 2025 20:54:43 +0800 Subject: [PATCH 1/2] Simplify service session's state. (#1676) --- src/kernel/Communicator.cc | 44 +++++++++++++++++--------------------- src/kernel/Communicator.h | 2 +- 2 files changed, 21 insertions(+), 25 deletions(-) diff --git a/src/kernel/Communicator.cc b/src/kernel/Communicator.cc index de47f4db..c76d7bcf 100644 --- a/src/kernel/Communicator.cc +++ b/src/kernel/Communicator.cc @@ -224,7 +224,7 @@ int CommService::init(const struct sockaddr *bind_addr, socklen_t addrlen, this->addrlen = addrlen; this->listen_timeout = listen_timeout; this->response_timeout = response_timeout; - INIT_LIST_HEAD(&this->alive_list); + INIT_LIST_HEAD(&this->keep_alive_list); this->ssl_ctx = NULL; this->ssl_accept_timeout = 0; @@ -253,9 +253,9 @@ int CommService::drain(int max) errno_bak = errno; pthread_mutex_lock(&this->mutex); - while (cnt != max && !list_empty(&this->alive_list)) + while (cnt != max && !list_empty(&this->keep_alive_list)) { - pos = this->alive_list.next; + pos = this->keep_alive_list.prev; entry = list_entry(pos, struct CommConnEntry, list); list_del(pos); cnt++; @@ -310,12 +310,7 @@ private: CommService *service; private: - virtual int create_connect_fd() - { - errno = EPERM; - return -1; - } - + virtual ~CommServiceTarget() { } friend class Communicator; }; @@ -359,7 +354,7 @@ CommSession::~CommSession() return; target = (CommServiceTarget *)this->target; - if (this->passive == 2) + if (!this->out && target->has_idle_conn()) target->shutdown(); target->decref(); @@ -485,7 +480,7 @@ int Communicator::send_message_sync(struct iovec vectors[], int cnt, if (service->listen_fd >= 0) { entry->state = CONN_STATE_KEEPALIVE; - list_add_tail(&entry->list, &service->alive_list); + list_add(&entry->list, &service->keep_alive_list); entry = NULL; } @@ -607,7 +602,6 @@ void Communicator::handle_incoming_request(struct poller_result *res) { case PR_ST_SUCCESS: session = entry->session; - session->passive = 2; state = CS_STATE_TOREPLY; pthread_mutex_lock(&target->mutex); if (entry->state == CONN_STATE_SUCCESS) @@ -649,7 +643,6 @@ void Communicator::handle_incoming_request(struct poller_result *res) state = CS_STATE_ERROR; case CONN_STATE_RECEIVING: session = entry->session; - session->passive = 3; break; case CONN_STATE_SUCCESS: @@ -795,7 +788,7 @@ void Communicator::handle_reply_result(struct poller_result *res) if (!this->stop_flag && service->listen_fd >= 0) { entry->state = CONN_STATE_KEEPALIVE; - list_add_tail(&entry->list, &service->alive_list); + list_add(&entry->list, &service->keep_alive_list); } else { @@ -1083,7 +1076,6 @@ void Communicator::handle_recvfrom_result(struct poller_result *res) target = entry->target; if (entry->state == CONN_STATE_SUCCESS) { - session->passive = 2; state = CS_STATE_TOREPLY; error = 0; entry->state = CONN_STATE_IDLE; @@ -1091,7 +1083,6 @@ void Communicator::handle_recvfrom_result(struct poller_result *res) } else { - session->passive = 3; state = CS_STATE_ERROR; if (entry->state == CONN_STATE_ERROR) error = entry->error; @@ -1974,14 +1965,19 @@ int Communicator::reply(CommSession *session) int errno_bak; int ret; - if (session->passive != 2) + if (!session->passive) { errno = EINVAL; return -1; } + if (session->out) + { + errno = ENOENT; + return -1; + } + errno_bak = errno; - session->passive = 3; target = (CommServiceTarget *)session->target; if (target->service->reliable) ret = this->reply_reliable(session, target); @@ -2023,9 +2019,8 @@ int Communicator::push(const void *buf, size_t size, CommSession *session) mutex = &in->entry->mutex; pthread_mutex_lock(mutex); - if ((session->passive == 2 && !list_empty(&session->target->idle_list)) || - (!session->passive && in->entry->session == session) || - session->passive == 1) + if ((!session->passive || session->target->has_idle_conn()) && + in->entry->session == session) { ret = in->inner()->feedback(buf, size); } @@ -2043,15 +2038,14 @@ int Communicator::shutdown(CommSession *session) { CommServiceTarget *target; - if (session->passive != 2) + if (!session->passive) { errno = EINVAL; return -1; } - session->passive = 3; target = (CommServiceTarget *)session->target; - if (!target->shutdown()) + if (session->out || !target->shutdown()) { errno = ENOENT; return -1; @@ -2068,7 +2062,9 @@ int Communicator::sleep(SleepSession *session) { if (mpoller_add_timer(&value, session, &session->timer, &session->index, this->mpoller) >= 0) + { return 0; + } } return -1; diff --git a/src/kernel/Communicator.h b/src/kernel/Communicator.h index 2c5f7e08..59628ecd 100644 --- a/src/kernel/Communicator.h +++ b/src/kernel/Communicator.h @@ -230,7 +230,7 @@ private: int ref; private: - struct list_head alive_list; + struct list_head keep_alive_list; pthread_mutex_t mutex; public: From be06666e4a2928e919d680909de22fd7dd9d555b Mon Sep 17 00:00:00 2001 From: Xie Han <63350856@qq.com> Date: Wed, 1 Jan 2025 20:55:42 +0800 Subject: [PATCH 2/2] 0.11.6 -> 0.11.7 --- CMakeLists.txt | 2 +- xmake.lua | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/CMakeLists.txt b/CMakeLists.txt index cac84e60..ff527e96 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -5,7 +5,7 @@ set(CMAKE_SKIP_RPATH TRUE) project( workflow - VERSION 0.11.6 + VERSION 0.11.7 LANGUAGES C CXX ) diff --git a/xmake.lua b/xmake.lua index 3317e4f5..a1682647 100644 --- a/xmake.lua +++ b/xmake.lua @@ -1,5 +1,5 @@ set_project("workflow") -set_version("0.11.6") +set_version("0.11.7") option("workflow_inc", {description = "workflow inc", default = "$(projectdir)/_include"}) option("workflow_lib", {description = "workflow lib", default = "$(projectdir)/_lib"})