#include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include "socket_wrapper.h" #include "timerhelper.h" #ifndef DEFAULT_TIMEOUT static const int TIMEOUT = 5; #else static const int TIMEOUT = DEFAULT_TIMEOUT; #endif #ifndef DEFAULT_PROGRESS_BAR_WIDTH 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); op.padding0 = 0; if(send(sock,&op,sizeof(op),0)<0){ perror("readop send fail"); return -1; } #ifdef MUL_CLIENT sleep(MUL_CLIENT); #endif if(send(sock,filename,op.file_url_size,0)<0){ perror("readop filename send fail"); return -1; } return 0; } /** * @arg cur_progress : it is percentage. */ bool isProgressBarNeedUpdate(size_t offset,size_t total,double cur_progress){ return ((double)offset / (double)total) > (cur_progress / 100.0); } void DisplayProgressBar(size_t offset,size_t total,double cur_progress){ char buf[PROGRESS_BAR_WIDTH]; size_t i; size_t cur_pos = (cur_progress / 100.0 * PROGRESS_BAR_WIDTH); //must be less than SIZE_MAX. other value is undefined behavior. for (i = 0; i < PROGRESS_BAR_WIDTH; i++) { if (i < cur_pos) buf[i] = '='; else if(i == cur_pos) buf[i] = '>'; else buf[i] = '.'; } printf("\r[%s]: %.2f%% bytes: %ld/%ld bytes",buf,cur_progress,total,offset); } void DisplayProgressBar100Percent(size_t total){ size_t i; char buf[PROGRESS_BAR_WIDTH+1]; for (i = 0; i < PROGRESS_BAR_WIDTH; i++){ buf[i] = '='; } buf[PROGRESS_BAR_WIDTH] = '\0'; printf("\r[%s]: 100%% bytes: %ld/%ld bytes\n",buf,total,total); } static bool DisplayProgress = true; int recvFile(int sock, const char * filename,size_t file_size){ int fd; size_t count = 0; int i; double cur_progress = 1; int return_value = 0; int buf_sz = getBufferSizeFrom(sock); uint8_t * buf = malloc(buf_sz*sizeof(*buf)); if (buf == NULL){ return -1; } fd = open(filename,O_WRONLY|O_CREAT|O_TRUNC,S_IRWXU|S_IRWXG|S_IROTH); if (fd < 0) { perror("file open fail"); return -1; } while (file_size - count > 0) { int readed = buf_sz < file_size - count ? buf_sz : file_size - count; if((i = recv_until_byte(sock,buf,readed,TIMEOUT)) < 0){ if(i == -2) fprintf(stderr,"recv file failed : timeout connetion lost\n"); else perror("recv file failed"); return_value = -1; goto END; } if(write(fd,buf,readed)<0){ perror("file write failed"); return_value = -1; goto END; } if( DisplayProgress && isProgressBarNeedUpdate(count,file_size,cur_progress) ){ DisplayProgressBar(count,file_size,cur_progress); cur_progress = (int)((((double)count / (double)file_size)) * 100.0 + 1.0); fflush(stdout); } count += readed; #ifdef SLOW_CLIENT usleep(SLOW_CLIENT); #endif } if(DisplayProgress) DisplayProgressBar100Percent(file_size); END: free(buf); close(fd); return return_value; } int recvData(int sock,const char * filename){ struct TransferResult res; int i=0; if((i=recv_until_byte(sock,&res,sizeof(res),TIMEOUT)) < 0){ if (i == -2) fprintf(stderr,"timeout\n"); else perror("recv fail"); return -1; } char error_meesage_buf[80] = ""; switch(res.res){ case RES_ERR: fprintf(stderr,"Server Fail: %s\n", strerror(res.err_number)); return -1; case RES_USR_ERR: assert(res.error_msg_size < 80);/*todo : fix*/ if((i=recv_until_byte(sock,error_meesage_buf,res.error_msg_size,TIMEOUT)) < 0){ if (i == -2) fprintf(stderr,"timeout\n"); else perror("recv fail"); return -1; } fprintf(stderr,"Error Message From Server: %s\n",error_meesage_buf); return -1; case RES_OK: return recvFile(sock,filename,res.file_size); break; default: fprintf(stderr,"unknown value!\n"); return -1; } 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; struct timespec end; clockid_t clock_id; struct timespec resolution; int op_count; } bench = {0,}; void init_bench_data(){ int i; memset(&bench,0,sizeof(bench)); bench.clock_id = CLOCK_PROCESS_CPUTIME_ID; i = clock_getres(bench.clock_id,&bench.resolution); if (i < 0){ bench.clock_id = CLOCK_REALTIME; clock_getres(bench.clock_id,&bench.resolution); } } //==== //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_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{ 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 0; } 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; 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){ 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,cmd_args.server_name,&addr.sin_addr); assert(check != -1); if (check == 0){ fprintf(stderr,"parsing fail : invaild format\n"); return -1; } } addr.sin_family = AF_INET; addr.sin_port = htons(cmd_args.server_port); if (bench.benchmode){ clock_gettime(bench.clock_id,&bench.begin); } 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++; } } 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); } 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); } } if (bench.benchmode){ struct timespec result; double avg; clock_gettime(bench.clock_id,&bench.end); result = timespec_sub(bench.end,bench.begin); 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 us/op\n",avg / 1000.0); fprintf(stdout,"resolution: %ld sec %ld ns\n",bench.resolution.tv_sec,bench.resolution.tv_nsec); } return retval; }