diff --git a/src/kernel/CommScheduler.cc b/src/kernel/CommScheduler.cc index dd0f2331..1558a09e 100644 --- a/src/kernel/CommScheduler.cc +++ b/src/kernel/CommScheduler.cc @@ -142,7 +142,7 @@ CommTarget *CommSchedTarget::acquire(int wait_timeout) return this; } -void CommSchedTarget::release() +void CommSchedTarget::release(int keep_alive) { pthread_mutex_t *mutex = &this->mutex; @@ -164,7 +164,7 @@ void CommSchedTarget::release() if (this->wait_cnt == 0 && this->group->wait_cnt > 0) pthread_cond_signal(&this->group->cond); - this->group->heap_adjust(this->index); + this->group->heap_adjust(this->index, keep_alive); } pthread_mutex_unlock(mutex); @@ -184,7 +184,7 @@ int CommSchedGroup::target_cmp(CommSchedTarget *target1, return 0; } -void CommSchedGroup::heap_adjust(int index) +void CommSchedGroup::heap_adjust(int index, int swap_on_equal) { CommSchedTarget *target = this->tg_heap[index]; CommSchedTarget *parent; @@ -192,7 +192,7 @@ void CommSchedGroup::heap_adjust(int index) while (index > 0) { parent = this->tg_heap[(index - 1) / 2]; - if (CommSchedGroup::target_cmp(target, parent) < 0) + if (CommSchedGroup::target_cmp(target, parent) < !!swap_on_equal) { this->tg_heap[index] = parent; parent->index = index; @@ -282,7 +282,7 @@ int CommSchedGroup::heap_insert(CommSchedTarget *target) this->tg_heap[this->heap_size] = target; target->index = this->heap_size; - this->heap_adjust(this->heap_size); + this->heap_adjust(this->heap_size, 0); this->heap_size++; return 0; } @@ -297,7 +297,7 @@ void CommSchedGroup::heap_remove(int index) target = this->tg_heap[this->heap_size]; this->tg_heap[index] = target; target->index = index; - this->heap_adjust(index); + this->heap_adjust(index, 0); this->heapify(target->index); } } diff --git a/src/kernel/CommScheduler.h b/src/kernel/CommScheduler.h index 62a43501..a4951a13 100644 --- a/src/kernel/CommScheduler.h +++ b/src/kernel/CommScheduler.h @@ -69,7 +69,7 @@ public: private: virtual CommTarget *acquire(int wait_timeout); /* final */ - virtual void release(); /* final */ + virtual void release(int keep_alive); /* final */ private: CommSchedGroup *group; @@ -102,7 +102,7 @@ private: private: static int target_cmp(CommSchedTarget *target1, CommSchedTarget *target2); void heapify(int top); - void heap_adjust(int from); + void heap_adjust(int index, int swap_on_equal); int heap_insert(CommSchedTarget *target); void heap_remove(int index); friend class CommSchedTarget; @@ -132,7 +132,7 @@ public: { ret = this->comm.request(session, *target); if (ret < 0) - (*target)->release(); + (*target)->release(0); } return ret; diff --git a/src/kernel/Communicator.cc b/src/kernel/Communicator.cc index 8770bd74..0924e1d2 100644 --- a/src/kernel/Communicator.cc +++ b/src/kernel/Communicator.cc @@ -688,7 +688,7 @@ void Communicator::handle_incoming_reply(struct poller_result *res) { if (session) { - target->release(); + target->release(entry->state == CONN_STATE_IDLE); session->handle(state, res->error); } @@ -810,7 +810,7 @@ void Communicator::handle_request_result(struct poller_result *res) case PR_ST_STOPPED: state = CS_STATE_STOPPED; - entry->target->release(); + entry->target->release(0); session->handle(state, res->error); pthread_mutex_lock(&entry->mutex); /* do nothing */ @@ -994,7 +994,7 @@ void Communicator::handle_connect_result(struct poller_result *res) case PR_ST_STOPPED: state = CS_STATE_STOPPED; - target->release(); + target->release(0); session->handle(state, res->error); this->release_conn(entry); break; diff --git a/src/kernel/Communicator.h b/src/kernel/Communicator.h index 443ce901..f67ce287 100644 --- a/src/kernel/Communicator.h +++ b/src/kernel/Communicator.h @@ -73,7 +73,7 @@ private: virtual int init_ssl(SSL *ssl) { return 0; } public: - virtual void release() { } + virtual void release(int keep_alive) { } private: struct sockaddr *addr;