diff --git a/p-client.c b/p-client.c index 6053cb8..8d69f66 100644 --- a/p-client.c +++ b/p-client.c @@ -13,18 +13,10 @@ #include #include #include -#include -#include +#include #include "socket_wrapper.h" +#include "timerhelper.h" -#ifdef USE_SENDFILE -#include -#ifndef DEFAULT_SEND_FILE_CHUNK_SIZE -const size_t SEND_FILE_CHUNK_SIZE = 0x100000; /*1MB*/ -#else -const size_t SEND_FILE_CHUNK_SIZE = DEFAULT_SEND_FILE_CHUNK_SIZE; /*1MB*/ -#endif -#endif #ifndef DEFAULT_TIMEOUT static const int TIMEOUT = 5; #else @@ -35,10 +27,14 @@ static const int PROGRESS_BAR_WIDTH = 30; #else static const int PROGRESS_BAR_WIDTH = DEFAULT_PROGRESS_BAR_WIDTH; #endif + +enum{ + FILENAME_BUF_SIZE = 1024 +}; /*======== *Operation *========*/ - +//if success, return zero int sendReadOp(int sock,const char * filename){ struct ReadOp op; op.file_url_size = strlen(filename); @@ -133,6 +129,7 @@ int recvFile(int sock, const char * filename,size_t file_size){ DisplayProgressBar100Percent(file_size); END: free(buf); + close(fd); return return_value; } @@ -144,7 +141,7 @@ int recvData(int sock,const char * filename){ else perror("recv fail"); return -1; } - static char error_meesage_buf[80] = ""; + char error_meesage_buf[80] = ""; switch(res.res){ case RES_ERR: fprintf(stderr,"Server Fail: %s\n", strerror(res.err_number)); @@ -168,6 +165,33 @@ int recvData(int sock,const char * filename){ return 0; } + +static bool quiet_mode = false; +int SendOpAndReceiveFile(const char * filename, struct sockaddr const * addr){ + int sock; + int ret = -1; + if(!quiet_mode) + fprintf(stdout,"request %s\n",filename); + + sock = socket(AF_INET,SOCK_STREAM,0); + if(sock < 0){ + perror("sock create fail"); + return -1; + } + + if(connect(sock,(struct sockaddr *)addr,sizeof(*addr)) < 0){ + perror("connect failed"); + return -1; + } + if(sendReadOp(sock,filename) == 0){ + ret = recvData(sock,filename); + } + close(sock); + return ret; +} +//==== +//bench +//==== struct benchmark_data{ bool benchmode; struct timespec begin; @@ -187,88 +211,247 @@ void init_bench_data(){ clock_getres(bench.clock_id,&bench.resolution); } } -static inline struct timespec timespec_sub(struct timespec a,struct timespec b){ - struct timespec ret; - ret.tv_sec = a.tv_sec - b.tv_sec; - ret.tv_nsec = a.tv_nsec - b.tv_nsec; - if (ret.tv_nsec < 0){ - ret.tv_sec--; - ret.tv_nsec += 1000000000L; +//==== +//simple queue +//==== +typedef enum{ + From_CharArray, + From_FileStream +} 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_FileStream; + 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{ + size_t thread_arr_size; + pthread_t * thread_arr; + pthread_mutex_t queueing_mutex; + simple_queue_t queue; +} 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{ + int retval; + int op_count; +} worker_return_t; + +__attribute_malloc__ worker_return_t * create_worker_return(){ + return (worker_return_t *)malloc(sizeof(worker_return_t)); +} +void destroy_worker_return(worker_return_t * r){ + free(r); +} + +typedef struct SimpleThreadArg{ + struct sockaddr addr; +} worker_arg_t; + +__attribute_malloc__ worker_arg_t * create_thread_arg(struct sockaddr * r){ + worker_arg_t * ret = (worker_arg_t *)malloc(sizeof(*ret)); + memcpy(&ret->addr,r,sizeof(*r)); return ret; } +void destroy_thread_arg(worker_arg_t * arg){ + free(arg); +} + +void * WorkerProc(void * args){ + worker_arg_t * arg = (worker_arg_t *)(args); + worker_return_t * ret = create_worker_return(); + char filename_buf[FILENAME_BUF_SIZE]; + const char * filename; + for(;;) + { + pthread_mutex_lock(&global_state.queueing_mutex); + filename = dequeue_from_simplecharqueue(&global_state.queue,filename_buf,FILENAME_BUF_SIZE); + pthread_mutex_unlock(&global_state.queueing_mutex); + if(filename == NULL) break; + ret->retval += SendOpAndReceiveFile(filename,&arg->addr); + ret->op_count++; + } + destroy_thread_arg(arg); + return ret; +} +//==== +// cmd parse +//==== +static struct { + const char * server_name; + in_port_t server_port; + bool stdinisatty; + int thread_number_option; + bool quiet_mode; +} cmd_args = { + .server_name = "", + .server_port = 0 +}; + +static const char * show_help_message(FILE * file,const char * argv0){ + 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; +} + +static int parse_arg(int argc,const char *argv[]){ + int cur; + cmd_args.stdinisatty = isatty(STDIN_FILENO); + if (argc < (cmd_args.stdinisatty ? 4 : 3)){ + show_help_message(stderr,argv[0]); + return 1; + } + cmd_args.server_name = argv[1]; + cmd_args.server_port = atoi(argv[2]); + if (cmd_args.server_port == 0){ + fprintf(stderr,"port invalid\n"); + return -1; + } + cmd_args.quiet_mode = false; + 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; + } + else if(strcmp("-nv",argv[cur]) == 0||strcmp("--no-verbose",argv[cur])==0){ + DisplayProgress = false; + } + else if(strcmp("-t",argv[cur]) == 0 || strcmp("--thread",argv[cur]) == 0){ + cur++; + if (cur >= argc){ + fprintf(stderr,"need number"); + return -2; + } + cmd_args.thread_number_option = atoi(argv[cur]); + if(cmd_args.thread_number_option == 0){ + fprintf(stderr,"not number or zero"); + return -2; + } + } + else if(strcmp("-h",argv[cur]) == 0 || strcmp("--help",argv[cur]) == 0){ + show_help_message(stdout,argv[0]); + return -1; + } + else if(strcmp("-q",argv[cur]) == 0 || strcmp("--quiet",argv[cur]) == 0){ + cmd_args.quiet_mode = true; + quiet_mode = true; + } + else break; + } + if (cmd_args.stdinisatty){ + init_queue_from_chararray(&global_state.queue, &argv[cur],&argv[argc]); + } + else{ + init_queue_from_file(&global_state.queue,stdin); + } + return 0; +} int main(int argc, const char *argv[]){ struct sockaddr_in addr; const char * filename; - const char * server_name; - in_port_t server_port = 0; - int arg_filename_start = 3; - int sock, err; + int err; int retval = 0; + init_bench_data(); + if(parse_arg(argc,argv) < 0) return EXIT_FAILURE; - - if (argc < 4){ - fprintf(stderr,"USAUE: %s SERVERNAME PORT [Option]... [FILENAME]...\n",argv[0]); - return 1; - } - server_name = argv[1]; - server_port = atoi(argv[2]); - for(;;){ - if (strcmp("-b",argv[arg_filename_start])==0 - ||strcmp("--benchmark",argv[arg_filename_start])==0){ - arg_filename_start++; - bench.benchmode = true; - } - else if(strcmp("--nogui",argv[arg_filename_start])==0){ - arg_filename_start++; - DisplayProgress = false; - } - else break; - } - - if (server_port == 0){ - fprintf(stderr,"port invalid\n"); - return 1; - } - - err = getsockaddrbyname(AF_INET,SOCK_STREAM,0,server_name,(struct sockaddr *)&addr); + err = getsockaddrbyname(AF_INET,SOCK_STREAM,0,cmd_args.server_name,(struct sockaddr *)&addr); if (err != 0){ int check; fprintf(stderr,"netdb fail: %s\n",gai_strerror(err)); //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); if (check == 0){ fprintf(stderr,"parsing fail : invaild format\n"); - return 1; + return -1; } } addr.sin_family = AF_INET; - addr.sin_port = htons(server_port); + addr.sin_port = htons(cmd_args.server_port); if (bench.benchmode){ clock_gettime(bench.clock_id,&bench.begin); } - while (arg_filename_start < argc){ - filename = argv[arg_filename_start++]; - sock = socket(AF_INET,SOCK_STREAM,0); - if(sock < 0){ - perror("sock create fail"); - return 1; + if(cmd_args.thread_number_option == 1){ + char filename_buf[FILENAME_BUF_SIZE]; + for (;;){ + filename = dequeue_from_simplecharqueue(&global_state.queue,filename_buf,FILENAME_BUF_SIZE); + if (filename == NULL) break; + + retval += SendOpAndReceiveFile(filename,(struct sockaddr *)&addr); + bench.op_count++; } - - if(connect(sock,(struct sockaddr *)&addr,sizeof(addr)) < 0){ - perror("connect failed"); - return 1; + } + else{ + int i = 0; + init_global_state(&global_state,cmd_args.thread_number_option); + for (i = 0; i < global_state.thread_arr_size; i++){ + worker_arg_t * arg = create_thread_arg((struct sockaddr *)&addr); + pthread_create(&global_state.thread_arr[i],NULL,WorkerProc,arg); } - if(sendReadOp(sock,filename) == 0){ - int ret = recvData(sock,filename); - retval += ret; + for (i = 0; i < global_state.thread_arr_size; i++){ + worker_return_t * ret; + pthread_join(global_state.thread_arr[i],(void **)&ret); + bench.op_count += ret->op_count; + retval += ret->retval; + destroy_worker_return(ret); } - close(sock); - bench.op_count++; } if (bench.benchmode){ struct timespec result; @@ -278,7 +461,7 @@ int main(int argc, const char *argv[]){ if (result.tv_sec == 0) avg = result.tv_nsec; else avg = result.tv_sec * 1e9 + result.tv_nsec; avg /= bench.op_count; - fprintf(stdout,"operation: %lf ns/op\n",avg); + fprintf(stdout,"operation: %lf us/op\n",avg / 1000.0); fprintf(stdout,"resolution: %ld sec %ld ns\n",bench.resolution.tv_sec,bench.resolution.tv_nsec); }