From 28ec387b2568b90c34cdf4f3bbd45435014c5fce Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=BA=90=E6=96=87=E9=9B=A8?= <41315874+fumiama@users.noreply.github.com> Date: Wed, 4 May 2022 21:13:57 +0800 Subject: [PATCH] fix deadlock --- dict.h | 57 +++++++++++++++++++++----------------- server.c | 83 ++++++++++++++++++++++++++------------------------------ 2 files changed, 70 insertions(+), 70 deletions(-) diff --git a/dict.h b/dict.h index b0f5dc2..f9335d6 100644 --- a/dict.h +++ b/dict.h @@ -20,17 +20,16 @@ struct dict_t { typedef struct dict_t dict_t; #define DICTSZ sizeof(dict_t) -#define DICT_LOCK_UN 0x00 -#define DICT_LOCK_SH 0x01 -#define DICT_LOCK_EX 0x02 -#define DICT_LOCKING_EX 0x04 - static char* dict_filepath; static uint8_t dict_md5[16]; +static volatile int is_ex_dict_open; + static FILE* dict_fp = NULL; //fp for EX static FILE* dict_fp_read = NULL; //fp for md5 static FILE* dict_thread_fp[THREADCNT]; +static pthread_rwlock_t mu; + #ifdef CPUBIT64 #define _dict_md5_2 ((uint64_t*)&dict_md5) @@ -56,7 +55,7 @@ static int fill_md5(pthread_rwlock_t* mu) { uint8_t* dict_buff = (uint8_t*)malloc(size); if(dict_buff) { if(pthread_rwlock_tryrdlock(mu)) { - perror("Readlock busy: "); + perror("Readlock busy"); return 1; } rewind(dict_fp_read); @@ -68,11 +67,11 @@ static int fill_md5(pthread_rwlock_t* mu) { } else { pthread_rwlock_unlock(mu); free(dict_buff); - perror("Read dict error: "); + perror("Read dict error"); return 2; } } else { - perror("Allocate memory error: "); + perror("Allocate memory error"); return 3; } } @@ -83,32 +82,34 @@ static int init_dict(char* file_path, pthread_rwlock_t* mu) { if(dict_fp) { int err = pthread_rwlock_init(mu, NULL); if(err) { - perror("Init lock error: "); + perror("Init lock error"); return 1; } dict_filepath = file_path; return fill_md5(mu); } - perror("Open dict error: "); + perror("Open dict error"); return 2; } -static FILE* open_dict(uint8_t lock_type, uint32_t index, pthread_rwlock_t* mu) { - if(lock_type & DICT_LOCK_EX) { - if(pthread_rwlock_wrlock(mu)) { - puts("Open dict: Writelock busy"); - return NULL; - } - if(!dict_fp) dict_fp = fopen(dict_filepath, "rb+"); - else rewind(dict_fp); - return dict_fp; +static inline FILE* open_ex_dict() { + if(pthread_rwlock_wrlock(&mu)) { + perror("Open dict: Writelock busy"); + return NULL; } + if(!dict_fp) dict_fp = fopen(dict_filepath, "rb+"); + else rewind(dict_fp); + if(dict_fp) is_ex_dict_open = 1; + return dict_fp; +} + +static inline FILE* open_shared_dict(uint32_t index) { if(index >= THREADCNT) { puts("Open dict: Index out of bounds"); return NULL; } - if(pthread_rwlock_tryrdlock(mu)) { - puts("Open dict: Readlock busy"); + if(pthread_rwlock_tryrdlock(&mu)) { + perror("Open dict: Readlock busy"); return NULL; } if(!dict_thread_fp[index]) dict_thread_fp[index] = fopen(dict_filepath, "rb"); @@ -125,10 +126,16 @@ static FILE* get_dict_fp_rd() { return dict_fp_read; } -static void close_dict(uint8_t lock_type, uint32_t index, pthread_rwlock_t* mu) { - if(lock_type & DICT_LOCK_EX) fflush(dict_fp); - pthread_rwlock_unlock(mu); - puts("Close dict"); +static inline void close_ex_dict() { + fflush(dict_fp); + is_ex_dict_open = 0; + pthread_rwlock_unlock(&mu); + puts("Close ex dict"); +} + +static inline void close_shared_dict() { + pthread_rwlock_unlock(&mu); + puts("Close shared dict"); } static inline int is_dict_md5_equal(uint8_t* digest) { diff --git a/server.c b/server.c index 5aefce6..17aaf81 100644 --- a/server.c +++ b/server.c @@ -32,12 +32,11 @@ struct thread_timer_t { uint32_t index; - uint32_t lock_type; + int accept_fd; char *dat, *ptr; time_t touch; ssize_t numbytes; pthread_t thread; - int accept_fd; }; typedef struct thread_timer_t thread_timer_t; static thread_timer_t timers[THREADCNT]; @@ -50,7 +49,6 @@ static dict_t* setdict; static uint32_t* items_len; static CONFIG cfg; static pthread_attr_t attr; -static pthread_rwlock_t mu; #define DICTPOOLSZ (((uint32_t)-1)>>((sizeof(uint32_t)*8-DICTPOOLBIT))) static dict_t* dict_pool[DICTPOOLSZ+1]; @@ -59,7 +57,6 @@ static void accept_client(); static void accept_timer(void *p); static uint16_t bind_server(uint16_t port); static void cleanup_thread(thread_timer_t* timer); -static int close_and_send(thread_timer_t* timer, enum SERVERACK cmd, char *data, size_t numbytes); static enum SERVERACK del(FILE *fp, char* key, int len, char ret[4]); static void handle_accept(void *accept_fd_p); static void handle_int(int signo); @@ -144,10 +141,9 @@ static int send_data(int accept_fd, int index, enum SERVERACK cmd, char *data, s static int send_all(thread_timer_t *timer) { int re = 1; - FILE *fp = open_dict(DICT_LOCK_SH, timer->index, &mu); + FILE *fp = open_shared_dict(timer->index); if(!fp) return 1; - pthread_cleanup_push((void*)&pthread_rwlock_unlock, (void*)&mu); - timer->lock_type = DICT_LOCK_SH; + pthread_cleanup_push((void*)&close_shared_dict, NULL); off_t len = 0, file_size = get_dict_size(); char* buf = (char*)malloc(file_size); if(buf) { @@ -171,9 +167,7 @@ static int send_all(thread_timer_t *timer) { } pthread_cleanup_pop(1); } - close_dict(DICT_LOCK_SH, timer->index, &mu); - timer->lock_type = DICT_LOCK_UN; - pthread_cleanup_pop(0); + pthread_cleanup_pop(1); return re; } @@ -218,11 +212,10 @@ static void init_dict_pool(FILE *fp) { static int s1_get(thread_timer_t *timer) { uint8_t digest[16]; - FILE *fp = open_dict(DICT_LOCK_SH, timer->index, &mu); + FILE *fp = open_shared_dict(timer->index); //timer->status = 0; while(fp) { int ch; - timer->lock_type = DICT_LOCK_SH; md5((uint8_t*)timer->dat, strlen(timer->dat)+1, digest); uint8_t* dp = digest; int p = ((*((uint32_t*)digest))>>(8*sizeof(uint32_t)-DICTPOOLBIT))&DICTPOOLSZ; @@ -232,7 +225,8 @@ static int s1_get(thread_timer_t *timer) { int notok = 1; while(dict_pool[p] && (notok=strcmp(timer->dat, dict_pool[p]->key)) && c-->0) p = ((*((uint32_t*)(++dp)))>>(8*sizeof(uint32_t)-DICTPOOLBIT))&DICTPOOLSZ; // 哈希碰撞 if(!notok) { - return close_and_send(timer, ACKSUCC, dict_pool[p]->data, last_nonnull(dict_pool[p]->data, DICTDATSZ)); + close_shared_dict(); + return send_data(timer->accept_fd, timer->index, ACKSUCC, dict_pool[p]->data, last_nonnull(dict_pool[p]->data, DICTDATSZ)); } while(has_next(fp, ch)) { @@ -242,7 +236,8 @@ static int s1_get(thread_timer_t *timer) { if(!strcmp(timer->dat, d->key)) { int r; pthread_cleanup_push((void*)free, (void*)spb); - r = close_and_send(timer, ACKSUCC, d->data, last_nonnull(d->data, DICTDATSZ)); + close_shared_dict(); + r = send_data(timer->accept_fd, timer->index, ACKSUCC, d->data, last_nonnull(d->data, DICTDATSZ)); pthread_cleanup_pop(1); return r; } else free(spb); @@ -250,16 +245,15 @@ static int s1_get(thread_timer_t *timer) { break; } - return close_and_send(timer, ACKNULL, "null", 4); + close_shared_dict(); + return send_data(timer->accept_fd, timer->index, ACKNULL, "null", 4); } static int s2_set(thread_timer_t *timer) { uint8_t digest[16]; - timer->lock_type = DICT_LOCKING_EX; - FILE *fp = open_dict(DICT_LOCK_EX, timer->index, &mu); + FILE *fp = open_ex_dict(); if(fp) { touch_timer(timer); - timer->lock_type = DICT_LOCK_EX; md5((uint8_t*)timer->dat, strlen(timer->dat)+1, digest); uint8_t* dp = digest; int p = ((*((uint32_t*)digest))>>(8*sizeof(uint32_t)-DICTPOOLBIT))&DICTPOOLSZ; @@ -274,7 +268,10 @@ static int s2_set(thread_timer_t *timer) { else { // 已有值 char ret[4]; // 先删去 - if(del(fp, timer->dat, timer->numbytes+1, ret) == ACKERRO) return close_and_send(timer, ACKERRO, "erro", 4); + if(del(fp, timer->dat, timer->numbytes+1, ret) == ACKERRO) { + close_ex_dict(); + return send_data(timer->accept_fd, timer->index, ACKERRO, "erro", 4); + } setdict = dict_pool[p]; } } @@ -288,7 +285,7 @@ static int s2_set(thread_timer_t *timer) { fseek(fp, 0, SEEK_END); return send_data(timer->accept_fd, timer->index, ACKDATA, "data", 4); } else { - timer->lock_type = DICT_LOCK_UN; + close_ex_dict(); //timer->status = 0; return send_data(timer->accept_fd, timer->index, ACKERRO, "erro", 4); } @@ -304,10 +301,12 @@ static int s3_set_data(thread_timer_t *timer) { if(!set_pb(get_dict_fp_wr(), items_len, sizeof(dict_t), setdict)) { fprintf(stderr, "Error set data: dict[%s]=%s\n", setdict->key, timer->dat); - return close_and_send(timer, ACKERRO, "erro", 4); + close_ex_dict(); + return send_data(timer->accept_fd, timer->index, ACKERRO, "erro", 4); } printf("Set data: dict[%s]=%s\n", setdict->key, timer->dat); - return close_and_send(timer, ACKSUCC, "succ", 4); + close_ex_dict(); + return send_data(timer->accept_fd, timer->index, ACKSUCC, "succ", 4); } static enum SERVERACK del(FILE *fp, char* key, int len, char ret[4]) { @@ -365,8 +364,7 @@ static enum SERVERACK del(FILE *fp, char* key, int len, char ret[4]) { static int s4_del(thread_timer_t *timer) { uint8_t digest[16]; char ret[4]; - timer->lock_type = DICT_LOCK_EX; - FILE *fp = open_dict(DICT_LOCK_EX, timer->index, &mu); + FILE *fp = open_ex_dict(); //timer->status = 0; if(fp) { md5((uint8_t*)timer->dat, strlen(timer->dat)+1, digest); @@ -375,12 +373,18 @@ static int s4_del(thread_timer_t *timer) { int c = 16-4; int notok = 1; while(dict_pool[p] && (notok=strcmp(timer->dat, dict_pool[p]->key)) && c-->0) p = ((*((uint32_t*)(++dp)))>>(8*sizeof(uint32_t)-DICTPOOLBIT))&DICTPOOLSZ; // 哈希碰撞 - if(notok) return close_and_send(timer, ACKNULL, "null", 4); + if(notok) { + close_ex_dict(); + return send_data(timer->accept_fd, timer->index, ACKNULL, "null", 4); + } free(dict_pool[p]); dict_pool[p] = NULL; - return close_and_send(timer, del(fp, timer->dat, timer->numbytes+1, ret), ret, 4); + int r = send_data(timer->accept_fd, timer->index, del(fp, timer->dat, timer->numbytes+1, ret), ret, 4); + close_ex_dict(); + return r; } - return close_and_send(timer, ACKNULL, "null", 4); + close_ex_dict(); + return send_data(timer->accept_fd, timer->index, ACKNULL, "null", 4); } static int s5_md5(thread_timer_t *timer) { @@ -403,9 +407,7 @@ static void accept_timer(void *p) { sleep(MAXWAITSEC / 4); time_t waitsec = time(NULL) - timer->touch; printf("Wait sec: %u, max: %u\n", (unsigned int)waitsec, MAXWAITSEC); - if(timer->lock_type >= DICT_LOCK_EX) { - if(waitsec > MAXWAITSEC*THREADCNT) break; - } else if(waitsec > MAXWAITSEC) break; + if(waitsec > MAXWAITSEC+2) break; } if(thread) { @@ -432,7 +434,6 @@ static void cleanup_thread(thread_timer_t* timer) { timer->ptr = NULL; puts("Free data"); } - if(timer->lock_type) close_dict(timer->lock_type, timer->index, &mu); puts("Finish cleaning"); } @@ -498,14 +499,14 @@ static void handle_accept(void *p) { switch(cp->cmd) { case CMDGET: //timer_pointer_of(p)->status = 1; - if(!s1_get(timer_pointer_of(p))) goto CONV_END; + if(!is_ex_dict_open && !s1_get(timer_pointer_of(p))) goto CONV_END; break; case CMDCAT: - if(!send_all(timer_pointer_of(p))) goto CONV_END; + if(!is_ex_dict_open && !send_all(timer_pointer_of(p))) goto CONV_END; break; case CMDMD5: //timer_pointer_of(p)->status = 5; - if(!s5_md5(timer_pointer_of(p))) goto CONV_END; + if(!is_ex_dict_open && !s5_md5(timer_pointer_of(p))) goto CONV_END; break; case CMDACK: break; case CMDEND: @@ -524,16 +525,14 @@ static void handle_accept(void *p) { switch(cp->cmd) { case CMDSET: //timer_pointer_of(p)->status = 2; - if(!s2_set(timer_pointer_of(p))) goto CONV_END; + if(!is_ex_dict_open && !s2_set(timer_pointer_of(p))) goto CONV_END; break; case CMDDEL: //timer_pointer_of(p)->status = 4; - if(!s4_del(timer_pointer_of(p))) goto CONV_END; + if(!is_ex_dict_open && !s4_del(timer_pointer_of(p))) goto CONV_END; break; case CMDDAT: - if(timer_pointer_of(p)->lock_type == DICT_LOCK_EX) { - if(!s3_set_data(timer_pointer_of(p))) goto CONV_END; - } + if(is_ex_dict_open && !s3_set_data(timer_pointer_of(p))) goto CONV_END; break; default: goto CONV_END; break; } @@ -625,12 +624,6 @@ static void accept_client() { } } -static int close_and_send(thread_timer_t* timer, enum SERVERACK cmd, char *data, size_t numbytes) { - close_dict(timer->lock_type, timer->index, &mu); - timer->lock_type = DICT_LOCK_UN; - return send_data(timer->accept_fd, timer->index, cmd, data, numbytes); -} - #define argequ(i, arg) (*(uint16_t*)argv[i] == *(uint16_t*)(arg)) #define showUsage(program) \ printf("Usage:\n%s [-d] listen_port dict_file [config_file | -]\n\t-d: As daemon\n\t- : Read config from env SDS_PWD & SDS_SPS\n", program)