Reduce CPU usage on idle in proxy mode:

- Scheduler timeout event now support millisecond granulity
- Use only READABLE event on proxy socket which reduce tasks CPU time
This commit is contained in:
lxsang 2021-01-30 22:16:11 +01:00
parent 474bccd587
commit 1883330bba
3 changed files with 34 additions and 27 deletions

View File

@ -762,31 +762,36 @@ static void *proxy_monitor(void *data)
rq->client->state = ANTD_CLIENT_PROXY_MONITOR; rq->client->state = ANTD_CLIENT_PROXY_MONITOR;
antd_client_t *proxy = (antd_client_t *)dvalue(rq->request, "PROXY_HANDLE"); antd_client_t *proxy = (antd_client_t *)dvalue(rq->request, "PROXY_HANDLE");
antd_task_t *task = antd_create_task(NULL, data, NULL, rq->client->last_io); antd_task_t *task = antd_create_task(NULL, data, NULL, rq->client->last_io);
int ret, sz1, sz2; int ret, sz1 = 0, sz2 = 0;
char *buf = NULL; char *buf = NULL;
buf = (char *)malloc(BUFFLEN); buf = (char *)malloc(BUFFLEN);
struct pollfd pfd[1]; struct pollfd pfd[1];
memset(pfd, 0, sizeof(pfd)); memset(pfd, 0, sizeof(pfd));
pfd[0].fd = proxy->sock; pfd[0].fd = proxy->sock;
pfd[0].events = POLLIN; pfd[0].events = POLLIN;
ret = 1; ret = 1;
if (poll(pfd, 1, 0) < 0)
{
(void)close(proxy->sock);
return task;
}
do
{
sz1 = antd_recv_upto(rq->client, buf, BUFFLEN); sz1 = antd_recv_upto(rq->client, buf, BUFFLEN);
if ((sz1 < 0) || (sz1 > 0 && antd_send(proxy, buf, sz1) != sz1)) if ((sz1 < 0) || (sz1 > 0 && antd_send(proxy, buf, sz1) != sz1))
{ {
ret = 0; ret = 0;
break;
} }
sz2 = read_buf(proxy, buf, BUFFLEN);
if (poll(pfd, 1, 0) < 0)
{
ret = 0;
}
sz2 = antd_recv_upto(proxy, buf, BUFFLEN);
if (sz2 < 0 || (sz2 > 0 && antd_send(rq->client, buf, sz2) != sz2)) if (sz2 < 0 || (sz2 > 0 && antd_send(rq->client, buf, sz2) != sz2))
{ {
ret = 0; ret = 0;
break;
} }
} while (sz1 > 0 || sz2 > 0);
free(buf); free(buf);
if (ret == 0) if (ret == 0)
@ -794,27 +799,23 @@ static void *proxy_monitor(void *data)
(void)close(proxy->sock); (void)close(proxy->sock);
return task; return task;
} }
if (sz2 == 0) if (sz2 == 0)
{ {
if ( if (
pfd[0].revents & POLLERR || pfd[0].revents & POLLERR ||
pfd[0].revents & POLLRDHUP || pfd[0].revents & POLLRDHUP ||
pfd[0].revents & POLLHUP || pfd[0].revents & POLLHUP ||
pfd[0].revents & POLLNVAL || pfd[0].revents & POLLNVAL) //||
pfd[0].revents & POLLIN) //pfd[0].revents & POLLIN)
{ {
(void)close(proxy->sock); (void)close(proxy->sock);
return task; return task;
} }
usleep(10000);
} }
task->handle = proxy_monitor; task->handle = proxy_monitor;
task->access_time = rq->client->last_io; task->access_time = rq->client->last_io;
antd_task_bind_event(task, proxy->sock, 0, TASK_EVT_ON_READABLE);
antd_task_bind_event(task, rq->client->sock, 0, TASK_EVT_ON_READABLE); antd_task_bind_event(task, rq->client->sock, 0, TASK_EVT_ON_READABLE);
antd_task_bind_event(task, proxy->sock, 0, TASK_EVT_ON_READABLE | TASK_EVT_ON_WRITABLE);
return task; return task;
} }

View File

@ -301,7 +301,6 @@ void antd_scheduler_destroy_data(void *data)
antd_client_t *proxy = (antd_client_t *)dvalue(rq->request, "PROXY_HANDLE"); antd_client_t *proxy = (antd_client_t *)dvalue(rq->request, "PROXY_HANDLE");
if(proxy) if(proxy)
{ {
printf("closing proxy \n");
close(proxy->sock); close(proxy->sock);
} }
finish_request(data); finish_request(data);
@ -309,12 +308,12 @@ void antd_scheduler_destroy_data(void *data)
int antd_task_data_id(void *data) int antd_task_data_id(void *data)
{ {
antd_request_t *rq = (antd_request_t *)data; /*antd_request_t *rq = (antd_request_t *)data;
if(!rq) if(!rq)
return 0; return 0;
return antd_scheduler_next_id(scheduler,rq->client->sock); return antd_scheduler_next_id(scheduler,rq->client->sock);*/
//UNUSED(data); UNUSED(data);
//return antd_scheduler_next_id(scheduler,0); return antd_scheduler_next_id(scheduler,0);
} }
int main(int argc, char *argv[]) int main(int argc, char *argv[])

View File

@ -5,6 +5,7 @@
#include <sys/types.h> #include <sys/types.h>
#include <sys/stat.h> #include <sys/stat.h>
#include <unistd.h> #include <unistd.h>
#include <sys/time.h>
#include <poll.h> #include <poll.h>
#include "scheduler.h" #include "scheduler.h"
#include "utils.h" #include "utils.h"
@ -24,6 +25,7 @@ typedef struct
{ {
int flags; int flags;
int fd; int fd;
struct timeval stamp;
int timeout; // seconds int timeout; // seconds
antd_task_t *task; antd_task_t *task;
} antd_task_evt_item_t; } antd_task_evt_item_t;
@ -655,6 +657,7 @@ static void task_event_collect(bst_node_t* node, void** argv, int argc)
antd_task_t* task = (antd_task_t*) node->data; antd_task_t* task = (antd_task_t*) node->data;
bst_node_t** exec_list = (bst_node_t**) argv[0]; bst_node_t** exec_list = (bst_node_t**) argv[0];
bst_node_t** poll_list = (bst_node_t**) argv[1]; bst_node_t** poll_list = (bst_node_t**) argv[1];
struct timeval now;
int* pollsize = (int*) argv[2]; int* pollsize = (int*) argv[2];
if(!task->events) if(!task->events)
{ {
@ -671,7 +674,10 @@ static void task_event_collect(bst_node_t* node, void** argv, int argc)
else if(it->evt->flags & TASK_EVT_ON_TIMEOUT) else if(it->evt->flags & TASK_EVT_ON_TIMEOUT)
{ {
// check if timeout // check if timeout
if(difftime(time(NULL),task->stamp) > it->evt->timeout ) gettimeofday(&now, NULL);
//do stuff
int diff = (int)(((now.tv_sec - it->evt->stamp.tv_sec) * 1000000 + now.tv_usec - it->evt->stamp.tv_usec) / 1000);
if( diff >= it->evt->timeout )
{ {
*exec_list = bst_insert(*exec_list,task->id, task); *exec_list = bst_insert(*exec_list,task->id, task);
} }
@ -692,6 +698,7 @@ void antd_task_bind_event(antd_task_t *task, int fd, int timeout, int flags)
eit->timeout = timeout; eit->timeout = timeout;
eit->flags = flags; eit->flags = flags;
eit->task = task; eit->task = task;
gettimeofday(&eit->stamp, NULL);
enqueue(&task->events, eit); enqueue(&task->events, eit);
} }