fumiama/simple-dict
1
0
Fork 0
mirror of https://github.com/fumiama/simple-dict.git synced 2026-06-11 21:50:58 +08:00
Code Activity

fix deadlock

Browse Source
This commit is contained in:
源文雨
2022-05-04 21:13:57 +08:00
parent 8389832951
commit 28ec387b25
2 changed files with 70 additions and 70 deletions
Download Patch File Download Diff File Expand all files Collapse all files

57
dict.h
View File

@@ -20,17 +20,16 @@ struct dict_t {
typedef struct dict_t dict_t; typedef struct dict_t dict_t;
#define DICTSZ sizeof(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 char* dict_filepath;
static uint8_t dict_md5[16]; 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 = NULL; //fp for EX
static FILE* dict_fp_read = NULL; //fp for md5 static FILE* dict_fp_read = NULL; //fp for md5
static FILE* dict_thread_fp[THREADCNT]; static FILE* dict_thread_fp[THREADCNT];
static pthread_rwlock_t mu;
#ifdef CPUBIT64 #ifdef CPUBIT64
#define _dict_md5_2 ((uint64_t*)&dict_md5) #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); uint8_t* dict_buff = (uint8_t*)malloc(size);
if(dict_buff) { if(dict_buff) {
if(pthread_rwlock_tryrdlock(mu)) { if(pthread_rwlock_tryrdlock(mu)) {
perror("Readlock busy: "); perror("Readlock busy");
return 1; return 1;
} }
rewind(dict_fp_read); rewind(dict_fp_read);
@@ -68,11 +67,11 @@ static int fill_md5(pthread_rwlock_t* mu) {
} else { } else {
pthread_rwlock_unlock(mu); pthread_rwlock_unlock(mu);
free(dict_buff); free(dict_buff);
perror("Read dict error: "); perror("Read dict error");
return 2; return 2;
} }
} else { } else {
perror("Allocate memory error: "); perror("Allocate memory error");
return 3; return 3;
} }
} }
@@ -83,32 +82,34 @@ static int init_dict(char* file_path, pthread_rwlock_t* mu) {
if(dict_fp) { if(dict_fp) {
int err = pthread_rwlock_init(mu, NULL); int err = pthread_rwlock_init(mu, NULL);
if(err) { if(err) {
perror("Init lock error: "); perror("Init lock error");
return 1; return 1;
} }
dict_filepath = file_path; dict_filepath = file_path;
return fill_md5(mu); return fill_md5(mu);
} }
perror("Open dict error: "); perror("Open dict error");
return 2; return 2;
} }
static FILE* open_dict(uint8_t lock_type, uint32_t index, pthread_rwlock_t* mu) { static inline FILE* open_ex_dict() {
if(lock_type & DICT_LOCK_EX) { if(pthread_rwlock_wrlock(&mu)) {
if(pthread_rwlock_wrlock(mu)) { perror("Open dict: Writelock busy");
puts("Open dict: Writelock busy"); return NULL;
return NULL;
}
if(!dict_fp) dict_fp = fopen(dict_filepath, "rb+");
else rewind(dict_fp);
return dict_fp;
} }
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) { if(index >= THREADCNT) {
puts("Open dict: Index out of bounds"); puts("Open dict: Index out of bounds");
return NULL; return NULL;
} }
if(pthread_rwlock_tryrdlock(mu)) { if(pthread_rwlock_tryrdlock(&mu)) {
puts("Open dict: Readlock busy"); perror("Open dict: Readlock busy");
return NULL; return NULL;
} }
if(!dict_thread_fp[index]) dict_thread_fp[index] = fopen(dict_filepath, "rb"); 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; return dict_fp_read;
} }
static void close_dict(uint8_t lock_type, uint32_t index, pthread_rwlock_t* mu) { static inline void close_ex_dict() {
if(lock_type & DICT_LOCK_EX) fflush(dict_fp); fflush(dict_fp);
pthread_rwlock_unlock(mu); is_ex_dict_open = 0;
puts("Close dict"); 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) { static inline int is_dict_md5_equal(uint8_t* digest) {

83
server.c
View File

@@ -32,12 +32,11 @@
struct thread_timer_t { struct thread_timer_t {
uint32_t index; uint32_t index;
uint32_t lock_type; int accept_fd;
char *dat, *ptr; char *dat, *ptr;
time_t touch; time_t touch;
ssize_t numbytes; ssize_t numbytes;
pthread_t thread; pthread_t thread;
int accept_fd;
}; };
typedef struct thread_timer_t thread_timer_t; typedef struct thread_timer_t thread_timer_t;
static thread_timer_t timers[THREADCNT]; static thread_timer_t timers[THREADCNT];
@@ -50,7 +49,6 @@ static dict_t* setdict;
static uint32_t* items_len; static uint32_t* items_len;
static CONFIG cfg; static CONFIG cfg;
static pthread_attr_t attr; static pthread_attr_t attr;
static pthread_rwlock_t mu;
#define DICTPOOLSZ (((uint32_t)-1)>>((sizeof(uint32_t)*8-DICTPOOLBIT))) #define DICTPOOLSZ (((uint32_t)-1)>>((sizeof(uint32_t)*8-DICTPOOLBIT)))
static dict_t* dict_pool[DICTPOOLSZ+1]; static dict_t* dict_pool[DICTPOOLSZ+1];
@@ -59,7 +57,6 @@ static void accept_client();
static void accept_timer(void *p); static void accept_timer(void *p);
static uint16_t bind_server(uint16_t port); static uint16_t bind_server(uint16_t port);
static void cleanup_thread(thread_timer_t* timer); 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 enum SERVERACK del(FILE *fp, char* key, int len, char ret[4]);
static void handle_accept(void *accept_fd_p); static void handle_accept(void *accept_fd_p);
static void handle_int(int signo); 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) { static int send_all(thread_timer_t *timer) {
int re = 1; int re = 1;
FILE *fp = open_dict(DICT_LOCK_SH, timer->index, &mu); FILE *fp = open_shared_dict(timer->index);
if(!fp) return 1; if(!fp) return 1;
pthread_cleanup_push((void*)&pthread_rwlock_unlock, (void*)&mu); pthread_cleanup_push((void*)&close_shared_dict, NULL);
timer->lock_type = DICT_LOCK_SH;
off_t len = 0, file_size = get_dict_size(); off_t len = 0, file_size = get_dict_size();
char* buf = (char*)malloc(file_size); char* buf = (char*)malloc(file_size);
if(buf) { if(buf) {
@@ -171,9 +167,7 @@ static int send_all(thread_timer_t *timer) {
} }
pthread_cleanup_pop(1); pthread_cleanup_pop(1);
} }
close_dict(DICT_LOCK_SH, timer->index, &mu); pthread_cleanup_pop(1);
timer->lock_type = DICT_LOCK_UN;
pthread_cleanup_pop(0);
return re; return re;
} }
@@ -218,11 +212,10 @@ static void init_dict_pool(FILE *fp) {
static int s1_get(thread_timer_t *timer) { static int s1_get(thread_timer_t *timer) {
uint8_t digest[16]; uint8_t digest[16];
FILE *fp = open_dict(DICT_LOCK_SH, timer->index, &mu); FILE *fp = open_shared_dict(timer->index);
//timer->status = 0; //timer->status = 0;
while(fp) { while(fp) {
int ch; int ch;
timer->lock_type = DICT_LOCK_SH;
md5((uint8_t*)timer->dat, strlen(timer->dat)+1, digest); md5((uint8_t*)timer->dat, strlen(timer->dat)+1, digest);
uint8_t* dp = digest; uint8_t* dp = digest;
int p = ((*((uint32_t*)digest))>>(8*sizeof(uint32_t)-DICTPOOLBIT))&DICTPOOLSZ; 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; 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; // 哈希碰撞 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) { 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)) { while(has_next(fp, ch)) {
@@ -242,7 +236,8 @@ static int s1_get(thread_timer_t *timer) {
if(!strcmp(timer->dat, d->key)) { if(!strcmp(timer->dat, d->key)) {
int r; int r;
pthread_cleanup_push((void*)free, (void*)spb); 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); pthread_cleanup_pop(1);
return r; return r;
} else free(spb); } else free(spb);
@@ -250,16 +245,15 @@ static int s1_get(thread_timer_t *timer) {
break; 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) { static int s2_set(thread_timer_t *timer) {
uint8_t digest[16]; uint8_t digest[16];
timer->lock_type = DICT_LOCKING_EX; FILE *fp = open_ex_dict();
FILE *fp = open_dict(DICT_LOCK_EX, timer->index, &mu);
if(fp) { if(fp) {
touch_timer(timer); touch_timer(timer);
timer->lock_type = DICT_LOCK_EX;
md5((uint8_t*)timer->dat, strlen(timer->dat)+1, digest); md5((uint8_t*)timer->dat, strlen(timer->dat)+1, digest);
uint8_t* dp = digest; uint8_t* dp = digest;
int p = ((*((uint32_t*)digest))>>(8*sizeof(uint32_t)-DICTPOOLBIT))&DICTPOOLSZ; int p = ((*((uint32_t*)digest))>>(8*sizeof(uint32_t)-DICTPOOLBIT))&DICTPOOLSZ;
@@ -274,7 +268,10 @@ static int s2_set(thread_timer_t *timer) {
else { // 已有值 else { // 已有值
char ret[4]; 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]; setdict = dict_pool[p];
} }
} }
@@ -288,7 +285,7 @@ static int s2_set(thread_timer_t *timer) {
fseek(fp, 0, SEEK_END); fseek(fp, 0, SEEK_END);
return send_data(timer->accept_fd, timer->index, ACKDATA, "data", 4); return send_data(timer->accept_fd, timer->index, ACKDATA, "data", 4);
} else { } else {
timer->lock_type = DICT_LOCK_UN; close_ex_dict();
//timer->status = 0; //timer->status = 0;
return send_data(timer->accept_fd, timer->index, ACKERRO, "erro", 4); 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)) { 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); 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); 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]) { 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) { static int s4_del(thread_timer_t *timer) {
uint8_t digest[16]; uint8_t digest[16];
char ret[4]; char ret[4];
timer->lock_type = DICT_LOCK_EX; FILE *fp = open_ex_dict();
FILE *fp = open_dict(DICT_LOCK_EX, timer->index, &mu);
//timer->status = 0; //timer->status = 0;
if(fp) { if(fp) {
md5((uint8_t*)timer->dat, strlen(timer->dat)+1, digest); 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 c = 16-4;
int notok = 1; 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; // 哈希碰撞 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]); free(dict_pool[p]);
dict_pool[p] = NULL; 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) { static int s5_md5(thread_timer_t *timer) {
@@ -403,9 +407,7 @@ static void accept_timer(void *p) {
sleep(MAXWAITSEC / 4); sleep(MAXWAITSEC / 4);
time_t waitsec = time(NULL) - timer->touch; time_t waitsec = time(NULL) - timer->touch;
printf("Wait sec: %u, max: %u\n", (unsigned int)waitsec, MAXWAITSEC); printf("Wait sec: %u, max: %u\n", (unsigned int)waitsec, MAXWAITSEC);
if(timer->lock_type >= DICT_LOCK_EX) { if(waitsec > MAXWAITSEC+2) break;
if(waitsec > MAXWAITSEC*THREADCNT) break;
} else if(waitsec > MAXWAITSEC) break;
} }
if(thread) { if(thread) {
@@ -432,7 +434,6 @@ static void cleanup_thread(thread_timer_t* timer) {
timer->ptr = NULL; timer->ptr = NULL;
puts("Free data"); puts("Free data");
} }
if(timer->lock_type) close_dict(timer->lock_type, timer->index, &mu);
puts("Finish cleaning"); puts("Finish cleaning");
} }
@@ -498,14 +499,14 @@ static void handle_accept(void *p) {
switch(cp->cmd) { switch(cp->cmd) {
case CMDGET: case CMDGET:
//timer_pointer_of(p)->status = 1; //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; break;
case CMDCAT: 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; break;
case CMDMD5: case CMDMD5:
//timer_pointer_of(p)->status = 5; //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; break;
case CMDACK: break; case CMDACK: break;
case CMDEND: case CMDEND:
@@ -524,16 +525,14 @@ static void handle_accept(void *p) {
switch(cp->cmd) { switch(cp->cmd) {
case CMDSET: case CMDSET:
//timer_pointer_of(p)->status = 2; //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; break;
case CMDDEL: case CMDDEL:
//timer_pointer_of(p)->status = 4; //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; break;
case CMDDAT: case CMDDAT:
if(timer_pointer_of(p)->lock_type == DICT_LOCK_EX) { if(is_ex_dict_open && !s3_set_data(timer_pointer_of(p))) goto CONV_END;
if(!s3_set_data(timer_pointer_of(p))) goto CONV_END;
}
break; break;
default: 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 argequ(i, arg) (*(uint16_t*)argv[i] == *(uint16_t*)(arg))
#define showUsage(program) \ #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) 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)