ant-http/libs/scheduler.c

369 lines
9.6 KiB
C
Raw Normal View History

2018-09-25 17:51:56 +02:00
#include "scheduler.h"
2018-09-27 15:00:19 +02:00
static void enqueue(antd_task_queue_t* q, antd_task_t* task)
2018-09-25 17:51:56 +02:00
{
2018-09-27 15:00:19 +02:00
antd_task_item_t it = *q;
2018-10-01 22:49:20 +02:00
while(it && it->next != NULL)
2018-09-25 17:51:56 +02:00
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
{
2018-09-27 15:00:19 +02:00
*q = taski;
2018-09-25 17:51:56 +02:00
}
else
{
it->next = taski;
}
}
2018-10-01 22:49:20 +02:00
static void stop(antd_scheduler_t* scheduler)
2018-09-25 17:51:56 +02:00
{
2018-10-01 22:49:20 +02:00
scheduler->status = 0;
2018-10-10 12:42:47 +02:00
// unlock all idle workers if any
2018-10-01 22:49:20 +02:00
for (int i = 0; i < scheduler->n_workers; i++)
2018-10-14 10:57:09 +02:00
sem_post(scheduler->worker_sem);
sem_post(scheduler->scheduler_sem);
2018-10-10 12:42:47 +02:00
for (int i = 0; i < scheduler->n_workers; i++)
if(scheduler->workers[i].id != -1)
pthread_join(scheduler->workers[i].tid, NULL);
2018-10-01 22:49:20 +02:00
if(scheduler->workers) free(scheduler->workers);
2018-09-27 15:00:19 +02:00
// destroy all the mutex
2018-10-01 22:49:20 +02:00
pthread_mutex_destroy(&scheduler->scheduler_lock);
pthread_mutex_destroy(&scheduler->worker_lock);
pthread_mutex_destroy(&scheduler->pending_lock);
2018-10-14 11:25:13 +02:00
sem_unlink("scheduler");
sem_unlink("worker");
2018-10-14 10:57:09 +02:00
sem_close(scheduler->scheduler_sem);
sem_close(scheduler->worker_sem);
2018-09-25 17:51:56 +02:00
}
2018-09-27 15:00:19 +02:00
static antd_task_item_t dequeue(antd_task_queue_t* q)
2018-09-25 17:51:56 +02:00
{
2018-09-27 15:00:19 +02:00
antd_task_item_t it = *q;
2018-09-25 17:51:56 +02:00
if(it)
{
2018-09-27 15:00:19 +02:00
*q = it->next;
it->next = NULL;
2018-09-25 17:51:56 +02:00
}
return it;
}
2018-09-27 15:00:19 +02:00
2018-09-25 17:51:56 +02:00
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;
}
2018-10-01 22:49:20 +02:00
static void execute_callback(antd_scheduler_t* scheduler, antd_task_t* task)
2018-09-25 17:51:56 +02:00
{
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);
2018-10-01 22:49:20 +02:00
antd_add_task(scheduler, task);
2018-09-25 17:51:56 +02:00
}
else
{
free(task);
}
}
2018-09-27 15:00:19 +02:00
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);
}
}
2018-10-10 12:42:47 +02:00
static void work(antd_worker_t* worker)
2018-09-27 17:18:31 +02:00
{
2018-10-10 12:42:47 +02:00
antd_scheduler_t* scheduler = (antd_scheduler_t*) worker->manager;
2018-10-01 22:49:20 +02:00
while(scheduler->status)
2018-09-27 17:18:31 +02:00
{
2018-10-01 22:49:20 +02:00
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);
2018-10-10 12:42:47 +02:00
// no task to execute, just sleep wait
2018-10-09 17:24:00 +02:00
if(!it)
{
2018-10-10 12:42:47 +02:00
//LOG("Worker %d goes to idle state\n", worker->id);
2018-10-14 10:57:09 +02:00
sem_wait(scheduler->worker_sem);
2018-10-09 17:24:00 +02:00
}
2018-10-10 12:42:47 +02:00
else
{
//LOG("task executed by worker %d\n", worker->id);
antd_execute_task(scheduler, it);
}
2018-09-27 17:18:31 +02:00
}
2018-09-27 15:00:19 +02:00
}
2018-10-01 22:49:20 +02:00
2018-09-27 17:18:31 +02:00
/*
Main API methods
init the main scheduler
*/
2018-09-26 10:30:04 +02:00
2018-10-01 22:49:20 +02:00
void antd_scheduler_init(antd_scheduler_t* scheduler, int n)
2018-09-25 17:51:56 +02:00
{
2018-10-01 22:49:20 +02:00
scheduler->n_workers = n;
scheduler->status = 1;
scheduler->workers_queue = NULL;
scheduler->pending_task = 0 ;
2019-07-31 15:11:59 +02:00
scheduler->validate_data = 0;
scheduler->destroy_data = NULL;
2018-10-10 12:42:47 +02:00
// init semaphore
2018-10-14 11:19:06 +02:00
scheduler->scheduler_sem = sem_open("scheduler", O_CREAT, 0600, 0);
if (scheduler->scheduler_sem == SEM_FAILED)
2018-10-10 12:42:47 +02:00
{
2018-10-14 10:57:09 +02:00
LOG("Cannot open semaphore for scheduler\n");
2018-10-10 12:42:47 +02:00
exit(-1);
}
2018-10-14 11:19:06 +02:00
scheduler->worker_sem = sem_open("worker", O_CREAT, 0600, 0);
2018-10-14 10:57:09 +02:00
if (!scheduler->worker_sem)
2018-10-10 12:42:47 +02:00
{
2018-10-14 10:57:09 +02:00
LOG("Cannot open semaphore for workers\n");
2018-10-10 12:42:47 +02:00
exit(-1);
}
// init lock
2018-10-01 22:49:20 +02:00
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;
2018-09-25 17:51:56 +02:00
// create scheduler.workers
2018-09-26 10:30:04 +02:00
if(n > 0)
2018-09-25 17:51:56 +02:00
{
2018-10-10 12:42:47 +02:00
scheduler->workers = (antd_worker_t*)malloc(n*(sizeof(antd_worker_t)));
2018-10-01 22:49:20 +02:00
if(!scheduler->workers)
2018-09-25 17:51:56 +02:00
{
2018-09-26 10:30:04 +02:00
LOG("Cannot allocate memory for worker\n");
exit(-1);
}
2018-10-01 22:49:20 +02:00
for(int i = 0; i < scheduler->n_workers;i++)
2018-09-26 10:30:04 +02:00
{
2018-10-10 12:42:47 +02:00
scheduler->workers[i].id = -1;
scheduler->workers[i].manager = (void*)scheduler;
if (pthread_create(&scheduler->workers[i].tid, NULL,(void *(*)(void *))work, (void*)&scheduler->workers[i]) != 0)
2018-09-26 10:30:04 +02:00
{
perror("pthread_create: cannot create worker\n");
}
2018-10-10 12:42:47 +02:00
else
{
scheduler->workers[i].id = i;
}
2018-09-25 17:51:56 +02:00
}
}
2018-10-01 22:49:20 +02:00
LOG("Antd scheduler initialized with %d worker\n", scheduler->n_workers);
2018-09-25 17:51:56 +02:00
}
/*
destroy all pending task
2018-09-27 15:00:19 +02:00
pthread_mutex_lock(&scheduler.queue_lock);
2018-09-25 17:51:56 +02:00
*/
2018-10-01 22:49:20 +02:00
void antd_scheduler_destroy(antd_scheduler_t* scheduler)
2018-09-25 17:51:56 +02:00
{
// free all the chains
2018-10-01 22:49:20 +02:00
stop(scheduler);
LOG("Destroy remaining queue\n");
2018-09-25 17:51:56 +02:00
for(int i=0; i < N_PRIORITY; i++)
{
2018-10-01 22:49:20 +02:00
destroy_queue(scheduler->task_queue[i]);
2018-09-25 17:51:56 +02:00
}
2018-10-01 22:49:20 +02:00
destroy_queue(scheduler->workers_queue);
2018-09-25 17:51:56 +02:00
}
/*
create a task
*/
2019-07-31 15:11:59 +02:00
antd_task_t* antd_create_task(void* (*handle)(void*), void *data, void* (*callback)(void*), time_t atime)
2018-09-25 17:51:56 +02:00
{
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;
2018-10-14 21:48:51 +02:00
task->type = LIGHT;
2019-07-31 15:11:59 +02:00
task->access_time = atime;
2018-09-25 17:51:56 +02:00
return task;
}
/*
scheduling a task
*/
2018-10-01 22:49:20 +02:00
void antd_add_task(antd_scheduler_t* scheduler, antd_task_t* task)
2018-09-25 17:51:56 +02:00
{
2018-09-27 15:00:19 +02:00
// check if task is exist
int prio = task->priority>N_PRIORITY-1?N_PRIORITY-1:task->priority;
2018-10-07 01:03:05 +02:00
//LOG("Prio is %d\n", prio);
2018-10-01 22:49:20 +02:00
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);
2018-10-10 12:42:47 +02:00
// wake up the scheduler if idle
2018-10-14 10:57:09 +02:00
sem_post(scheduler->scheduler_sem);
2018-09-25 17:51:56 +02:00
}
2018-10-01 22:49:20 +02:00
void antd_execute_task(antd_scheduler_t* scheduler, antd_task_item_t taski)
2018-09-25 17:51:56 +02:00
{
2018-10-01 22:49:20 +02:00
if(!taski)
return;
2018-09-25 17:51:56 +02:00
// 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
2018-10-01 22:49:20 +02:00
execute_callback(scheduler, taski->task);
2018-09-25 17:51:56 +02:00
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
2018-10-01 22:49:20 +02:00
execute_callback(scheduler, rtask);
2018-09-25 17:51:56 +02:00
free(taski->task);
free(taski);
}
else
{
2018-10-01 22:49:20 +02:00
antd_add_task(scheduler, rtask);
2018-09-25 17:51:56 +02:00
free(taski->task);
free(taski);
}
}
2018-10-01 22:49:20 +02:00
pthread_mutex_lock(&scheduler->pending_lock);
scheduler->pending_task--;
pthread_mutex_unlock(&scheduler->pending_lock);
2018-09-25 17:51:56 +02:00
}
2018-09-27 17:18:31 +02:00
2018-10-01 22:49:20 +02:00
int antd_scheduler_busy(antd_scheduler_t* scheduler)
2018-09-25 17:51:56 +02:00
{
2018-10-01 22:49:20 +02:00
return scheduler->pending_task != 0;
2018-09-25 17:51:56 +02:00
}
2018-10-01 22:49:20 +02:00
2018-10-09 17:24:00 +02:00
int antd_task_schedule(antd_scheduler_t* scheduler)
2018-09-27 15:00:19 +02:00
{
// fetch next task from the task_queue
antd_task_item_t it = NULL;
2018-10-01 22:49:20 +02:00
pthread_mutex_lock(&scheduler->scheduler_lock);
2018-09-27 15:00:19 +02:00
for(int i = 0; i< N_PRIORITY; i++)
{
2018-10-01 22:49:20 +02:00
it = dequeue(&scheduler->task_queue[i]);
2018-09-27 15:00:19 +02:00
if(it)
break;
}
2018-10-01 22:49:20 +02:00
pthread_mutex_unlock(&scheduler->scheduler_lock);
2019-07-31 15:11:59 +02:00
// no task
2018-09-27 15:00:19 +02:00
if(!it)
{
2018-10-09 17:24:00 +02:00
return 0;
2018-09-27 15:00:19 +02:00
}
// has the task now
2019-07-31 15:11:59 +02:00
// 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;
}
2018-10-10 12:42:47 +02:00
// check the type of task
2018-10-01 22:49:20 +02:00
if(it->task->type == LIGHT || scheduler->n_workers <= 0)
2018-09-27 15:00:19 +02:00
{
// do it by myself
2018-10-01 22:49:20 +02:00
antd_execute_task( scheduler, it);
2018-09-27 15:00:19 +02:00
}
else
{
// delegate to other workers by
//pushing to the worker queue
2018-10-01 22:49:20 +02:00
pthread_mutex_lock(&scheduler->worker_lock);
enqueue(&scheduler->workers_queue, it->task);
pthread_mutex_unlock(&scheduler->worker_lock);
2018-10-10 12:42:47 +02:00
// wake up idle worker
2018-10-14 10:57:09 +02:00
sem_post(scheduler->worker_sem);
2018-09-27 15:00:19 +02:00
free(it);
}
2018-10-09 17:24:00 +02:00
return 1;
2018-10-10 12:42:47 +02:00
}
void antd_wait(antd_scheduler_t* scheduler)
{
int stat;
while(scheduler->status)
{
stat = antd_task_schedule(scheduler);
if(!stat)
{
// no task found, go to idle state
2018-10-14 10:57:09 +02:00
sem_wait(scheduler->scheduler_sem);
2018-10-10 12:42:47 +02:00
}
}
}