Enhance scheduler, cleanup lib

configure.ac

@@ -1,5 +1,5 @@
 # initialise autoconf and set up some basic information about the program we’re packaging
-AC_INIT([antd], [1.0.4b], [xsang.le@gmail.com])
+AC_INIT([antd], [1.0.5b], [xsang.le@gmail.com])
 
 # We’re going to use automake for this project
 AM_INIT_AUTOMAKE([subdir-objects])

http_server.c

@@ -1,4 +1,22 @@
+#include <sys/socket.h>
+#include <sys/select.h>
+#include <netinet/in.h>
+#include <dlfcn.h>
+#include <sys/stat.h>
+#include <unistd.h>
+
+#ifdef USE_OPENSSL
+#include <openssl/ssl.h>
+#include <openssl/err.h>
+#endif
+
 #include "http_server.h"
+#include "lib/handle.h"
+#include "plugin_manager.h"
+#include "lib/scheduler.h"
+#include "lib/utils.h"
+#include "lib/ini.h"
+#include "lib/base64.h"
 
 //define all basic mime here
 static mime_t _mimes[] = {
@@ -254,7 +272,6 @@ void *accept_request(void *data)
 	antd_request_t *rq = (antd_request_t *)data;
 
 	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
 	antd_client_t *client = (antd_client_t *)rq->client;
@@ -274,15 +291,6 @@ 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);
-		// retry it later
-		if(time(NULL) - client->last_wait > MAX_WAIT_S)
-		{
-			LOG("Read and write timeout, give up on %d\n", client->sock);
-			server_config.connection++;
-			unknow(rq->client);
-			return task;
-		}*/
 		task->handle = accept_request;
 		return task;
 	}
@@ -300,16 +308,6 @@ void *accept_request(void *data)
 			case SSL_ERROR_WANT_READ:
 			case SSL_ERROR_WANT_WRITE:
 			case SSL_ERROR_NONE:
-				//LOG("RETRY SSL %d\n", client->sock);
-				/*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;
 				return task;
 			default:
@@ -329,14 +327,6 @@ void *accept_request(void *data)
 	{
 		if (!FD_ISSET(client->sock, &read_flags))
 		{
-			/*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;
 			return task;
 		}
@@ -386,7 +376,6 @@ void *resolve_request(void *data)
 	char path[2 * BUFFLEN];
 	antd_request_t *rq = (antd_request_t *)data;
 	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;
 	char *rqp = NULL;
@@ -593,7 +582,6 @@ void *serve_file(void *data)
 {
 	antd_request_t *rq = (antd_request_t *)data;
 	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");
 
@@ -638,7 +626,7 @@ void *serve_file(void *data)
 #endif
 			dput(rhd.header, "Content-Length", strdup(ibuf));
 			gmtime_r(&st.st_ctime, &tm);
-			strftime(ibuf, 255, "%a, %d %b %Y %H:%M:%S GMT", &tm);
+			strftime(ibuf, 64, "%a, %d %b %Y %H:%M:%S GMT", &tm);
 			dput(rhd.header, "Last-Modified", strdup(ibuf));
 			dput(rhd.header, "Cache-Control", strdup("no-cache"));
 			antd_send_header(rq->client, &rhd);
@@ -827,8 +815,6 @@ void *decode_request_header(void *data)
 		free(host);
 	// header ok, now checkmethod
 	antd_task_t *task = antd_create_task(decode_request, (void *)rq, NULL,rq->client->last_io);
-
-	task->priority++;
 	return task;
 }
 
@@ -847,7 +833,6 @@ void *decode_request(void *data)
 		ws = 1;
 	method = (char *)dvalue(rq->request, "METHOD");
 	task = antd_create_task(NULL, (void *)rq, NULL, rq->client->last_io);
-	task->priority++;
 	if (strcmp(method, "GET") == 0 || strcmp(method, "HEAD") == 0)
 	{
 		//if(ctype) free(ctype);
@@ -891,7 +876,6 @@ void *decode_post_request(void *data)
 		clen = atoi(tmp);
 	char *method = (char *)dvalue(rq->request, "METHOD");
 	task = antd_create_task(NULL, (void *)rq, NULL, rq->client->last_io);
-	task->priority++;
 	task->type = HEAVY;
 	if (!method || strcmp(method, "POST") != 0)
 		return task;
@@ -1005,7 +989,6 @@ void *decode_multi_part_request(void *data, const char *ctype)
 	int len;
 	antd_request_t *rq = (antd_request_t *)data;
 	antd_task_t *task = antd_create_task(NULL, (void *)rq, NULL, rq->client->last_io);
-	task->priority++;
 	//dictionary dic = NULL;
 	boundary = strsep(&str_copy, "="); //discard first part
 	boundary = str_copy;
@@ -1039,7 +1022,6 @@ void *decode_multi_part_request_data(void *data)
 	char *token, *keytoken, *valtoken;
 	antd_request_t *rq = (antd_request_t *)data;
 	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_t dic = (dictionary_t)dvalue(rq->request, "REQUEST_DATA");
 	// search for content disposition:
@@ -1258,7 +1240,6 @@ void *execute_plugin(void *data, const char *pname)
 	char *error;
 	antd_request_t *rq = (antd_request_t *)data;
 	antd_task_t *task = antd_create_task(NULL, (void *)rq, NULL, rq->client->last_io);
-	task->priority++;
 	//LOG("Plugin name '%s'", pname);
 
 	//load the plugin
@@ -1297,7 +1278,6 @@ void *execute_plugin(void *data, const char *pname)
 	{
 		free(task);
 		task = antd_create_task(decode_post_request, (void *)rq, fn, rq->client->last_io);
-		task->priority++;
 	}
 	return task;
 }

http_server.h

@@ -1,16 +1,9 @@
 #ifndef HTTP_SERVER
 #define HTTP_SERVER
-#include <sys/types.h>
-#include <netinet/in.h>
-#include <arpa/inet.h>
-#include <sys/stat.h>
-#include <pthread.h>
-#include <signal.h>
-#include <sys/socket.h>
-#include <sys/select.h>
+
+#include <stdio.h>
+#include <string.h>
 #include "lib/handle.h"
-#include "lib/scheduler.h"
-#include "plugin_manager.h"
 
 #define PLUGIN_HANDLER	 "handle"
 #define WS_MAGIC_STRING	 "258EAFA5-E914-47DA-95CA-C5AB0DC85B11"

httpd.c

@@ -1,7 +1,18 @@
-
-#include <dirent.h>
+#include <pthread.h>
+#include <signal.h>
+#ifdef USE_OPENSSL
+#include <openssl/ssl.h>
+#include <openssl/err.h>
+#endif
+#include <netinet/in.h>
+#include <arpa/inet.h>
+#include <sys/socket.h>
 #include "http_server.h"
 #include "lib/ini.h"
+#include "lib/scheduler.h"
+#include "plugin_manager.h"
+#include "lib/utils.h"
+
 
 static  antd_scheduler_t scheduler;
 
@@ -238,6 +249,11 @@ static void* antd_monitor(port_config_t* pcnf)
 
 int main(int argc, char* argv[])
 {
+	pthread_t monitor_th;
+	// startup port
+	chain_t it;
+	port_config_t * pcnf;
+	int nlisten = 0;
 // load the config first
 	if(argc==1)
 		load_config(CONFIG_FILE);
@@ -268,12 +284,12 @@ int main(int argc, char* argv[])
 	// default to 4 workers
 	scheduler.validate_data = 1;
 	scheduler.destroy_data = finish_request;
-	antd_scheduler_init(&scheduler, conf->n_workers);
-	pthread_t monitor_th;
-	// startup port
-	chain_t it;
-	port_config_t * pcnf;
-	int nlisten = 0;
+	if(antd_scheduler_init(&scheduler, conf->n_workers) == -1)
+	{
+		ERROR("Unable to initialise scheduler. Exit");
+		stop_serve(0);
+		exit(1);
+	}
 	for_each_assoc(it, conf->ports)
 	{
 		pcnf = (port_config_t*)it->value;

lib/dbhelper.c

@@ -1,4 +1,8 @@
+#include <string.h>
+#include <stdio.h>
+
 #include "dbhelper.h"
+#include "utils.h"
 
 sqlite3 * database(const char* file)
 {

lib/dbhelper.h

@@ -1,7 +1,6 @@
 #ifndef DB_HELPER
 #define DB_HELPER
 #include <sqlite3.h>
-#include "utils.h"
 
 sqlite3 * database(const char*);
 typedef struct _dbfield

lib/dictionary.c

@@ -21,8 +21,9 @@ LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
 OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
 THE SOFTWARE.
 */
-#include "dictionary.h"
+#include <string.h>
 #include "utils.h"
+#include "dictionary.h"
 
 dictionary_t dict()
 {

lib/handle.c

@@ -1,4 +1,27 @@
 #include "handle.h"
+#include "utils.h"
+
+#include <time.h>
+#include <string.h>
+#include <sys/socket.h>
+#include <netinet/in.h>
+#include <arpa/inet.h>
+#include <errno.h>
+//open ssl
+#ifdef USE_OPENSSL
+#include <openssl/ssl.h>
+#include <openssl/err.h>
+#endif
+#ifdef USE_ZLIB
+#include <zlib.h>
+#endif
+#ifdef USE_DB
+#include "dbhelper.h"
+#endif
+
+#include <fcntl.h>
+#include <stdlib.h>
+
 #define HTML_TPL "<HTML><HEAD><TITLE>%s</TITLE></HEAD><BODY><h2>%s</h2></BODY></HTML>"
 
 static const char* S_100 =  "Continue";

lib/handle.h

@@ -1,26 +1,11 @@
 #ifndef HANDLE_H
 #define HANDLE_H
-#include <sys/socket.h>
-#include <netinet/in.h>
-#include <arpa/inet.h>
-#include <errno.h>
-//open ssl
-#ifdef USE_OPENSSL
-#include <openssl/ssl.h>
-#include <openssl/err.h>
-#endif
-#ifdef USE_ZLIB
-#include <zlib.h>
-#endif
-#ifdef USE_DB
-#include "dbhelper.h"
-#endif
 
-#include <fcntl.h>
-#include <stdlib.h>
-#include "dictionary.h"
+#include <time.h>
+
 #include "list.h"
-#include "ini.h"
+#include "dictionary.h"
+
 
 #define SERVER_NAME "Antd"
 #define IS_POST(method) (strcmp(method,"POST")== 0)

lib/list.c

@@ -21,7 +21,11 @@ LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
 OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
 THE SOFTWARE.
 */
+
+#include <string.h>
+
 #include "list.h"
+#include "utils.h"
 
 list_t list_init()
 {

lib/list.h

@@ -23,7 +23,7 @@ THE SOFTWARE.
 */
 #ifndef LIST_H
 #define LIST_H
-#include "utils.h"
+
 
 #define LIST_TYPE_ARRAY		0x5
 #define LIST_TYPE_POINTER	0x4

lib/plugin.h

@@ -1,11 +1,8 @@
 #ifndef PLUGIN_H
 #define PLUGIN_H
 
-#ifdef USE_DB
-#include "dbhelper.h"
-#endif
-#include "ws.h"
-#include "scheduler.h"
+#include "utils.h"
+#include "handle.h"
 
  
 

lib/scheduler.c

@@ -1,4 +1,8 @@
+#include <fcntl.h>
+#include <errno.h>
+#include <string.h>
 #include "scheduler.h"
+#include "utils.h"
 
 static void enqueue(antd_task_queue_t* q, antd_task_t* task)
 {
@@ -94,6 +98,11 @@ static void execute_callback(antd_scheduler_t* scheduler, antd_task_t* task)
         // call the first come call back
         task->handle = cb->handle;
         task->callback = task->callback->next;
+        task->priority = task->priority + 1;
+        if(task->priority > N_PRIORITY - 1)
+        {
+            task->priority = N_PRIORITY - 1;
+        }
         free(cb);
         antd_add_task(scheduler, task);
     }
@@ -150,7 +159,7 @@ static void* work(antd_worker_t* worker)
     init the main scheduler
 */
 
-void antd_scheduler_init(antd_scheduler_t* scheduler, int n)
+int antd_scheduler_init(antd_scheduler_t* scheduler, int n)
 {
     scheduler->n_workers = n;
     scheduler->status = 1;
@@ -163,13 +172,13 @@ void antd_scheduler_init(antd_scheduler_t* scheduler, int n)
     if (scheduler->scheduler_sem == SEM_FAILED)
     {
         ERROR("Cannot open semaphore for scheduler");
-        exit(-1);
+        return -1;
     }
     scheduler->worker_sem = sem_open("worker", O_CREAT, 0600, 0);
     if (!scheduler->worker_sem)
     {
         ERROR("Cannot open semaphore for workers");
-        exit(-1);
+        return -1;
     }
     // init lock 
     pthread_mutex_init(&scheduler->scheduler_lock,NULL);
@@ -183,7 +192,7 @@ void antd_scheduler_init(antd_scheduler_t* scheduler, int n)
         if(!scheduler->workers)
         {
             ERROR("Cannot allocate memory for worker");
-            exit(-1);
+            return -1;
         }
         for(int i = 0; i < scheduler->n_workers;i++)
         {
@@ -191,7 +200,8 @@ void antd_scheduler_init(antd_scheduler_t* scheduler, int n)
             scheduler->workers[i].manager = (void*)scheduler;
             if (pthread_create(&scheduler->workers[i].tid, NULL,(void *(*)(void *))work, (void*)&scheduler->workers[i]) != 0)
             {
-                perror("pthread_create: cannot create worker\n");
+                ERROR("pthread_create: cannot create worker: %s", strerror(errno));
+                return -1;
             }
             else
             {
@@ -200,6 +210,7 @@ void antd_scheduler_init(antd_scheduler_t* scheduler, int n)
         }
     }
     LOG("Antd scheduler initialized with %d worker", scheduler->n_workers);
+    return 0;
 }
 /* 
     destroy all pending task
@@ -227,7 +238,7 @@ antd_task_t* antd_create_task(void* (*handle)(void*), void *data, void* (*callba
     task->data = data;
     task->handle = handle;
     task->callback = callback_of(callback);
-    task->priority = NORMAL_PRIORITY;
+    task->priority = HIGH_PRIORITY;
     task->type = HEAVY;
     //task->type = LIGHT;
     task->access_time = atime;
@@ -258,6 +269,7 @@ void antd_execute_task(antd_scheduler_t* scheduler, antd_task_item_t taski)
     if(!taski)
         return;
     // execute the task
+    LOG("Execute task with priority: %d", taski->task->priority);
     void *ret = (*(taski->task->handle))(taski->task->data);
     // check the return data if it is a new task
     if(!ret)
@@ -289,6 +301,11 @@ void antd_execute_task(antd_scheduler_t* scheduler, antd_task_item_t taski)
         }
         else
         {
+            rtask->priority = taski->task->priority + 1;
+            if(rtask->priority > N_PRIORITY - 1)
+            {
+                rtask->priority = N_PRIORITY - 1;
+            }
             antd_add_task(scheduler, rtask);
             free(taski->task);
             free(taski);
@@ -324,10 +341,10 @@ int antd_task_schedule(antd_scheduler_t* scheduler)
     }
     // has the task now
     // validate the task
-    if(scheduler->validate_data && difftime( time(NULL), it->task->access_time) > MAX_VALIDITY_INTERVAL)
+    if(scheduler->validate_data && difftime( time(NULL), it->task->access_time) > MAX_VALIDITY_INTERVAL && it->task->priority == N_PRIORITY - 1)
     {
         // data task is not valid
-        LOG("Task data is not valid, task will be killed");
+        // LOG("Task is no longer valid and will be killed");
         if(scheduler->destroy_data)
             scheduler->destroy_data(it->task->data);
         if(it->task->callback)

lib/scheduler.h

@@ -1,10 +1,9 @@
 #ifndef ANT_SCHEDULER
 #define ANT_SCHEDULER
 
-#include "utils.h"
 #include <pthread.h>
 #include <semaphore.h>
-#include <fcntl.h>
+#include <stdint.h>
 #define N_PRIORITY 10
 #define NORMAL_PRIORITY ((int)((N_PRIORITY - 1) / 2))
 #define LOW_PRIORITY (N_PRIORITY - 1)
@@ -98,7 +97,7 @@ typedef struct
 /*
     init the main scheduler
 */
-void antd_scheduler_init(antd_scheduler_t *, int);
+int antd_scheduler_init(antd_scheduler_t *, int);
 /*
     destroy all pending task
 */

lib/utils.c

@@ -23,6 +23,29 @@ THE SOFTWARE.
 */
 #include "utils.h"
 
+#include <dirent.h>
+#include <unistd.h>
+#include <sys/types.h>
+#include <sys/stat.h>
+#include <stdlib.h>
+#include <stdarg.h>
+#include <errno.h>
+#include <string.h>
+#include <stdio.h>
+#include <stdint.h>
+#include <ctype.h>
+
+#ifdef USE_OPENSSL
+#include <openssl/sha.h>
+#else
+#include "sha1.h"
+#endif
+
+
+#include "dictionary.h"
+#include <time.h>
+
+
 
 void error_log(const char* fmt, ...)
 {

lib/utils.h

@@ -23,27 +23,12 @@ THE SOFTWARE.
 */
 #ifndef UTILS_H
 #define UTILS_H
-#include <stdio.h>
-#include <stdarg.h>
-#include <unistd.h>
-#include <strings.h>
-#include <string.h>
-#include <stdlib.h>
-#include <dirent.h>
-#include <sys/stat.h>
-#include <sys/types.h>
-#include <ctype.h>
+
 #include <regex.h>
-#include <time.h>
-#include <stdint.h>
-#include <errno.h>
 #include <syslog.h>
-#ifdef USE_OPENSSL
-#include <openssl/sha.h>
-#else
-#include "sha1.h"
-#endif
-#include "base64.h"
+#include <stdint.h>
+#include <stdlib.h>
+
 #include "dictionary.h"
 
 #define LEFTROTATE(x, c) (((x) << (c)) | ((x) >> (32 - (c))))

lib/ws.c

@@ -1,7 +1,20 @@
+
+#include <stdio.h>
+#include <unistd.h>
+#include <string.h>
+#include <errno.h>
+#include<netdb.h> //hostent
 #ifdef USE_OPENSSL
 #include <openssl/ssl.h>
 #include <openssl/err.h>
 #endif
+#include <sys/time.h>
+#include <ifaddrs.h>
+#include <arpa/inet.h>
+
+#include "utils.h"
+#include "handle.h"
+
 #include "ws.h"
 static void ws_gen_mask_key(ws_msg_header_t * header)
 {

lib/ws.h

@@ -1,19 +1,10 @@
 #ifndef WS_H
 #define WS_H
-#include <resolv.h>
-#include <errno.h>
-#include <ifaddrs.h>
-#include <sys/socket.h>
-#include <stdio.h>
-#include <stdint.h>
-#include<netdb.h> //hostent
-#include <netinet/in.h>
-#include <arpa/inet.h>
-#include <libgen.h>
-#include <sys/time.h>
 
-#include "utils.h"
+#include <stdint.h>
+
 #include "handle.h"
+
 #define CONN_TIME_OUT_S 3
 #define BITV(v,i)  ((v & (1 << i)) >> i) 
 #define WS_TEXT	0x1

plugin_manager.c

@@ -1,4 +1,9 @@
+#include <dlfcn.h>
+#include <string.h>
 #include "plugin_manager.h"
+#include "lib/utils.h"
+#include "lib/handle.h"
+#include "http_server.h"
 /**
  * Plugin table to store the loaded plugin
  */

plugin_manager.h

@@ -1,9 +1,5 @@
 #ifndef PLUGIN_MANAGER_H
 #define PLUGIN_MANAGER_H
-#include <dlfcn.h>
-#include "lib/utils.h"
-#include "lib/handle.h"
-#include "http_server.h"
 struct plugin_entry { 
     struct plugin_entry *next; 
     char *pname;