mirror of
https://github.com/sogou/workflow.git
synced 2026-10-22 23:39:54 +08:00
optimize CommSchedGroup target selection
This commit is contained in:
@@ -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);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -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;
|
||||
|
||||
Reference in New Issue
Block a user