proxy task is not closed imediately, result in hight CPU usage

http_server.c

@@ -1,5 +1,6 @@
+//#define _GNU_SOURCE
 #include <sys/socket.h>
-#include <sys/select.h>
+#include <poll.h>
 #include <netinet/in.h>
 #include <dlfcn.h>
 #include <sys/stat.h>
@@ -301,24 +302,26 @@ void *accept_request(void *data)
 
 	task = antd_create_task(NULL, (void *)rq, NULL, rq->client->last_io);
 	antd_task_bind_event(task, rq->client->sock, 0, TASK_EVT_ON_WRITABLE | TASK_EVT_ON_READABLE);
-	fd_set read_flags, write_flags;
 	// first verify if the socket is ready
 	antd_client_t *client = (antd_client_t *)rq->client;
-	FD_ZERO(&read_flags);
+	
-	FD_SET(rq->client->sock, &read_flags);
+	struct pollfd pfd[1];
-	FD_ZERO(&write_flags);
+
-	FD_SET(rq->client->sock, &write_flags);
+	pfd[0].fd = client->sock;
-	struct timeval timeout;
+	pfd[0].events = POLLIN | POLLOUT;
-	timeout.tv_sec = 0;
+
-	timeout.tv_usec = 500;
+	int sel = poll(pfd, 1, POLL_EVENT_TO);
-	// select
-	int sel = select(client->sock + 1, &read_flags, &write_flags, (fd_set *)0, &timeout);
 	if (sel == -1)
 	{
 		antd_error(rq->client, 400, "Bad request");
 		return task;
 	}
-	if (sel == 0 || (!FD_ISSET(client->sock, &read_flags) && !FD_ISSET(client->sock, &write_flags)))
+	if(pfd[0].revents & POLLERR || pfd[0].revents & POLLHUP)
+	{
+		antd_error(rq->client, 400, "Bad request");
+		return task;
+	}
+	if (sel == 0 || (!(pfd[0].revents & POLLIN) && !(pfd[0].revents & POLLOUT)))
 	{
 		task->handle = accept_request;
 		return task;
@@ -354,7 +357,7 @@ void *accept_request(void *data)
 	}
 	else
 	{
-		if (!FD_ISSET(client->sock, &read_flags))
+		if (!((pfd[0].revents & POLLIN)))
 		{
 			task->handle = accept_request;
 			return task;
@@ -756,59 +759,63 @@ static void *proxy_monitor(void *data)
 {
 	antd_request_t *rq = (antd_request_t *)data;
 	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);
+	char* method = (char *)dvalue(rq->request, "METHOD");
-	int ret, max_fd;
+	antd_task_t *task = antd_create_task(NULL, data, NULL, rq->client->last_io);
-	fd_set read_flags, write_flags;
+	int ret, sz;
-	// first verify if the socket is ready
+	struct pollfd pfd[2];
-	FD_ZERO(&read_flags);
-	FD_SET(rq->client->sock, &read_flags);
-	FD_SET(proxy->sock, &read_flags);
-	FD_ZERO(&write_flags);
-	FD_SET(rq->client->sock, &write_flags);
-	//FD_SET(proxy->sock, &write_flags);
 	char *buf = NULL;
-	struct timeval timeout;
+
-	timeout.tv_sec = 0;
+	pfd[0].fd = rq->client->sock;
-	timeout.tv_usec = 5000;
+	pfd[1].fd = proxy->sock;
-	
+	pfd[1].events = POLLIN;
-	max_fd = proxy->sock > rq->client->sock ? proxy->sock: rq->client->sock;
+	pfd[0].events = POLLIN;
-	buf = (char *)malloc(BUFFLEN);
+	if(rq->client->ssl)
-	//printf("start proxy monitor\n");
-	// select
-	do
 	{
-		ret = select(max_fd + 1, &read_flags, &write_flags, (fd_set *)0, &timeout);
+		pfd[0].events |= POLLOUT;
-		if(ret > 0)
+	}
+	// select
+	ret = poll(pfd, 2, POLL_EVENT_TO);
+	if (ret > 0)
+	{
+		if(
+			pfd[0].revents & POLLERR ||
+			pfd[0].revents & POLLRDHUP ||
+			pfd[0].revents & POLLHUP ||
+			pfd[0].revents & POLLNVAL||
+			pfd[1].revents & POLLERR ||
+			pfd[1].revents & POLLHUP ||
+			pfd[1].revents & POLLRDHUP ||
+			pfd[1].revents & POLLNVAL 
+			)
 		{
-			
+			(void)close(proxy->sock);
-			memset(buf, '\0', BUFFLEN);
+			return task;
-			if (FD_ISSET(rq->client->sock, &read_flags) || FD_ISSET(rq->client->sock, &write_flags))
+		}
+		buf = (char *)malloc(BUFFLEN);
+		memset(buf, '\0', BUFFLEN);
+		if ((pfd[0].revents & POLLIN) || (pfd[0].revents & POLLOUT))
+		{
+			sz = antd_recv_upto(rq->client, buf, BUFFLEN);
+			if ( (sz < 0) || (sz == 0 && !rq->client->ssl) || ( antd_send(proxy, buf, sz) != sz ))
 			{
-				ret = antd_recv_upto(rq->client, buf, BUFFLEN);
+				free(buf);
-				if(ret == -1)
+				(void)close(proxy->sock);
-				{
+				return task;
-					free(buf);
-					(void)close(proxy->sock);
-					return task;
-				}
-				antd_send(proxy, buf, ret);
-			}
-			if (FD_ISSET(proxy->sock, &read_flags))
-			{
-				ret = antd_recv_upto(proxy, buf, BUFFLEN);
-				if(ret == -1)
-				{
-					free(buf);
-					(void)close(proxy->sock);
-					return task;
-				}
-				antd_send(rq->client, buf, ret);
 			}
 		}
-	} while (ret > 0);
+		if (pfd[1].revents & POLLIN)
-	free(buf);
+		{
-	//printf("monitor return: %d\n", ret);
+			if ((sz = antd_recv_upto(proxy, buf, BUFFLEN)) <= 0 || antd_send(rq->client, buf, sz) != sz)
-	if(ret == -1)
+			{
+				free(buf);
+				(void)close(proxy->sock);
+				return task;
+			}
+		}
+		free(buf);
+	}
+	
+	if (ret < 0)
 	{
 		(void)close(proxy->sock);
 		return task;
@@ -817,7 +824,14 @@ static void *proxy_monitor(void *data)
 	task->handle = proxy_monitor;
 	task->access_time = rq->client->last_io;
 	// register event
-	antd_task_bind_event(task, rq->client->sock, 0, TASK_EVT_ON_READABLE | TASK_EVT_ON_WRITABLE);
+	if(rq->client->ssl && (EQU(method, "POST") || ws_enable(rq->request)))
+	{
+		antd_task_bind_event(task, rq->client->sock, 0, TASK_EVT_ON_READABLE | TASK_EVT_ON_WRITABLE);
+	}
+	else
+	{
+		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);
 	return task;
 }
@@ -834,6 +848,7 @@ static void *proxify(void *data)
 	char *path = dvalue(rq->request, "PROXY_PATH");
 	char *query = dvalue(rq->request, "PROXY_QUERY");
 	dictionary_t xheader = dvalue(rq->request, "REQUEST_HEADER");
+	char* method = (char *)dvalue(rq->request, "METHOD");
 	antd_task_t *task = antd_create_task(NULL, data, NULL, rq->client->last_io);
 	if (!xheader)
 	{
@@ -852,6 +867,7 @@ static void *proxify(void *data)
 	proxy->ssl = NULL;
 	proxy->zstream = NULL;
 	proxy->z_level = ANTD_CNONE;
+	// store content length here
 	dput(rq->request, "PROXY_HANDLE", proxy);
 
 	str = __s("%s %s?%s HTTP/1.1\r\n", (char *)dvalue(rq->request, "METHOD"), path, query);
@@ -882,7 +898,15 @@ static void *proxify(void *data)
 	task->handle = proxy_monitor;
 	task->access_time = rq->client->last_io;
 	// register event
-	antd_task_bind_event(task, rq->client->sock, 0, TASK_EVT_ON_READABLE | TASK_EVT_ON_WRITABLE);
+	
+	if(rq->client->ssl && (EQU(method, "POST") || ws_enable(rq->request)))
+	{
+		antd_task_bind_event(task, rq->client->sock, 0, TASK_EVT_ON_READABLE | TASK_EVT_ON_WRITABLE);
+	}
+	else
+	{
+		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);
 	return task;
 }
@@ -1127,12 +1151,12 @@ void *decode_post_request(void *data)
 	if (strstr(ctype, FORM_URL_ENCODE))
 	{
 		char *pquery = post_data_decode(rq->client, clen);
-		if(pquery)
+		if (pquery)
 		{
 			decode_url_request(pquery, request);
 			free(pquery);
 		}
-		else if(clen > 0)
+		else if (clen > 0)
 		{
 			task->handle = decode_post_request;
 		}
@@ -1156,7 +1180,7 @@ void *decode_post_request(void *data)
 			dput(request, key, strdup(pquery));
 			free(pquery);
 		}
-		else if(clen > 0)
+		else if (clen > 0)
 		{
 			task->handle = decode_post_request;
 		}

lib/scheduler.c

@@ -12,7 +12,6 @@
 
 #define MAX_VALIDITY_INTERVAL 60 // 1minute
 #define MAX_FIFO_NAME_SZ 255
-#define POLL_EVENT_TO 100 // ms
 
 // callback definition
 struct _antd_callback_t
@@ -210,6 +209,7 @@ static void destroy_queue(antd_queue_t q, int is_task)
             if (it->raw_ptr)
             {
                 free(it->raw_ptr);
+                it->raw_ptr = NULL;
             }
         }
         // then free the placeholder
@@ -637,7 +637,15 @@ static void task_polls_collect(bst_node_t* node, void** argv, int argc)
         }
     }
 }
-
+static void antd_deploy_task(bst_node_t* node, void** argv, int argc)
+{
+    UNUSED(argc);
+    if(!node || !node->data)
+        return;
+    antd_scheduler_t* sched = (antd_scheduler_t*) argv[0];
+    antd_task_t* task = node->data;
+    antd_task_schedule(sched, task);
+}
 static void task_event_collect(bst_node_t* node, void** argv, int argc)
 {
     UNUSED(argc);
@@ -690,6 +698,7 @@ void *antd_scheduler_wait(void *ptr)
     void *argv[3];
     antd_queue_t exec_list = NULL;
     bst_node_t* poll_list = NULL;
+    bst_node_t* scheduled_list = NULL;
     antd_queue_item_t it = NULL;
     antd_queue_item_t curr = NULL;
     antd_task_evt_item_t *eit = NULL;
@@ -745,9 +754,12 @@ void *antd_scheduler_wait(void *ptr)
                     {
                         // find the event
                         task_node = NULL;
+                        eit = NULL;
                         node = bst_find(poll_list,i);
                         if(node)
+                        {
                             eit = (antd_task_evt_item_t *)node->data;
+                        }
                         if(eit)
                         {
                             if( ((eit->flags & TASK_EVT_ON_READABLE) && (pfds[i].revents & POLLIN))
@@ -760,7 +772,8 @@ void *antd_scheduler_wait(void *ptr)
                                     scheduler->task_queue = bst_delete(scheduler->task_queue, eit->task->id);
                                 pthread_mutex_unlock(&scheduler->scheduler_lock);
                                 if(task_node)
-                                    antd_task_schedule(scheduler, eit->task);
+                                    scheduled_list = bst_insert(scheduled_list, eit->task->id, eit->task);
+                                    //antd_task_schedule(scheduler, eit->task);
                             }
                             else if( (pfds[i].revents & POLLERR) || (pfds[i].revents & POLLHUP) ) {
                                 // task is no longer available
@@ -774,7 +787,13 @@ void *antd_scheduler_wait(void *ptr)
                             }
                         }
                     }
-                    
+                    if(scheduled_list)
+                    {
+                        argv[0] = scheduler;
+                        bst_for_each(scheduled_list, antd_deploy_task, argv, 1);
+                        bst_free(scheduled_list);
+                        scheduled_list = NULL;
+                    }
                 }
                 free(pfds);
             }
@@ -801,10 +820,14 @@ int antd_scheduler_ok(antd_scheduler_t *scheduler)
 int antd_scheduler_next_id(antd_scheduler_t *sched, int input)
 {
     int id = input;
+    if(sched->id_allocator < 0)
+    {
+        sched->id_allocator = 0;
+    }
     pthread_mutex_lock(&sched->scheduler_lock);
     if (id == 0)
     {
-         sched->id_allocator++;
+        sched->id_allocator++;
         id = sched->id_allocator;
     }
 

lib/scheduler.h

@@ -11,6 +11,7 @@
 #define TASK_EVT_ON_READABLE 0x02
 #define TASK_EVT_ON_WRITABLE 0x04
 #define TASK_EVT_ON_TIMEOUT 0x08
+#define POLL_EVENT_TO 100 // ms
 
 typedef struct _antd_scheduler_t antd_scheduler_t;
 typedef struct _antd_callback_t antd_callback_t;