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

优化处理流程

Browse Source
This commit is contained in:
源文雨
2022-05-06 13:17:03 +08:00
parent 21773b3710
commit 4ff61bf495
5 changed files with 219 additions and 137 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
client.c
View File

@@ -32,7 +32,7 @@ static char sps[64] = "minamoto";
void getMessage(void *p) { void getMessage(void *p) {
int c = 0, offset = 0; int c = 0, offset = 0;
CMDPACKET* cp = (CMDPACKET*)bufr; cmdpacket_t cp = (cmdpacket_t)bufr;
while(offset >= CMDPACKET_HEAD_LEN || (c = recv(sockfd, bufr+offset, CMDPACKET_HEAD_LEN-offset, MSG_WAITALL)) > 0) { while(offset >= CMDPACKET_HEAD_LEN || (c = recv(sockfd, bufr+offset, CMDPACKET_HEAD_LEN-offset, MSG_WAITALL)) > 0) {
#ifdef DEBUG #ifdef DEBUG
printf("Recv %d bytes.\n", c); printf("Recv %d bytes.\n", c);
@@ -127,7 +127,7 @@ off_t file_size_of(const char* fname) {
else return -1; else return -1;
} }
void send_cmd(int accept_fd, CMDPACKET* p) { void send_cmd(int accept_fd, cmdpacket_t p) {
#ifdef DEBUG #ifdef DEBUG
printf("send %d bytes encrypted data with %d bytes head.\n", p->datalen, CMDPACKET_HEAD_LEN); printf("send %d bytes encrypted data with %d bytes head.\n", p->datalen, CMDPACKET_HEAD_LEN);
printf("raw packet: "); printf("raw packet: ");
@@ -188,7 +188,7 @@ int main(int argc,char *argv[]) { //usage: ./client host port
} }
else { else {
buf[3] = 0; buf[3] = 0;
CMDPACKET* p = malloc(CMDPACKET_LEN_MAX); cmdpacket_t p = malloc(CMDPACKET_LEN_MAX);
if(!strcmp(buf, "set")) { if(!strcmp(buf, "set")) {
p->cmd = CMDSET; p->cmd = CMDSET;
p->datalen = strlen(buf+4); p->datalen = strlen(buf+4);

4
crypto.h
View File

@@ -93,7 +93,7 @@ static char* raw_decrypt(const char* buf, off_t* len, int index, const char pwd[
return decbuf; return decbuf;
} }
static void cmdpacket_encrypt(CMDPACKET* p, int index, const char pwd[64]) { static void cmdpacket_encrypt(cmdpacket_t p, int index, const char pwd[64]) {
TEADAT tin = {p->datalen, p->data}; TEADAT tin = {p->datalen, p->data};
TEADAT tout; TEADAT tout;
TEA tea[4]; TEA tea[4];
@@ -134,7 +134,7 @@ static void cmdpacket_encrypt(CMDPACKET* p, int index, const char pwd[64]) {
return; return;
} }
static int cmdpacket_decrypt(CMDPACKET* p, int index, const char pwd[64]) { static int cmdpacket_decrypt(cmdpacket_t p, int index, const char pwd[64]) {
TEADAT tin = {p->datalen, p->data}; TEADAT tin = {p->datalen, p->data};
TEADAT tout; TEADAT tout;
TEA tea[4]; TEA tea[4];

8
dict.h
View File

@@ -24,6 +24,7 @@ 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 is_ex_dict_open;
static volatile int is_ex_dict_opening;
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
@@ -92,10 +93,13 @@ static int init_dict(char* file_path, pthread_rwlock_t* mu) {
} }
static inline FILE* open_ex_dict() { static inline FILE* open_ex_dict() {
is_ex_dict_opening = 1;
if(pthread_rwlock_wrlock(&mu)) { if(pthread_rwlock_wrlock(&mu)) {
perror("Open dict: Writelock busy"); perror("Open dict: Writelock busy");
is_ex_dict_opening = 0;
return NULL; return NULL;
} }
is_ex_dict_opening = 0;
if(!dict_fp) dict_fp = fopen(dict_filepath, "rb+"); if(!dict_fp) dict_fp = fopen(dict_filepath, "rb+");
else rewind(dict_fp); else rewind(dict_fp);
if(dict_fp) is_ex_dict_open = 1; if(dict_fp) is_ex_dict_open = 1;
@@ -116,10 +120,6 @@ static inline FILE* open_shared_dict(uint32_t index) {
return dict_thread_fp[index]; return dict_thread_fp[index];
} }
static FILE* get_dict_fp_wr() {
return dict_fp;
}
static FILE* get_dict_fp_rd() { static FILE* get_dict_fp_rd() {
rewind(dict_fp_read); rewind(dict_fp_read);
return dict_fp_read; return dict_fp_read;

325
server.c
View File

@@ -37,16 +37,14 @@ struct thread_timer_t {
ssize_t numbytes; ssize_t numbytes;
char *dat; char *dat;
pthread_t thread; pthread_t thread;
uint8_t buf[BUFSIZ]; uint8_t buf[BUFSIZ-sizeof(uint32_t)-sizeof(int)-sizeof(time_t)-sizeof(ssize_t)-sizeof(char*)-sizeof(pthread_t)];
}; };
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];
#define timer_pointer_of(x) ((thread_timer_t*)(x)) #define timer_pointer_of(x) ((thread_timer_t*)(x))
#define touch_timer(x) (timer_pointer_of(x)->touch = time(NULL)) #define touch_timer(x) (timer_pointer_of(x)->touch = time(NULL))
static int fd; // server fd static dict_t setdicts[THREADCNT];
static dict_t d;
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;
@@ -54,86 +52,87 @@ static pthread_attr_t attr;
#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];
static void accept_client(); static void accept_client(int fd);
static void accept_timer(void *p); static void accept_timer(void *p);
static uint16_t bind_server(uint16_t port); static int bind_server(uint16_t* port);
static void cleanup_thread(thread_timer_t* timer); static void cleanup_thread(thread_timer_t* timer);
static enum SERVERACK del(FILE *fp, char* key, int len, char ret[4]); static server_ack_t del(FILE *fp, const 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);
static void handle_pipe(int signo); static void handle_pipe(int signo);
static void handle_quit(int signo); static void handle_quit(int signo);
static void init_dict_pool(FILE *fp); static void init_dict_pool(FILE *fp);
static int insert_item(FILE *fp, const dict_t* dict, int keysize, int datasize);
static void kill_timer(pthread_t thread); static void kill_timer(pthread_t thread);
static uint32_t last_nonnull(char* p, uint32_t max_size); static inline uint32_t last_nonnull(const char* p, uint32_t max_size);
static int listen_socket(); static int listen_socket(int fd);
static int send_all(thread_timer_t *timer); static int send_all(thread_timer_t *timer);
static int send_data(int accept_fd, int index, enum SERVERACK cmd, char *data, size_t length); static int send_data(int accept_fd, int index, server_ack_t cmd, const char *data, size_t length);
static int s1_get(thread_timer_t *timer); static int s1_get(thread_timer_t *timer);
static int s2_set(thread_timer_t *timer); static int s2_set(thread_timer_t *timer);
static int s3_set_data(thread_timer_t *timer); static int s3_set_data(thread_timer_t *timer);
static int s4_del(thread_timer_t *timer); static int s4_del(thread_timer_t *timer);
static int s5_md5(thread_timer_t *timer); static int s5_md5(thread_timer_t *timer);
static uint16_t bind_server(uint16_t port) { static int bind_server(uint16_t* port) {
#ifdef LISTEN_ON_IPV6 #ifdef LISTEN_ON_IPV6
server_addr.sin6_family = AF_INET6; server_addr.sin6_family = AF_INET6;
server_addr.sin6_port = htons(port); server_addr.sin6_port = htons(*port);
bzero(&(server_addr.sin6_addr), sizeof(server_addr.sin6_addr)); bzero(&(server_addr.sin6_addr), sizeof(server_addr.sin6_addr));
fd = socket(PF_INET6, SOCK_STREAM, 0); int fd = socket(PF_INET6, SOCK_STREAM, 0);
#else #else
server_addr.sin_family = AF_INET; server_addr.sin_family = AF_INET;
server_addr.sin_port = htons(port); server_addr.sin_port = htons(*port);
server_addr.sin_addr.s_addr = INADDR_ANY; server_addr.sin_addr.s_addr = INADDR_ANY;
bzero(&(server_addr.sin_zero), 8); bzero(&(server_addr.sin_zero), 8);
fd = socket(AF_INET, SOCK_STREAM, 0); int fd = socket(AF_INET, SOCK_STREAM, 0);
#endif #endif
if(!~bind(fd, (struct sockaddr *)&server_addr, struct_len)) { if(!~bind(fd, (struct sockaddr *)&server_addr, struct_len)) {
perror("Bind server failure: "); perror("Bind server failure");
return 0; return 0;
} }
#ifdef LISTEN_ON_IPV6 #ifdef LISTEN_ON_IPV6
port = ntohs(server_addr.sin6_port); *port = ntohs(server_addr.sin6_port);
struct in6_addr in = server_addr.sin6_addr; struct in6_addr in = server_addr.sin6_addr;
char str[INET6_ADDRSTRLEN]; // 46 char str[INET6_ADDRSTRLEN]; // 46
inet_ntop(AF_INET6, &in, str, sizeof(str)); inet_ntop(AF_INET6, &in, str, sizeof(str));
#else #else
port = ntohs(server_addr.sin_port); *port = ntohs(server_addr.sin_port);
struct in_addr in = server_addr.sin_addr; struct in_addr in = server_addr.sin_addr;
char str[INET_ADDRSTRLEN]; // 16 char str[INET_ADDRSTRLEN]; // 16
inet_ntop(AF_INET, &in, str, sizeof(str)); inet_ntop(AF_INET, &in, str, sizeof(str));
#endif #endif
printf("Bind server successfully on %s:%u\n", str, port); printf("Bind server successfully on %s:%u\n", str, *port);
return port; return fd;
} }
static int listen_socket() { static int listen_socket(int fd) {
if(!~listen(fd, THREADCNT)) { if(!~listen(fd, THREADCNT)) {
perror("Listen failed: "); perror("Listen failed");
return 0; return 0;
} }
puts("Listening..."); puts("Listening...");
return 1; return fd;
} }
static uint32_t last_nonnull(char* p, uint32_t max_size) { static inline uint32_t last_nonnull(const char* p, uint32_t max_size) {
if(max_size > 1) while(!p[max_size - 1]) max_size--; if(max_size > 1) while(!p[max_size - 1]) max_size--;
return max_size; return max_size;
} }
static int send_data(int accept_fd, int index, enum SERVERACK cmd, char *data, size_t length) { static int send_data(int accept_fd, int index, server_ack_t cmd, const char *data, size_t length) {
char buf[CMDPACKET_LEN_MAX]; char buf[CMDPACKET_LEN_MAX];
CMDPACKET* p = (CMDPACKET*)buf; cmdpacket_t p = (cmdpacket_t)buf;
p->cmd = (uint8_t)cmd; p->cmd = (uint8_t)cmd;
p->datalen = length; p->datalen = length;
memcpy(p->data, data, p->datalen); memcpy(p->data, data, p->datalen);
cmdpacket_encrypt(p, index, cfg.pwd); cmdpacket_encrypt(p, index, cfg.pwd);
int total = CMDPACKET_HEAD_LEN+p->datalen; int total = CMDPACKET_HEAD_LEN+p->datalen;
if(!~send(accept_fd, buf, total, 0)) { if(!~send(accept_fd, buf, total, 0)) {
perror("Send data error: "); perror("Send data error");
return 0; return 0;
} else { } else {
printf("Send %d bytes data: ", total); printf("Send %d bytes data", total);
for(int i = 0; i < length; i++) putchar(data[i]); for(int i = 0; i < length; i++) putchar(data[i]);
putchar('\n'); putchar('\n');
return 1; return 1;
@@ -176,10 +175,11 @@ static int send_all(thread_timer_t *timer) {
static void init_dict_pool(FILE *fp) { static void init_dict_pool(FILE *fp) {
uint8_t digest[16]; uint8_t digest[16];
uint8_t buf[8+DICTSZ];
int ch; int ch;
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 = get_pb(fp); SIMPLE_PB* spb = read_pb_into(fp, (SIMPLE_PB*)buf);
if(!spb) continue; // skip error bytes if(!spb) continue; // skip error bytes
dict_t* d = (dict_t*)spb->target; dict_t* d = (dict_t*)spb->target;
md5((uint8_t *)d->key, strlen(d->key)+1, digest); md5((uint8_t *)d->key, strlen(d->key)+1, digest);
@@ -206,13 +206,12 @@ static void init_dict_pool(FILE *fp) {
memcpy(dnew, d, sizeof(dict_t)); memcpy(dnew, d, sizeof(dict_t));
dict_pool[p] = dnew; // 解决哈希冲突 dict_pool[p] = dnew; // 解决哈希冲突
} }
free(spb);
} }
} }
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];
FILE *fp = open_shared_dict(timer->index); FILE *fp = open_shared_dict(timer->index);
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 ret = -1; int ret = -1;
@@ -235,14 +234,13 @@ static int s1_get(thread_timer_t *timer) {
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 = get_pb(fp); SIMPLE_PB* spb = read_pb_into(fp, (SIMPLE_PB*)buf);
if(!spb) continue; // skip error bytes
dict_t* d = (dict_t*)spb->target; dict_t* d = (dict_t*)spb->target;
if(!strcmp(timer->dat, d->key)) { if(!strcmp(timer->dat, d->key)) {
pthread_cleanup_push((void*)free, (void*)spb);
ret = send_data(timer->accept_fd, timer->index, ACKSUCC, d->data, last_nonnull(d->data, DICTDATSZ)); ret = send_data(timer->accept_fd, timer->index, ACKSUCC, d->data, last_nonnull(d->data, DICTDATSZ));
pthread_cleanup_pop(1);
break; break;
} else free(spb); }
} }
break; break;
@@ -253,92 +251,154 @@ static int s1_get(thread_timer_t *timer) {
} }
static int s2_set(thread_timer_t *timer) { static int s2_set(thread_timer_t *timer) {
uint8_t digest[16]; memset(&setdicts[timer->index], 0, sizeof(dict_t));
FILE *fp = open_ex_dict(); strncpy(setdicts[timer->index].key, timer->dat, DICTKEYSZ-1);
int r; md5((uint8_t*)timer->dat, strlen(timer->dat)+1, setdicts[timer->index].data);
if(fp == NULL) return send_data(timer->accept_fd, timer->index, ACKERRO, "erro", 4); return send_data(timer->accept_fd, timer->index, ACKDATA, "data", 4);
pthread_cleanup_push((void*)&close_ex_dict, NULL); }
while(1) {
touch_timer(timer);
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;
if(!dict_pool[p]) setdict = dict_pool[p] = (dict_t*)malloc(sizeof(dict_t)); static int insert_item(FILE *fp, const dict_t* dict, int keysize, int datasize) {
else { int ch;
int c = 16-4; const char* key = dict->key;
int notok; const char* data = dict->data;
while(dict_pool[p] && (notok=strcmp(timer->dat, dict_pool[p]->key)) && c-->0) p = ((*((uint32_t*)(++dp)))>>(8*sizeof(uint32_t)-DICTPOOLBIT))&DICTPOOLSZ; // 哈希碰撞 uint8_t buf[8+DICTSZ];
if(!dict_pool[p]) setdict = dict_pool[p] = (dict_t*)malloc(sizeof(dict_t)); // 无值 while(has_next(fp, ch)) {
else if(notok) setdict = &d; // 全部冲突 if(!ch) continue; // skip null bytes
else { // 已有值 SIMPLE_PB* spb = read_pb_into(fp, (SIMPLE_PB*)buf);
char ret[4]; if(!spb) continue; // skip error bytes
// 先删去 dict_t* d = (dict_t*)spb->target;
if(del(fp, timer->dat, timer->numbytes+1, ret) == ACKERRO) { if(memcmp(key, d->key, keysize)) continue;
close_ex_dict(); int datalen = last_nonnull(d->data, DICTDATSZ);
r = send_data(timer->accept_fd, timer->index, ACKERRO, "erro", 4); if(datalen == datasize) { // 新增 data 可以直接覆盖到原 data
if(fseek(fp, -(int)(spb->real_len), SEEK_CUR)) goto ERR_INSERT_ITEM;
goto TRY_INSERT_ITEM;
}
uint32_t next = ftell(fp);
uint32_t this = next - spb->real_len;
if(fseek(fp, 0, SEEK_END)) goto ERR_INSERT_ITEM;
uint32_t end = ftell(fp);
if(next == end) {
if(!ftruncate(fileno(fp), end - spb->real_len)) {
if(fseek(fp, 0, SEEK_END)) goto ERR_INSERT_ITEM;
goto TRY_INSERT_ITEM;
}
goto ERR_INSERT_ITEM;
}
uint32_t cap = end - next;
#ifdef DEBUG
printf("this: %u, next: %u, end: %u, cap: %u\n", this, next, end, cap);
#endif
char* data = malloc(cap);
int iserr = 1;
if(data) {
pthread_cleanup_push((void*)&free, data);
while(1) {
if(fseek(fp, next, SEEK_SET)) {
iserr = 1;
break; break;
} }
setdict = dict_pool[p]; if(fread(data, cap, 1, fp) == 1) {
if(!ftruncate(fileno(fp), end - spb->real_len)) {
if(fseek(fp, this, SEEK_SET)) {
iserr = 1;
break;
}
if(fwrite(data, cap, 1, fp) == 1) {
iserr = 0;
break;
}
}
}
break;
} }
pthread_cleanup_pop(1);
} }
#ifdef DEBUG if(iserr) goto ERR_INSERT_ITEM;
printf("item %s(%0*x) will fill into %p.\n", timer->dat, DICTPOOLBIT/4, p, dict_pool[p]);
#endif
//timer->status = 3;
memset(setdict, 0, sizeof(dict_t));
strncpy(setdict->key, timer->dat, DICTKEYSZ-1);
fseek(fp, 0, SEEK_END);
r = send_data(timer->accept_fd, timer->index, ACKDATA, "data", 4);
break; break;
} }
pthread_cleanup_pop(0); if(fseek(fp, 0, SEEK_END)) goto ERR_INSERT_ITEM;
return r; TRY_INSERT_ITEM:
if(set_pb(fp, items_len, sizeof(dict_t), dict))
return 0;
ERR_INSERT_ITEM:
return 1;
} }
static int s3_set_data(thread_timer_t *timer) { static int s3_set_data(thread_timer_t *timer) {
//timer->status = 0; //timer->status = 0;
uint32_t datasize = (timer->numbytes > (DICTDATSZ-1))?(DICTDATSZ-1):timer->numbytes; FILE *fp = open_ex_dict();
int ret; if(fp == NULL) return send_data(timer->accept_fd, timer->index, ACKERRO, "erro", 4);
int datasize = (timer->numbytes > (DICTDATSZ-1))?(DICTDATSZ-1):timer->numbytes;
#ifdef DEBUG #ifdef DEBUG
printf("Set data size: %u\n", datasize); printf("Set data size: %u\n", datasize);
#endif #endif
if(datasize <= 0 || datasize > sizeof(setdicts[timer->index].data))
return send_data(timer->accept_fd, timer->index, ACKERRO, "erro", 4);
int r;
pthread_cleanup_push((void*)&close_ex_dict, NULL); pthread_cleanup_push((void*)&close_ex_dict, NULL);
memcpy(setdict->data, timer->dat, datasize);
if(!set_pb(get_dict_fp_wr(), items_len, sizeof(dict_t), setdict)) { uint8_t* dp = (uint8_t*)setdicts[timer->index].data;
touch_timer(timer);
int p = ((*((uint32_t*)dp))>>(8*sizeof(uint32_t)-DICTPOOLBIT))&DICTPOOLSZ;
dict_t* setdict;
if(!dict_pool[p]) {
setdict = dict_pool[p] = (dict_t*)malloc(sizeof(dict_t));
memcpy(setdict->key, setdicts[timer->index].key, DICTKEYSZ);
}
else {
int c = 16-4;
int notok;
while(dict_pool[p] && (notok=strcmp(setdicts[timer->index].key, dict_pool[p]->key)) && c-->0) p = ((*((uint32_t*)(++dp)))>>(8*sizeof(uint32_t)-DICTPOOLBIT))&DICTPOOLSZ; // 哈希碰撞
if(!dict_pool[p]) {
setdict = dict_pool[p] = (dict_t*)malloc(sizeof(dict_t)); // 无值
memcpy(setdict->key, setdicts[timer->index].key, DICTKEYSZ);
}
else if(notok) setdict = &setdicts[timer->index]; // 全部冲突
else setdict = dict_pool[p]; // 已有值
}
#ifdef DEBUG
printf("item %s(%0*x) will fill into %p.\n", setdicts[timer->index].key, DICTPOOLBIT/4, p, dict_pool[p]);
#endif
memset(setdict->data, 0, DICTDATSZ);
memcpy(setdict->data, timer->dat, timer->numbytes);
if(insert_item(fp, setdict, strlen(setdict->key)+1, datasize)) {
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);
ret = send_data(timer->accept_fd, timer->index, ACKERRO, "erro", 4); r = send_data(timer->accept_fd, timer->index, ACKERRO, "erro", 4);
} else { } else {
printf("Set data: dict[%s]=%s\n", setdict->key, timer->dat); printf("Set data: dict[%s]=%s\n", setdict->key, timer->dat);
ret = send_data(timer->accept_fd, timer->index, ACKSUCC, "succ", 4); r = send_data(timer->accept_fd, timer->index, ACKSUCC, "succ", 4);
} }
pthread_cleanup_pop(1); pthread_cleanup_pop(1);
return ret; return r;
} }
static enum SERVERACK del(FILE *fp, char* key, int len, char ret[4]) { static server_ack_t del(FILE *fp, const char* key, int len, char ret[4]) {
int ch; int ch;
uint8_t buf[8+DICTSZ];
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 = get_pb(fp); SIMPLE_PB* spb = read_pb_into(fp, (SIMPLE_PB*)buf);
if(!spb) continue; // skip error bytes
dict_t* d = (dict_t*)spb->target; dict_t* d = (dict_t*)spb->target;
if(memcmp(key, d->key, len)) { if(memcmp(key, d->key, len)) continue;
free(spb);
continue;
}
uint32_t next = ftell(fp); uint32_t next = ftell(fp);
uint32_t this = next - spb->real_len; uint32_t this = next - spb->real_len;
fseek(fp, 0, SEEK_END); if(fseek(fp, 0, SEEK_END)) {
*(uint32_t*)ret = *(uint32_t*)"erro";
return ACKERRO;
}
uint32_t end = ftell(fp); uint32_t end = ftell(fp);
if(next == end) { if(next == end) {
if(!ftruncate(fileno(fp), end - spb->real_len)) { if(!ftruncate(fileno(fp), end - spb->real_len)) {
free(spb);
*(uint32_t*)ret = *(uint32_t*)"succ"; *(uint32_t*)ret = *(uint32_t*)"succ";
return ACKSUCC; return ACKSUCC;
} }
free(spb);
*(uint32_t*)ret = *(uint32_t*)"erro"; *(uint32_t*)ret = *(uint32_t*)"erro";
return ACKERRO; return ACKERRO;
} }
@@ -347,22 +407,34 @@ static enum SERVERACK del(FILE *fp, char* key, int len, char ret[4]) {
printf("this: %u, next: %u, end: %u, cap: %u\n", this, next, end, cap); printf("this: %u, next: %u, end: %u, cap: %u\n", this, next, end, cap);
#endif #endif
char* data = malloc(cap); char* data = malloc(cap);
server_ack_t ack = 0;
if(data) { if(data) {
fseek(fp, next, SEEK_SET); pthread_cleanup_push((void*)&free, data);
if(fread(data, cap, 1, fp) == 1) { while(1) {
if(!ftruncate(fileno(fp), end - spb->real_len)) { if(fseek(fp, next, SEEK_SET)) {
fseek(fp, this, SEEK_SET); *(uint32_t*)ret = *(uint32_t*)"erro";
if(fwrite(data, cap, 1, fp) == 1) { ack = ACKERRO;
free(data); break;
free(spb); }
*(uint32_t*)ret = *(uint32_t*)"succ"; if(fread(data, cap, 1, fp) == 1) {
return ACKSUCC; if(!ftruncate(fileno(fp), end - spb->real_len)) {
if(fseek(fp, this, SEEK_SET)) {
*(uint32_t*)ret = *(uint32_t*)"erro";
ack = ACKERRO;
break;
}
if(fwrite(data, cap, 1, fp) == 1) {
*(uint32_t*)ret = *(uint32_t*)"succ";
ack = ACKSUCC;
break;
}
} }
} }
break;
} }
free(data); pthread_cleanup_pop(1);
} }
free(spb); if(ack) return ack;
*(uint32_t*)ret = *(uint32_t*)"erro"; *(uint32_t*)ret = *(uint32_t*)"erro";
return ACKERRO; return ACKERRO;
} }
@@ -422,9 +494,10 @@ 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);
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+is_ex_dict_open?MAXWAITSEC:0) { if(waitsec > MAXWAITSEC) {
pthread_kill(thread, SIGQUIT); pthread_kill(thread, SIGQUIT);
puts("Kill thread"); puts("Kill thread");
break; break;
@@ -455,7 +528,7 @@ static void cleanup_thread(thread_timer_t* timer) {
static void handle_int(int signo) { static void handle_int(int signo) {
puts("Keyboard interrupted"); puts("Keyboard interrupted");
exit(0); pthread_exit(NULL);
} }
static void handle_pipe(int signo) { static void handle_pipe(int signo) {
@@ -467,7 +540,7 @@ static void handle_accept(void *p) {
puts("\nConnected to the client, handling accept..."); puts("\nConnected to the client, handling accept...");
pthread_t thread; pthread_t thread;
if (pthread_create(&thread, &attr, (void *)&accept_timer, p)) { if (pthread_create(&thread, &attr, (void *)&accept_timer, p)) {
perror("Error creating timer thread: "); perror("Error creating timer thread");
cleanup_thread(timer_pointer_of(p)); cleanup_thread(timer_pointer_of(p));
return; return;
} }
@@ -477,7 +550,7 @@ static void handle_accept(void *p) {
int accept_fd = timer_pointer_of(p)->accept_fd; int accept_fd = timer_pointer_of(p)->accept_fd;
uint32_t index = timer_pointer_of(p)->index; uint32_t index = timer_pointer_of(p)->index;
uint8_t *buff = timer_pointer_of(p)->buf; uint8_t *buff = timer_pointer_of(p)->buf;
CMDPACKET* cp = (CMDPACKET*)buff; cmdpacket_t cp = (cmdpacket_t)buff;
ssize_t numbytes = 0, offset = 0; ssize_t numbytes = 0, offset = 0;
while( while(
offset >= CMDPACKET_HEAD_LEN offset >= CMDPACKET_HEAD_LEN
@@ -546,7 +619,7 @@ static void handle_accept(void *p) {
if(!is_ex_dict_open && !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(is_ex_dict_open && !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; break;
default: goto CONV_END; break; default: goto CONV_END; break;
} }
@@ -573,17 +646,17 @@ static void handle_accept(void *p) {
puts("Thread exited normally"); puts("Thread exited normally");
} }
static void accept_client() { static void accept_client(int fd) {
pid_t pid = fork(); /*pid_t pid = fork();
while (pid > 0) { //主进程监控子进程状态，如果子进程异常终止则重启之 while (pid > 0) { //主进程监控子进程状态，如果子进程异常终止则重启之
wait(NULL); wait(NULL);
puts("Server subprocess exited. Restart..."); puts("Server subprocess exited. Restart...");
pid = fork(); pid = fork();
} }
while(pid < 0) { while(pid < 0) {
perror("Error when forking a subprocess: "); perror("Error when forking a subprocess");
sleep(1); sleep(1);
} }*/
signal(SIGINT, handle_int); signal(SIGINT, handle_int);
signal(SIGQUIT, handle_quit); signal(SIGQUIT, handle_quit);
signal(SIGKILL, exit); signal(SIGKILL, exit);
@@ -609,7 +682,7 @@ static void accept_client() {
#endif #endif
int accept_fd; int accept_fd;
while((accept_fd=accept(fd, (struct sockaddr *)&client_addr, &struct_len))<=0) while((accept_fd=accept(fd, (struct sockaddr *)&client_addr, &struct_len))<=0)
perror("Accept client error: "); perror("Accept client error");
#ifdef LISTEN_ON_IPV6 #ifdef LISTEN_ON_IPV6
uint16_t port = ntohs(client_addr.sin6_port); uint16_t port = ntohs(client_addr.sin6_port);
struct in6_addr in = client_addr.sin6_addr; struct in6_addr in = client_addr.sin6_addr;
@@ -628,7 +701,7 @@ static void accept_client() {
timer->touch = time(NULL); timer->touch = time(NULL);
reset_seq(p); reset_seq(p);
if (pthread_create(&timer->thread, &attr, (void *)&handle_accept, timer)) { if (pthread_create(&timer->thread, &attr, (void *)&handle_accept, timer)) {
perror("Error creating thread: "); perror("Error creating thread");
cleanup_thread(timer); cleanup_thread(timer);
continue; continue;
} }
@@ -638,7 +711,7 @@ static void accept_client() {
#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)
int main(int argc, char *argv[]) { int main(int argc, char *argv[]) {
if(argc != 4 && argc != 5) { if(argc != 4 && argc != 5) {
showUsage(argv[0]); showUsage(argv[0]);
@@ -652,14 +725,14 @@ int main(int argc, char *argv[]) {
return 1; return 1;
} }
if(as_daemon && daemon(1, 1)<0) { if(as_daemon && daemon(1, 1)<0) {
perror("Start daemon error: "); perror("Start daemon error");
return 2; return 2;
} }
FILE *fp = NULL; FILE *fp = NULL;
fp = fopen(argv[as_daemon?3:2], "rb+"); fp = fopen(argv[as_daemon?3:2], "rb+");
if(!fp) fp = fopen(argv[as_daemon?3:2], "wb+"); if(!fp) fp = fopen(argv[as_daemon?3:2], "wb+");
if(!fp) { if(!fp) {
fprintf(stderr, "Error opening dict file: %s : ", argv[as_daemon?3:2]); fprintf(stderr, "Error opening dict file: %s", argv[as_daemon?3:2]);
perror(""); perror("");
return 3; return 3;
} }
@@ -690,24 +763,30 @@ int main(int argc, char *argv[]) {
} }
if(!fp) fp = fopen(argv[as_daemon?4:3], "rb"); if(!fp) fp = fopen(argv[as_daemon?4:3], "rb");
if(fp == NULL) { if(fp == NULL) {
fprintf(stderr, "Error opening config file: %s : ", argv[as_daemon?4:3]); fprintf(stderr, "Error opening config file: %s", argv[as_daemon?4:3]);
perror(""); perror("fopen");
return 7; return 7;
} }
if(~((int)fp)) { if(~((int)fp)) {
SIMPLE_PB* spb = get_pb(fp); uint8_t buf[8+sizeof(CONFIG)];
cfg = *(CONFIG*)spb->target; SIMPLE_PB* spb = read_pb_into(fp, (SIMPLE_PB*)buf);
if(!spb) {
fprintf(stderr, "Error reading config file: %s\n", argv[as_daemon?4:3]);
return 8;
}
cfg = *(CONFIG*)(spb->target);
fclose(fp); fclose(fp);
free(spb);
} }
items_len = align_struct(sizeof(dict_t), 2, d.key, d.data); items_len = align_struct(sizeof(dict_t), 2, setdicts[0].key, setdicts[0].data);
if(!items_len) { if(!items_len) {
fputs("Align struct error", stderr); fputs("Align struct error", stderr);
return 8; return 9;
} }
if(!bind_server((uint16_t)port)) return 9; int fd;
if(!listen_socket()) return 10; if(!(fd=bind_server((uint16_t*)&port))) return 10;
accept_client(); if(!listen_socket(fd)) return 11;
close(fd); pthread_cleanup_push((void*)&close, (void*)((long long)fd));
return 11; accept_client(fd);
pthread_cleanup_pop(1);
return 0;
} }

11
server.h
View File

@@ -9,16 +9,19 @@
// DICTPOOLBIT must be lower than 4*8 = 32 // DICTPOOLBIT must be lower than 4*8 = 32
#define DICTPOOLBIT 16 #define DICTPOOLBIT 16
enum SERVERCMD {CMDGET, CMDCAT, CMDMD5, CMDACK, CMDEND, CMDSET, CMDDEL, CMDDAT}; enum server_cmd_t {CMDGET, CMDCAT, CMDMD5, CMDACK, CMDEND, CMDSET, CMDDEL, CMDDAT};
enum SERVERACK {ACKNONE=0b0000011, ACKSUCC=0b0010011, ACKDATA=0b0100011, ACKNULL=0b0110011, ACKNEQU=0b1000011, ACKERRO=0b1010011}; enum server_ack_t {ACKNONE=0b0000011, ACKSUCC=0b0010011, ACKDATA=0b0100011, ACKNULL=0b0110011, ACKNEQU=0b1000011, ACKERRO=0b1010011};
struct CMDPACKET { typedef enum server_cmd_t server_cmd_t;
typedef enum server_ack_t server_ack_t;
struct cmdpacket_t {
uint8_t cmd; // SERVERCMD or SERVERACK uint8_t cmd; // SERVERCMD or SERVERACK
uint8_t datalen; // data len is less than 255 uint8_t datalen; // data len is less than 255
uint8_t md5[16]; // md5 digest of data below uint8_t md5[16]; // md5 digest of data below
uint8_t data[]; // with TEA encoding, 64 bytes will be 160 bytes uint8_t data[]; // with TEA encoding, 64 bytes will be 160 bytes
}; };
typedef struct CMDPACKET CMDPACKET; typedef struct cmdpacket_t* cmdpacket_t;
#define CMDPACKET_HEAD_LEN (1+1+16) #define CMDPACKET_HEAD_LEN (1+1+16)
#define CMDPACKET_LEN_MAX (CMDPACKET_HEAD_LEN+255) #define CMDPACKET_LEN_MAX (CMDPACKET_HEAD_LEN+255)