add statistic log to scheduler, good for debug

antd-config.ini

@@ -8,6 +8,7 @@ database=/opt/www/database/
 ; tmp dir
 tmpdir=/opt/www/tmp/
 ; max concurent connection
+statistic_fifo=/opt/www/tmp/antd_stat
 maxcon=200
 ; server backlocg
 backlog=5000

http_server.c

@@ -4,10 +4,15 @@
 #include <dlfcn.h>
 #include <sys/stat.h>
 #include <unistd.h>
+#include <string.h>
+#include <errno.h>
 
 #ifdef USE_OPENSSL
 #include <openssl/ssl.h>
 #include <openssl/err.h>
+#include <openssl/sha.h>
+#else
+#include "sha1.h"
 #endif
 
 #include "http_server.h"
@@ -62,6 +67,8 @@ void destroy_config()
 		list_free(&server_config.gzip_types);
 	if (server_config.mimes)
 		freedict(server_config.mimes);
+	if (server_config.stat_fifo_path)
+		free(server_config.stat_fifo_path);
 	if (server_config.ports)
 	{
 		chain_t it;
@@ -103,16 +110,22 @@ static int config_handler(void *conf, const char *section, const char *name,
 	}
 	else if (MATCH("SERVER", "plugins_ext"))
 	{
+		if(pconfig->plugins_ext)
+			free(pconfig->plugins_ext);
 		pconfig->plugins_ext = strdup(value);
 	}
 	else if (MATCH("SERVER", "database"))
 	{
+		if(pconfig->db_path)
+			free(pconfig->db_path);
 		pconfig->db_path = strdup(value);
 		if (stat(pconfig->db_path, &st) == -1)
 			mkdirp(pconfig->db_path, 0755);
 	}
 	else if (MATCH("SERVER", "tmpdir"))
 	{
+		if(pconfig->tmpdir)
+			free(pconfig->tmpdir);
 		pconfig->tmpdir = strdup(value);
 		if (stat(pconfig->tmpdir, &st) == -1)
 			mkdirp(pconfig->tmpdir, 0755);
@@ -121,6 +134,12 @@ static int config_handler(void *conf, const char *section, const char *name,
 			removeAll(pconfig->tmpdir, 0);
 		}
 	}
+	else if (MATCH("SERVER", "statistic_fifo"))
+	{
+		if(pconfig->stat_fifo_path)
+			free(pconfig->stat_fifo_path);
+		pconfig->stat_fifo_path = strdup(value);
+	}
 	else if (MATCH("SERVER", "max_upload_size"))
 	{
 		pconfig->max_upload_size = atoi(value);
@@ -150,14 +169,20 @@ static int config_handler(void *conf, const char *section, const char *name,
 #ifdef USE_OPENSSL
 	else if (MATCH("SERVER", "ssl.cert"))
 	{
+		if(pconfig->sslcert)
+			free(pconfig->sslcert);
 		pconfig->sslcert = strdup(value);
 	}
 	else if (MATCH("SERVER", "ssl.key"))
 	{
+		if(pconfig->sslkey)
+			free(pconfig->sslkey);
 		pconfig->sslkey = strdup(value);
 	}
 	else if (MATCH("SERVER", "ssl.cipher"))
 	{
+		if(pconfig->ssl_cipher)
+			free(pconfig->ssl_cipher);
 		pconfig->ssl_cipher = strdup(value);
 	}
 #endif
@@ -220,11 +245,12 @@ static int config_handler(void *conf, const char *section, const char *name,
 void load_config(const char *file)
 {
 	server_config.ports = dict();
-	server_config.plugins_dir = "plugins/";
+	server_config.plugins_dir = strdup("plugins/");
-	server_config.plugins_ext = ".dylib";
+	server_config.plugins_ext = strdup(".dylib");
-	server_config.db_path = "databases/";
+	server_config.db_path = strdup("databases/");
 	//server_config.htdocs = "htdocs/";
-	server_config.tmpdir = "tmp/";
+	server_config.tmpdir = strdup("tmp/");
+	server_config.stat_fifo_path = strdup("/var/run/antd_stat");
 	server_config.n_workers = 4;
 	server_config.backlog = 1000;
 	server_config.handlers = dict();
@@ -233,8 +259,8 @@ void load_config(const char *file)
 	server_config.connection = 0;
 	server_config.mimes = dict();
 	server_config.enable_ssl = 0;
-	server_config.sslcert = "cert.pem";
+	server_config.sslcert = strdup("cert.pem");
-	server_config.sslkey = "key.pem";
+	server_config.sslkey = strdup("key.pem");
 	server_config.ssl_cipher = NULL;
 	server_config.gzip_enable = 0;
 	server_config.gzip_types = NULL;
@@ -297,7 +323,7 @@ void *accept_request(void *data)
 	// perform the ssl handshake if enabled
 #ifdef USE_OPENSSL
 	int ret = -1, stat;
-	if (client->ssl && client->status == 0)
+	if (client->ssl && client->state == ANTD_CLIENT_ACCEPT)
 	{
 		//LOG("Atttempt %d\n", client->attempt);
 		if (SSL_accept((SSL *)client->ssl) == -1)
@@ -318,7 +344,7 @@ void *accept_request(void *data)
 				return task;
 			}
 		}
-		client->status = 1;
+		client->state = ANTD_CLIENT_HANDSHAKE;
 		task->handle = accept_request;
 		//LOG("Handshake finish for %d\n", client->sock);
 		return task;
@@ -334,6 +360,7 @@ void *accept_request(void *data)
 #endif
 	//LOG("Ready for reading %d\n", client->sock);
 	//server_config.connection++;
+	client->state = ANTD_CLIENT_PROTO_CHECK;
 	read_buf(rq->client, buf, sizeof(buf));
 	line = buf;
 	LOG("Request (%d): %s", rq->client->sock, line);
@@ -381,6 +408,7 @@ void *resolve_request(void *data)
 	char *newurl = NULL;
 	char *rqp = NULL;
 	char *oldrqp = NULL;
+	rq->client->state = ANTD_CLIENT_RESOLVE_REQUEST;
 	htdocs(rq, path);
 	strcat(path, url);
 	//LOG("Path is : %s", path);
@@ -585,7 +613,7 @@ void *serve_file(void *data)
 	antd_task_t *task = antd_create_task(NULL, (void *)rq, NULL, rq->client->last_io);
 	char *path = (char *)dvalue(rq->request, "ABS_RESOURCE_PATH");
 	char *mime_type = (char *)dvalue(rq->request, "RESOURCE_MIME");
-
+	rq->client->state = ANTD_CLIENT_SERVE_FILE;
 	struct stat st;
 	int s = stat(path, &st);
 
@@ -721,6 +749,7 @@ char *apply_rules(dictionary_t rules, const char *host, char *url)
 void *decode_request_header(void *data)
 {
 	antd_request_t *rq = (antd_request_t *)data;
+	rq->client->state = ANTD_CLIENT_HEADER_DECODE;
 	dictionary_t cookie = NULL;
 	char *line;
 	char *token;
@@ -1256,7 +1285,7 @@ void *execute_plugin(void *data, const char *pname)
 	antd_request_t *rq = (antd_request_t *)data;
 	antd_task_t *task = antd_create_task(NULL, (void *)rq, NULL, rq->client->last_io);
 	//LOG("Plugin name '%s'", pname);
-
+	rq->client->state = ANTD_CLIENT_PLUGIN_EXEC;
 	//load the plugin
 	if ((plugin = plugin_lookup((char *)pname)) == NULL)
 	{

httpd.c

@@ -4,6 +4,7 @@
 #include <openssl/ssl.h>
 #include <openssl/err.h>
 #endif
+#include <unistd.h>
 #include <netinet/in.h>
 #include <arpa/inet.h>
 #include <sys/socket.h>
@@ -13,6 +14,9 @@
 #include "plugin_manager.h"
 #include "lib/utils.h"
 
+#define SEND_STAT(fd, buff, ret, ...)       \
+	snprintf(buff, BUFFLEN, ##__VA_ARGS__); \
+	ret = write(fd, buff, strlen(buff));
 
 static antd_scheduler_t scheduler;
 
@@ -30,7 +34,6 @@ static void init_openssl()
 	OpenSSL_add_ssl_algorithms();
 }
 
-
 static SSL_CTX *create_context()
 {
 	const SSL_METHOD *method;
@@ -39,7 +42,8 @@ static SSL_CTX *create_context()
 	method = SSLv23_server_method();
 
 	ctx = SSL_CTX_new(method);
-    if (!ctx) {
+	if (!ctx)
+	{
 		ERROR("Unable to create SSL context");
 		ERR_print_errors_fp(stderr);
 		exit(EXIT_FAILURE);
@@ -50,8 +54,7 @@ static SSL_CTX *create_context()
 #if OPENSSL_VERSION_NUMBER >= 0x10002000L
 static unsigned char antd_protocols[] = {
 	//TODO: add support to HTTP/2 protocol: 2,'h', '2',
-    8, 'h', 't', 't', 'p', '/', '1', '.', '1'
+	8, 'h', 't', 't', 'p', '/', '1', '.', '1'};
-};
 static int alpn_advertise_protos_cb(SSL *ssl, const unsigned char **out, unsigned int *outlen, void *arg)
 {
 	UNUSED(ssl);
@@ -103,18 +106,21 @@ static void configure_context(SSL_CTX *ctx)
 	/* Set the key and cert */
 	/* use the full chain bundle of certificate */
 	//if (SSL_CTX_use_certificate_file(ctx, server_config->sslcert, SSL_FILETYPE_PEM) <= 0) {
-	if (SSL_CTX_use_certificate_chain_file(ctx, cnf->sslcert) <= 0) {
+	if (SSL_CTX_use_certificate_chain_file(ctx, cnf->sslcert) <= 0)
+	{
 		ERROR("Fail to read SSL certificate chain file: %s", cnf->sslcert);
 		ERR_print_errors_fp(stderr);
 		exit(EXIT_FAILURE);
 	}
 
-    if (SSL_CTX_use_PrivateKey_file(ctx, cnf->sslkey, SSL_FILETYPE_PEM) <= 0 ) {
+	if (SSL_CTX_use_PrivateKey_file(ctx, cnf->sslkey, SSL_FILETYPE_PEM) <= 0)
+	{
 		ERROR("Fail to read SSL private file: %s", cnf->sslkey);
 		ERR_print_errors_fp(stderr);
 		exit(EXIT_FAILURE);
 	}
-	if (!SSL_CTX_check_private_key(ctx)) {
+	if (!SSL_CTX_check_private_key(ctx))
+	{
 		ERROR("Failed to validate SSL certificate");
 		ERR_print_errors_fp(stderr);
 		exit(EXIT_FAILURE);
@@ -127,8 +133,8 @@ static void configure_context(SSL_CTX *ctx)
 
 #endif
 
-
+static void stop_serve(int dummy)
-static void stop_serve(int dummy) {
+{
 	UNUSED(dummy);
 	// close log server
 	closelog();
@@ -213,12 +219,14 @@ static void* antd_monitor(port_config_t* pcnf)
 				client->sock = client_sock;
 				time(&client->last_io);
 				client->ssl = NULL;
+				client->state = ANTD_CLIENT_ACCEPT;
+				client->z_status = 0;
 #ifdef USE_OPENSSL
-				client->status = 0;
 				if (pcnf->usessl == 1)
 				{
 					client->ssl = (void *)SSL_new(ctx);
-					if(!client->ssl) continue;
+					if (!client->ssl)
+						continue;
 					SSL_set_fd((SSL *)client->ssl, client->sock);
 					// this can be used in the protocol select callback to
 					// set the protocol selected by the server
@@ -247,6 +255,54 @@ static void* antd_monitor(port_config_t* pcnf)
 	return NULL;
 }
 
+static void client_statistic(int fd, void *user_data)
+{
+	antd_request_t *request = (antd_request_t *)user_data;
+	chain_t it, it1;
+	dictionary_t tmp;
+	int ret;
+	char buff[BUFFLEN];
+	if (request == NULL)
+	{
+		SEND_STAT(fd, buff, ret, "Data is null\n");
+		return;
+	}
+	// send client general infomation
+	SEND_STAT(fd, buff, ret, "Client id: %d\n", request->client->sock);
+	SEND_STAT(fd, buff, ret, "Last IO: %lu\n", (unsigned long)request->client->last_io);
+	SEND_STAT(fd, buff, ret, "Current state: %d\n", request->client->state);
+	SEND_STAT(fd, buff, ret, "z_level: %d\n", request->client->z_level);
+	if (request->client->ssl)
+	{
+		SEND_STAT(fd, buff, ret, "SSL is enabled\n");
+	}
+	// send client request detail
+	if (request->request)
+	{
+		for_each_assoc(it, request->request)
+		{
+			if (strcmp(it->key, "REQUEST_HEADER") == 0 ||
+				strcmp(it->key, "REQUEST_DATA") == 0 ||
+				strcmp(it->key, "COOKIE") == 0)
+			{
+				tmp = (dictionary_t)it->value;
+				if (tmp)
+				{
+					for_each_assoc(it1, tmp)
+					{
+						SEND_STAT(fd, buff, ret, "%s: %s\n", it1->key, (char *)it1->value);
+					}
+				}
+			}
+			else
+			{
+				SEND_STAT(fd, buff, ret, "%s: %s\n", it->key, (char *)it->value);
+			}
+		}
+	}
+	UNUSED(ret);
+}
+
 int main(int argc, char *argv[])
 {
 	pthread_t monitor_th;
@@ -284,6 +340,8 @@ int main(int argc, char* argv[])
 	// default to 4 workers
 	scheduler.validate_data = 1;
 	scheduler.destroy_data = finish_request;
+	strncpy(scheduler.stat_fifo, conf->stat_fifo_path, MAX_FIFO_NAME_SZ);
+	scheduler.stat_data_cb = client_statistic;
 	if (antd_scheduler_init(&scheduler, conf->n_workers) == -1)
 	{
 		ERROR("Unable to initialise scheduler. Exit");

lib/handle.c

@@ -6,6 +6,7 @@
 #include <sys/socket.h>
 #include <netinet/in.h>
 #include <arpa/inet.h>
+#include <stdio.h>
 #include <errno.h>
 //open ssl
 #ifdef USE_OPENSSL
@@ -295,7 +296,7 @@ void antd_send_header(void *cl, antd_response_header_t *res)
 						}
 						else
 						{
-							client->status = Z_NO_FLUSH;
+							client->z_status = Z_NO_FLUSH;
 							dput(res->header, "Content-Encoding", strdup("gzip"));
 						}
 					}
@@ -309,7 +310,7 @@ void antd_send_header(void *cl, antd_response_header_t *res)
 						}
 						else
 						{
-							client->status = Z_NO_FLUSH;
+							client->z_status = Z_NO_FLUSH;
 							dput(res->header, "Content-Encoding", strdup("deflate"));
 						}
 					}
@@ -384,7 +385,7 @@ int antd_send(void *src, const void *data_in, int len_in)
 		{
 			zstream->avail_out = BUFFLEN;
 			zstream->next_out = buf;
-			if (deflate(zstream, source->status) == Z_STREAM_ERROR)
+			if (deflate(zstream, source->z_status) == Z_STREAM_ERROR)
 			{
 				source->z_level = current_zlevel;
 				data = NULL;
@@ -729,9 +730,9 @@ int antd_close(void *src)
 	//TODO: send finish data to the socket before quit
 	if (source->zstream)
 	{
-		if (source->status == Z_NO_FLUSH && source->z_level != ANTD_CNONE)
+		if (source->z_status == Z_NO_FLUSH && source->z_level != ANTD_CNONE)
 		{
-			source->status = Z_FINISH;
+			source->z_status = Z_FINISH;
 			antd_send(source, "", 0);
 		}
 		deflateEnd(source->zstream);

lib/handle.h

@@ -6,7 +6,6 @@
 #include "list.h"
 #include "dictionary.h"
 
-
 #define SERVER_NAME "Antd"
 #define IS_POST(method) (strcmp(method, "POST") == 0)
 #define IS_GET(method) (strcmp(method, "GET") == 0)
@@ -19,12 +18,25 @@
 #define FORM_MULTI_PART "multipart/form-data"
 #define MAX_IO_WAIT_TIME 5 // second
 
+#define ANTD_CLIENT_ACCEPT 0x0
+#define ANTD_CLIENT_HANDSHAKE 0x1
+#define ANTD_CLIENT_HEADER_DECODE 0x2
+#define ANTD_CLIENT_PLUGIN_EXEC 0x3
+#define ANTD_CLIENT_PROTO_CHECK 0x4
+#define ANTD_CLIENT_RESOLVE_REQUEST 0x5
+#define ANTD_CLIENT_SERVE_FILE 0x6
 
-typedef enum {ANTD_CGZ, ANTD_CDEFL, ANTD_CNONE} antd_compress_t;
+typedef enum
+{
+    ANTD_CGZ,
+    ANTD_CDEFL,
+    ANTD_CNONE
+} antd_compress_t;
 
 //extern config_t server_config;
 
-typedef struct {
+typedef struct
+{
     unsigned int port;
     int usessl;
     char *htdocs;
@@ -32,17 +44,20 @@ typedef struct {
     dictionary_t rules;
 } port_config_t;
 
-typedef struct{
+typedef struct
+{
     int sock;
     void *ssl;
-    int status;
+    int state;
     time_t last_io;
     // compress
     antd_compress_t z_level;
     void *zstream;
+    int z_status;
 } antd_client_t;
 
-typedef struct {
+typedef struct
+{
     antd_client_t *client;
     dictionary_t request;
 } antd_request_t;
@@ -55,15 +70,15 @@ typedef struct
 
 } antd_response_header_t;
 
-
+typedef struct
-
+{
-typedef struct  { 
     //int port;
     char *plugins_dir;
     char *plugins_ext;
     char *db_path;
     //char* htdocs;
     char *tmpdir;
+    char *stat_fifo_path;
     dictionary_t handlers;
     int backlog;
     int maxcon;
@@ -82,7 +97,8 @@ typedef struct  {
     // #endif
 } config_t;
 
-typedef struct  { 
+typedef struct
+{
     char name[128];
     char dbpath[512];
     char tmpdir[512];
@@ -90,7 +106,6 @@ typedef struct  {
     int raw_body;
 } plugin_header_t;
 
-
 int __attribute__((weak)) require_plugin(const char *);
 void __attribute__((weak)) htdocs(antd_request_t *rq, char *dest);
 void __attribute__((weak)) dbdir(char *dest);

lib/scheduler.c

@@ -1,9 +1,24 @@
 #include <fcntl.h>
 #include <errno.h>
 #include <string.h>
+#include <stdio.h>
+#include <sys/types.h>
+#include <sys/stat.h>
+#include <poll.h>
+#include <unistd.h>
 #include "scheduler.h"
 #include "utils.h"
 
+static void set_nonblock(int fd)
+{
+    int flags;
+    flags = fcntl(fd, F_GETFL, 0);
+    if(flags == -1)
+    {
+        ERROR("Unable to set flag");
+    }
+    fcntl(fd, F_SETFL, flags | O_NONBLOCK);
+}
 static void enqueue(antd_task_queue_t *q, antd_task_t *task)
 {
     antd_task_item_t it = *q;
@@ -22,7 +37,6 @@ static void enqueue(antd_task_queue_t* q, antd_task_t* task)
     }
 }
 
-
 static void stop(antd_scheduler_t *scheduler)
 {
     scheduler->status = 0;
@@ -34,7 +48,9 @@ static void stop(antd_scheduler_t* scheduler)
     for (int i = 0; i < scheduler->n_workers; i++)
         if (scheduler->workers[i].id != -1)
             pthread_join(scheduler->workers[i].tid, NULL);
-    if(scheduler->workers) free(scheduler->workers);
+    if (scheduler->workers)
+        free(scheduler->workers);
+    (void)pthread_join(scheduler->stat_tid, NULL);
     // destroy all the mutex
     pthread_mutex_destroy(&scheduler->scheduler_lock);
     pthread_mutex_destroy(&scheduler->worker_lock);
@@ -56,7 +72,6 @@ static antd_task_item_t dequeue(antd_task_queue_t* q)
     return it;
 }
 
-
 antd_callback_t *callback_of(void *(*callback)(void *))
 {
     antd_callback_t *cb = NULL;
@@ -86,7 +101,8 @@ static void enqueue_callback(antd_callback_t* cb, antd_callback_t* el)
     antd_callback_t *it = cb;
     while (it && it->next != NULL)
         it = it->next;
-    if(!it) return; // this should not happend
+    if (!it)
+        return; // this should not happend
     it->next = el;
 }
 
@@ -119,8 +135,10 @@ static void destroy_queue(antd_task_queue_t q)
     while (it)
     {
         // first free the task
-        if(it->task && it->task->callback) free_callback(it->task->callback);
+        if (it->task && it->task->callback)
-        if(it->task) free(it->task);
+            free_callback(it->task->callback);
+        if (it->task)
+            free(it->task);
         // then free the placeholder
         curr = it;
         it = it->next;
@@ -149,7 +167,86 @@ static void* work(antd_worker_t* worker)
             //LOG("task executed by worker %d\n", worker->id);
             antd_execute_task(scheduler, it);
         }
+    }
+    return NULL;
+}
 
+static void *statistic(antd_scheduler_t *scheduler)
+{
+    struct pollfd fdp;
+    int ret;
+    char buffer[MAX_FIFO_NAME_SZ];
+    antd_task_item_t it;
+    while (scheduler->status)
+    {
+        if (scheduler->stat_fd == -1)
+        {
+            scheduler->stat_fd = open(scheduler->stat_fifo, O_RDWR);
+            if (scheduler->stat_fd == -1)
+            {
+                ERROR("Unable to open FIFO %s: %s", scheduler->stat_fifo, strerror(errno));
+                return NULL;
+            }
+        }
+        fdp.fd = scheduler->stat_fd;
+        fdp.events = POLLOUT;
+        // poll the fd in blocking mode
+        ret = poll(&fdp, 1, -1);
+
+        if (ret > 0 && (fdp.revents & POLLOUT) && scheduler->pending_task > 0)
+        {
+            pthread_mutex_lock(&scheduler->scheduler_lock);
+            // write statistic data
+            snprintf(buffer, MAX_FIFO_NAME_SZ, "Pending task: %d. Detail:\n", scheduler->pending_task);
+            ret = write(scheduler->stat_fd, buffer, strlen(buffer));
+
+            for (int i = 0; i < N_PRIORITY; i++)
+            {
+                snprintf(buffer, MAX_FIFO_NAME_SZ, "#### PRIORITY: %d\n", i);
+                ret = write(scheduler->stat_fd, buffer, strlen(buffer));
+
+                it = scheduler->task_queue[i];
+                while (it)
+                {
+                    // send statistic on task data
+                    snprintf(buffer, MAX_FIFO_NAME_SZ, "---- Task created at: %lu ----\n", it->task->stamp);
+                    ret = write(scheduler->stat_fd, buffer, strlen(buffer));
+
+                    // send statistic on task data
+                    snprintf(buffer, MAX_FIFO_NAME_SZ, "Access time: %lu\nn", (unsigned long)it->task->access_time);
+                    ret = write(scheduler->stat_fd, buffer, strlen(buffer));
+
+                    snprintf(buffer, MAX_FIFO_NAME_SZ, "Current time: %lu\n", (unsigned long)time(NULL));
+                    ret = write(scheduler->stat_fd, buffer, strlen(buffer));
+
+                    snprintf(buffer, MAX_FIFO_NAME_SZ, "Task type: %d\n", it->task->type);
+                    ret = write(scheduler->stat_fd, buffer, strlen(buffer));
+
+                    if (it->task->handle)
+                    {
+                        snprintf(buffer, MAX_FIFO_NAME_SZ, "Has handle: yes\n");
+                        ret = write(scheduler->stat_fd, buffer, strlen(buffer));
+                    }
+
+                    if (it->task->callback)
+                    {
+                        snprintf(buffer, MAX_FIFO_NAME_SZ, "Has callback: yes\n");
+                        ret = write(scheduler->stat_fd, buffer, strlen(buffer));
+                    }
+
+                    // now print all task data statistic
+                    if (scheduler->stat_data_cb)
+                    {
+                        scheduler->stat_data_cb(scheduler->stat_fd, it->task->data);
+                    }
+                    it = it->next;
+                }
+            }
+            pthread_mutex_unlock(&scheduler->scheduler_lock);
+            ret = close(scheduler->stat_fd);
+            scheduler->stat_fd = -1;
+            usleep(5000);
+        }
     }
     return NULL;
 }
@@ -167,6 +264,9 @@ int antd_scheduler_init(antd_scheduler_t* scheduler, int n)
     scheduler->pending_task = 0;
     scheduler->validate_data = 0;
     scheduler->destroy_data = NULL;
+    scheduler->stat_fd = -1;
+    //scheduler->stat_data_cb = NULL;
+    //memset(scheduler->stat_fifo, 0, MAX_FIFO_NAME_SZ);
     // init semaphore
     scheduler->scheduler_sem = sem_open("scheduler", O_CREAT, 0600, 0);
     if (scheduler->scheduler_sem == SEM_FAILED)
@@ -184,7 +284,8 @@ int antd_scheduler_init(antd_scheduler_t* scheduler, int n)
     pthread_mutex_init(&scheduler->scheduler_lock, NULL);
     pthread_mutex_init(&scheduler->worker_lock, NULL);
     pthread_mutex_init(&scheduler->pending_lock, NULL);
-    for(int i = 0; i < N_PRIORITY; i++) scheduler->task_queue[i] = NULL;
+    for (int i = 0; i < N_PRIORITY; i++)
+        scheduler->task_queue[i] = NULL;
     // create scheduler.workers
     if (n > 0)
     {
@@ -209,6 +310,34 @@ int antd_scheduler_init(antd_scheduler_t* scheduler, int n)
             }
         }
     }
+    // delete the fifo if any
+    if (scheduler->stat_fifo[0] != '\0')
+    {
+        LOG("Statistic fifo at: %s", scheduler->stat_fifo);
+        (void)remove(scheduler->stat_fifo);
+        // create the fifo file
+        if (mkfifo(scheduler->stat_fifo, 0666) == -1)
+        {
+            ERROR("Unable to create statictis FIFO %s: %s", scheduler->stat_fifo, strerror(errno));
+        }
+        else
+        {
+            // open the fifo in write mode
+            scheduler->stat_fd = open(scheduler->stat_fifo, O_RDWR);
+            if (scheduler->stat_fd == -1)
+            {
+                ERROR("Unable to open FIFO %s: %s", scheduler->stat_fifo, strerror(errno));
+            }
+            else
+            {
+                set_nonblock(scheduler->stat_fd);
+                if (pthread_create(&scheduler->stat_tid, NULL, (void *(*)(void *))statistic, scheduler) != 0)
+                {
+                    ERROR("pthread_create: cannot create statistic thread: %s", strerror(errno));
+                }
+            }
+        }
+    }
     LOG("Antd scheduler initialized with %d worker", scheduler->n_workers);
     return 0;
 }
@@ -263,7 +392,6 @@ void antd_add_task(antd_scheduler_t* scheduler, antd_task_t* task)
     sem_post(scheduler->scheduler_sem);
 }
 
-
 void antd_execute_task(antd_scheduler_t *scheduler, antd_task_item_t taski)
 {
     if (!taski)
@@ -310,9 +438,6 @@ void antd_execute_task(antd_scheduler_t* scheduler, antd_task_item_t taski)
             free(taski);
         }
     }
-    pthread_mutex_lock(&scheduler->pending_lock);
-    scheduler->pending_task--;
-    pthread_mutex_unlock(&scheduler->pending_lock);
 }
 
 int antd_scheduler_busy(antd_scheduler_t *scheduler)
@@ -338,12 +463,15 @@ int antd_task_schedule(antd_scheduler_t* scheduler)
     {
         return 0;
     }
+    pthread_mutex_lock(&scheduler->pending_lock);
+    scheduler->pending_task--;
+    pthread_mutex_unlock(&scheduler->pending_lock);
     // has the task now
     // validate the task
     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 is no longer valid and 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

@@ -9,6 +9,7 @@
 #define LOW_PRIORITY (N_PRIORITY - 1)
 #define HIGH_PRIORITY 0
 #define MAX_VALIDITY_INTERVAL 20 // 10 s for task validity
+#define MAX_FIFO_NAME_SZ 255
 typedef enum
 {
     LIGHT,
@@ -92,6 +93,13 @@ typedef struct
     */
     void* (*destroy_data)(void*);
     int validate_data;
+    /**
+     * statistic infomation
+    */
+    char stat_fifo[MAX_FIFO_NAME_SZ];
+    int stat_fd;
+    pthread_t stat_tid;
+    void (*stat_data_cb)(int, void *);
 } antd_scheduler_t;
 
 /*

lib/ws.c

@@ -415,7 +415,7 @@ int ws_client_connect(ws_client_t* wsclient, port_config_t pcnf)
 	}
 	// will be free
 	wsclient->antdsock->sock = sock;
-	wsclient->antdsock->status = 0;
+	wsclient->antdsock->z_status = 0;
 	wsclient->antdsock->last_io = time(NULL);
 	wsclient->antdsock->zstream = NULL;
 #ifdef USE_OPENSSL