dany/ant-http
1
0
Fork 0
mirror of https://github.com/lxsang/ant-http synced 2024-11-17 17:08:20 +01:00
Code Issues Actions Projects Releases Wiki Activity

merge with async

Browse Source
This commit is contained in:
lxsang 2018-10-07 14:05:57 +02:00
commit 76b9abc4b2
18 changed files with 1435 additions and 769 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
Makefile
View File

@ -16,7 +16,8 @@ LIBOBJS = libs/ini.o \
libs/utils.o \
libs/ws.o \
libs/sha1.o \
libs/list.o
libs/list.o \
libs/scheduler.o
PLUGINSDEP = libs/plugin.o
@ -30,6 +31,9 @@ httpd: lib $(SERVER_O)
$(CC) $(CFLAGS) $(SERVER_O) -o $(BUILDIRD)/httpd httpd.c $(SERVERLIB)
cp antd $(BUILDIRD)
relay: lib $(SERVER_O)
$(CC) $(CFLAGS) $(SERVER_O) -o $(BUILDIRD)/relay relay.c $(SERVERLIB)
cp forward $(BUILDIRD)
lib: $(LIBOBJS)
$(CC) $(CFLAGS) $(DB_LIB) $(SSL_LIB) -shared -o $(LIB_NAME).$(EXT) $(LIBOBJS)
cp $(LIB_NAME).$(EXT) $(LIB_PATH$)/

8
forward Normal file
View File

@ -0,0 +1,8 @@
#!/bin/sh
UNAME=`uname -s`
if [ "$UNAME" = "Darwin" ]; then
DYLD_LIBRARY_PATH=$(dirname "$0")/plugins/ $(dirname "$0")/relay
else
LD_LIBRARY_PATH=$(dirname "$0")/plugins/ $(dirname "$0")/relay
fi

1191
http_server.c
View File

File diff suppressed because it is too large Load Diff

30
http_server.h
View File

@ -7,29 +7,28 @@
#include <pthread.h>
#include <signal.h>
#include <sys/socket.h>
#include <sys/select.h>
#include "libs/handle.h"
#include "libs/scheduler.h"
#include "plugin_manager.h"
#define FORM_URL_ENCODE "application/x-www-form-urlencoded"
#define FORM_MULTI_PART "multipart/form-data"
#define PLUGIN_HANDLER "handle"
#define WS_MAGIC_STRING "258EAFA5-E914-47DA-95CA-C5AB0DC85B11"
#define ISspace(x) isspace((int)(x))
#define SERVER_STRING "Server: ant-httpd"
#define MATCH(s, n) strcmp(section, s) == 0 && strcmp(name, n) == 0
#define CONFIG "config.ini"
extern config_t server_config;
void accept_request(void*);
config_t* config();
void destroy_config();
void* accept_request(void*);
void* finish_request(void*);
void cat(void*, FILE *);
void cannot_execute(void*);
void error_die(const char *);
//int get_line(int, char *, int);
void not_found(void*);
void serve_file(void*, const char *);
void* serve_file(void*);
int startup(unsigned *);
void unimplemented(void*);
void badrequest(void*);
@ -37,12 +36,15 @@ int rule_check(const char*, const char*, const char* , const char* , const char*
void ws_confirm_request(void*, const char*);
char* post_url_decode(void* client,int len);
void decode_url_request(const char* query, dictionary);
dictionary decode_request(void* client,const char* method, char* url);
void decode_multi_part_request(void*,const char*, dictionary);
void* decode_request_header(void* data);
void* decode_request(void* data);
void* decode_post_request(void* data);
void* resolve_request(void* data);
void* decode_multi_part_request(void*,const char*, dictionary);
void* decode_multi_part_request_data(void* data);
dictionary decode_cookie(const char*);
char* post_data_decode(void*,int);
int execute_plugin(void* client, const char *path,
const char *method, dictionary rq);
void set_nonblock(int);
void* execute_plugin(void* data, const char *path);
#endif

183
httpd.c
View File

@ -2,9 +2,9 @@
#include <dirent.h>
#include "http_server.h"
#include "libs/ini.h"
static antd_scheduler_t scheduler;
static int server_sock = -1;
#define MATCH(s, n) strcmp(section, s) == 0 && strcmp(name, n) == 0
int server_sock = -1;
#ifdef USE_OPENSSL
static int ssl_session_ctx_id = 1;
SSL_CTX *ctx;
@ -53,13 +53,13 @@ void configure_context(SSL_CTX *ctx)
SSL_CTX_set_session_id_context(ctx, (void *)&ssl_session_ctx_id, sizeof(ssl_session_ctx_id));
/* 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, server_config.sslcert) <= 0) {
//if (SSL_CTX_use_certificate_file(ctx, server_config->sslcert, SSL_FILETYPE_PEM) <= 0) {
if (SSL_CTX_use_certificate_chain_file(ctx, config()->sslcert) <= 0) {
ERR_print_errors_fp(stderr);
exit(EXIT_FAILURE);
}
if (SSL_CTX_use_PrivateKey_file(ctx, server_config.sslkey, SSL_FILETYPE_PEM) <= 0 ) {
if (SSL_CTX_use_PrivateKey_file(ctx, config()->sslkey, SSL_FILETYPE_PEM) <= 0 ) {
ERR_print_errors_fp(stderr);
exit(EXIT_FAILURE);
}
@ -72,117 +72,23 @@ void configure_context(SSL_CTX *ctx)
#endif
static int config_handler(void* conf, const char* section, const char* name,
const char* value)
{
config_t* pconfig = (config_t*)conf;
//char * ppath = NULL;
if (MATCH("SERVER", "port")) {
pconfig->port = atoi(value);
} else if (MATCH("SERVER", "plugins")) {
pconfig->plugins_dir = strdup(value);
} else if (MATCH("SERVER", "plugins_ext")) {
pconfig->plugins_ext = strdup(value);
} else if(MATCH("SERVER", "database")) {
pconfig->db_path = strdup(value);
} else if(MATCH("SERVER", "htdocs")) {
pconfig->htdocs = strdup(value);
} else if(MATCH("SERVER", "tmpdir")) {
pconfig->tmpdir = strdup(value);
}
else if(MATCH("SERVER", "maxcon")) {
pconfig->maxcon = atoi(value);
}
else if(MATCH("SERVER", "backlog")) {
pconfig->backlog = atoi(value);
}
#ifdef USE_OPENSSL
else if(MATCH("SERVER", "ssl.enable")) {
pconfig->usessl = atoi(value);
}
else if(MATCH("SERVER", "ssl.cert")) {
pconfig->sslcert = strdup(value);
}
else if(MATCH("SERVER", "ssl.key")) {
pconfig->sslkey = strdup(value);
}
#endif
else if (strcmp(section, "RULES") == 0)
{
list_put_s(&pconfig->rules, name);
list_put_s(&pconfig->rules, value);
}
else if (strcmp(section, "FILEHANDLER") == 0)
{
dput( pconfig->handlers, name ,strdup(value));
}
else if(strcmp(section,"AUTOSTART")==0){
// The server section must be added before the autostart section
// auto start plugin
plugin_load(value);
} else {
return 0; /* unknown section/name, error */
}
return 1;
}
void init_file_system()
{
struct stat st;
if (stat(server_config.plugins_dir, &st) == -1)
mkdir(server_config.plugins_dir, 0755);
if (stat(server_config.db_path, &st) == -1)
mkdir(server_config.db_path, 0755);
if (stat(server_config.htdocs, &st) == -1)
mkdir(server_config.htdocs, 0755);
if (stat(server_config.tmpdir, &st) == -1)
mkdir(server_config.tmpdir, 0755);
else
{
removeAll(server_config.tmpdir,0);
}
}
void load_config(const char* file)
{
server_config.port = 8888;
server_config.plugins_dir = "plugins/";
server_config.plugins_ext = ".dylib";
server_config.db_path = "databases/";
server_config.htdocs = "htdocs";
server_config.tmpdir = "tmp";
server_config.backlog = 100;
server_config.rules = list_init();
server_config.handlers = dict();
server_config.maxcon = 1000;
server_config.connection = 0;
#ifdef USE_OPENSSL
server_config.usessl = 0;
server_config.sslcert = "cert.pem";
server_config.sslkey = "key.pem";
#endif
if (ini_parse(file, config_handler, &server_config) < 0) {
LOG("Can't load '%s'\n. Used defaut configuration", file);
}
else
{
LOG("Using configuration : %s\n", file);
#ifdef USE_OPENSSL
LOG("SSL enable %d\n", server_config.usessl);
LOG("SSL cert %s\n", server_config.sslcert);
LOG("SSL key %s\n", server_config.sslkey);
#endif
}
init_file_system();
}
void stop_serve(int dummy) {
list_free(&(server_config.rules));
freedict(server_config.handlers);
LOG("Unclosed connection: %d\n", server_config.connection);
unload_all_plugin();
UNUSED(dummy);
sigset_t mask;
sigemptyset(&mask);
//Blocks the SIG_IGN signal (by adding SIG_IGN to newMask)
sigaddset(&mask, SIGINT);
sigaddset(&mask, SIGPIPE);
sigaddset(&mask, SIGABRT);
sigprocmask(SIG_BLOCK, &mask, NULL);
antd_scheduler_destroy(&scheduler);
unload_all_plugin();
destroy_config();
#ifdef USE_OPENSSL
SSL_CTX_free(ctx);
#endif
close(server_sock);
sigprocmask(SIG_UNBLOCK, &mask, NULL);
}
int main(int argc, char* argv[])
{
@ -191,11 +97,10 @@ int main(int argc, char* argv[])
load_config(CONFIG);
else
load_config(argv[1]);
unsigned port = server_config.port;
unsigned port = config()->port;
int client_sock = -1;
struct sockaddr_in client_name;
socklen_t client_name_len = sizeof(client_name);
pthread_t newthread;
char* client_ip = NULL;
// ignore the broken PIPE error when writing
//or reading to/from a closed socked connection
@ -204,7 +109,7 @@ int main(int argc, char* argv[])
signal(SIGINT, stop_serve);
#ifdef USE_OPENSSL
if( server_config.usessl == 1 )
if( config()->usessl == 1 )
{
init_openssl();
ctx = create_context();
@ -213,24 +118,24 @@ int main(int argc, char* argv[])
}
#endif
server_sock = startup(&port);
LOG("httpd running on port %d\n", port);
while (1)
// default to 4 workers
antd_scheduler_init(&scheduler, config()->n_workers);
set_nonblock(server_sock);
while (scheduler.status)
{
if( server_config.connection >= server_config.maxcon )
{
LOG("Too many unclosed connection (%d). Wait for it\n", server_config.connection);
continue;
}
antd_client_t* client = (antd_client_t*)malloc(sizeof(antd_client_t));
antd_task_schedule(&scheduler);
client_sock = accept(server_sock,(struct sockaddr *)&client_name,&client_name_len);
if (client_sock == -1)
{
perror("Cannot accept client request\n");
//perror("Cannot accept client request\n");
continue;
}
antd_client_t* client = (antd_client_t*)malloc(sizeof(antd_client_t));
antd_request_t* request = (antd_request_t*)malloc(sizeof(*request));
request->client = client;
request->request = dict();
/*
get the remote IP
*/
@ -240,40 +145,46 @@ 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);
LOG("socket: %d\n", client_sock);
}
//return &(((struct sockaddr_in6*)sa)->sin6_addr);
/* accept_request(client_sock); */
// set timeout to socket
struct timeval timeout;
timeout.tv_sec = 20;
timeout.tv_usec = 0;
set_nonblock(client_sock);
/*struct timeval timeout;
timeout.tv_sec = 0;
timeout.tv_usec = 5000;
if (setsockopt (client_sock, SOL_SOCKET, SO_RCVTIMEO, (char *)&timeout,sizeof(timeout)) < 0)
perror("setsockopt failed\n");
if (setsockopt (client_sock, SOL_SOCKET, SO_SNDTIMEO, (char *)&timeout,sizeof(timeout)) < 0)
perror("setsockopt failed\n");
*/
client->sock = client_sock;
server_config.connection++;
//LOG("Unclosed connection: %d\n", server_config.connection);
// 100 times retry connection before abort
//LOG("Unclosed connection: %d\n", server_config->connection);
#ifdef USE_OPENSSL
client->ssl = NULL;
if(server_config.usessl == 1)
client->status = 0;
if(config()->usessl == 1)
{
client->ssl = (void*)SSL_new(ctx);
if(!client->ssl) continue;
SSL_set_fd((SSL*)client->ssl, client_sock);
SSL_set_fd((SSL*)client->ssl, client->sock);
if (SSL_accept((SSL*)client->ssl) <= 0) {
/*if (SSL_accept((SSL*)client->ssl) <= 0) {
LOG("EROOR accept\n");
ERR_print_errors_fp(stderr);
antd_close(client);
continue;
}
}*/
}
#endif
if (pthread_create(&newthread , NULL,(void *(*)(void *))accept_request, (void *)client) != 0)
// create callback for the server
antd_add_task(&scheduler, antd_create_task(accept_request,(void*)request, finish_request ));
/*if (pthread_create(&newthread , NULL,(void *(*)(void *))accept_request, (void *)client) != 0)
{
perror("pthread_create");
antd_close(client);
@ -282,7 +193,7 @@ int main(int argc, char* argv[])
{
//reclaim the stack data when thread finish
pthread_detach(newthread) ;
}
}*/
//accept_request(&client);
}

9
libs/dictionary.c
View File

@ -52,18 +52,25 @@ association __put_el_with_key(dictionary dic, const char* key)
if(dic == NULL) return NULL;
if ((np = dlookup(dic,key)) == NULL) { /* not found */
np = (association) malloc(sizeof(*np));
np->value = NULL;
if (np == NULL || (np->key = strdup(key)) == NULL)
return NULL;
hashval = hash(key, DHASHSIZE);
np->next = dic[hashval];
dic[hashval] = np;
}
// found
return np;
}
association dput(dictionary dic,const char* key, void* value)
{
association np = __put_el_with_key(dic,key);
if(np == NULL) return NULL;
if(np == NULL)
{
if(value) free(value);
return NULL;
}
if(np->value && value) free(np->value);
np->value = value;
return np;
}

1
libs/dictionary.h
View File

@ -46,6 +46,5 @@ void* dvalue(dictionary, const char*);
association dput(dictionary,const char*, void*);
int dremove(dictionary, const char*);
void freedict(dictionary);
void stest(const char* );
#endif

161
libs/handle.c
View File

@ -1,5 +1,4 @@
#include "handle.h"
config_t server_config;
#include "handle.h"
#ifdef USE_OPENSSL
int usessl()
{
@ -96,18 +95,112 @@ int antd_send(void *src, const void* data, int len)
int antd_recv(void *src, void* data, int len)
{
if(!src) return -1;
int ret;
int read;
antd_client_t * source = (antd_client_t *) src;
#ifdef USE_OPENSSL
if(usessl())
{
//LOG("SSL READ\n");
ret = SSL_read((SSL*) source->ssl, data, len);
int received;
char* ptr = (char* )data;
int readlen = len > BUFFLEN?BUFFLEN:len;
read = 0;
fd_set fds;
struct timeval timeout;
while (readlen > 0)
{
received = SSL_read (source->ssl, ptr+read, readlen);
if (received > 0)
{
read += received;
readlen = (len - read) > BUFFLEN?BUFFLEN:(len-read);
}
else
{
//printf(" received equal to or less than 0\n")
int err = SSL_get_error(source->ssl, received);
switch (err)
{
case SSL_ERROR_NONE:
{
// no real error, just try again...
//LOG("SSL_ERROR_NONE \n");
continue;
}
case SSL_ERROR_ZERO_RETURN:
{
// peer disconnected...
//printf("SSL_ERROR_ZERO_RETURN \n");
break;
}
case SSL_ERROR_WANT_READ:
{
// no data available right now, wait a few seconds in case new data arrives...
//printf("SSL_ERROR_WANT_READ\n");
int sock = SSL_get_rfd(source->ssl);
FD_ZERO(&fds);
FD_SET(sock, &fds);
timeout.tv_sec = 0;
timeout.tv_usec = 500;
err = select(sock+1, &fds, NULL, NULL, &timeout);
if (err == 0 || (err > 0 && FD_ISSET(sock, &fds)))
continue; // more data to read...
break;
}
case SSL_ERROR_WANT_WRITE:
{
// socket not writable right now, wait a few seconds and try again...
//printf("SSL_ERROR_WANT_WRITE \n");
int sock = SSL_get_wfd(source->ssl);
FD_ZERO(&fds);
FD_SET(sock, &fds);
timeout.tv_sec = 0;
timeout.tv_usec = 500;
err = select(sock+1, NULL, &fds, NULL, &timeout);
if (err == 0 || (err > 0 && FD_ISSET(sock, &fds)))
continue; // can write more data now...
break;
}
default:
{
// other error
break;
}
}
break;
}
}
/*
int stat, r, st;
do{
ret = SSL_read((SSL*) source->ssl, data, len);
stat = SSL_get_error((SSL*)source->ssl, r);
} while(ret == -1 &&
(
stat == SSL_ERROR_WANT_READ ||
stat == SSL_ERROR_WANT_WRITE ||
stat == SSL_ERROR_NONE ||
(stat == SSL_ERROR_SYSCALL && r== 0 && !ERR_get_error())
));
if(ret == -1)
{
LOG("Problem reading %d %d %d\n", ret, stat, r);
}
//set_nonblock(source->sock);
*/
}
else
{
#endif
ret = recv(((int) source->sock), data, len, 0);
read = recv(((int) source->sock), data, len, 0);
#ifdef USE_OPENSSL
}
#endif
@ -115,7 +208,20 @@ int antd_recv(void *src, void* data, int len)
{
antd_close(src);
}*/
return ret;
return read;
}
void set_nonblock(int socket) {
int flags;
flags = fcntl(socket,F_GETFL,0);
//assert(flags != -1);
fcntl(socket, F_SETFL, flags | O_NONBLOCK);
}
void set_block()
{
int flags;
flags = fcntl(socket,F_GETFL,0);
//assert(flags != -1);
fcntl(socket, F_SETFL, flags & (~O_NONBLOCK));
}
int antd_close(void* src)
{
@ -133,9 +239,7 @@ int antd_close(void* src)
#endif
//printf("Close sock %d\n", source->sock);
int ret = close(source->sock);
if(source->ip) free(source->ip);
server_config.connection--;
LOG("Remaining connection %d\n", server_config.connection);
if(source->ip) free(source->ip);
free(src);
src = NULL;
return ret;
@ -299,8 +403,41 @@ void clear_cookie(void* client, dictionary dic)
}
void unknow(void* client)
{
html(client);
__t(client,"404 API not found");
set_status(client,520,"Unknown Error");
__t(client,"Content-Type: text/html; charset=utf-8");
response(client,"");
__t(client,"520 Unknow request");
}
void notfound(void* client)
{
set_status(client,404,"Not found");
__t(client,"Content-Type: text/html; charset=utf-8");
response(client,"");
__t(client,"Resource not found");
}
void badrequest(void* client)
{
set_status(client,400,"Bad request");
__t(client,"Content-Type: text/html; charset=utf-8");
response(client,"");
__t(client,"400 Bad request");
}
void unimplemented(void* client)
{
set_status(client,501,"Method Not Implemented");
__t(client,"Content-Type: text/html");
response(client,"");
__t(client, "<HTML><HEAD><TITLE>Method Not Implemented");
__t(client, "</TITLE></HEAD>");
__t(client, "<BODY><P>HTTP request method not supported.");
__t(client, "</BODY></HTML>");
}
void cannot_execute(void* client)
{
set_status(client,500,"Internal Server Error");
__t(client,"Content-Type: text/html");
response(client,"");
__t(client, "<P>Error prohibited CGI execution.");
}
int ws_enable(dictionary dic)
{

31
libs/handle.h
View File

@ -11,11 +11,12 @@
#ifdef USE_DB
#include "dbhelper.h"
#endif
#include <fcntl.h>
#include "dictionary.h"
#include "list.h"
#include "ini.h"
#define SERVER_NAME "antd"
#define SERVER_NAME "Antd"
#define IS_POST(method) (strcmp(method,"POST")== 0)
#define IS_GET(method) (strcmp(method,"GET")== 0)
#define R_STR(d,k) ((char*)dvalue(d,k))
@ -28,6 +29,21 @@
#ifdef USE_OPENSSL
int __attribute__((weak)) usessl();
#endif
//extern config_t server_config;
typedef struct{
int sock;
void* ssl;
char* ip;
#ifdef USE_OPENSSL
int status;
#endif
} antd_client_t;
typedef struct {
antd_client_t* client;
dictionary request;
} antd_request_t;
typedef struct {
int port;
@ -41,19 +57,15 @@ typedef struct {
int backlog;
int maxcon;
int connection;
int n_workers;
#ifdef USE_OPENSSL
int usessl;
char* sslcert;
char* sslkey;
#endif
}config_t;
//extern config_t server_config;
typedef struct{
int sock;
void* ssl;
char* ip;
} antd_client_t;
void set_nonblock(int socket);
void set_block(int socket);
int response(void*, const char*);
void ctype(void*,const char*);
void redirect(void*,const char*);
@ -75,6 +87,9 @@ void set_status(void*,int,const char*);
void clear_cookie(void*, dictionary);
/*Default function for plugin*/
void unknow(void*);
void badrequest(void* client);
void unimplemented(void* client);
void notfound(void* client);
int ws_enable(dictionary);
char* read_line(void* sock);
int read_buf(void* sock,char* buf,int i);

15
libs/plugin.c
View File

@ -1,8 +1,7 @@
#include "plugin.h"
plugin_header __plugin__;
plugin_header_t __plugin__;
// private function
call __init__;
void __init_plugin__(const char* pl,config_t* conf){
__plugin__.name = strdup(pl);
__plugin__.dbpath= strdup(conf->db_path);
@ -12,7 +11,7 @@ void __init_plugin__(const char* pl,config_t* conf){
#ifdef USE_OPENSSL
__plugin__.usessl = conf->usessl;
#endif
if(__init__ != NULL) __init__();
init();
};
#ifdef USE_DB
@ -41,7 +40,10 @@ int usessl()
return __plugin__.usessl;
}
#endif*/
plugin_header_t* meta()
{
return &__plugin__;
}
char* route(const char* repath)
{
int len = strlen(__plugin__.name) + 2;
@ -75,9 +77,10 @@ char* config_dir()
return path;
}
void __release()
void __release__()
{
printf("Releasing plugin\n");
destroy();
LOG("Releasing plugin\n");
if(__plugin__.name) free(__plugin__.name);
if(__plugin__.dbpath) free(__plugin__.dbpath);
if(__plugin__.htdocs) free(__plugin__.htdocs);

14
libs/plugin.h
View File

@ -5,6 +5,7 @@
#include "dbhelper.h"
#endif
#include "ws.h"
#include "scheduler.h"
typedef struct {
char *name;
@ -15,16 +16,14 @@ typedef struct {
#ifdef USE_OPENSSL
int usessl;
#endif
} plugin_header;
} plugin_header_t;
typedef void(*call)();
//typedef void(*call)();
#ifdef USE_DB
typedef sqlite3* sqldb;
#endif
extern plugin_header __plugin__;
extern call __init__;
#ifdef USE_DB
@ -36,6 +35,9 @@ char* route(const char*);
char* htdocs(const char*);
char* config_dir();
/*Default function for plugin*/
void handle(void*, const char*,const char*,dictionary);
void __release();
// init the plugin
void init();
void destroy();
void* handle(void*);
plugin_header_t* meta();
#endif

292
libs/scheduler.c Normal file
View File

@ -0,0 +1,292 @@
#include "scheduler.h"
static void enqueue(antd_task_queue_t* q, antd_task_t* task)
{
antd_task_item_t it = *q;
while(it && it->next != NULL)
it = it->next;
antd_task_item_t taski = (antd_task_item_t)malloc(sizeof *taski);
taski->task = task;
taski->next = NULL;
if(!it) // first task
{
*q = taski;
}
else
{
it->next = taski;
}
}
static void stop(antd_scheduler_t* scheduler)
{
scheduler->status = 0;
for (int i = 0; i < scheduler->n_workers; i++)
pthread_join(scheduler->workers[i], NULL);
if(scheduler->workers) free(scheduler->workers);
// destroy all the mutex
pthread_mutex_destroy(&scheduler->scheduler_lock);
pthread_mutex_destroy(&scheduler->worker_lock);
pthread_mutex_destroy(&scheduler->pending_lock);
}
static antd_task_item_t dequeue(antd_task_queue_t* q)
{
antd_task_item_t it = *q;
if(it)
{
*q = it->next;
it->next = NULL;
}
return it;
}
static antd_callback_t* callback_of( void* (*callback)(void*) )
{
antd_callback_t* cb = NULL;
if(callback)
{
cb = (antd_callback_t*)malloc(sizeof *cb);
cb->handle = callback;
cb->next = NULL;
}
return cb;
}
static void free_callback(antd_callback_t* cb)
{
antd_callback_t* it = cb;
antd_callback_t* curr;
while(it)
{
curr = it;
it = it->next;
free(curr);
}
}
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
it->next = el;
}
static void execute_callback(antd_scheduler_t* scheduler, antd_task_t* task)
{
antd_callback_t* cb = task->callback;
if(cb)
{
// call the first come call back
task->handle = cb->handle;
task->callback = task->callback->next;
free(cb);
antd_add_task(scheduler, task);
}
else
{
free(task);
}
}
static void destroy_queue(antd_task_queue_t q)
{
antd_task_item_t it, curr;
it = q;
while(it)
{
// first free the task
if(it->task && it->task->callback) free_callback(it->task->callback);
if(it->task) free(it->task);
// then free the placeholder
curr = it;
it = it->next;
free(curr);
}
}
static void work(antd_scheduler_t* scheduler)
{
while(scheduler->status)
{
antd_task_item_t it;
pthread_mutex_lock(&scheduler->worker_lock);
it = dequeue(&scheduler->workers_queue);
pthread_mutex_unlock(&scheduler->worker_lock);
// execute the task
//LOG("task executed by worker %d\n", worker->pid);
antd_execute_task(scheduler, it);
}
}
/*
Main API methods
init the main scheduler
*/
void antd_scheduler_init(antd_scheduler_t* scheduler, int n)
{
scheduler->n_workers = n;
scheduler->status = 1;
scheduler->workers_queue = NULL;
scheduler->pending_task = 0 ;
// init lock
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;
// create scheduler.workers
if(n > 0)
{
scheduler->workers = (pthread_t*)malloc(n*(sizeof(pthread_t)));
if(!scheduler->workers)
{
LOG("Cannot allocate memory for worker\n");
exit(-1);
}
for(int i = 0; i < scheduler->n_workers;i++)
{
if (pthread_create(&scheduler->workers[i], NULL,(void *(*)(void *))work, (void*)scheduler) != 0)
{
perror("pthread_create: cannot create worker\n");
}
}
}
LOG("Antd scheduler initialized with %d worker\n", scheduler->n_workers);
}
/*
destroy all pending task
pthread_mutex_lock(&scheduler.queue_lock);
*/
void antd_scheduler_destroy(antd_scheduler_t* scheduler)
{
// free all the chains
stop(scheduler);
LOG("Destroy remaining queue\n");
for(int i=0; i < N_PRIORITY; i++)
{
destroy_queue(scheduler->task_queue[i]);
}
destroy_queue(scheduler->workers_queue);
}
/*
create a task
*/
antd_task_t* antd_create_task(void* (*handle)(void*), void *data, void* (*callback)(void*))
{
antd_task_t* task = (antd_task_t*)malloc(sizeof *task);
task->stamp = (unsigned long)time(NULL);
task->data = data;
task->handle = handle;
task->callback = callback_of(callback);
task->priority = NORMAL_PRIORITY;
task->type = LIGHT;
return task;
}
/*
scheduling a task
*/
void antd_add_task(antd_scheduler_t* scheduler, antd_task_t* task)
{
// check if task is exist
int prio = task->priority>N_PRIORITY-1?N_PRIORITY-1:task->priority;
//LOG("Prio is %d\n", prio);
pthread_mutex_lock(&scheduler->scheduler_lock);
enqueue(&scheduler->task_queue[prio], task);
pthread_mutex_unlock(&scheduler->scheduler_lock);
pthread_mutex_lock(&scheduler->pending_lock);
scheduler->pending_task++;
pthread_mutex_unlock(&scheduler->pending_lock);
}
void antd_execute_task(antd_scheduler_t* scheduler, antd_task_item_t taski)
{
if(!taski)
return;
// execute the task
void *ret = (*(taski->task->handle))(taski->task->data);
// check the return data if it is a new task
if(!ret)
{
// call the first callback
execute_callback(scheduler, taski->task);
free(taski);
}
else
{
antd_task_t* rtask = (antd_task_t*) ret;
if(taski->task->callback)
{
if(rtask->callback)
{
enqueue_callback(rtask->callback, taski->task->callback);
}
else
{
rtask->callback = taski->task->callback;
}
}
if(!rtask->handle)
{
// call the first callback
execute_callback(scheduler, rtask);
free(taski->task);
free(taski);
}
else
{
antd_add_task(scheduler, rtask);
free(taski->task);
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)
{
return scheduler->pending_task != 0;
}
void antd_task_schedule(antd_scheduler_t* scheduler)
{
// fetch next task from the task_queue
antd_task_item_t it = NULL;
pthread_mutex_lock(&scheduler->scheduler_lock);
for(int i = 0; i< N_PRIORITY; i++)
{
it = dequeue(&scheduler->task_queue[i]);
if(it)
break;
}
pthread_mutex_unlock(&scheduler->scheduler_lock);
if(!it)
{
return;
}
// has the task now
// check the type of tas
if(it->task->type == LIGHT || scheduler->n_workers <= 0)
{
// do it by myself
antd_execute_task( scheduler, it);
}
else
{
// delegate to other workers by
//pushing to the worker queue
pthread_mutex_lock(&scheduler->worker_lock);
enqueue(&scheduler->workers_queue, it->task);
pthread_mutex_unlock(&scheduler->worker_lock);
free(it);
}
}

96
libs/scheduler.h Normal file
View File

@ -0,0 +1,96 @@
#ifndef ANT_SCHEDULER
#define ANT_SCHEDULER
#include "utils.h"
#include <pthread.h>
#define N_PRIORITY 10
#define NORMAL_PRIORITY ((int)((N_PRIORITY - 1) / 2))
#define LOW_PRIORITY (N_PRIORITY - 1)
#define HIGH_PRIORITY 0
typedef enum { LIGHT, HEAVY } antd_task_type_t;
// callback definition
typedef struct __callback_t{
void* (*handle)(void*);
struct __callback_t * next;
} antd_callback_t;
// task definition
typedef struct {
/*
creation time of a task
*/
unsigned long stamp;
/*
priority from 0 to N_PRIORITY - 1
higher value is lower priority
*/
uint8_t priority;
/*
the callback
*/
void* (*handle)(void*);
antd_callback_t* callback;
/*
user data if any
*/
void * data;
/*
type of a task
light tasks are executed directly
heavy tasks are delegated to workers
*/
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 antd_task_item_t antd_task_queue_t;
typedef struct {
pthread_mutex_t scheduler_lock;
pthread_mutex_t worker_lock;
pthread_mutex_t pending_lock;
antd_task_queue_t task_queue[N_PRIORITY];
antd_task_queue_t workers_queue;
uint8_t status; // 0 stop, 1 working
pthread_t* workers;
int n_workers;
int pending_task;
} antd_scheduler_t;
/*
init the main scheduler
*/
void antd_scheduler_init(antd_scheduler_t*, int);
/*
destroy all pending task
*/
void antd_scheduler_destroy(antd_scheduler_t*);
/*
create a task
*/
antd_task_t* antd_create_task(void* (*handle)(void*), void *data, void* (*callback)(void*));
/*
add a task
*/
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);
/*
scheduler status
*/
int antd_scheduler_busy(antd_scheduler_t*);
/*
schedule a task
*/
void antd_task_schedule(antd_scheduler_t*);
#endif

7
libs/utils.c
View File

@ -145,7 +145,11 @@ char* ext(const char* file)
if(file == NULL) return NULL;
char* str_cpy = strdup(file);
char* str_org = str_cpy;
if(strstr(str_cpy,".")<= 0) return strdup("");
if(strstr(str_cpy,".")<= 0)
{
free(str_org);
return NULL;
}
if(*file == '.')
trim(str_cpy,'.');
@ -184,6 +188,7 @@ mime_t mime_from_type(const char* type)
char* mime(const char* file)
{
char * ex = ext(file);
if(!ex) return "application/octet-stream";
mime_t m = mime_from_ext(ex);
if(ex)
free(ex);

3
libs/utils.h
View File

@ -58,6 +58,9 @@ THE SOFTWARE.
#else
#define LOG(a,...) do{}while(0)
#endif
// add this to the utils
#define UNUSED(x) (void)(x)
#define BUFFLEN 1024
#define HASHSIZE 1024
#define DHASHSIZE 50

26
plugin_manager.c
View File

@ -31,10 +31,16 @@ struct plugin_entry *plugin_load(char *name)
if ((np = plugin_lookup(name)) == NULL) { /* not found */
np = (struct plugin_entry *) malloc(sizeof(*np));
if (np == NULL || (np->pname = strdup(name)) == NULL)
return NULL;
{
if(np) free(np);
return NULL;
}
if ((np->handle = plugin_from_file(name)) == NULL)
{
if(np) free(np);
return NULL;
hashval = hash(name,HASHSIZE);
}
hashval = hash(name,HASHSIZE);
np->next = plugin_table[hashval];
plugin_table[hashval] = np;
} else /* already there */
@ -53,7 +59,7 @@ void * plugin_from_file(char* name)
{
void *lib_handle;
char* error;
char* path = __s("%s%s%s",server_config.plugins_dir,name,server_config.plugins_ext);
char* path = __s("%s%s%s",config()->plugins_dir,name,config()->plugins_ext);
void (*fn)(const char*, config_t*);
lib_handle = dlopen(path, RTLD_LAZY);
if (!lib_handle)
@ -68,7 +74,7 @@ void * plugin_from_file(char* name)
if ((error = dlerror()) != NULL)
LOG("Problem when setting data path for %s : %s \n", name,error);
else
(*fn)(name,&server_config);
(*fn)(name,config());
if(path)
free(path);
return lib_handle;
@ -79,17 +85,7 @@ void unload_plugin(struct plugin_entry* np)
char* error;
void (*fn)() = NULL;
// find and execute the exit function
fn = (void (*)())dlsym(np->handle, "pexit");
if ((error = dlerror()) != NULL)
{
LOG("Cant not find exit method from %s : %s \n", np->pname,error);
}
else
{
// execute it
(*fn)();
}
fn = (void(*)()) dlsym(np->handle, "__release");
fn = (void(*)()) dlsym(np->handle, "__release__");
if ((error = dlerror()) != NULL)
{
LOG("Cant not release plugin %s : %s \n", np->pname,error);

3
plugin_manager.h
View File

@ -3,13 +3,12 @@
#include <dlfcn.h>
#include "libs/utils.h"
#include "libs/handle.h"
#include "http_server.h"
struct plugin_entry {
struct plugin_entry *next;
char *pname;
void *handle;
};
extern config_t server_config;
/* lookup: look for s in hashtab */
struct plugin_entry *plugin_lookup(char *s);
/* install: put (name, defn) in hashtab */

128
relay.c Normal file
View File

@ -0,0 +1,128 @@
#include "http_server.h"
#include "libs/scheduler.h"
#include <fcntl.h>
static antd_scheduler_t scheduler;
/*
this node is a relay from the http
to https
*/
#define MATCH(s, n) strcmp(section, s) == 0 && strcmp(name, n) == 0
int server_sock = -1;
void stop_serve(int dummy) {
UNUSED(dummy);
antd_scheduler_destroy(&scheduler);
close(server_sock);
}
/*
HTTP/1.1 301 Moved Permanently
Location: http://www.example.org/
Content-Type: text/html
Content-Length: 174
*/
void* antd_redirect(void* user_data)
{
void** data = (void**)user_data;
void* client = data[0];
char* host = (char*)data[1];
__t(client,"%s", "HTTP/1.1 301 Moved Permanently");
__t(client, "Location: https://%s", host);
__t(client, "%s", "Content-Type: text/html");
__t(client, "");
__t(client, "This page has moved to https://%s", host);
free(host);
free(user_data);
return antd_create_task(NULL,client, NULL);
}
void* antd_free_client(void* client)
{
antd_client_t * source = (antd_client_t *) client;
if(source->ip) free(source->ip);
close(source->sock);
free(client);
return NULL;
}
void* antd_get_host(void * client)
{
char buff[1024];
char* line, *token;
char* host = NULL;
while((read_buf(client,buff,sizeof(buff))) && strcmp("\r\n",buff))
{
line = buff;
trim(line, '\n');
trim(line, '\r');
token = strsep(&line, ":");
trim(token,' ');
trim(line,' ');
if(token && strcasecmp(token,"Host")==0)
if(line)
{
host = strdup(line);
break;
}
}
if(!host) host = strdup("lxsang.me");
void** data = (void**)malloc(2*(sizeof *data));
data[0] = client;
data[1] = (void*)host;
LOG("[%s] Request for: %s --> https://%s\n", ((antd_client_t*)client)->ip, host, host);
return antd_create_task(antd_redirect,data, NULL);
}
int main(int argc, char* argv[])
{
UNUSED(argc);
UNUSED(argv);
// load the config first
unsigned port = 80;
int client_sock = -1;
struct sockaddr_in client_name;
socklen_t client_name_len = sizeof(client_name);
// ignore the broken PIPE error when writing
//or reading to/from a closed socked connection
signal(SIGPIPE, SIG_IGN);
signal(SIGABRT, SIG_IGN);
signal(SIGINT, stop_serve);
server_sock = startup(&port);
//struct timeval timeout;
//timeout.tv_sec = 0;
//timeout.tv_usec = 500;
// 0 worker
antd_scheduler_init(&scheduler, 0);
// set server socket to non blocking
fcntl(server_sock, F_SETFL, O_NONBLOCK); /* Change the socket into non-blocking state */
LOG("relayd running on port %d\n", port);
while (scheduler.status)
{
// execute task
antd_task_schedule(&scheduler);
client_sock = accept(server_sock,(struct sockaddr *)&client_name,&client_name_len);
if (client_sock == -1)
{
continue;
}
antd_client_t* client = (antd_client_t*)malloc(sizeof(antd_client_t));
// set timeout to socket
//if (setsockopt (client_sock, SOL_SOCKET, SO_RCVTIMEO, (char *)&timeout,sizeof(timeout)) < 0)
// perror("setsockopt failed\n");
//if (setsockopt (client_sock, SOL_SOCKET, SO_SNDTIMEO, (char *)&timeout,sizeof(timeout)) < 0)
// perror("setsockopt failed\n");
/*
get the remote IP
*/
client->ip = NULL;
if (client_name.sin_family == AF_INET)
client->ip = strdup(inet_ntoa(client_name.sin_addr));
client->sock = client_sock;
//accept_request(&client);
antd_add_task(&scheduler, antd_create_task(antd_get_host,(void*)client, antd_free_client ));
}
return(0);
}