monoid/HW12
1
0
Fork 0
Code Issues 2 Pull Requests Projects Releases Wiki Activity

refactor client.c

Browse Source
This commit is contained in:
ubuntu201711081 2020-12-07 09:17:01 +00:00
parent 35aa8243b3
commit 50eeba56a4
1 changed files with 135 additions and 76 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

211
client.c
View File

@ -187,7 +187,9 @@ int SendOpAndReceiveFile(const char * filename, struct sockaddr const * addr){
close(sock); close(sock);
return ret; return ret;
} }
//====
//bench
//====
struct benchmark_data{ struct benchmark_data{
bool benchmode; bool benchmode;
struct timespec begin; struct timespec begin;
@ -207,22 +209,75 @@ void init_bench_data(){
clock_getres(bench.clock_id,&bench.resolution); clock_getres(bench.clock_id,&bench.resolution);
} }
} }
//====
static size_t thread_number_option = 1; //simple queue
//====
typedef enum{ typedef enum{
From_CharArray, From_CharArray,
From_FileStream From_FileStream
} queueing_method_t; } queueing_method_t;
typedef struct SimpleCharQueue{
queueing_method_t method;
const char ** filename_begin;
const char ** filename_end;
FILE * fs;
} simple_queue_t;
//
// q have lifetime of arg b or arg e.
// do not deallocate before queue closed.
static bool init_queue_from_chararray(simple_queue_t * restrict q,const char ** b, const char ** e){
q->method = From_CharArray;
q->filename_begin = b;
q->filename_end = e;
return true;
}
//
// q have lifetime of arg f.
// do not close file f before queue closed.
static bool init_queue_from_file(simple_queue_t * restrict q, FILE * f){
q->method = From_CharArray;
q->fs = f;
return true;
}
static const char * dequeue_from_simplecharqueue(simple_queue_t * restrict q,char * restrict buf, size_t size){
const char * filename;
if (q->method == From_CharArray){
if (__glibc_unlikely(q->filename_begin == q->filename_end)) {
return NULL;
}
filename = *(q->filename_begin++);
}
else{
//unsafe.
int t = fscanf(q->fs,"%s",buf);
if (__glibc_unlikely(t != 1)) {
return NULL;
}
filename = buf;
}
return filename;
}
//====
//thread
//====
struct SimpleThreadGlobal{ struct SimpleThreadGlobal{
size_t thread_arr_size; size_t thread_arr_size;
pthread_t * thread_arr; pthread_t * thread_arr;
pthread_mutex_t queueing_mutex; pthread_mutex_t queueing_mutex;
queueing_method_t method; simple_queue_t queue;
const char ** filename_begin;
const char ** filename_end;
} global_state; } global_state;
//if initialization is success, return true
bool init_global_state(struct SimpleThreadGlobal * t, size_t size){
pthread_mutex_init(&t->queueing_mutex,NULL);
t->thread_arr_size = size;
t->thread_arr = (pthread_t *)malloc(sizeof(*global_state.thread_arr) * global_state.thread_arr_size);
if (t->thread_arr == NULL){
return false;
}
return true;
}
typedef struct SimpleThreadReturn{ typedef struct SimpleThreadReturn{
int retval; int retval;
@ -249,7 +304,6 @@ void destroy_thread_arg(worker_arg_t * arg){
free(arg); free(arg);
} }
void * WorkerProc(void * args){ void * WorkerProc(void * args){
worker_arg_t * arg = (worker_arg_t *)(args); worker_arg_t * arg = (worker_arg_t *)(args);
worker_return_t * ret = create_worker_return(); worker_return_t * ret = create_worker_return();
@ -258,123 +312,128 @@ void * WorkerProc(void * args){
for(;;) for(;;)
{ {
pthread_mutex_lock(&global_state.queueing_mutex); pthread_mutex_lock(&global_state.queueing_mutex);
if (global_state.method == From_CharArray){ filename = dequeue_from_simplecharqueue(&global_state.queue,filename_buf,FILENAME_BUF_SIZE);
if (__glibc_unlikely(global_state.filename_begin == global_state.filename_end)) {
pthread_mutex_unlock(&global_state.queueing_mutex); pthread_mutex_unlock(&global_state.queueing_mutex);
break; if(filename == NULL) break;
}
filename = *(global_state.filename_begin++);
}
else{
//unsafe.
int t = fscanf(stdin,"%s",filename_buf);
if (__glibc_unlikely(t != 1)) {
pthread_mutex_unlock(&global_state.queueing_mutex);
break;
}
filename = filename_buf;
}
pthread_mutex_unlock(&global_state.queueing_mutex);
ret->retval += SendOpAndReceiveFile(filename,&arg->addr); ret->retval += SendOpAndReceiveFile(filename,&arg->addr);
ret->op_count++; ret->op_count++;
} }
destroy_thread_arg(arg); destroy_thread_arg(arg);
return ret; return ret;
} }
//====
static bool stdinisatty; // cmd parse
//====
int main(int argc, const char *argv[]){ static struct {
struct sockaddr_in addr;
const char * filename;
const char * server_name; const char * server_name;
in_port_t server_port = 0; in_port_t server_port;
int arg_filename_start = 3; bool stdinisatty;
int err; int thread_number_option;
int retval = 0; } cmd_args = {
.server_name = "",
.server_port = 0
};
init_bench_data(); static const char * show_help_message(FILE * file,const char * argv0){
stdinisatty = isatty(STDIN_FILENO); static const char * msg =
"USAUE: %s SERVERNAME PORT [Option]... [FILENAME]...\n"
"Options and arguments: \n"
"-b or --benchmark\t:benchmark mode\n"
"-nv or --no-verbose\t:no progress bar\n"
"-h\t:print help message.\n";
fprintf(file,msg,argv0);
return msg;
}
if (argc < (stdinisatty ? 4 : 3)){ static int parse_arg(int argc,const char *argv[]){
fprintf(stderr,"USAUE: %s SERVERNAME PORT [Option]... [FILENAME]...\n",argv[0]); int cur;
cmd_args.stdinisatty = isatty(STDIN_FILENO);
if (argc < (cmd_args.stdinisatty ? 4 : 3)){
show_help_message(stderr,argv[0]);
return 1; return 1;
} }
server_name = argv[1]; cmd_args.server_name = argv[1];
server_port = atoi(argv[2]); cmd_args.server_port = atoi(argv[2]);
while(arg_filename_start < argc){ if (cmd_args.server_port == 0){
if (strcmp("-b",argv[arg_filename_start])==0 fprintf(stderr,"port invalid\n");
||strcmp("--benchmark",argv[arg_filename_start])==0){ return -1;
arg_filename_start++; }
cmd_args.thread_number_option = 1;
for(cur = 3;cur < argc; cur++){
if (strcmp("-b",argv[cur])==0||strcmp("--benchmark",argv[cur])==0){
bench.benchmode = true; bench.benchmode = true;
} }
else if(strcmp("-nv",argv[arg_filename_start]) == 0||strcmp("--no-verbose",argv[arg_filename_start])==0){ else if(strcmp("-nv",argv[cur]) == 0||strcmp("--no-verbose",argv[cur])==0){
arg_filename_start++;
DisplayProgress = false; DisplayProgress = false;
} }
else if(strcmp("-t",argv[arg_filename_start]) == 0 || strcmp("--thread",argv[arg_filename_start]) == 0){ else if(strcmp("-t",argv[cur]) == 0 || strcmp("--thread",argv[cur]) == 0){
arg_filename_start++; cur++;
if (arg_filename_start >= argc){ if (cur >= argc){
fprintf(stderr,"need number"); fprintf(stderr,"need number");
return -2; return -2;
} }
thread_number_option = atoi(argv[arg_filename_start++]); cmd_args.thread_number_option = atoi(argv[cur]);
if(thread_number_option == 0){ if(cmd_args.thread_number_option == 0){
fprintf(stderr,"not number or zero"); fprintf(stderr,"not number or zero");
return -2; return -2;
} }
} }
else if(strcmp("-h",argv[cur]) == 0 || strcmp("--help",argv[cur]) == 0){
show_help_message(stdout,argv[0]);
return 0;
}
else break; else break;
} }
if (server_port == 0){ if (cmd_args.stdinisatty){
fprintf(stderr,"port invalid\n"); init_queue_from_chararray(&global_state.queue, &argv[cur],&argv[argc]);
return 1; }
else{
init_queue_from_file(&global_state.queue,stdin);
} }
err = getsockaddrbyname(AF_INET,SOCK_STREAM,0,server_name,(struct sockaddr *)&addr); return 0;
}
int main(int argc, const char *argv[]){
struct sockaddr_in addr;
const char * filename;
int err;
int retval = 0;
init_bench_data();
if(parse_arg(argc,argv) < 0) return EXIT_FAILURE;
err = getsockaddrbyname(AF_INET,SOCK_STREAM,0,cmd_args.server_name,(struct sockaddr *)&addr);
if (err != 0){ if (err != 0){
int check; int check;
fprintf(stderr,"netdb fail: %s\n",gai_strerror(err)); fprintf(stderr,"netdb fail: %s\n",gai_strerror(err));
//assume that sernmae is *.*.*.* and try to parse addr //assume that sernmae is *.*.*.* and try to parse addr
check = inet_pton(AF_INET,server_name,&addr.sin_addr); check = inet_pton(AF_INET,cmd_args.server_name,&addr.sin_addr);
assert(check != -1); assert(check != -1);
if (check == 0){ if (check == 0){
fprintf(stderr,"parsing fail : invaild format\n"); fprintf(stderr,"parsing fail : invaild format\n");
return 1; return -1;
} }
} }
addr.sin_family = AF_INET; addr.sin_family = AF_INET;
addr.sin_port = htons(server_port); addr.sin_port = htons(cmd_args.server_port);
if (bench.benchmode){ if (bench.benchmode){
clock_gettime(bench.clock_id,&bench.begin); clock_gettime(bench.clock_id,&bench.begin);
} }
if(thread_number_option == 1){ if(cmd_args.thread_number_option == 1){
char filename_buf[FILENAME_BUF_SIZE]; char filename_buf[FILENAME_BUF_SIZE];
for (;;){ for (;;){
if (stdinisatty){ filename = dequeue_from_simplecharqueue(&global_state.queue,filename_buf,FILENAME_BUF_SIZE);
if (arg_filename_start >= argc) break; if (filename == NULL) break;
filename = argv[arg_filename_start++];
}
else{
//unsafe.
int t = fscanf(stdin,"%s",filename_buf);
if (t != 1) break;
filename = filename_buf;
}
retval += SendOpAndReceiveFile(filename,(struct sockaddr *)&addr); retval += SendOpAndReceiveFile(filename,(struct sockaddr *)&addr);
bench.op_count++; bench.op_count++;
} }
} }
else{ else{
int i = 0; int i = 0;
global_state.method = stdinisatty ? From_CharArray : From_FileStream; init_global_state(&global_state,cmd_args.thread_number_option);
global_state.filename_begin = &argv[arg_filename_start];
global_state.filename_end = &argv[argc];
pthread_mutex_init(&global_state.queueing_mutex,NULL);
global_state.thread_arr_size = thread_number_option;
global_state.thread_arr = (pthread_t *)malloc(sizeof(*global_state.thread_arr) * global_state.thread_arr_size);
for (i = 0; i < global_state.thread_arr_size; i++){ for (i = 0; i < global_state.thread_arr_size; i++){
worker_arg_t * arg = create_thread_arg((struct sockaddr *)&addr); worker_arg_t * arg = create_thread_arg((struct sockaddr *)&addr);
pthread_create(&global_state.thread_arr[i],NULL,WorkerProc,arg); pthread_create(&global_state.thread_arr[i],NULL,WorkerProc,arg);