fumiama/simple-dict
1
0
Fork 0
mirror of https://github.com/fumiama/simple-dict.git synced 2026-06-23 12:40:25 +08:00
Code Activity

use single dict fp

Browse Source
This commit is contained in:
源文雨
2022-05-08 15:37:18 +08:00
parent 16c839e7d5
commit 9d49aa2849
2 changed files with 56 additions and 83 deletions
Download Patch File Download Diff File Expand all files Collapse all files

96
dict.h
View File

@@ -23,12 +23,11 @@ typedef struct dict_t dict_t;
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 volatile int has_dict_opened;
static volatile int is_ex_dict_opening; static volatile int is_dict_opening;
static volatile uint32_t ex_dict_owner_index = (uint32_t)-1; static volatile uint32_t dict_owner_index = (uint32_t)-1;
static FILE* dict_fp = NULL; // fp for EX static FILE* dict_fp = NULL;
static FILE* dict_thread_fp[THREADCNT];
static pthread_rwlock_t mu; static pthread_rwlock_t mu;
#ifdef CPUBIT64 #ifdef CPUBIT64
@@ -54,17 +53,11 @@ static int fill_md5(FILE* fp) {
} }
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)) {
perror("Readlock busy");
return 1;
}
if(fread(dict_buff, size, 1, fp) == 1) { if(fread(dict_buff, size, 1, fp) == 1) {
pthread_rwlock_unlock(&mu);
md5(dict_buff, size, dict_md5); md5(dict_buff, size, dict_md5);
free(dict_buff); free(dict_buff);
return 0; return 0;
} else { } else {
pthread_rwlock_unlock(&mu);
free(dict_buff); free(dict_buff);
perror("Read dict error"); perror("Read dict error");
return 2; return 2;
@@ -76,61 +69,43 @@ static int fill_md5(FILE* fp) {
} }
static int init_dict(char* file_path) { static int init_dict(char* file_path) {
dict_fp = fopen(file_path, "rb+"); FILE* fp = fopen(file_path, "rb+");
if(dict_fp) { if(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(dict_fp); err = fill_md5(fp);
fclose(fp);
return err;
} }
perror("Open dict error"); perror("Open dict error");
return 2; return 2;
} }
static inline FILE* open_ex_dict(uint32_t index) { static inline FILE* open_dict(uint32_t index, int isro) {
is_ex_dict_opening = 1; is_dict_opening = 1;
if(pthread_rwlock_wrlock(&mu)) { if(pthread_rwlock_wrlock(&mu)) {
perror("Open ex dict: Writelock busy"); perror("Open dict: Writelock busy");
is_ex_dict_opening = 0; is_dict_opening = 0;
return NULL; return NULL;
} }
is_ex_dict_opening = 0; is_dict_opening = 0;
if(!dict_fp) dict_fp = fopen(dict_filepath, "rb+"); dict_fp = fopen(dict_filepath, isro?"rb":"rb+");
else rewind(dict_fp);
if(!dict_fp) { if(!dict_fp) {
perror("Open ex dict: fopen"); perror("Open dict: fopen");
pthread_rwlock_unlock(&mu); pthread_rwlock_unlock(&mu);
return NULL; return NULL;
} }
is_ex_dict_open = 1; has_dict_opened = 1;
ex_dict_owner_index = index; dict_owner_index = index;
puts("Open ex dict"); puts("Open dict");
return dict_fp; return dict_fp;
} }
static inline FILE* open_shared_dict(uint32_t index, int requirelock) { static inline int require_shared_lock() {
if(index >= THREADCNT) {
puts("Open dict: Index out of bounds");
return NULL;
}
if(requirelock && pthread_rwlock_tryrdlock(&mu)) {
perror("Open dict: Readlock busy");
return NULL;
}
if(!dict_thread_fp[index]) dict_thread_fp[index] = fopen(dict_filepath, "rb");
else rewind(dict_thread_fp[index]);
puts("Open shared dict");
return dict_thread_fp[index];
}
static inline int require_shared_lock(uint32_t index) {
if(index >= THREADCNT) {
puts("Open dict: Index out of bounds");
return 1;
}
if(pthread_rwlock_tryrdlock(&mu)) { if(pthread_rwlock_tryrdlock(&mu)) {
perror("Open dict: Readlock busy"); perror("Open dict: Readlock busy");
return 1; return 1;
@@ -139,26 +114,21 @@ static inline int require_shared_lock(uint32_t index) {
return 0; return 0;
} }
static inline void close_ex_dict(uint32_t index) { static inline void release_shared_lock() {
if(index != ex_dict_owner_index) return; pthread_rwlock_unlock(&mu);
if(is_ex_dict_open) { puts("Release shared lock");
fflush(dict_fp);
for(int i = 0; i < THREADCNT; i++) {
if(dict_thread_fp[i]) {
fclose(dict_thread_fp[i]); // 关闭所有 fp 以同步数据
dict_thread_fp[i] = NULL;
}
}
is_ex_dict_open = 0;
ex_dict_owner_index = (uint32_t)-1;
pthread_rwlock_unlock(&mu);
puts("Close ex dict");
} else puts("Ex dict already closed");
} }
static inline void close_shared_dict() { static inline void close_dict(uint32_t index) {
pthread_rwlock_unlock(&mu); if(index != dict_owner_index) return;
puts("Close shared dict"); if(has_dict_opened) {
fclose(dict_fp);
dict_fp = NULL;
has_dict_opened = 0;
dict_owner_index = (uint32_t)-1;
pthread_rwlock_unlock(&mu);
puts("Close dict");
} else puts("Dict already closed");
} }
static inline int is_dict_md5_equal(uint8_t* digest) { static inline int is_dict_md5_equal(uint8_t* digest) {

43
server.c
View File

@@ -141,9 +141,9 @@ static int send_data(int accept_fd, int index, server_ack_t cmd, const char *dat
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_shared_dict(timer->index, 1); FILE *fp = open_dict(timer->index, 1);
if(fp == NULL) return 1; if(fp == NULL) return 1;
pthread_cleanup_push((void*)&close_shared_dict, NULL); pthread_cleanup_push((void*)&close_dict, (void*)(uintptr_t)timer->index);
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) {
@@ -215,10 +215,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];
uint8_t buf[8+DICTSZ]; uint8_t buf[8+DICTSZ];
if(require_shared_lock(timer->index)) // busy if(require_shared_lock()) // busy
return send_data(timer->accept_fd, timer->index, ACKERRO, "erro", 4); return send_data(timer->accept_fd, timer->index, ACKERRO, "erro", 4);
int ret = -1; int ret = -1;
pthread_cleanup_push((void*)&close_shared_dict, NULL); pthread_cleanup_push((void*)&release_shared_lock, NULL);
while(1) { while(1) {
int ch; int ch;
md5((uint8_t*)timer->dat, strlen(timer->dat)+1, digest); md5((uint8_t*)timer->dat, strlen(timer->dat)+1, digest);
@@ -246,12 +246,13 @@ static int s1_get(thread_timer_t *timer) {
printf("cannot find any empty slot for digest of %s: %08x, open dict to find it.\n", timer->dat, p); printf("cannot find any empty slot for digest of %s: %08x, open dict to find it.\n", timer->dat, p);
#endif #endif
FILE *fp = open_shared_dict(timer->index, 0); // really open FILE *fp = open_dict(timer->index, 1); // really open
if(fp == NULL) { if(fp == NULL) {
ret = send_data(timer->accept_fd, timer->index, ACKERRO, "erro", 4); ret = send_data(timer->accept_fd, timer->index, ACKERRO, "erro", 4);
break; break;
} }
pthread_cleanup_push((void*)&close_dict, (void*)(uintptr_t)timer->index);
while(has_next(fp, ch)) { while(has_next(fp, ch)) {
if(!ch) continue; // skip null bytes if(!ch) continue; // skip null bytes
SIMPLE_PB* spb = read_pb_into(fp, (SIMPLE_PB*)buf); SIMPLE_PB* spb = read_pb_into(fp, (SIMPLE_PB*)buf);
@@ -262,6 +263,7 @@ static int s1_get(thread_timer_t *timer) {
break; break;
} }
} }
pthread_cleanup_pop(1);
break; break;
} }
@@ -349,7 +351,7 @@ ERR_INSERT_ITEM:
static int s3_set_data(thread_timer_t *timer) { static int s3_set_data(thread_timer_t *timer) {
if(!setdicts[timer->index].data[0]) return send_data(timer->accept_fd, timer->index, ACKERRO, "erro", 4); if(!setdicts[timer->index].data[0]) return send_data(timer->accept_fd, timer->index, ACKERRO, "erro", 4);
FILE *fp = open_ex_dict(timer->index); FILE *fp = open_dict(timer->index, 0);
if(fp == NULL) return send_data(timer->accept_fd, timer->index, ACKERRO, "erro", 4); if(fp == NULL) return send_data(timer->accept_fd, timer->index, ACKERRO, "erro", 4);
int datasize = (timer->numbytes > (DICTDATSZ-1))?(DICTDATSZ-1):timer->numbytes; int datasize = (timer->numbytes > (DICTDATSZ-1))?(DICTDATSZ-1):timer->numbytes;
@@ -360,7 +362,7 @@ static int s3_set_data(thread_timer_t *timer) {
return send_data(timer->accept_fd, timer->index, ACKERRO, "erro", 4); return send_data(timer->accept_fd, timer->index, ACKERRO, "erro", 4);
int r; int r;
pthread_cleanup_push((void*)&close_ex_dict, (void*)(uintptr_t)timer->index); pthread_cleanup_push((void*)&close_dict, (void*)(uintptr_t)timer->index);
uint8_t* dp = (uint8_t*)setdicts[timer->index].data; uint8_t* dp = (uint8_t*)setdicts[timer->index].data;
touch_timer(timer); touch_timer(timer);
@@ -482,10 +484,10 @@ static int s4_del(thread_timer_t *timer) {
uint8_t digest[16]; uint8_t digest[16];
char ret[4]; char ret[4];
int r; int r;
FILE *fp = open_ex_dict(timer->index); FILE *fp = open_dict(timer->index, 0);
if(fp == NULL) return send_data(timer->accept_fd, timer->index, ACKERRO, "erro", 4); if(fp == NULL) return send_data(timer->accept_fd, timer->index, ACKERRO, "erro", 4);
pthread_cleanup_push((void*)&close_ex_dict, (void*)(uintptr_t)timer->index); pthread_cleanup_push((void*)&close_dict, (void*)(uintptr_t)timer->index);
while(1) { while(1) {
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;
@@ -515,10 +517,10 @@ static int s4_del(thread_timer_t *timer) {
} }
static int s5_md5(thread_timer_t *timer) { static int s5_md5(thread_timer_t *timer) {
FILE* fp = open_shared_dict(timer->index, 1); FILE* fp = open_dict(timer->index, 1);
if(fp == NULL) return send_data(timer->accept_fd, timer->index, ACKERRO, "erro", 4); if(fp == NULL) return send_data(timer->accept_fd, timer->index, ACKERRO, "erro", 4);
int r; int r;
pthread_cleanup_push((void*)&close_shared_dict, NULL); pthread_cleanup_push((void*)&close_dict, (void*)(uintptr_t)timer->index);
fill_md5(fp); fill_md5(fp);
r = is_dict_md5_equal((uint8_t*)timer->dat); r = is_dict_md5_equal((uint8_t*)timer->dat);
pthread_cleanup_pop(1); pthread_cleanup_pop(1);
@@ -549,7 +551,7 @@ static void accept_timer(void *p) {
while(!pthread_kill(thread, 0)) { while(!pthread_kill(thread, 0)) {
sleep(MAXWAITSEC / 4); sleep(MAXWAITSEC / 4);
if(is_ex_dict_opening) touch_timer(p); if(is_dict_opening) touch_timer(p);
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(waitsec > MAXWAITSEC) { if(waitsec > MAXWAITSEC) {
@@ -573,7 +575,7 @@ static void cleanup_thread(thread_timer_t* timer) {
timer->accept_fd = 0; timer->accept_fd = 0;
puts("Close accept"); puts("Close accept");
} }
close_ex_dict(timer->index); close_dict(timer->index);
puts("Finish cleaning"); puts("Finish cleaning");
} }
@@ -636,13 +638,13 @@ static void handle_accept(void *p) {
printf("[normal] Get %zd bytes packet with cmd: %d, data: %s\n", offset, cp->cmd, cp->data); printf("[normal] Get %zd bytes packet with cmd: %d, data: %s\n", offset, cp->cmd, cp->data);
switch(cp->cmd) { switch(cp->cmd) {
case CMDGET: case CMDGET:
if(!is_ex_dict_open && !s1_get(timer_pointer_of(p))) goto CONV_END; if(!has_dict_opened && !s1_get(timer_pointer_of(p))) goto CONV_END;
break; break;
case CMDCAT: case CMDCAT:
if(!is_ex_dict_open && !send_all(timer_pointer_of(p))) goto CONV_END; if(!has_dict_opened && !send_all(timer_pointer_of(p))) goto CONV_END;
break; break;
case CMDMD5: case CMDMD5:
if(!is_ex_dict_open && !s5_md5(timer_pointer_of(p))) goto CONV_END; if(!has_dict_opened && !s5_md5(timer_pointer_of(p))) goto CONV_END;
break; break;
case CMDACK: case CMDACK:
case CMDEND: case CMDEND:
@@ -660,13 +662,13 @@ static void handle_accept(void *p) {
printf("[super] Get %zd bytes packet with data: %s\n", offset, cp->data); printf("[super] Get %zd bytes packet with data: %s\n", offset, cp->data);
switch(cp->cmd) { switch(cp->cmd) {
case CMDSET: case CMDSET:
if(!is_ex_dict_open && !s2_set(timer_pointer_of(p))) goto CONV_END; if(!has_dict_opened && !s2_set(timer_pointer_of(p))) goto CONV_END;
break; break;
case CMDDEL: case CMDDEL:
if(!is_ex_dict_open && !s4_del(timer_pointer_of(p))) goto CONV_END; if(!has_dict_opened && !s4_del(timer_pointer_of(p))) goto CONV_END;
break; break;
case CMDDAT: case CMDDAT:
if(!is_ex_dict_open && !s3_set_data(timer_pointer_of(p))) goto CONV_END; if(!has_dict_opened && !s3_set_data(timer_pointer_of(p))) goto CONV_END;
break; break;
default: goto CONV_END; break; default: goto CONV_END; break;
} }
@@ -712,7 +714,8 @@ static void accept_client(int fd) {
pthread_attr_init(&attr); pthread_attr_init(&attr);
pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED); pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);
init_crypto(); init_crypto();
init_dict_pool(open_shared_dict(0, 0)); init_dict_pool(open_dict(0, 1));
close_dict(0);
while(1) { while(1) {
puts("Ready for accept, waitting..."); puts("Ready for accept, waitting...");
int p = 0; int p = 0;