fumiama/simple-kanban
1
0
Fork 0
mirror of https://github.com/fumiama/simple-kanban.git synced 2026-06-18 16:30:25 +08:00
Code Activity

使用tcpool全新优化

Browse Source
This commit is contained in:
源文雨
2023-11-08 18:00:51 +09:00
parent d9e1c7374e
commit 5bc6279ee4
3 changed files with 708 additions and 512 deletions
Download Patch File Download Diff File Expand all files Collapse all files

98
file.h Normal file
View File

@@ -0,0 +1,98 @@
#ifndef _FILE_H_
#define _FILE_H_
#include <stdint.h>
#include <stdio.h>
#include <sys/stat.h>
#include <stdlib.h>
#include <string.h>
#include <pthread.h>
static char* file_filepath[2];
static volatile uint16_t has_file_opened[2];
static volatile uint16_t is_file_opening[2];
static volatile uint32_t file_owner_index[2] = {(uint32_t)-1, (uint32_t)-1};
static FILE* file_fp[2];
static pthread_rwlock_t mu[2];
static inline off_t get_file_size(int isdata) {
struct stat statbuf;
if(stat(file_filepath[!!isdata], &statbuf)==0) {
return statbuf.st_size;
}
return -1;
}
static int init_file(char* file_path[2]) {
for (int i = 0; i < 2; i++) {
FILE* fp = fopen(file_path[i], "rb+");
if(!fp) {
perror("Open file error");
return 2;
}
int err = pthread_rwlock_init(&mu[i], NULL);
if(err) {
perror("Init lock error");
return 1;
}
file_filepath[i] = file_path[i];
fclose(fp);
}
return 0;
}
static inline FILE* open_file(uint32_t index, int isdata, int isro) {
isdata = !!isdata;
is_file_opening[isdata] = 1;
if(pthread_rwlock_wrlock(&mu[isdata])) {
perror("Open file: Writelock busy");
is_file_opening[isdata] = 0;
return NULL;
}
is_file_opening[isdata] = 0;
file_fp[isdata] = fopen(file_filepath[isdata], isro?"rb":"rb+");
if(!file_fp[isdata]) {
perror("Open file: fopen");
pthread_rwlock_unlock(&mu[isdata]);
return NULL;
}
has_file_opened[isdata] = 1;
file_owner_index[isdata] = index;
puts("Open file");
return file_fp[isdata];
}
static inline int require_shared_lock(int isdata) {
if(pthread_rwlock_rdlock(&mu[!!isdata])) {
perror("Open file: Readlock busy");
return 1;
}
puts("Shared lock required");
return 0;
}
static inline void release_shared_lock(int isdata) {
pthread_rwlock_unlock(&mu[!!isdata]);
puts("Release shared lock");
}
static inline void close_file(uint32_t index, int isdata) {
isdata = !!isdata;
if(index != file_owner_index[isdata]) return;
if(has_file_opened[isdata]) {
fclose(file_fp[isdata]);
file_fp[isdata] = NULL;
has_file_opened[isdata] = 0;
file_owner_index[isdata] = (uint32_t)-1;
pthread_rwlock_unlock(&mu[isdata]);
puts("Close file");
} else puts("file already closed");
}
static void close_file_wrap(uint32_t index_isdata[2]) {
close_file(index_isdata[0], (int)index_isdata[1]);
}
#endif

692
server.c
View File

@@ -1,6 +1,5 @@
#include <arpa/inet.h> #include <arpa/inet.h>
#include <errno.h> #include <errno.h>
#include <fcntl.h>
#include <netdb.h> #include <netdb.h>
#include <netinet/in.h> #include <netinet/in.h>
#include <signal.h> #include <signal.h>
@@ -18,298 +17,59 @@
#include <time.h> #include <time.h>
#include <ctype.h> #include <ctype.h>
#include <unistd.h> #include <unistd.h>
#include "config.h"
#if !__APPLE__ #if !__APPLE__
#include <sys/sendfile.h> #include <sys/sendfile.h>
#include <endian.h> #include <endian.h>
#ifndef FD_COPY
#define FD_COPY(f, t) bcopy(f, t, sizeof(*(f)))
#endif
#else #else
#include <machine/endian.h> #include <machine/endian.h>
#endif #endif
#include "config.h"
static uint8_t _cfg[sizeof(simple_pb_t)+sizeof(config_t)]; #include "file.h"
#define cfg ((const const_config_t*)(_cfg+sizeof(simple_pb_t))) // 存储 pwd 和 sps
static int fd; // server socket fd
#ifdef LISTEN_ON_IPV6
static socklen_t struct_len = sizeof(struct sockaddr_in6);
static struct sockaddr_in6 server_addr;
static struct sockaddr_in6 client_addr;
#else
static socklen_t struct_len = sizeof(struct sockaddr_in);
static struct sockaddr_in server_addr;
static struct sockaddr_in client_addr;
#endif
static char *data_path; // cat 命令读取的文件位置 static char *data_path; // cat 命令读取的文件位置
static char *kanban_path; // get 命令读取的文件位置 static char *kanban_path; // get 命令读取的文件位置
/* lock operations for open_file */ static uint8_t _cfg[sizeof(simple_pb_t)+sizeof(config_t)];
#define LOCK_ALLF_UN 0x00 /* unlock all file */ #define cfg ((const const_config_t*)(_cfg+sizeof(simple_pb_t))) // 存储 pwd 和 sps
#define LOCK_DATA_SH 0x01 /* shared data file lock */
#define LOCK_DATA_EX 0x02 /* exclusive data file lock */
#define LOCK_KANB_SH 0x04 /* shared kanban file lock */
#define LOCK_KANB_EX 0x08 /* exclusive kanban file lock */
static int file_mode = LOCK_ALLF_UN; #define TCPOOL_THREAD_TIMER_T_SZ 65536
static int kanb_file_ro_cnt, data_file_ro_cnt; #define TCPOOL_MAXWAITSEC 16
#define SERVER_THREAD_BUFSZ ( \
TCPOOL_THREAD_TIMER_T_SZ \
-TCPOOL_THREAD_TIMER_T_HEAD_SZ \
-sizeof(ssize_t)-2*sizeof(uint8_t) \
)
#define TCPOOL_THREAD_CONTEXT \
ssize_t numbytes; /* 本次接收的数据长度 */ \
int8_t status; /* 本会话所处的状态 */ \
uint8_t isdata; /* 是否为 datfile */ \
char data[SERVER_THREAD_BUFSZ]
#define TCPOOL_TOUCH_TIMER_CONDITION (*(volatile uint32_t*)(is_file_opening))
#define TCPOOL_INIT_ACTION init_file((char*[]){kanban_path, data_path})
#define TCPOOL_PREHANDLE_ACCEPT_ACTION(timer) \
timer->status = -1; \
timer->isdata = 0;
#define TCPOOL_CLEANUP_THREAD_ACTION(timer) \
close_file(timer->index, timer->isdata); \
timer->status = -1;
// THREADCNT 并行的监听队列/select队列长度 #include "tcpool.h"
#define THREADCNT 16
// MAXWAITSEC 最大等待时间
#define MAXWAITSEC 16
#define TIMERDATSZ (BUFSIZ-sizeof(int)-sizeof(time_t)-sizeof(int)-sizeof(ssize_t)-sizeof(int8_t)-sizeof(uint8_t)*3-sizeof(FILE*))
static struct timeval timeout;
// 会话的上下文
// 包含了本次 accept 的全部信息
struct threadtimer_t {
int index; // 自身位置
int accept_fd; // 本次 accept 的 fd
time_t touch; // 最后访问时间，与当前时间差超过 MAXWAITSEC 将强行中断连接
ssize_t numbytes; // 本次接收的数据长度
FILE *fp; // 本会话打开的文件
int8_t status; // 本会话所处的状态
uint8_t is_open; // 标识 fp 是否正在使用
uint8_t lock_type; // 打开文件类型
uint8_t again_cnt; // EAGAIN 次数
char data[TIMERDATSZ];
};
typedef struct threadtimer_t threadtimer_t;
static threadtimer_t timers[THREADCNT];
static fd_set rdfds, wrfds, erfds, tmpfds;
#define show_usage(program) printf("Usage: %s (-d) port kanban.txt data.bin config.sp\n\t-d: As daemon\n", program)
/*
* accept_client 接受新连接，
* 调用 select 处理
* 处理入口点为 handle_accept
* 当 client 超过 MAXWAITSEC 未响应时
* 调用 clean_timer 回收
* 未被释放的资源以防止内存泄漏等
*/
static void accept_client();
static int bind_server(int* port);
/* /*
* check_buffer 检查接收到的数据，结合 * check_buffer 检查接收到的数据，结合
* 当前会话所处状态决定接下来的处理流程 * 当前会话所处状态决定接下来的处理流程
*/ */
static int check_buffer(threadtimer_t *timer); static int check_buffer(tcpool_thread_timer_t *timer);
static void clean_timer(threadtimer_t* timer); static int send_all(tcpool_thread_timer_t *timer);
static void close_file(threadtimer_t *timer);
static int close_file_and_send(threadtimer_t *timer, char *data, size_t numbytes);
// handle_accept 初步解析指令，处理部分粘连
static int handle_accept(threadtimer_t* p);
static void handle_end(int signo);
static void handle_int(int signo);
static void handle_quit(int signo);
static void handle_segv(int signo);
static int listen_socket();
static FILE *open_file(char* file_path, int lock_type, char* mode);
static int send_all(char* file_path, threadtimer_t *timer);
static int send_data(int accept_fd, char *data, size_t length); static int send_data(int accept_fd, char *data, size_t length);
static off_t size_of_file(const char* fname); static int sm1_pwd(tcpool_thread_timer_t *timer);
static int sm1_pwd(threadtimer_t *timer); static int s0_init(tcpool_thread_timer_t *timer);
static int s0_init(threadtimer_t *timer); static int s1_get(tcpool_thread_timer_t *timer);
static int s1_get(threadtimer_t *timer); static int s2_set(tcpool_thread_timer_t *timer);
static int s2_set(threadtimer_t *timer); static int s3_set_data(tcpool_thread_timer_t *timer);
static int s3_set_data(threadtimer_t *timer);
static void accept_client() {
int i; char c;
signal(SIGINT, handle_int);
signal(SIGQUIT, handle_quit);
signal(SIGKILL, handle_end);
signal(SIGSEGV, handle_segv);
signal(SIGPIPE, SIG_IGN);
signal(SIGTERM, handle_end);
FD_SET(fd, &tmpfds);
while(1) {
FD_COPY(&tmpfds, &rdfds);
FD_COPY(&tmpfds, &erfds);
puts("Selecting...");
timeout.tv_sec = MAXWAITSEC/4;
int r = select(THREADCNT+8, &rdfds, &wrfds, &erfds, &timeout);
if(r < 0) {
perror("select");
return;
}
if(r == 0) { // 超时
for(i = 0; i < THREADCNT; i++) {
if(timers[i].touch && timers[i].accept_fd) {
time_t waitsec = time(NULL) - timers[i].touch;
if(waitsec > MAXWAITSEC) {
printf("Close@%d, wait sec: %u, max: %u\n", i, (unsigned int)waitsec, MAXWAITSEC);
clean_timer(&timers[i]);
}
}
}
continue;
}
puts("\nSelected");
// 正常
if(FD_ISSET(fd, &rdfds)) { // 有新连接
int p = 0;
while(p < THREADCNT && timers[p].touch) p++;
if(p >= THREADCNT) {
puts("Max thread cnt exceeded");
int nfd = accept(fd, (struct sockaddr *)&client_addr, &struct_len);
if(nfd > 0) close(nfd);
goto HANDLE_CLIENTS;
}
threadtimer_t* timer = &timers[p];
timer->accept_fd = accept(fd, (struct sockaddr *)&client_addr, &struct_len);
if(timer->accept_fd <= 0) {
perror("accept");
goto HANDLE_CLIENTS;
}
#ifdef LISTEN_ON_IPV6
uint16_t port = ntohs(client_addr.sin6_port);
struct in6_addr in = client_addr.sin6_addr;
char str[INET6_ADDRSTRLEN]; // 46
inet_ntop(AF_INET6, &in, str, sizeof(str));
#else
uint16_t port = ntohs(client_addr.sin_port);
struct in_addr in = client_addr.sin_addr;
char str[INET_ADDRSTRLEN]; // 16
inet_ntop(AF_INET, &in, str, sizeof(str));
#endif
time_t t = time(NULL);
printf("> %sAccept client(%d) %s:%u at slot No.%d, ", ctime(&t), timer->accept_fd, str, port, p);
timer->index = p;
timer->touch = time(NULL);
timer->is_open = 0;
timer->fp = NULL;
timer->status = -1;
timer->again_cnt = 0;
if(send_data(timer->accept_fd, "Welcome to simple kanban server.", 33) <= 0) {
puts("Send banner to new client failed");
clean_timer(timer);
goto HANDLE_CLIENTS;
}
FD_SET(timer->accept_fd, &tmpfds);
puts("Add new client into select list");
} else if(FD_ISSET(fd, &erfds)) { // 主套接字错误
int nfd = accept(fd, (struct sockaddr *)&client_addr, &struct_len);
perror("main fd in erfds");
if(nfd > 0) close(nfd);
return;
}
HANDLE_CLIENTS:
for(i = 0; i < THREADCNT; i++) {
if(timers[i].touch && timers[i].accept_fd) {
if(FD_ISSET(timers[i].accept_fd, &rdfds)) {
if(recv(timers[i].accept_fd, &c, 1, MSG_PEEK) <= 0 || !handle_accept(&timers[i])) clean_timer(&timers[i]);
else FD_SET(timers[i].accept_fd, &tmpfds);
} else if(FD_ISSET(timers[i].accept_fd, &erfds)) {
printf("Close@%d due to error\n", i);
clean_timer(&timers[i]);
}
}
}
}
}
static int bind_server(int* port) {
#ifdef LISTEN_ON_IPV6
server_addr.sin6_family = AF_INET6;
server_addr.sin6_port = htons((uint16_t)(*port));
bzero(&(server_addr.sin6_addr), sizeof(server_addr.sin6_addr));
int fd = socket(PF_INET6, SOCK_STREAM, 0);
#else
server_addr.sin_family = AF_INET;
server_addr.sin_port = htons((uint16_t)(*port));
server_addr.sin_addr.s_addr = INADDR_ANY;
bzero(&(server_addr.sin_zero), 8);
int fd = socket(AF_INET, SOCK_STREAM, 0);
#endif
int on = 1;
if(setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &on, sizeof(on))) {
perror("Set socket option failure");
return 0;
}
if(!~bind(fd, (struct sockaddr *)&server_addr, struct_len)) {
perror("Bind server failure");
return 0;
}
#ifdef LISTEN_ON_IPV6
*port = ntohs(server_addr.sin6_port);
struct in6_addr in = server_addr.sin6_addr;
char str[INET6_ADDRSTRLEN]; // 46
inet_ntop(AF_INET6, &in, str, sizeof(str));
#else
*port = ntohs(server_addr.sin_port);
struct in_addr in = server_addr.sin_addr;
char str[INET_ADDRSTRLEN]; // 16
inet_ntop(AF_INET, &in, str, sizeof(str));
#endif
printf("Bind server successfully on %s:%u\n", str, *port);
return fd;
}
static int check_buffer(threadtimer_t *timer) {
printf("Status: %d\n", (int)timer->status);
switch(timer->status) {
case -1: return sm1_pwd(timer); break;
case 0: return s0_init(timer); break;
case 1: return s1_get(timer); break;
case 2: return s2_set(timer); break;
case 3: return s3_set_data(timer); break;
default: return -1; break;
}
}
static void clean_timer(threadtimer_t* timer) {
printf("Start cleaning: ");
if(timer->is_open) {
close_file(timer);
printf("Close file, ");
}
FD_CLR(timer->accept_fd, &tmpfds);
if(timer->accept_fd) {
close(timer->accept_fd);
timer->accept_fd = 0;
printf("Close accept, ");
}
timer->touch = 0;
timer->status = -1;
timer->lock_type = 0;
puts("Finish cleaning");
}
static void close_file(threadtimer_t *timer) {
if(timer->is_open && timer->fp != NULL) {
int lock_type = timer->lock_type;
puts("Close file");
fclose(timer->fp);
timer->is_open = 0;
timer->fp = NULL;
file_mode &= ~lock_type;
if((lock_type&LOCK_KANB_SH) > 0 && --kanb_file_ro_cnt > 0) {
file_mode |= LOCK_KANB_SH;
}
if((lock_type&LOCK_DATA_SH) > 0 && --data_file_ro_cnt > 0) {
file_mode |= LOCK_DATA_SH;
}
timer->lock_type = 0;
}
}
static int close_file_and_send(threadtimer_t *timer, char *data, size_t numbytes) {
close_file(timer);
return send_data(timer->accept_fd, data, numbytes);
}
#define take_word(p, w, buff) if((p)->numbytes >= strlen(w) && !strncmp(buff, w, strlen(w))) {\ #define take_word(p, w, buff) if((p)->numbytes >= strlen(w) && !strncmp(buff, w, strlen(w))) {\
printf("<--- Taking: %s in %zd --->\n", w, (p)->numbytes);\ printf("<--- Taking: %s in %zd --->\n", w, (p)->numbytes);\
@@ -318,7 +78,7 @@ static int close_file_and_send(threadtimer_t *timer, char *data, size_t numbytes
buff[l] = 0;\ buff[l] = 0;\
ssize_t n = (p)->numbytes - l;\ ssize_t n = (p)->numbytes - l;\
(p)->numbytes = l;\ (p)->numbytes = l;\
if(!(r = check_buffer((p)))) {\ if(!check_buffer((p))) {\
printf("<--- break in %zd --->\n", (p)->numbytes); \ printf("<--- break in %zd --->\n", (p)->numbytes); \
break; \ break; \
} \ } \
@@ -330,160 +90,86 @@ static int close_file_and_send(threadtimer_t *timer, char *data, size_t numbytes
(p)->numbytes = n;\ (p)->numbytes = n;\
printf("<--- pass in %zd --->\n", (p)->numbytes); \ printf("<--- pass in %zd --->\n", (p)->numbytes); \
} }
#define touch_timer(x) ((x)->touch = time(NULL)) static void accept_action(tcpool_thread_timer_t *p) {
#define my_fd(x) ((x)->accept_fd) if(send_data(p->accept_fd, "Welcome to simple kanban server.", 33) <= 0) {
#define my_dat(x) ((x)->data) puts("Send banner to new client failed");
static int handle_accept(threadtimer_t* p) { return;
int r = 1; }
printf("Recv data from client@%d, ", p->index); printf("Recv data from client@%d, ", p->index);
if((p)->status == 3) return s3_set_data(p); int is_first = 1;
while(((p)->numbytes = recv(my_fd(p), my_dat(p), TIMERDATSZ, MSG_DONTWAIT)) > 0) { while(((p)->numbytes = recv(p->accept_fd, p->data, SERVER_THREAD_BUFSZ, 0)) > 0) {
touch_timer(p); tcpool_touch_timer(p);
my_dat(p)[(p)->numbytes] = 0; p->data[p->numbytes] = 0;
printf("Get %d bytes: %s, Check buffer...\n", (int)(p)->numbytes, my_dat(p)); printf("Get %d bytes: %s, Check buffer...\n", (int)(p)->numbytes, p->data);
//处理允许的粘连 if(is_first) {
take_word(p, cfg->pwd, my_dat(p)); is_first = 0;
take_word(p, "get", my_dat(p)); //处理允许的粘连
take_word(p, "cat", my_dat(p)); take_word(p, cfg->pwd, p->data);
take_word(p, "quit", my_dat(p)); take_word(p, "get", p->data);
take_word(p, cfg->sps, my_dat(p)); take_word(p, "cat", p->data);
take_word(p, "ver", my_dat(p)); take_word(p, "quit", p->data);
take_word(p, "dat", my_dat(p)); take_word(p, cfg->sps, p->data);
take_word(p, "ver", p->data);
take_word(p, "dat", p->data);
}
if((p)->numbytes <= 0) { if((p)->numbytes <= 0) {
puts("Taking words finished"); puts("Taking words finished");
continue;
}
if(!check_buffer((p))) {
printf("<--- break normal in %zd --->\n", p->numbytes);
break; break;
} }
puts("Last check_buffer");
r = check_buffer((p));
break;
} }
if((p)->numbytes <= 0) { printf("Recv finished, remain: %zd\n", (p)->numbytes);
if(errno == EAGAIN || errno == EINVAL) { }
if(!++(p)->again_cnt) {
r = 0; static int check_buffer(tcpool_thread_timer_t *timer) {
puts("Max EAGAIN/EINVAL cnt exceeded"); printf("Status: %d\n", (int)timer->status);
} switch(timer->status) {
} else if(errno) { case -1: return sm1_pwd(timer); break;
perror("recv"); case 0: return s0_init(timer); break;
r = 0; case 1: return s1_get(timer); break;
} case 2: return s2_set(timer); break;
case 3: return s3_set_data(timer); break;
default: return -1; break;
} }
printf("Recv finished, remain: %zd, continue: %s\n", (p)->numbytes, r?"true":"false");
return r;
} }
static void handle_end(int signo) { static int send_all(tcpool_thread_timer_t *timer) {
puts("Handle kill/term");
close(fd);
fflush(stdout);
exit(0);
}
static void handle_int(int signo) {
puts("Keyboard interrupted");
close(fd);
fflush(stdout);
exit(0);
}
static void handle_quit(int signo) {
puts("Handle sigquit");
close(fd);
fflush(stdout);
exit(0);
}
static void handle_segv(int signo) {
puts("Handle sigsegv");
close(fd);
fflush(stdout);
exit(0);
}
static int listen_socket() {
int flags = fcntl(fd, F_GETFL, 0);
if(!~listen(fd, THREADCNT)) return 1;
return fcntl(fd, F_SETFL, flags | O_NONBLOCK);
}
static FILE *open_file(char* file_path, int lock_type, char* mode) {
FILE *fp = NULL;
if((lock_type&LOCK_KANB_SH)) {
if((file_mode&LOCK_KANB_EX) > 0) {
puts("open_file(KANB_SH): file is busy");
return NULL;
}
file_mode |= LOCK_KANB_SH;
kanb_file_ro_cnt++;
} else if((lock_type&LOCK_DATA_SH)) {
if((file_mode&LOCK_DATA_EX) > 0) {
puts("open_file(DATA_SH): file is busy");
return NULL;
}
file_mode |= LOCK_DATA_SH;
data_file_ro_cnt++;
} else if(lock_type&LOCK_KANB_EX) {
if((file_mode&(LOCK_KANB_EX|LOCK_KANB_SH)) > 0) {
puts("open_file(KANB_EX): file is busy");
return NULL;
}
file_mode |= LOCK_KANB_EX;
} else if(lock_type&LOCK_DATA_EX) {
if((file_mode&(LOCK_DATA_EX|LOCK_DATA_SH)) > 0) {
puts("open_file(DATA_EX): file is busy");
return NULL;
}
file_mode |= LOCK_DATA_EX;
}
fp = fopen(file_path, mode);
if(!fp) {
perror("fopen");
file_mode &= ~lock_type;
return NULL;
}
printf("Open file %s in mode %s\n", file_path, mode);
return fp;
}
static int send_all(char* file_path, threadtimer_t *timer) {
int re = 1; int re = 1;
FILE *fp = open_file(file_path, timer->lock_type, "rb"); FILE *fp = open_file(timer->index, timer->isdata, 1);
if(fp) { if(fp == NULL) return 1;
timer->fp = fp; uint32_t close_file_wrap_data[2] = {timer->index, (uint32_t)timer->isdata};
timer->is_open = 1; pthread_cleanup_push((void (*)(void*))&close_file_wrap, (void*)close_file_wrap_data);
uint32_t file_size = (uint32_t)size_of_file(file_path); off_t len = 0, file_size = get_file_size(timer->isdata);
printf("Get file size: %d bytes, ", (int)file_size); printf("Get file size: %d bytes, ", (int)file_size);
off_t len = 0; #if __APPLE__
int flags = fcntl(timer->accept_fd, F_GETFL, 0); #ifdef WORDS_BIGENDIAN
fcntl(timer->accept_fd, F_SETFL, flags & ~O_NONBLOCK); file_size = __DARWIN_OSSwapInt32(file_size);
#if __APPLE__
#ifdef WORDS_BIGENDIAN
file_size = __DARWIN_OSSwapInt32(file_size);
#endif
struct sf_hdtr hdtr;
struct iovec headers;
headers.iov_base = &file_size;
headers.iov_len = sizeof(uint32_t);
hdtr.headers = &headers;
hdtr.hdr_cnt = 1;
hdtr.trailers = NULL;
hdtr.trl_cnt = 0;
re = !sendfile(fileno(fp), timer->accept_fd, 0, &len, &hdtr, 0);
if(!re) perror("sendfile");
#else
#ifdef WORDS_BIGENDIAN
uint32_t little_fs = __builtin_bswap32(file_size);
send(timer->accept_fd, &little_fs, sizeof(uint32_t), 0);
#else
send(timer->accept_fd, &file_size, sizeof(uint32_t), 0);
#endif
re = sendfile(timer->accept_fd, fileno(fp), &len, file_size) > 0;
if(!re) perror("sendfile");
#endif #endif
printf("Send %d bytes\n", (int)len); struct sf_hdtr hdtr;
close_file(timer); struct iovec headers;
fcntl(timer->accept_fd, F_SETFL, flags); headers.iov_base = &file_size;
} headers.iov_len = sizeof(uint32_t);
hdtr.headers = &headers;
hdtr.hdr_cnt = 1;
hdtr.trailers = NULL;
hdtr.trl_cnt = 0;
re = !sendfile(fileno(fp), timer->accept_fd, 0, &len, &hdtr, 0);
if(!re) perror("sendfile");
#else
#ifdef WORDS_BIGENDIAN
uint32_t little_fs = __builtin_bswap32(file_size);
send(timer->accept_fd, &little_fs, sizeof(uint32_t), 0);
#else
send(timer->accept_fd, &file_size, sizeof(uint32_t), 0);
#endif
re = sendfile(timer->accept_fd, fileno(fp), &len, file_size) > 0;
if(!re) perror("sendfile");
#endif
printf("Send %d bytes\n", (int)len);
pthread_cleanup_pop(1);
return re; return re;
} }
@@ -493,7 +179,7 @@ static int send_data(int accept_fd, char *data, size_t length) {
puts("Send data error: zero length"); puts("Send data error: zero length");
return 0; return 0;
} }
if(!~send(accept_fd, data, length, MSG_DONTWAIT)) { if(!~send(accept_fd, data, length, 0)) {
puts("Send data error"); puts("Send data error");
return 0; return 0;
} }
@@ -510,13 +196,7 @@ static int send_data(int accept_fd, char *data, size_t length) {
return 1; return 1;
} }
static off_t size_of_file(const char* fname) { static int sm1_pwd(tcpool_thread_timer_t *timer) {
struct stat statbuf;
if(stat(fname, &statbuf)==0) return statbuf.st_size;
else return -1;
}
static int sm1_pwd(threadtimer_t *timer) {
if(!strncmp(timer->data, cfg->pwd, strlen(cfg->pwd))) { if(!strncmp(timer->data, cfg->pwd, strlen(cfg->pwd))) {
timer->status = 0; timer->status = 0;
puts("Password check passed"); puts("Password check passed");
@@ -524,49 +204,48 @@ static int sm1_pwd(threadtimer_t *timer) {
return !timer->status; return !timer->status;
} }
static int s0_init(threadtimer_t *timer) { static int s0_init(tcpool_thread_timer_t *timer) {
if(!strncmp(timer->data, "get", 3)) timer->status = 1; if(!strncmp(timer->data, "get", 3)) {
else if(!strncmp(timer->data, cfg->sps, strlen(cfg->sps))) timer->status = 2; timer->status = 1;
else if(!strncmp(timer->data, "cat", 3)) { return 1;
timer->lock_type = LOCK_DATA_SH;
return send_all(data_path, timer);
} }
else if(!strncmp(timer->data, "quit", 4)) return 0; if(!strncmp(timer->data, cfg->sps, strlen(cfg->sps))) {
return send_data(timer->accept_fd, timer->data, timer->numbytes); timer->status = 2;
return 2;
}
if(!strncmp(timer->data, "cat", 3)) {
timer->isdata = 1;
return send_all(timer);
}
//if(!strncmp(timer->data, "quit", 4)) return 0;
return 0;
} }
static int s1_get(threadtimer_t *timer) { //get kanban // s1_get scan getxxx
FILE *fp = open_file(kanban_path, LOCK_KANB_SH, "rb"); static int s1_get(tcpool_thread_timer_t *timer) {
FILE *fp = open_file(timer->index, 0, 1);
timer->status = 0; timer->status = 0;
if(fp) { if(!fp) goto GET_END;
timer->fp = fp;
timer->is_open = 1; uint32_t close_file_wrap_data[2] = {timer->index, (uint32_t)timer->isdata};
timer->lock_type = LOCK_KANB_SH; int r; uint32_t ver, cli_ver;
uint32_t ver, cli_ver;
if(fscanf(fp, "%u", &ver) > 0) { pthread_cleanup_push((void (*)(void*))&close_file_wrap, (void*)close_file_wrap_data);
if(sscanf(timer->data, "%u", &cli_ver) > 0) { timer->isdata = 0;
if(cli_ver < ver) { //need to send a new kanban r = fscanf(fp, "%u", &ver);
timer->is_open = 0; pthread_cleanup_pop(1);
close_file(timer);
int r = send_all(kanban_path, timer); if(r <= 0) goto GET_END;
if(strstr(timer->data, "quit")) { if(sscanf(timer->data, "%u", &cli_ver) <= 0) goto GET_END;
puts("Found last cmd is quit"); if(cli_ver >= ver) goto GET_END;
timer->numbytes = 0;
return 0; //need to send a new kanban
} r = send_all(timer);
int i = 0; goto GET_SKIP;
for(; i < timer->numbytes; i++) {
if(!isdigit(timer->data[i])) { GET_END:
timer->numbytes -= i; r = send_data(timer->accept_fd, "null", 4);
break; GET_SKIP:
}
}
return r;
}
}
}
}
int r = close_file_and_send(timer, "null", 4);
if(strstr(timer->data, "quit")) { if(strstr(timer->data, "quit")) {
puts("Found last cmd is quit"); puts("Found last cmd is quit");
timer->numbytes = 0; timer->numbytes = 0;
@@ -582,37 +261,33 @@ static int s1_get(threadtimer_t *timer) { //get kanban
return r; return r;
} }
static int s2_set(threadtimer_t *timer) { static int s2_set(tcpool_thread_timer_t *timer) {
FILE *fp = NULL; FILE *fp = NULL;
int lktp;
if(!strncmp(timer->data, "ver", 3)) { if(!strncmp(timer->data, "ver", 3)) {
fp = open_file(kanban_path, LOCK_KANB_EX, "wb"); timer->isdata = 0;
lktp = LOCK_KANB_EX;
} else if(!strncmp(timer->data, "dat", 3)) { } else if(!strncmp(timer->data, "dat", 3)) {
fp = open_file(data_path, LOCK_DATA_EX, "wb"); timer->isdata = 1;
lktp = LOCK_DATA_EX;
}
if(fp) {
timer->status = 3;
timer->fp = fp;
timer->is_open = 1;
timer->lock_type = lktp;
return send_data(timer->accept_fd, "data", 4);
} else { } else {
timer->status = 0; timer->status = 0;
return send_data(timer->accept_fd, "erro", 4); return send_data(timer->accept_fd, "erro", 4);
} }
timer->status = 3;
return send_data(timer->accept_fd, "data", 4);
} }
static int s3_set_data(threadtimer_t *timer) { static int s3_set_data(tcpool_thread_timer_t *timer) {
char ret[4] = "succ"; char ret[4];
int flags = fcntl(timer->accept_fd, F_GETFL, 0); *(uint32_t*)ret = *(uint32_t*)"succ";
fcntl(timer->accept_fd, F_SETFL, flags & ~O_NONBLOCK);
timer->status = 0; timer->status = 0;
ssize_t n = recv(timer->accept_fd, timer->data, 4, MSG_WAITALL); FILE* fp = open_file(timer->index, timer->isdata, 0);
if(n < 4) { uint32_t close_file_wrap_data[2] = {timer->index, (uint32_t)timer->isdata};
*(uint32_t*)ret = *(uint32_t*)"erro"; pthread_cleanup_push((void (*)(void*))&close_file_wrap, (void*)close_file_wrap_data);
goto S3_RETURN; if(timer->numbytes < 4) {
ssize_t n = recv(timer->accept_fd, timer->data+timer->numbytes, 4-timer->numbytes, MSG_WAITALL);
if(n < 4-timer->numbytes) {
*(uint32_t*)ret = *(uint32_t*)"erro";
goto S3_RETURN;
}
} }
#ifdef WORDS_BIGENDIAN #ifdef WORDS_BIGENDIAN
uint32_t file_size = __builtin_bswap32(*(uint32_t*)(timer->data)); uint32_t file_size = __builtin_bswap32(*(uint32_t*)(timer->data));
@@ -620,27 +295,27 @@ static int s3_set_data(threadtimer_t *timer) {
uint32_t file_size = *(uint32_t*)(timer->data); uint32_t file_size = *(uint32_t*)(timer->data);
#endif #endif
printf("Set data size: %u\n", file_size); printf("Set data size: %u\n", file_size);
if((timer->numbytes = recv(timer->accept_fd, timer->data, TIMERDATSZ, 0)) < 0 && errno != EAGAIN) { if((timer->numbytes = recv(timer->accept_fd, timer->data, SERVER_THREAD_BUFSZ, 0)) < 0 && errno != EAGAIN) {
*(uint32_t*)ret = *(uint32_t*)"erro"; *(uint32_t*)ret = *(uint32_t*)"erro";
goto S3_RETURN; goto S3_RETURN;
} }
printf("Get data size: %d\n", (int)timer->numbytes); printf("Get data size: %d\n", (int)timer->numbytes);
if(file_size <= TIMERDATSZ) { if(file_size <= SERVER_THREAD_BUFSZ) {
while(timer->numbytes != file_size) { while(timer->numbytes != file_size) {
ssize_t n = recv(timer->accept_fd, timer->data + timer->numbytes, TIMERDATSZ - timer->numbytes, MSG_WAITALL); ssize_t n = recv(timer->accept_fd, timer->data + timer->numbytes, SERVER_THREAD_BUFSZ - timer->numbytes, MSG_WAITALL);
if(n <= 0) { if(n <= 0) {
*(uint32_t*)ret = *(uint32_t*)"erro"; *(uint32_t*)ret = *(uint32_t*)"erro";
goto S3_RETURN; goto S3_RETURN;
} }
timer->numbytes += n; timer->numbytes += n;
} }
if(fwrite(timer->data, file_size, 1, timer->fp) != 1) { if(fwrite(timer->data, file_size, 1, fp) != 1) {
perror("fwrite"); perror("fwrite");
*(uint32_t*)ret = *(uint32_t*)"erro"; *(uint32_t*)ret = *(uint32_t*)"erro";
} }
goto S3_RETURN; goto S3_RETURN;
} }
if(timer->numbytes > 0 && fwrite(timer->data, timer->numbytes, 1, timer->fp) != 1) { if(timer->numbytes > 0 && fwrite(timer->data, timer->numbytes, 1, fp) != 1) {
perror("fwrite"); perror("fwrite");
*(uint32_t*)ret = *(uint32_t*)"erro"; *(uint32_t*)ret = *(uint32_t*)"erro";
goto S3_RETURN; goto S3_RETURN;
@@ -648,7 +323,7 @@ static int s3_set_data(threadtimer_t *timer) {
int32_t remain = file_size - timer->numbytes; int32_t remain = file_size - timer->numbytes;
while(remain > 0) { while(remain > 0) {
// printf("remain:%d\n", (int)remain); // printf("remain:%d\n", (int)remain);
ssize_t n = recv(timer->accept_fd, timer->data, (remain>TIMERDATSZ)?TIMERDATSZ:remain, MSG_WAITALL); ssize_t n = recv(timer->accept_fd, timer->data, (remain>SERVER_THREAD_BUFSZ)?SERVER_THREAD_BUFSZ:remain, MSG_WAITALL);
if(n < 0) { if(n < 0) {
*(uint32_t*)ret = *(uint32_t*)"erro"; *(uint32_t*)ret = *(uint32_t*)"erro";
goto S3_RETURN; goto S3_RETURN;
@@ -657,7 +332,7 @@ static int s3_set_data(threadtimer_t *timer) {
usleep(10000); // 10 ms usleep(10000); // 10 ms
continue; continue;
} }
if(fwrite(timer->data, n, 1, timer->fp) != 1) { if(fwrite(timer->data, n, 1, fp) != 1) {
perror("fwrite"); perror("fwrite");
*(uint32_t*)ret = *(uint32_t*)"erro"; *(uint32_t*)ret = *(uint32_t*)"erro";
goto S3_RETURN; goto S3_RETURN;
@@ -665,22 +340,25 @@ static int s3_set_data(threadtimer_t *timer) {
remain -= n; remain -= n;
} }
S3_RETURN: S3_RETURN:
fcntl(timer->accept_fd, F_SETFL, flags); pthread_cleanup_pop(1);
return close_file_and_send(timer, ret, 4); return send_data(timer->accept_fd, ret, 4);
} }
#define show_usage(program) printf("Usage: %s (-d) port kanban.txt data.bin config.sp\n\t-d: As daemon\n", program)
int main(int argc, char *argv[]) { int main(int argc, char *argv[]) {
if(argc != 5 && argc != 6) { if(argc != 5 && argc != 6) {
show_usage(argv[0]); show_usage(argv[0]);
return 0; return 0;
} }
int port = 0; uint16_t port = 0; int tmp;
int as_daemon = !strncmp(argv[1], "-d", 3); int as_daemon = !strncmp(argv[1], "-d", 3);
sscanf(argv[as_daemon?2:1], "%d", &port); sscanf(argv[as_daemon?2:1], "%d", &tmp);
if(port < 0 || port >= 65536) { if(tmp < 0 || tmp >= 65536) {
printf("Error port: %d\n", port); printf("Error port: %d\n", tmp);
return 1; return 1;
} }
port = (uint16_t)tmp;
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;
@@ -714,20 +392,10 @@ int main(int argc, char *argv[]) {
} }
read_pb_into(fp, (simple_pb_t*)(&_cfg)); read_pb_into(fp, (simple_pb_t*)(&_cfg));
fclose(fp); fclose(fp);
if(!(fd = bind_server(&port))) { if(!(tmp = bind_server(&port))) return 6;
return 6; if(!listen_socket(tmp)) return 7;
} pthread_cleanup_push((void (*)(void*))&close, (void*)((long long)tmp));
if(listen_socket()) { accept_client(tmp);
perror("Listen failed"); pthread_cleanup_pop(1);
return 7;
}
/*
printf("password: ");
puts(cfg->pwd);
printf("set password: ");
puts(cfg->sps);
*/
accept_client();
close(fd);
return 99; return 99;
} }

430
tcpool.h Normal file
View File

@@ -0,0 +1,430 @@
#ifndef _TCPOOL_H_
#define _TCPOOL_H_
/* See feature_test_macros(7) */
#define _GNU_SOURCE 1
#include <arpa/inet.h>
#include <netinet/in.h>
#include <pthread.h>
#include <setjmp.h>
#include <signal.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/socket.h>
#include <sys/types.h>
#include <sys/uio.h>
#include <sys/wait.h>
#include <time.h>
#include <unistd.h>
#ifndef TCPOOL_THREAD_TIMER_T_SZ
#define TCPOOL_THREAD_TIMER_T_SZ 1024
#endif
#define TCPOOL_THREAD_TIMER_T_HEAD_SZ ( \
sizeof(uint32_t) \
+sizeof(int) \
+sizeof(time_t) \
+sizeof(pthread_rwlock_t) \
+2*sizeof(pthread_t) \
+2*sizeof(pthread_cond_t) \
+2*sizeof(pthread_mutex_t) \
+sizeof(pthread_rwlock_t) \
+2*sizeof(uint8_t) \
)
#ifndef TCPOOL_THREADCNT
#define TCPOOL_THREADCNT 32
#endif
#ifndef TCPOOL_MAXWAITSEC
#define TCPOOL_MAXWAITSEC 8
#endif
#ifndef TCPOOL_THREAD_CONTEXT
#define TCPOOL_THREAD_CONTEXT uint8_t __padding[ \
TCPOOL_THREAD_TIMER_T_SZ \
-TCPOOL_THREAD_TIMER_T_HEAD_SZ \
]
#endif
#ifndef TCPOOL_TOUCH_TIMER_CONDITION
#define TCPOOL_TOUCH_TIMER_CONDITION (0)
#endif
#ifndef TCPOOL_INIT_ACTION
#define TCPOOL_INIT_ACTION ;
#endif
#ifndef TCPOOL_PREHANDLE_ACCEPT_ACTION
#define TCPOOL_PREHANDLE_ACCEPT_ACTION(timer) ;
#endif
#ifndef TCPOOL_CLEANUP_THREAD_ACTION
#define TCPOOL_CLEANUP_THREAD_ACTION(timer) ;
#endif
struct tcpool_thread_timer_t {
uint32_t index;
int accept_fd;
time_t touch; // lock by mt
pthread_rwlock_t mt; // lock touch
pthread_t thread;
pthread_t timerthread;
pthread_cond_t c; // lock by mc
pthread_mutex_t mc; // lock c
pthread_cond_t tc; // lock by tmc
pthread_mutex_t tmc; // lock tc&hastimerslept
pthread_rwlock_t mb; // lock isbusy
TCPOOL_THREAD_CONTEXT;
uint8_t isbusy; // lock by mb
uint8_t hastimerslept; // lock by tmc
};
typedef struct tcpool_thread_timer_t tcpool_thread_timer_t;
static tcpool_thread_timer_t tcpool_timers[TCPOOL_THREADCNT];
#define tcpool_timer_pointer_of(x) ((tcpool_thread_timer_t*)(x))
#define tcpool_touch_timer(x) { \
pthread_rwlock_wrlock(&tcpool_timer_pointer_of(x)->mt); \
tcpool_timer_pointer_of(x)->touch = time(NULL); \
printf("Touch timer@%d\n", tcpool_timer_pointer_of(x)->index);\
pthread_rwlock_unlock(&tcpool_timer_pointer_of(x)->mt); \
}
#ifdef LISTEN_ON_IPV6
static socklen_t tcpool_struct_len = sizeof(struct sockaddr_in6);
static struct sockaddr_in6 tcpool_server_addr;
#else
static socklen_t tcpool_struct_len = sizeof(struct sockaddr_in);
static struct sockaddr_in tcpool_server_addr;
#endif
static pthread_attr_t __tcpool_thread_attr;
static pthread_key_t __tcpool_pthread_key_index;
static sigjmp_buf __tcpool_jmp2convend[TCPOOL_THREADCNT];
static void accept_action(tcpool_thread_timer_t *timer);
static void accept_client(int fd);
static void accept_timer(void *p);
static int bind_server(uint16_t* port);
static void cleanup_thread(tcpool_thread_timer_t* timer);
static void handle_accept(void *accept_fd_p);
static void handle_int(int signo);
static void handle_kill(int signo);
static void handle_pipe(int signo);
static void handle_quit(int signo);
static void handle_segv(int signo);
static int listen_socket(int fd);
static int bind_server(uint16_t* port) {
#ifdef LISTEN_ON_IPV6
tcpool_server_addr.sin6_family = AF_INET6;
tcpool_server_addr.sin6_port = htons(*port);
bzero(&(tcpool_server_addr.sin6_addr), sizeof(tcpool_server_addr.sin6_addr));
int fd = socket(PF_INET6, SOCK_STREAM, 0);
#else
tcpool_server_addr.sin_family = AF_INET;
tcpool_server_addr.sin_port = htons(*port);
tcpool_server_addr.sin_addr.s_addr = INADDR_ANY;
bzero(&(tcpool_server_addr.sin_zero), 8);
int fd = socket(AF_INET, SOCK_STREAM, 0);
#endif
int on = 1;
if(setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &on, sizeof(on))) {
perror("Set socket option failure");
return 0;
}
if(!~bind(fd, (struct sockaddr *)&tcpool_server_addr, tcpool_struct_len)) {
perror("Bind server failure");
return 0;
}
#ifdef LISTEN_ON_IPV6
*port = ntohs(tcpool_server_addr.sin6_port);
struct in6_addr in = tcpool_server_addr.sin6_addr;
char str[INET6_ADDRSTRLEN]; // 46
inet_ntop(AF_INET6, &in, str, sizeof(str));
#else
*port = ntohs(tcpool_server_addr.sin_port);
struct in_addr in = tcpool_server_addr.sin_addr;
char str[INET_ADDRSTRLEN]; // 16
inet_ntop(AF_INET, &in, str, sizeof(str));
#endif
printf("Bind server successfully on %s:%u\n", str, *port);
return fd;
}
static int listen_socket(int fd) {
if(!~listen(fd, TCPOOL_THREADCNT)) {
perror("Listen failed");
return 0;
}
puts("Listening...");
return fd;
}
static void handle_quit(int signo) {
uint32_t index = (uint32_t)((uintptr_t)pthread_getspecific(__tcpool_pthread_key_index));
printf("Handle sigquit@%d\n", index-1);
fflush(stdout);
if(index) {
sigaction(SIGQUIT, &(const struct sigaction){handle_quit}, NULL);
siglongjmp(__tcpool_jmp2convend[index-1], signo);
}
else pthread_exit(NULL);
}
static void handle_segv(int signo) {
uint32_t index = (uint32_t)((uintptr_t)pthread_getspecific(__tcpool_pthread_key_index));
printf("Handle sigsegv@%d\n", index-1);
fflush(stdout);
if(index) {
sigaction(SIGSEGV, &(const struct sigaction){handle_segv}, NULL);
siglongjmp(__tcpool_jmp2convend[index-1], signo);
}
else pthread_exit(NULL);
}
static void handle_kill(int signo) {
puts("Handle sigkill/sigterm");
fflush(stdout);
exit(signo);
}
static void handle_int(int signo) {
puts("Keyboard interrupted");
fflush(stdout);
exit(signo);
}
static void handle_pipe(int signo) {
uint32_t index = (uint32_t)((uintptr_t)pthread_getspecific(__tcpool_pthread_key_index));
printf("Pipe error@%d, break loop...\n", index-1);
fflush(stdout);
if(index) {
sigaction(SIGPIPE, &(const struct sigaction){handle_pipe}, NULL);
siglongjmp(__tcpool_jmp2convend[index-1], signo);
}
else pthread_exit(NULL);
}
static void accept_timer(void *p) {
tcpool_thread_timer_t *timer = tcpool_timer_pointer_of(p);
uint32_t index = timer->index;
pthread_t thread = timer->thread;
uint8_t isbusy;
sleep(TCPOOL_MAXWAITSEC / 4);
while(thread && !pthread_kill(thread, 0)) {
pthread_rwlock_rdlock(&timer->mb);
isbusy = timer->isbusy;
pthread_rwlock_unlock(&timer->mb);
if(!isbusy) {
TIMER_SLEEP:
pthread_mutex_lock(&timer->tmc);
timer->hastimerslept = 1;
printf("Timer@%d sleep\n", timer->index);
pthread_cond_wait(&timer->tc, &timer->tmc);
timer->hastimerslept = 0;
pthread_mutex_unlock(&timer->tmc);
printf("Timer@%d wake up\n", timer->index);
sleep(TCPOOL_MAXWAITSEC / 4);
thread = timer->thread;
}
if(TCPOOL_TOUCH_TIMER_CONDITION) tcpool_touch_timer(p);
pthread_rwlock_rdlock(&timer->mt);
time_t waitsec = time(NULL) - timer->touch;
pthread_rwlock_unlock(&timer->mt);
printf("Wait@%d sec: %u, max: %u\n", timer->index, (unsigned int)waitsec, TCPOOL_MAXWAITSEC);
if(waitsec > TCPOOL_MAXWAITSEC) {
if(thread) {
pthread_kill(thread, SIGQUIT);
printf("Kill thread@%d\n", timer->index);
}
break;
}
sleep(TCPOOL_MAXWAITSEC / 4);
thread = timer->thread;
}
goto TIMER_SLEEP;
}
static void cleanup_thread(tcpool_thread_timer_t* timer) {
printf("Start cleaning@%d, ", timer->index);
if(timer->accept_fd) {
close(timer->accept_fd);
timer->accept_fd = 0;
printf("Close accept, ");
}
TCPOOL_CLEANUP_THREAD_ACTION(timer);
timer->thread = 0;
printf("Clear thread, ");
pthread_cond_destroy(&timer->c);
printf("Destroy accept cond, ");
pthread_mutex_destroy(&timer->mc);
printf("Destroy accept mutex, ");
pthread_rwlock_wrlock(&timer->mb);
timer->isbusy = 0;
printf("Clear busy, ");
pthread_rwlock_unlock(&timer->mb);
puts("Finish cleaning");
}
static void handle_accept(void *p) {
#ifdef DEBUG
printf("accept ptr: %p\n", p);
#endif
pthread_cleanup_push((void (*)(void*))&cleanup_thread, p);
puts("Handling accept...");
pthread_setspecific(__tcpool_pthread_key_index, (void*)((uintptr_t)tcpool_timer_pointer_of(p)->index+1));
if(sigsetjmp(__tcpool_jmp2convend[tcpool_timer_pointer_of(p)->index], 1)) {
printf("Long Jump@%d\n", tcpool_timer_pointer_of(p)->index);
goto CONV_END;
}
while(1) {
accept_action(tcpool_timer_pointer_of(p));
CONV_END: puts("Conversation end");
if(tcpool_timer_pointer_of(p)->accept_fd) {
close(tcpool_timer_pointer_of(p)->accept_fd);
tcpool_timer_pointer_of(p)->accept_fd = 0;
puts("Close accept");
}
TCPOOL_CLEANUP_THREAD_ACTION(tcpool_timer_pointer_of(p));
pthread_mutex_lock(&tcpool_timer_pointer_of(p)->mc);
pthread_rwlock_wrlock(&tcpool_timer_pointer_of(p)->mb);
tcpool_timer_pointer_of(p)->isbusy = 0;
pthread_rwlock_unlock(&tcpool_timer_pointer_of(p)->mb);
puts("Set thread status to idle");
pthread_cond_wait(&tcpool_timer_pointer_of(p)->c, &tcpool_timer_pointer_of(p)->mc);
pthread_mutex_unlock(&tcpool_timer_pointer_of(p)->mc);
puts("Thread wakeup");
}
pthread_cleanup_pop(1);
}
static void accept_client(int fd) {
sigaction(SIGINT , &(const struct sigaction){handle_int}, NULL);
sigaction(SIGQUIT, &(const struct sigaction){handle_quit}, NULL);
sigaction(SIGKILL, &(const struct sigaction){handle_kill}, NULL);
sigaction(SIGSEGV, &(const struct sigaction){handle_segv}, NULL);
sigaction(SIGPIPE, &(const struct sigaction){handle_pipe}, NULL);
sigaction(SIGTERM, &(const struct sigaction){handle_kill}, NULL);
pthread_attr_init(&__tcpool_thread_attr);
pthread_attr_setdetachstate(&__tcpool_thread_attr, PTHREAD_CREATE_DETACHED);
TCPOOL_INIT_ACTION;
for(int i = 0; i < TCPOOL_THREADCNT; i++) {
pthread_rwlock_init(&tcpool_timers[i].mt, NULL);
pthread_rwlock_init(&tcpool_timers[i].mb, NULL);
}
pthread_key_create(&__tcpool_pthread_key_index, NULL);
while(1) {
int p = 0;
while(p < TCPOOL_THREADCNT) {
pthread_rwlock_rdlock(&tcpool_timers[p].mb);
if(!tcpool_timers[p].isbusy) break;
pthread_rwlock_unlock(&tcpool_timers[p].mb);
p++;
}
if(p >= TCPOOL_THREADCNT) {
puts("Max thread cnt exceeded");
sleep(1);
continue;
}
printf("Ready for accept on slot No.%d\n", p);
tcpool_thread_timer_t* timer = &tcpool_timers[p];
pthread_rwlock_unlock(&timer->mb);
#ifdef LISTEN_ON_IPV6
struct sockaddr_in6 client_addr;
#else
struct sockaddr_in client_addr;
#endif
int accept_fd;
if((accept_fd=accept(fd, (struct sockaddr *)&client_addr, &tcpool_struct_len))<=0) {
perror("Accept client error");
continue;
}
pthread_rwlock_wrlock(&timer->mb);
timer->isbusy = 1;
pthread_rwlock_unlock(&timer->mb);
#ifdef LISTEN_ON_IPV6
uint16_t port = ntohs(client_addr.sin6_port);
struct in6_addr in = client_addr.sin6_addr;
char str[INET6_ADDRSTRLEN]; // 46
inet_ntop(AF_INET6, &in, str, sizeof(str));
#else
uint16_t port = ntohs(client_addr.sin_port);
struct in_addr in = client_addr.sin_addr;
char str[INET_ADDRSTRLEN]; // 16
inet_ntop(AF_INET, &in, str, sizeof(str));
#endif
time_t t = time(NULL);
printf("\n> %sAccept client %s:%u at slot No.%d, ", ctime(&t), str, port, p);
timer->accept_fd = accept_fd;
timer->index = p;
pthread_rwlock_wrlock(&timer->mt);
timer->touch = time(NULL);
pthread_rwlock_unlock(&timer->mt);
TCPOOL_PREHANDLE_ACCEPT_ACTION(timer);
// start or wakeup accept thread
pthread_t thread = timer->thread;
if(thread && !pthread_kill(thread, 0)) {
pthread_mutex_lock(&timer->mc);
pthread_cond_signal(&timer->c); // wakeup thread
pthread_mutex_unlock(&timer->mc);
puts("Pick thread from pool");
} else {
pthread_cond_init(&timer->c, NULL);
pthread_mutex_init(&timer->mc, NULL);
if (pthread_create(&timer->thread, &__tcpool_thread_attr, (void (*)(void*))&handle_accept, timer)) {
perror("Error creating thread");
cleanup_thread(timer);
putchar('\n');
continue;
}
puts("Thread created");
}
// start or wakeup timer thread
thread = timer->timerthread;
if(!thread || pthread_kill(thread, 0)) {
printf("Creating timer thread...");
pthread_cond_init(&timer->tc, NULL);
pthread_mutex_init(&timer->tmc, NULL);
timer->hastimerslept = 0;
if (pthread_create(&timer->timerthread, &__tcpool_thread_attr, (void (*)(void*))&accept_timer, timer)) {
perror("Error creating timer thread");
cleanup_thread(timer);
putchar('\n');
continue;
}
puts("succeeded");
} else {
pthread_mutex_lock(&timer->tmc);
uint8_t hastimerslept = timer->hastimerslept;
pthread_mutex_unlock(&timer->tmc);
if(hastimerslept) {
printf("Waking up timer thread...");
pthread_mutex_lock(&timer->tmc);
pthread_cond_signal(&timer->tc); // wakeup thread
pthread_mutex_unlock(&timer->tmc);
puts("succeeded");
} else puts("Timer already running");
}
}
}
#endif /* _TCPOOL_H_ */