dany/ant-http
1
0
Fork 0
mirror of https://github.com/lxsang/ant-http synced 2025-07-25 10:09:46 +02:00
Code Issues Actions Projects Releases Wiki Activity

fix deadlock

Browse Source
This commit is contained in:
lxsang
2018-10-14 10:57:09 +02:00
parent 9f239197df
commit ba4bfadec0
7 changed files with 44 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files

28
libs/handle.c
View File

@ -84,7 +84,7 @@ int antd_send(void *src, const void* data, int len)
written = 0;
fd_set fds;
struct timeval timeout;
while (writelen > 0)
while (writelen > 0 && source->attempt < MAX_ATTEMPT)
{
count = SSL_write (source->ssl, ptr+written, writelen);
if (count > 0)
@ -102,6 +102,7 @@ int antd_send(void *src, const void* data, int len)
{
// no real error, just try again...
//LOG("SSL_ERROR_NONE \n");
source->attempt++;
continue;
}
@ -125,8 +126,11 @@ int antd_send(void *src, const void* data, int len)
timeout.tv_usec = 500;
err = select(sock+1, &fds, NULL, NULL, &timeout);
if (err == 0 || (err > 0 && FD_ISSET(sock, &fds)))
{
source->attempt++;
continue; // more data to read...
break;
}
break;
}
case SSL_ERROR_WANT_WRITE:
@ -142,8 +146,11 @@ int antd_send(void *src, const void* data, int len)
err = select(sock+1, NULL, &fds, NULL, &timeout);
if (err == 0 || (err > 0 && FD_ISSET(sock, &fds)))
{
source->attempt++;
continue; // can write more data now...
break;
}
break;
}
default:
@ -156,6 +163,7 @@ int antd_send(void *src, const void* data, int len)
break;
}
}
source->attempt = 0;
}
else
{
@ -184,7 +192,7 @@ int antd_recv(void *src, void* data, int len)
read = 0;
fd_set fds;
struct timeval timeout;
while (readlen > 0)
while (readlen > 0 && source->attempt < MAX_ATTEMPT)
{
received = SSL_read (source->ssl, ptr+read, readlen);
if (received > 0)
@ -202,6 +210,7 @@ int antd_recv(void *src, void* data, int len)
{
// no real error, just try again...
//LOG("SSL_ERROR_NONE \n");
source->attempt++;
continue;
}
@ -225,8 +234,11 @@ int antd_recv(void *src, void* data, int len)
timeout.tv_usec = 500;
err = select(sock+1, &fds, NULL, NULL, &timeout);
if (err == 0 || (err > 0 && FD_ISSET(sock, &fds)))
{
source->attempt++;
continue; // more data to read...
break;
}
break;
}
case SSL_ERROR_WANT_WRITE:
@ -242,8 +254,11 @@ int antd_recv(void *src, void* data, int len)
err = select(sock+1, NULL, &fds, NULL, &timeout);
if (err == 0 || (err > 0 && FD_ISSET(sock, &fds)))
{
source->attempt++;
continue; // can write more data now...
break;
}
break;
}
default:
@ -256,6 +271,7 @@ int antd_recv(void *src, void* data, int len)
break;
}
}
source->attempt = 0;
/*
int stat, r, st;
do{

3
libs/handle.h
View File

@ -26,7 +26,7 @@
#define __RESULT__ "{\"result\":%d,\"msg\":\"%s\"}"
#define FORM_URL_ENCODE "application/x-www-form-urlencoded"
#define FORM_MULTI_PART "multipart/form-data"
#define MAX_ATTEMPT 500
#ifdef USE_OPENSSL
int __attribute__((weak)) usessl();
#endif
@ -37,6 +37,7 @@ typedef struct{
char* ip;
#ifdef USE_OPENSSL
int status;
int attempt;
#endif
} antd_client_t;

26
libs/scheduler.c
View File

@ -24,8 +24,8 @@ static void stop(antd_scheduler_t* scheduler)
scheduler->status = 0;
// unlock all idle workers if any
for (int i = 0; i < scheduler->n_workers; i++)
sem_post(&scheduler->worker_sem);
sem_post(&scheduler->scheduler_sem);
sem_post(scheduler->worker_sem);
sem_post(scheduler->scheduler_sem);
for (int i = 0; i < scheduler->n_workers; i++)
if(scheduler->workers[i].id != -1)
pthread_join(scheduler->workers[i].tid, NULL);
@ -34,8 +34,8 @@ static void stop(antd_scheduler_t* scheduler)
pthread_mutex_destroy(&scheduler->scheduler_lock);
pthread_mutex_destroy(&scheduler->worker_lock);
pthread_mutex_destroy(&scheduler->pending_lock);
sem_destroy(&scheduler->scheduler_sem);
sem_destroy(&scheduler->worker_sem);
sem_close(scheduler->scheduler_sem);
sem_close(scheduler->worker_sem);
}
static antd_task_item_t dequeue(antd_task_queue_t* q)
@ -130,7 +130,7 @@ static void work(antd_worker_t* worker)
if(!it)
{
//LOG("Worker %d goes to idle state\n", worker->id);
sem_wait(&scheduler->worker_sem);
sem_wait(scheduler->worker_sem);
}
else
{
@ -153,14 +153,16 @@ void antd_scheduler_init(antd_scheduler_t* scheduler, int n)
scheduler->workers_queue = NULL;
scheduler->pending_task = 0 ;
// init semaphore
if (sem_init(&scheduler->scheduler_sem, 0, 0) == -1)
scheduler->scheduler_sem = shm_open("scheduler", O_RDWR | O_CREAT, S_IRWXU);
if (!scheduler->scheduler_sem)
{
LOG("Cannot init semaphore for scheduler\n");
LOG("Cannot open semaphore for scheduler\n");
exit(-1);
}
if (sem_init(&scheduler->worker_sem, 0, 0) == -1)
scheduler->worker_sem = shm_open("worker", O_RDWR | O_CREAT, S_IRWXU);
if (!scheduler->worker_sem)
{
LOG("Cannot init semaphore for workers\n");
LOG("Cannot open semaphore for workers\n");
exit(-1);
}
// init lock
@ -239,7 +241,7 @@ void antd_add_task(antd_scheduler_t* scheduler, antd_task_t* task)
scheduler->pending_task++;
pthread_mutex_unlock(&scheduler->pending_lock);
// wake up the scheduler if idle
sem_post(&scheduler->scheduler_sem);
sem_post(scheduler->scheduler_sem);
}
@ -326,7 +328,7 @@ int antd_task_schedule(antd_scheduler_t* scheduler)
enqueue(&scheduler->workers_queue, it->task);
pthread_mutex_unlock(&scheduler->worker_lock);
// wake up idle worker
sem_post(&scheduler->worker_sem);
sem_post(scheduler->worker_sem);
free(it);
}
return 1;
@ -340,7 +342,7 @@ void antd_wait(antd_scheduler_t* scheduler)
if(!stat)
{
// no task found, go to idle state
sem_wait(&scheduler->scheduler_sem);
sem_wait(scheduler->scheduler_sem);
}
}
}

4
libs/scheduler.h
View File

@ -64,8 +64,8 @@ typedef struct {
pthread_mutex_t worker_lock;
pthread_mutex_t pending_lock;
// event handle
sem_t scheduler_sem;
sem_t worker_sem;
sem_t *scheduler_sem;
sem_t *worker_sem;
// worker and data
antd_task_queue_t task_queue[N_PRIORITY];
antd_task_queue_t workers_queue;