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/src/kernel/Communicator.cc b/src/kernel/Communicator.cc index c1bcd93c..dcd2774f 100644 --- a/src/kernel/Communicator.cc +++ b/src/kernel/Communicator.cc @@ -177,7 +177,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); return 0; } @@ -203,9 +203,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++; @@ -260,12 +260,7 @@ private: CommService *service; private: - virtual int create_connect_fd() - { - errno = EPERM; - return -1; - } - + virtual ~CommServiceTarget() { } friend class Communicator; }; @@ -309,7 +304,7 @@ CommSession::~CommSession() return; target = (CommServiceTarget *)this->target; - if (this->passive == 2) + if (!this->out && target->has_idle_conn()) target->shutdown(); target->decref(); @@ -424,7 +419,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; } @@ -545,7 +540,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) @@ -587,7 +581,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: @@ -733,7 +726,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 { @@ -986,7 +979,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; @@ -994,7 +986,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; @@ -1842,14 +1833,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); @@ -1891,9 +1887,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); } @@ -1911,15 +1906,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; @@ -1936,7 +1930,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 55142fa3..c99a1133 100644 --- a/src/kernel/Communicator.h +++ b/src/kernel/Communicator.h @@ -203,7 +203,7 @@ private: int ref; private: - struct list_head alive_list; + struct list_head keep_alive_list; pthread_mutex_t mutex; public: diff --git a/xmake.lua b/xmake.lua index 318562b5..ddf2dd08 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"})