From 52ede859bd2bd5f2d3c20d8ac326a776a17ed553 Mon Sep 17 00:00:00 2001 From: lxsang Date: Wed, 31 Jul 2019 15:11:59 +0200 Subject: [PATCH] add data validity to scheduler --- http_server.c | 61 ++++++++++++++------------------- http_server.h | 15 --------- httpd.c | 33 ++++-------------- libs/handle.c | 4 +++ libs/handle.h | 4 +-- libs/scheduler.c | 21 ++++++++++-- libs/scheduler.h | 88 +++++++++++++++++++++++++++++------------------- 7 files changed, 111 insertions(+), 115 deletions(-) diff --git a/http_server.c b/http_server.c index fe70048..e87fee5 100644 --- a/http_server.c +++ b/http_server.c @@ -159,7 +159,7 @@ void *accept_request(void *data) antd_task_t *task; antd_request_t *rq = (antd_request_t *)data; - task = antd_create_task(NULL, (void *)rq, NULL); + task = antd_create_task(NULL, (void *)rq, NULL, rq->client->last_io); task->priority++; fd_set read_flags, write_flags; // first verify if the socket is ready @@ -180,7 +180,7 @@ void *accept_request(void *data) } if (sel == 0 || (!FD_ISSET(client->sock, &read_flags) && !FD_ISSET(client->sock, &write_flags))) { - if(client->last_wait == 0) client->last_wait = time(NULL); + /*if(client->last_wait == 0) client->last_wait = time(NULL); // retry it later if(time(NULL) - client->last_wait > MAX_WAIT_S) { @@ -188,9 +188,8 @@ void *accept_request(void *data) server_config.connection++; unknow(rq->client); return task; - } + }*/ task->handle = accept_request; - task->status = TASK_ACCEPT_PEND; return task; } // perform the ssl handshake if enabled @@ -208,48 +207,49 @@ void *accept_request(void *data) case SSL_ERROR_WANT_WRITE: case SSL_ERROR_NONE: //LOG("RETRY SSL %d\n", client->sock); - task->status = TASK_ACCEPT_SSL_CONT; + /*if(client->last_wait == 0) client->last_wait = time(NULL); + if(time(NULL) - client->last_wait > MAX_WAIT_S) + { + server_config.connection++; + unknow(rq->client); + LOG("SSL timeout, give up on %d\n", client->sock); + return task; + } + task->status = TASK_ACCEPT_SSL_CONT;*/ task->handle = accept_request; - //task->priority = HIGH_PRIORITY; - //task->type = LIGHT; return task; default: LOG("Error performing SSL handshake %d %d %lu\n", stat, ret, ERR_get_error()); - server_config.connection++; + //server_config.connection++; ERR_print_errors_fp(stderr); unknow(rq->client); return task; } } client->status = 1; - // reset the waiting - client->last_wait = 0; task->handle = accept_request; - task->status = TASK_ACCEPT_HS_DONE; - LOG("Handshake finish for %d\n", client->sock); + //LOG("Handshake finish for %d\n", client->sock); return task; } else { if (!FD_ISSET(client->sock, &read_flags)) { - if(client->last_wait == 0) client->last_wait = time(NULL); + /*if(client->last_wait == 0) client->last_wait = time(NULL); if(time(NULL) - client->last_wait > MAX_WAIT_S) { server_config.connection++; unknow(rq->client); LOG("Read timeout, give up on %d\n", client->sock); return task; - } + }*/ task->handle = accept_request; - task->status = TASK_ACCEPT_READWAIT; return task; } } #endif LOG("Ready for reading %d\n", client->sock); - client->last_wait = time(NULL); - server_config.connection++; + //server_config.connection++; read_buf(rq->client, buf, sizeof(buf)); line = buf; // get the method string @@ -283,7 +283,6 @@ void *accept_request(void *data) // decode request // now return the task task->handle = decode_request_header; - task->status = TASK_DECODE_HEADER; return task; } @@ -292,7 +291,7 @@ void *resolve_request(void *data) struct stat st; char path[2 * BUFFLEN]; antd_request_t *rq = (antd_request_t *)data; - antd_task_t *task = antd_create_task(NULL, (void *)rq, NULL); + antd_task_t *task = antd_create_task(NULL, (void *)rq, NULL, rq->client->last_io); task->priority++; char *url = (char *)dvalue(rq->request, "RESOURCE_PATH"); char *newurl = NULL; @@ -383,7 +382,6 @@ void *resolve_request(void *data) { task->type = HEAVY; task->handle = serve_file; - task->status = TASK_SERVE_FILE; } return task; } @@ -480,7 +478,7 @@ static void error_die(const char *sc) void *serve_file(void *data) { antd_request_t *rq = (antd_request_t *)data; - antd_task_t *task = antd_create_task(NULL, (void *)rq, NULL); + antd_task_t *task = antd_create_task(NULL, (void *)rq, NULL, rq->client->last_io); task->priority++; char *path = (char *)dvalue(rq->request, "ABS_RESOURCE_PATH"); char *mime_type = (char *)dvalue(rq->request, "RESOURCE_MIME"); @@ -615,8 +613,7 @@ void *decode_request_header(void *data) if (host) free(host); // header ok, now checkmethod - antd_task_t *task = antd_create_task(decode_request, (void *)rq, NULL); - task->status = TASK_DECODE_RQ; + antd_task_t *task = antd_create_task(decode_request, (void *)rq, NULL,rq->client->last_io); task->priority++; return task; @@ -636,7 +633,7 @@ void *decode_request(void *data) if (tmp && strcasecmp(tmp, "websocket") == 0) ws = 1; method = (char *)dvalue(rq->request, "METHOD"); - task = antd_create_task(NULL, (void *)rq, NULL); + task = antd_create_task(NULL, (void *)rq, NULL, rq->client->last_io); task->priority++; if (strcmp(method, "GET") == 0 || strcmp(method, "HEAD") == 0) { @@ -651,13 +648,11 @@ void *decode_request(void *data) } // resolve task task->handle = resolve_request; - task->status = TASK_RESOLVE_RQ; return task; } else if (strcmp(method, "POST") == 0) { task->handle = resolve_request; - task->status = TASK_RESOLVE_RQ; //task->type = HEAVY; return task; } @@ -682,7 +677,7 @@ void *decode_post_request(void *data) if (tmp) clen = atoi(tmp); char *method = (char *)dvalue(rq->request, "METHOD"); - task = antd_create_task(NULL, (void *)rq, NULL); + task = antd_create_task(NULL, (void *)rq, NULL, rq->client->last_io); task->priority++; task->type = HEAVY; if (!method || strcmp(method, "POST") != 0) @@ -799,9 +794,8 @@ void *decode_multi_part_request(void *data, const char *ctype) char *str_copy = strdup(ctype); char *orgcpy = str_copy; antd_request_t *rq = (antd_request_t *)data; - antd_task_t *task = antd_create_task(NULL, (void *)rq, NULL); + antd_task_t *task = antd_create_task(NULL, (void *)rq, NULL, rq->client->last_io); task->priority++; - task->status = TASK_DECODE_MP_DATA; //dictionary dic = NULL; boundary = strsep(&str_copy, "="); //discard first part boundary = str_copy; @@ -840,7 +834,7 @@ void *decode_multi_part_request_data(void *data) FILE *fp = NULL; char *token, *keytoken, *valtoken; antd_request_t *rq = (antd_request_t *)data; - antd_task_t *task = antd_create_task(NULL, (void *)rq, NULL); + antd_task_t *task = antd_create_task(NULL, (void *)rq, NULL, rq->client->last_io); task->priority++; char *boundary = (char *)dvalue(rq->request, "MULTI_PART_BOUNDARY"); dictionary dic = (dictionary)dvalue(rq->request, "REQUEST_DATA"); @@ -979,7 +973,6 @@ void *decode_multi_part_request_data(void *data) // continue upload task->type = HEAVY; task->handle = decode_multi_part_request_data; - task->status = TASK_DECODE_MP_DATA; } free(line); free(boundend); @@ -1072,7 +1065,7 @@ void *execute_plugin(void *data, const char *pname) struct plugin_entry *plugin; char *error; antd_request_t *rq = (antd_request_t *)data; - antd_task_t *task = antd_create_task(NULL, (void *)rq, NULL); + antd_task_t *task = antd_create_task(NULL, (void *)rq, NULL, rq->client->last_io); task->priority++; LOG("Plugin name '%s'\n", pname); @@ -1107,14 +1100,12 @@ void *execute_plugin(void *data, const char *pname) { task->handle = fn; task->type = HEAVY; - task->status = TASK_EXEC_PLUGIN_RAW; } else { free(task); - task = antd_create_task(decode_post_request, (void *)rq, fn); + task = antd_create_task(decode_post_request, (void *)rq, fn, rq->client->last_io); task->priority++; - task->status = TASK_EXEC_PLUGIN_COOK; } return task; } diff --git a/http_server.h b/http_server.h index edde09f..da8a30f 100644 --- a/http_server.h +++ b/http_server.h @@ -17,21 +17,6 @@ #define MATCH(s, n) strcmp(section, s) == 0 && strcmp(name, n) == 0 #define CONFIG "config.ini" -// define all task status here -// for debug purpose -#define TASK_ACCEPT 0x01 -#define TASK_ACCEPT_PEND 0xA0 -#define TASK_ACCEPT_SSL_CONT 0xA1 -#define TASK_ACCEPT_HS_DONE 0xA2 -#define TASK_ACCEPT_READWAIT 0xA3 -#define TASK_DECODE_HEADER 0x02 -#define TASK_DECODE_RQ 0x03 -#define TASK_RESOLVE_RQ 0x04 -#define TASK_EXEC_PLUGIN_RAW 0x05 // with raw data -#define TASK_EXEC_PLUGIN_COOK 0x06 // with decoded post request data -#define TASK_SERVE_FILE 0x07 -#define TASK_DECODE_MP_DATA 0x08 - config_t* config(); void destroy_config(); void load_config(const char* file); diff --git a/httpd.c b/httpd.c index 1fe5436..6cc0e10 100644 --- a/httpd.c +++ b/httpd.c @@ -72,6 +72,7 @@ void configure_context(SSL_CTX *ctx) #endif + void stop_serve(int dummy) { UNUSED(dummy); sigset_t mask; @@ -99,25 +100,7 @@ void stop_serve(int dummy) { close(server_sock); sigprocmask(SIG_UNBLOCK, &mask, NULL); } -void antd_scheduler_dump(antd_scheduler_t* scheduler) -{ - antd_task_queue_t queue = NULL; - antd_task_item_t it = NULL; - LOG("[[[[SCHEDULER]]]] : dumping all value:\n"); - pthread_mutex_lock(&scheduler->scheduler_lock); - for(int i = 0; i < N_PRIORITY; i++) - { - queue = scheduler->task_queue[i]; - for(it = queue; it != NULL && it->next != NULL; it = it->next) - { - antd_request_t* request = it->task->data; - LOG("From: %s [%d]\n", request->client->ip, request->client->sock); - LOG("\tStamp: %ul\n", it->task->stamp); - LOG("\tstatus: %x\n", it->task->status); - } - } - pthread_mutex_unlock(&scheduler->scheduler_lock); -} + int main(int argc, char* argv[]) { // load the config first @@ -150,6 +133,8 @@ int main(int argc, char* argv[]) LOG("httpd running on port %d\n", port); // default to 4 workers antd_scheduler_init(&scheduler, config()->n_workers); + scheduler.validate_data = 1; + scheduler.destroy_data = finish_request; // use blocking server_sock // make the scheduler wait for event on another thread // this allow to ged rid of high cpu usage on @@ -189,10 +174,6 @@ int main(int argc, char* argv[]) client_ip = inet_ntoa(client_name.sin_addr); client->ip = strdup(client_ip); LOG("Client IP: %s\n", client_ip); - if(strcmp(client->ip, "193.48.235.2") == 0) - { - antd_scheduler_dump(&scheduler); - } //LOG("socket: %d\n", client_sock); } @@ -209,10 +190,10 @@ int main(int argc, char* argv[]) perror("setsockopt failed\n"); */ client->sock = client_sock; + time(&client->last_io); #ifdef USE_OPENSSL client->ssl = NULL; client->status = 0; - client->last_wait = 0; if(config()->usessl == 1) { client->ssl = (void*)SSL_new(ctx); @@ -227,9 +208,9 @@ int main(int argc, char* argv[]) }*/ } #endif + config()->connection++; // create callback for the server - task = antd_create_task(accept_request,(void*)request, finish_request ); - task->status = TASK_ACCEPT; + task = antd_create_task(accept_request,(void*)request, finish_request, client->last_io); //task->type = LIGHT; antd_add_task(&scheduler, task); } diff --git a/libs/handle.c b/libs/handle.c index 799d056..da0ba36 100644 --- a/libs/handle.c +++ b/libs/handle.c @@ -193,6 +193,8 @@ int antd_send(void *src, const void* data, int len) { antd_close(src); }*/ + if(written > 0) + time(&source->last_io); return written; } int antd_recv(void *src, void* data, int len) @@ -340,6 +342,8 @@ int antd_recv(void *src, void* data, int len) { antd_close(src); }*/ + if(read > 0) + time(&source->last_io); return read; } void set_nonblock(int socket) { diff --git a/libs/handle.h b/libs/handle.h index 15ade27..eefbd94 100644 --- a/libs/handle.h +++ b/libs/handle.h @@ -26,7 +26,7 @@ #define __RESULT__ "{\"result\":%d,\"msg\":\"%s\"}" #define FORM_URL_ENCODE "application/x-www-form-urlencoded" #define FORM_MULTI_PART "multipart/form-data" -#define MAX_WAIT_S 20 // 1/3 minute +#define MAX_WAIT_S 2 // 1/3 minute #ifdef USE_OPENSSL int __attribute__((weak)) usessl(); #endif @@ -38,7 +38,7 @@ typedef struct{ #ifdef USE_OPENSSL int status; #endif - time_t last_wait; + time_t last_io; } antd_client_t; typedef struct { diff --git a/libs/scheduler.c b/libs/scheduler.c index c3dc833..bb499a9 100644 --- a/libs/scheduler.c +++ b/libs/scheduler.c @@ -154,6 +154,8 @@ void antd_scheduler_init(antd_scheduler_t* scheduler, int n) scheduler->status = 1; scheduler->workers_queue = NULL; scheduler->pending_task = 0 ; + scheduler->validate_data = 0; + scheduler->destroy_data = NULL; // init semaphore scheduler->scheduler_sem = sem_open("scheduler", O_CREAT, 0600, 0); if (scheduler->scheduler_sem == SEM_FAILED) @@ -216,7 +218,7 @@ void antd_scheduler_destroy(antd_scheduler_t* scheduler) /* create a task */ -antd_task_t* antd_create_task(void* (*handle)(void*), void *data, void* (*callback)(void*)) +antd_task_t* antd_create_task(void* (*handle)(void*), void *data, void* (*callback)(void*), time_t atime) { antd_task_t* task = (antd_task_t*)malloc(sizeof *task); task->stamp = (unsigned long)time(NULL); @@ -225,7 +227,7 @@ antd_task_t* antd_create_task(void* (*handle)(void*), void *data, void* (*callba task->callback = callback_of(callback); task->priority = NORMAL_PRIORITY; task->type = LIGHT; - task->status = NOSTATUS; + task->access_time = atime; return task; } @@ -312,11 +314,26 @@ int antd_task_schedule(antd_scheduler_t* scheduler) break; } pthread_mutex_unlock(&scheduler->scheduler_lock); + // no task if(!it) { return 0; } // has the task now + // validate the task + if(scheduler->validate_data && difftime( time(NULL), it->task->access_time) > MAX_VALIDITY_INTERVAL) + { + // data task is not valid + LOG("Task data is not valid \n"); + if(scheduler->destroy_data) + scheduler->destroy_data(it->task->data); + if(it->task->callback) + free_callback(it->task->callback); + free(it->task); + free(it); + return 0; + } + // check the type of task if(it->task->type == LIGHT || scheduler->n_workers <= 0) { diff --git a/libs/scheduler.h b/libs/scheduler.h index 4627276..687a6b3 100644 --- a/libs/scheduler.h +++ b/libs/scheduler.h @@ -9,63 +9,69 @@ #define NORMAL_PRIORITY ((int)((N_PRIORITY - 1) / 2)) #define LOW_PRIORITY (N_PRIORITY - 1) #define HIGH_PRIORITY 0 -#define NOSTATUS 0x00 -typedef enum { LIGHT, HEAVY } antd_task_type_t; +#define MAX_VALIDITY_INTERVAL 20 // 10 s for task validity +typedef enum +{ + LIGHT, + HEAVY +} antd_task_type_t; // callback definition -typedef struct __callback_t{ - void* (*handle)(void*); - struct __callback_t * next; +typedef struct __callback_t +{ + void *(*handle)(void *); + struct __callback_t *next; } antd_callback_t; // task definition -typedef struct { +typedef struct +{ /* creation time of a task */ unsigned long stamp; + /* + Last access time of + task data + */ + time_t access_time; /* priority from 0 to N_PRIORITY - 1 higher value is lower priority */ uint8_t priority; - /* - The status of a task - should be set by the application - default value 0x00 means that - there is no status set - */ - uint8_t status; /* the callback */ - void* (*handle)(void*); - antd_callback_t* callback; - /* + void *(*handle)(void *); + antd_callback_t *callback; + /* user data if any */ - void * data; + void *data; /* type of a task light tasks are executed directly heavy tasks are delegated to workers */ - antd_task_type_t type; + antd_task_type_t type; } antd_task_t; - -typedef struct __task_item_t{ - antd_task_t* task; - struct __task_item_t* next; -}* antd_task_item_t; +typedef struct __task_item_t +{ + antd_task_t *task; + struct __task_item_t *next; +} * antd_task_item_t; typedef antd_task_item_t antd_task_queue_t; -typedef struct { +typedef struct +{ int id; pthread_t tid; - void* manager; + void *manager; } antd_worker_t; -typedef struct { +typedef struct +{ // data lock pthread_mutex_t scheduler_lock; pthread_mutex_t worker_lock; @@ -77,44 +83,56 @@ typedef struct { antd_task_queue_t task_queue[N_PRIORITY]; antd_task_queue_t workers_queue; uint8_t status; // 0 stop, 1 working - antd_worker_t* workers; + antd_worker_t *workers; int n_workers; int pending_task; + /* + function pointer that free data in a task if + the task is not valid + default to NULL + */ + void* (*destroy_data)(void*); + int validate_data; } antd_scheduler_t; /* init the main scheduler */ -void antd_scheduler_init(antd_scheduler_t*, int); +void antd_scheduler_init(antd_scheduler_t *, int); /* destroy all pending task */ -void antd_scheduler_destroy(antd_scheduler_t*); +void antd_scheduler_destroy(antd_scheduler_t *); /* create a task + parameter: + - handle + - data + - callback + - last data access time */ -antd_task_t* antd_create_task(void* (*handle)(void*), void *data, void* (*callback)(void*)); +antd_task_t *antd_create_task(void *(*handle)(void *), void *data, void *(*callback)(void *), time_t); /* add a task */ -void antd_add_task(antd_scheduler_t*, antd_task_t*); +void antd_add_task(antd_scheduler_t *, antd_task_t *); /* execute and free a task a task */ -void antd_execute_task(antd_scheduler_t*, antd_task_item_t); +void antd_execute_task(antd_scheduler_t *, antd_task_item_t); /* scheduler status */ -int antd_scheduler_busy(antd_scheduler_t*); +int antd_scheduler_busy(antd_scheduler_t *); /* schedule a task */ -int antd_task_schedule(antd_scheduler_t*); +int antd_task_schedule(antd_scheduler_t *); /* wait for event */ -void antd_wait(antd_scheduler_t*); +void antd_wait(antd_scheduler_t *); #endif \ No newline at end of file