diff --git a/example/demo/hos_write_complete.cpp b/example/demo/hos_write_complete.cpp index 6fc40409..00a78109 100644 --- a/example/demo/hos_write_complete.cpp +++ b/example/demo/hos_write_complete.cpp @@ -70,18 +70,15 @@ int main(int argc, char *argv[]) char *bucket = argv[1]; char *object = argv[2]; int test_times = atoi(argv[3]); - //int test_times = 10000; char *buf = (char *)malloc(1024 * 1024 * 40); - //char buf[1024 * 1024 * 4]; - //char buf[1024 * 4]; size_t buf_size; int mode = FILE_MODE; size_t fd[10001] = {0}; userdata_t data = {&finished}; - //printf("finished:%lu\n", finished.tv_sec * 1000 * 1000 * 1000 + finished.tv_nsec); file_to_buffer(object, buf, &buf_size); + hos_init_api(); debuginfo("hos_client_init start ...\n"); hos_client_handle handle = hos_client_create("http://192.168.40.223:9098/hos/", "default", "default", 4000); //hos_client_handle handle = hos_client_create("http://192.168.32.10:9098/hos/", "default", "default", 4); @@ -110,11 +107,6 @@ int main(int argc, char *argv[]) #if 1 mode = FILE_MODE; -#if 0 - for (i = 0; i < test_times; i++) - { - } -#endif debuginfo("hos_upload_file start ...\n"); clock_gettime(CLOCK_MONOTONIC, &start); for (i = 0; i < test_times; i++) @@ -165,6 +157,7 @@ int main(int argc, char *argv[]) } debuginfo("hos_client_close end ...\n"); + hos_shutdown_api(); free(buf); return 0; } diff --git a/example/performance/HosClientPerformance.cpp b/example/performance/HosClientPerformance.cpp index c9b10359..e2aa7b23 100644 --- a/example/performance/HosClientPerformance.cpp +++ b/example/performance/HosClientPerformance.cpp @@ -13,10 +13,11 @@ extern "C" #include #include #include +#include } #include"../../src/hos_client.h" -#define MAX_THREAD_NUM 16 +#define MAX_THREAD_NUM 32 #ifndef MIN #define MIN(a,b) ((a) > (b)) ? (b) : (a) #endif @@ -30,7 +31,9 @@ typedef struct conf_s char file[STRING_SIZE]; size_t pool_size; size_t thread_sum; - size_t append_size; + size_t size; + float append_size; + size_t slice; int mode; }conf_t; @@ -51,6 +54,7 @@ static void configuration_init(conf_t *conf) conf->append_size = 1024; conf->thread_sum = 1; conf->mode = BUFF_MODE; + conf->slice = 0; return ; } @@ -136,14 +140,20 @@ static int upload_file(char *file, char *buff, int buff_len, int test_times, thr size_t fd[3000]; conf_t *conf = &thread_info->conf; struct timespec tstart, tend, twrite; - size_t time_write, time_upload; + long time_write = 0, time_upload = 0; size_t len = strlen(performance_info); char file_size[128]; + long record[1000] = {0}; + double variance = 0.00; + double average = 0.00; + long time = 0; + int success_cnt; //写文件 - clock_gettime(CLOCK_MONOTONIC, &tstart); + //clock_gettime(CLOCK_MONOTONIC, &tstart); for (i = 0; i < test_times; i++) { + clock_gettime(CLOCK_MONOTONIC, &tstart); fp = fopen(file, "w+"); if (fp == NULL) { @@ -157,31 +167,44 @@ static int upload_file(char *file, char *buff, int buff_len, int test_times, thr return -1; } fclose(fp); + clock_gettime(CLOCK_MONOTONIC, &twrite); + record[i] = calc_time(tstart, twrite); + time_write += record[i]; } - clock_gettime(CLOCK_MONOTONIC, &twrite); - time_write = calc_time(tstart, twrite); + //clock_gettime(CLOCK_MONOTONIC, &twrite); + //time_write = calc_time(tstart, twrite); time_write /= test_times; //上传文件 - clock_gettime(CLOCK_MONOTONIC, &tstart); + //clock_gettime(CLOCK_MONOTONIC, &tstart); for (i = 0; i < test_times; i++) { + clock_gettime(CLOCK_MONOTONIC, &tstart); fd[i] = hos_open_fd(thread_info->handle, conf->bucket, conf->object, callback, NULL, thread_info->thread_num, conf->mode); if (hos_write(fd[i], file, 0, thread_info->thread_num) != HOS_CLIENT_OK) { printf("error:hos_write file:%s\n", file); return -1; } + clock_gettime(CLOCK_MONOTONIC, &tend); + time = calc_time(tstart, tend); + time_upload += time; + record[i] += time; } - clock_gettime(CLOCK_MONOTONIC, &tend); - time_upload = calc_time(tstart, tend); + //clock_gettime(CLOCK_MONOTONIC, &tend); + //time_upload = calc_time(tstart, tend); time_upload /= test_times; - sprintf(file_size, "%dk", buff_len / 1024); - sprintf(&performance_info[len], "%-20lu%-20s%-20lu%-20lu%-20lu\n", - thread_info->thread_num, file_size, time_write, time_upload, time_write + time_upload); + average = time_write + time_upload; + for (i = 0; i < test_times; i++) + { + variance += pow((record[i] - average), 2); + } + variance /= test_times; - //hos_close_fd(fd, thread_info->thread_num); + sprintf(file_size, "%dk", buff_len / 1024); + sprintf(&performance_info[len], "%-20lu%-20s%-20ld%-20ld%-20lf%-20lf\n", + thread_info->thread_num, file_size, time_write, time_upload, average, sqrt(variance)); return 0; } @@ -201,109 +224,127 @@ static int upload_buff(char * buff, int buff_len, int test_times, thread_info_t char append_size[128]; size_t success_cnt = 0; int ret = 0; + int drop = 0; + double variance = 0.00; + double average = 0.00; + long record[1000] = {0}; -#if 0 - size_t time_tmp = 0; - while(1) - { - tmp = j * conf->append_size; - rest = buff_len - tmp; - if (rest < conf->append_size) - { - clock_gettime(CLOCK_MONOTONIC, &tstart); - for (i = 0; i < test_times; i++) - { - hos_write(fd, &buff[tmp], rest, thread_info->thread_num); - } - clock_gettime(CLOCK_MONOTONIC, &ttmp); - time_tmp = calc_time(tstart, ttmp); - time_tmp /= test_times; - time += time_tmp; - - sprintf(file_size, "%dk", buff_len / 1024); - sprintf(append_size, "%luk", conf->append_size / 1024); - len = strlen(performance_info); - sprintf(&performance_info[len], "%-20lu%-20s%-20s%-20lu%-20lu\n", - thread_info->thread_num, file_size, append_size, time_tmp, time); - break; - } - clock_gettime(CLOCK_MONOTONIC, &tstart); - for (i = 0; i < test_times; i++) - { - hos_write(fd, &buff[tmp], conf->append_size, thread_info->thread_num); - } - clock_gettime(CLOCK_MONOTONIC, &ttmp); - time_tmp = calc_time(tstart, ttmp); - time_tmp /= test_times; - time += time_tmp; - - sprintf(file_size, "%dk", buff_len / 1024); - sprintf(append_size, "%luk", conf->append_size / 1024); - len = strlen(performance_info); - sprintf(&performance_info[len], "%-20lu%-20s%-20s%-20lu%-20lu\n", - thread_info->thread_num, file_size, append_size, time_tmp, time); - - j++; - } -#else - if (conf->mode & APPEND_MODE) + if (conf->slice) { fd = hos_open_fd(thread_info->handle, conf->bucket, conf->object, callback, NULL, thread_info->thread_num, conf->mode); - clock_gettime(CLOCK_MONOTONIC, &tstart); for (i = 0; i < test_times; i++) { + j = 0; + clock_gettime(CLOCK_MONOTONIC, &tstart); while(1) { - tmp = j * conf->append_size; + tmp = j * conf->slice; rest = buff_len - tmp; - if (rest < conf->append_size) + if (rest < conf->slice) { hos_write(fd, &buff[tmp], rest, thread_info->thread_num); break; } - hos_write(fd, &buff[tmp], conf->append_size, thread_info->thread_num); + hos_write(fd, &buff[tmp], conf->slice, thread_info->thread_num); j++; } + clock_gettime(CLOCK_MONOTONIC, &ttmp); + record[i] = calc_time(tstart, ttmp); + average += record[i]; } - clock_gettime(CLOCK_MONOTONIC, &ttmp); - time = calc_time(tstart, ttmp); - time /= test_times; + average /= test_times; - sprintf(file_size, "%dk", buff_len / 1024); - sprintf(append_size, "%luk", conf->append_size / 1024); - len = strlen(performance_info); - sprintf(&performance_info[len], "%-20lu%-20s%-20s%-20d%-20lu\n", - thread_info->thread_num, file_size, append_size, 0, time); - }else - { - clock_gettime(CLOCK_MONOTONIC, &tstart); for (i = 0; i < test_times; i++) { - fd = hos_open_fd(thread_info->handle, conf->bucket, conf->object, callback, NULL, thread_info->thread_num, conf->mode); - ret = hos_write(fd, buff, buff_len, thread_info->thread_num); - if (ret == HOS_CLIENT_OK) - { - success_cnt++; - }else - { - printf("error code:%d\n", ret); - } + variance += pow((record[i] - average), 2); } - clock_gettime(CLOCK_MONOTONIC, &ttmp); - time = calc_time(tstart, ttmp); - if (success_cnt) - time /= success_cnt; - else - time /= test_times; + variance /= test_times; sprintf(file_size, "%dk", buff_len / 1024); - sprintf(append_size, "%luk", conf->append_size / 1024); + sprintf(append_size, "%gk", conf->append_size / 1024); len = strlen(performance_info); - sprintf(&performance_info[len], "%-20lu%-20s%-20d%-20d%-20lu\n", - thread_info->thread_num, file_size, 0, 0, time); + sprintf(&performance_info[len], "%-20lu%-20s%-20s%-20d%-20lf%-20lf\n", + thread_info->thread_num, file_size, append_size, 0, average, sqrt(variance)); + }else{ + + if (conf->mode & APPEND_MODE) + { + fd = hos_open_fd(thread_info->handle, conf->bucket, conf->object, callback, NULL, thread_info->thread_num, conf->mode); +#if 1 + for (i = 0; i < test_times; i++) + { + clock_gettime(CLOCK_MONOTONIC, &tstart); +#endif + j = 0; + while(1) + { + tmp = j * conf->append_size; + rest = buff_len - tmp; + if (rest < conf->append_size) + { + hos_write(fd, &buff[tmp], rest, thread_info->thread_num); + break; + } + hos_write(fd, &buff[tmp], conf->append_size, thread_info->thread_num); + j++; + } +#if 1 + clock_gettime(CLOCK_MONOTONIC, &ttmp); + record[i] = calc_time(tstart, ttmp); + average += record[i]; + } + average /= test_times; +#endif + + for (i = 0; i < test_times; i++) + { + variance += pow((record[i] - average), 2); + } + variance /= test_times; + + sprintf(file_size, "%dk", buff_len / 1024); + sprintf(append_size, "%gk", conf->append_size / 1024); + len = strlen(performance_info); + sprintf(&performance_info[len], "%-20lu%-20s%-20s%-20d%-20lf%-20lf\n", + thread_info->thread_num, file_size, append_size, 0, average,sqrt(variance)); + }else + { + for (i = 0; i < test_times; i++) + { + clock_gettime(CLOCK_MONOTONIC, &tstart); + fd = hos_open_fd(thread_info->handle, conf->bucket, conf->object, callback, NULL, thread_info->thread_num, conf->mode); + ret = hos_write(fd, buff, buff_len, thread_info->thread_num); + if (ret == HOS_CLIENT_OK) + { + success_cnt++; + }else + { + printf("error code:%d, thread_id:%d\n", ret, thread_info->thread_num); + //break; + } + clock_gettime(CLOCK_MONOTONIC, &ttmp); + record[i] = calc_time(tstart, ttmp); + average += record[i]; + } + if (success_cnt) + average /= success_cnt; + else + average /= test_times; + + for (i = 0; i < test_times; i++) + { + variance += pow((record[i] - average), 2); + } + variance /= test_times; + + sprintf(file_size, "%dk", buff_len / 1024); + sprintf(append_size, "%luk", conf->append_size / 1024); + len = strlen(performance_info); + sprintf(&performance_info[len], "%-20lu%-20s%-20d%-20d%-20lf%-20lf\n", + thread_info->thread_num, file_size, 0, 0, average, sqrt(variance)); + } } -#endif hos_close_fd(fd, thread_info->thread_num); return 0; @@ -415,12 +456,12 @@ int main(int argc, char *argv[]) #endif configuration_init(&conf); //读取命令行配置 - while((ch = getopt(argc, argv, "a:e:b:o:f:p:t:k:BFAh")) != -1) + while((ch = getopt(argc, argv, "a:e:b:o:f:p:t:k:s:S:BFAh")) != -1) { switch(ch) { case 'a': - conf.append_size = 1024 * atoi(optarg); + conf.append_size = 1024 * atof(optarg); break; case 'e': //endpoint @@ -461,6 +502,12 @@ int main(int argc, char *argv[]) case 'k': conf.append_size = 1024 * atoi(optarg); break; + case 's': + conf.size = atoi(optarg); + break; + case 'S': + conf.slice = atoi(optarg); + break; case 'h': default: printf("usage: HosClientPerformance \n[-e set endpoint] \n[-b set bucket] \n" @@ -492,8 +539,8 @@ int main(int argc, char *argv[]) return -1; } - set_thread_sum(handle, conf.thread_sum); - set_cache_size(handle, conf.append_size); + hos_set_thread_sum(handle, conf.thread_sum); + hos_set_cache_size(handle, conf.append_size); //创建bucket if (hos_create_bucket(handle, conf.bucket)) @@ -529,7 +576,7 @@ int main(int argc, char *argv[]) } CPU_ZERO(&mask); - CPU_SET(thread_num + 8, &mask); + CPU_SET(thread_num, &mask); if (pthread_setaffinity_np(thread[thread_num], sizeof(mask), &mask) != 0) { printf("warning:could not set CPU affinity, continuing...\n"); diff --git a/example/performance/file/c.sh b/example/performance/file/c.sh index 974d0741..b98adf7d 100755 --- a/example/performance/file/c.sh +++ b/example/performance/file/c.sh @@ -4,7 +4,7 @@ # Created Time: Tue 13 Oct 2020 06:28:57 PM CST ######################################################################### #!/bin/bash -size=("1k" "10k" "100k" "1M" "2M" "3M" "4M" "5M" "10M" "20M" "30M") +size=("100" "200" "300" "400" "500" "600" "700" "800" "900" "1k" "1500" "10k" "100k" "1M" "2M" "3M" "4M" "5M" "10M" "20M" "30M") num=0 diff --git a/example/performance/test.sh b/example/performance/test.sh index 1c18bc39..19a04113 100755 --- a/example/performance/test.sh +++ b/example/performance/test.sh @@ -4,14 +4,22 @@ # Created Time: Tue 13 Oct 2020 02:28:31 PM CST ######################################################################### #!/bin/bash -size=(1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 30, 40, 50, 60, 70, 80, 90, 100, 110, 120, 130, 140, 150, 200, 300, 400, 500) +#size=(0.1 0.2 0.3 0.4 0.5 0.6 0.7 0.8 0.9 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 30 40 50 60 70 80 90 100 110 120 130 140 150 200 300 400 500) +#size=(100 200 300 400 500 600 700 800 900 1k 1500 10k 100k 1M 2M 3M 4M 5M 10M 20M) +size=(1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32) num=0 while((${num} < 37)) do #./HosClientPerformance -a ${size[$num]} - echo "./HosClientPerformance -a ${size[$num]} -A -f ./file/1M.data" - ./HosClientPerformance -a ${size[$num]} -A -f ./file/1M.data + #echo "./HosClientPerformance -a ${size[$num]} -A -f ./file/100k.data" + #./HosClientPerformance -a ${size[$num]} -A -f ./file/100k.data + #echo ./HosClientPerformance -f ./file/${size[$num]}.data -F + #./HosClientPerformance -f ./file/${size[$num]}.data -F + #echo ./HosClientPerformance -a 100 -A -f ./file/100k.data -t ${size[$num]} -S 1024 + #./HosClientPerformance -a 100 -A -f ./file/100k.data -t ${size[$num]} -S 1024 + echo ./HosClientPerformance -a 100 -A -f ./file/100k.data -S ${size[$num]} -t 32 + ./HosClientPerformance -a 100 -A -f ./file/100k.data -S ${size[$num]} -t 32 let "num++" done diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt index 5bcb016a..2228c01c 100644 --- a/src/CMakeLists.txt +++ b/src/CMakeLists.txt @@ -9,7 +9,7 @@ link_directories(${SUPPORT_INSTALL_PREFIX}) link_directories(${CMAKE_INSTALL_PREFIX}/lib) set(CMKAE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -shared -fPIC") add_library(${lib_name}_shared SHARED hos_client.cpp hos_hash.cpp) -target_link_libraries(${lib_name}_shared libaws-cpp-sdk-s3.so libaws-cpp-sdk-core.so libMESA_field_stat2.so) +target_link_libraries(${lib_name}_shared libaws-cpp-sdk-s3.so libaws-cpp-sdk-core.so libMESA_field_stat2.so libcurl.so) set_target_properties(${lib_name}_shared PROPERTIES OUTPUT_NAME ${lib_name}) add_dependencies(${lib_name}_shared aws-sdk-cpp-master-static) diff --git a/src/hos_client.cpp b/src/hos_client.cpp index 290c4311..5d251aeb 100644 --- a/src/hos_client.cpp +++ b/src/hos_client.cpp @@ -42,10 +42,14 @@ typedef struct hos_client_handle_s #define HOS_FS2_STOP 2 int *line_ids; int *column_ids; - size_t tx_pkts; - size_t tx_bytes; - size_t rx_pkts; - size_t rx_bytes; + int *tx_pkts; + int *tx_bytes; + int *rx_pkts; + int *rx_bytes; + int *tx_pkts_last; + int *tx_bytes_last; + int *rx_pkts_last; + int *rx_bytes_last; }hos_client_handle_t; hos_info_t *hash_hos_info[MAX_HOS_CLIENT_THREAD_NUM]; @@ -98,19 +102,19 @@ static void PutObjectAsyncFinished(const Aws::S3::S3Client* S3Client, } } -void set_cache_size(hos_client_handle client, size_t cache_size) +void hos_set_cache_size(hos_client_handle client, size_t cache_size) { client->cache_size = cache_size; return ; } -void set_cache_times(hos_client_handle client, size_t cache_times) +void hos_set_cache_times(hos_client_handle client, size_t cache_times) { client->cache_times = cache_times; return ; } -void set_thread_sum(hos_client_handle client, size_t thread_sum) +void hos_set_thread_sum(hos_client_handle client, size_t thread_sum) { client->thread_sum = thread_sum; return ; @@ -145,7 +149,7 @@ hos_client_handle hos_client_create(const char *endpoint, const char *accesskeyi config.endpointOverride = endpoint; config.verifySSL = false; config.enableEndpointDiscovery = true; - config.executor = std::shared_ptr(std::make_shared(pool_size));//支持线程池 + config.executor = std::shared_ptr(std::make_shared(pool_size, Aws::Utils::Threading::OverflowPolicy::REJECT_IMMEDIATELY));//支持线程池 handle->S3Client = new Aws::S3::S3Client(credentials, config, Aws::Client::AWSAuthV4Signer::PayloadSigningPolicy::Never, false); handle->options = options; @@ -167,6 +171,15 @@ hos_client_handle hos_client_create(const char *endpoint, const char *accesskeyi static void *fs2_statistics(void *ptr) { hos_client_handle handle = (hos_client_handle)ptr; + int i = 0; + int rx_pkts_sum = 0; + int rx_bytes_sum = 0; + int tx_pkts_sum = 0; + int tx_bytes_sum = 0; + int rx_pkts_sum_interval = 0; + int rx_bytes_sum_interval = 0; + int tx_pkts_sum_interval = 0; + int tx_bytes_sum_interval = 0; while(1) { @@ -174,17 +187,53 @@ static void *fs2_statistics(void *ptr) { break; } - FS_operate(handle->fs2_handle, handle->line_ids[0], handle->column_ids[0], FS_OP_ADD, handle->tx_pkts); - FS_operate(handle->fs2_handle, handle->line_ids[0], handle->column_ids[1], FS_OP_ADD, handle->tx_bytes); - FS_operate(handle->fs2_handle, handle->line_ids[0], handle->column_ids[2], FS_OP_ADD, handle->rx_pkts); - FS_operate(handle->fs2_handle, handle->line_ids[0], handle->column_ids[3], FS_OP_ADD, handle->rx_bytes); - FS_operate(handle->fs2_handle, handle->line_ids[1], handle->column_ids[0], FS_OP_SET, handle->tx_pkts * 10); - FS_operate(handle->fs2_handle, handle->line_ids[1], handle->column_ids[1], FS_OP_SET, handle->tx_bytes * 10); - FS_operate(handle->fs2_handle, handle->line_ids[1], handle->column_ids[2], FS_OP_SET, handle->rx_pkts * 10); - FS_operate(handle->fs2_handle, handle->line_ids[1], handle->column_ids[3], FS_OP_SET, handle->rx_bytes * 10); + rx_pkts_sum = 0; + rx_bytes_sum = 0; + tx_pkts_sum = 0; + tx_bytes_sum = 0; + rx_pkts_sum_interval = 0; + rx_bytes_sum_interval = 0; + tx_pkts_sum_interval = 0; + tx_bytes_sum_interval = 0; - usleep(100); + for (i = 0; i < (int)handle->thread_sum; i++) + { + rx_pkts_sum += handle->rx_pkts[i]; + rx_bytes_sum += handle->rx_bytes[i]; + tx_pkts_sum += handle->tx_pkts[i]; + tx_bytes_sum += handle->tx_bytes[i]; + rx_pkts_sum_interval += (handle->rx_pkts[i] - handle->rx_pkts_last[i]); + rx_bytes_sum_interval += (handle->rx_bytes[i] - handle->rx_bytes_last[i]); + tx_pkts_sum_interval += (handle->tx_pkts[i] - handle->tx_pkts_last[i]); + tx_bytes_sum_interval += (handle->tx_bytes[i] - handle->tx_bytes_last[i]); + + FS_operate(handle->fs2_handle, handle->line_ids[2 * i], handle->column_ids[0], FS_OP_SET, handle->rx_pkts[i]); + FS_operate(handle->fs2_handle, handle->line_ids[2 * i], handle->column_ids[1], FS_OP_SET, handle->rx_bytes[i]); + FS_operate(handle->fs2_handle, handle->line_ids[2 * i], handle->column_ids[2], FS_OP_SET, handle->tx_pkts[i]); + FS_operate(handle->fs2_handle, handle->line_ids[2 * i], handle->column_ids[3], FS_OP_SET, handle->tx_bytes[i]); + + FS_operate(handle->fs2_handle, handle->line_ids[2 * i + 1], handle->column_ids[0], FS_OP_SET, (handle->rx_pkts[i] - handle->rx_pkts_last[i])); + FS_operate(handle->fs2_handle, handle->line_ids[2 * i + 1], handle->column_ids[1], FS_OP_SET, (handle->rx_bytes[i] - handle->rx_bytes_last[i])); + FS_operate(handle->fs2_handle, handle->line_ids[2 * i + 1], handle->column_ids[2], FS_OP_SET, (handle->tx_pkts[i] - handle->tx_pkts_last[i])); + FS_operate(handle->fs2_handle, handle->line_ids[2 * i + 1], handle->column_ids[3], FS_OP_SET, (handle->tx_bytes[i] - handle->tx_bytes_last[i])); + + handle->rx_pkts_last[i] = handle->rx_pkts[i]; + handle->rx_bytes_last[i] = handle->rx_bytes[i]; + handle->tx_pkts_last[i] = handle->tx_pkts[i]; + handle->tx_bytes_last[i] = handle->tx_bytes[i]; + } + FS_operate(handle->fs2_handle, handle->line_ids[2 * handle->thread_sum], handle->column_ids[0], FS_OP_SET, rx_pkts_sum); + FS_operate(handle->fs2_handle, handle->line_ids[2 * handle->thread_sum], handle->column_ids[1], FS_OP_SET, rx_bytes_sum); + FS_operate(handle->fs2_handle, handle->line_ids[2 * handle->thread_sum], handle->column_ids[2], FS_OP_SET, tx_pkts_sum); + FS_operate(handle->fs2_handle, handle->line_ids[2 * handle->thread_sum], handle->column_ids[3], FS_OP_SET, tx_bytes_sum); + + FS_operate(handle->fs2_handle, handle->line_ids[2 * handle->thread_sum + 1], handle->column_ids[0], FS_OP_SET, rx_pkts_sum_interval); + FS_operate(handle->fs2_handle, handle->line_ids[2 * handle->thread_sum + 1], handle->column_ids[1], FS_OP_SET, rx_bytes_sum_interval); + FS_operate(handle->fs2_handle, handle->line_ids[2 * handle->thread_sum + 1], handle->column_ids[2], FS_OP_SET, tx_pkts_sum_interval); + FS_operate(handle->fs2_handle, handle->line_ids[2 * handle->thread_sum + 1], handle->column_ids[3], FS_OP_SET, tx_bytes_sum_interval); + + sleep(1); } pthread_exit(NULL); } @@ -193,10 +242,9 @@ void hos_expand_fs2(hos_client_handle handle, const char * path, int format, cha { screen_stat_handle_t fs2_handle = NULL; const char *app_name = "hos-sdk-client-cpp"; - int *line_ids = (int *)malloc(sizeof(int) * 2); - int *column_ids = (int *)malloc(sizeof(int) * 4); int value = 0; char buff[128]; + int i = 0; fs2_handle = FS_create_handle(); @@ -212,6 +260,7 @@ void hos_expand_fs2(hos_client_handle handle, const char * path, int format, cha value = 1; FS_set_para(fs2_handle, CREATE_THREAD, &value, sizeof(value)); FS_set_para(fs2_handle, METRIS_FORMAT, &format, sizeof(format)); + FS_set_para(fs2_handle, STAT_CYCLE, &value, sizeof(value)); value = 4096; FS_set_para(fs2_handle, MAX_STAT_FIELD_NUM, &value, sizeof(value)); if (server_ip == NULL) @@ -224,25 +273,47 @@ void hos_expand_fs2(hos_client_handle handle, const char * path, int format, cha FS_set_para(fs2_handle, STATS_SERVER_PORT, &port, sizeof(port)); + value = FS_OUTPUT_STATSD; + FS_set_para(fs2_handle, STATS_FORMAT, &value, sizeof(value)); + + int *line_ids = (int *)calloc(2 * handle->thread_sum + 2, sizeof(int)); + int *column_ids = (int *)calloc(4, sizeof(int)); + //line info - snprintf(buff, sizeof(buff), "tx_pkts"); - column_ids[0] = FS_register(fs2_handle, FS_STYLE_COLUMN, FS_CALC_CURRENT, buff); - snprintf(buff, sizeof(buff), "tx_bytes"); - column_ids[1] = FS_register(fs2_handle, FS_STYLE_COLUMN, FS_CALC_CURRENT, buff); snprintf(buff, sizeof(buff), "rx_pkts"); - column_ids[2] = FS_register(fs2_handle, FS_STYLE_COLUMN, FS_CALC_CURRENT, buff); + column_ids[0] = FS_register(fs2_handle, FS_STYLE_COLUMN, FS_CALC_CURRENT, buff); snprintf(buff, sizeof(buff), "rx_bytes"); + column_ids[1] = FS_register(fs2_handle, FS_STYLE_COLUMN, FS_CALC_CURRENT, buff); + snprintf(buff, sizeof(buff), "tx_pkts"); + column_ids[2] = FS_register(fs2_handle, FS_STYLE_COLUMN, FS_CALC_CURRENT, buff); + snprintf(buff, sizeof(buff), "tx_bytes"); column_ids[3] = FS_register(fs2_handle, FS_STYLE_COLUMN, FS_CALC_CURRENT, buff); + for (i = 0; i < (int)handle->thread_sum; i++) + { + snprintf(buff, sizeof(buff), "total(%d)", i); + line_ids[2 * i] = FS_register(fs2_handle, FS_STYLE_LINE, FS_CALC_CURRENT, buff); + snprintf(buff, sizeof(buff), "rate(%d)", i); + line_ids[2 * i + 1] = FS_register(fs2_handle, FS_STYLE_LINE, FS_CALC_CURRENT, buff); + } snprintf(buff, sizeof(buff), "total"); - line_ids[0] = FS_register(fs2_handle, FS_STYLE_LINE, FS_CALC_CURRENT, buff); - snprintf(buff, sizeof(buff), "speed/s"); - line_ids[1] = FS_register(fs2_handle, FS_STYLE_LINE, FS_CALC_CURRENT, buff); + line_ids[2 * handle->thread_sum] = FS_register(fs2_handle, FS_STYLE_LINE, FS_CALC_CURRENT, buff); + snprintf(buff, sizeof(buff), "rate"); + line_ids[2 * handle->thread_sum + 1] = FS_register(fs2_handle, FS_STYLE_LINE, FS_CALC_CURRENT, buff); handle->fs2_handle = fs2_handle; handle->line_ids = line_ids; handle->column_ids = column_ids; handle->fs2_status = HOS_FS2_START; + handle->tx_pkts = (int *)calloc(handle->thread_sum, sizeof(int)); + handle->tx_bytes = (int *)calloc(handle->thread_sum, sizeof(int)); + handle->rx_pkts = (int *)calloc(handle->thread_sum, sizeof(int)); + handle->rx_bytes = (int *)calloc(handle->thread_sum, sizeof(int)); + handle->tx_pkts_last = (int *)calloc(handle->thread_sum, sizeof(int)); + handle->tx_bytes_last = (int *)calloc(handle->thread_sum, sizeof(int)); + handle->rx_pkts_last = (int *)calloc(handle->thread_sum, sizeof(int)); + handle->rx_bytes_last = (int *)calloc(handle->thread_sum, sizeof(int)); + FS_start(fs2_handle); pthread_create(&handle->fs2_thread, NULL, fs2_statistics, handle); @@ -382,7 +453,7 @@ int hos_open_fd(hos_client_handle handle, const char *bucket, const char *object return HOS_FD_NOT_ENOUGH; } - hos_info_t info = {fd, mode, handle, bucket, object, (void *)callback, userdata, NULL, handle->cache_size, handle->cache_times, 0, }; + hos_info_t info = {fd, mode, handle, bucket, object, (void *)callback, userdata, NULL, handle->cache_times, handle->cache_size, 0, }; add_hos_info(&hash_hos_info[thread_id], &info); fd_info[thread_id][fd] = 1; @@ -416,8 +487,11 @@ int hos_write(size_t fd, const char *stream, size_t stream_len, size_t thread_id handle = (hos_client_handle)hos_info->handle; //field_stat2 record - handle->rx_pkts++; - handle->rx_bytes += stream_len; + if (handle->fs2_handle) + { + handle->rx_pkts[thread_id]++; + handle->rx_bytes[thread_id] += stream_len; + } Aws::S3::S3Client& S3Client = *(handle->S3Client); @@ -507,19 +581,22 @@ int hos_write(size_t fd, const char *stream, size_t stream_len, size_t thread_id } if (ret) { - handle->tx_pkts++; - if (hos_info->mode & BUFF_MODE) + if (handle->fs2_handle) { - if (hos_info->mode & APPEND_MODE) + handle->tx_pkts[thread_id]++; + if (hos_info->mode & BUFF_MODE) { - handle->tx_bytes += handle->cache_size; + if (hos_info->mode & APPEND_MODE) + { + handle->tx_bytes[thread_id] += handle->cache_size; + }else + { + handle->tx_bytes[thread_id] += stream_len; + } }else { - handle->tx_bytes += stream_len; + handle->tx_bytes[thread_id] += buffer.st_size; } - }else - { - handle->tx_bytes += buffer.st_size; } while (flag == 1) { @@ -562,19 +639,31 @@ int hos_client_destory(hos_client_handle handle) { delete_all(&hash_hos_info[i]); } - handle->fs2_status = HOS_FS2_STOP; - pthread_join(handle->fs2_thread, NULL); if (handle->fs2_handle) { + handle->fs2_status = HOS_FS2_STOP; + pthread_join(handle->fs2_thread, NULL); FS_stop(&handle->fs2_handle); - } - if (handle->line_ids) - { - free(handle->line_ids); - } - if (handle->column_ids) - { - free(handle->column_ids); + if (handle->rx_pkts) + free(handle->rx_pkts); + if (handle->rx_bytes) + free(handle->rx_bytes); + if (handle->tx_pkts) + free(handle->tx_pkts); + if (handle->tx_bytes) + free(handle->tx_bytes); + if (handle->rx_pkts_last) + free(handle->rx_pkts_last); + if (handle->rx_bytes_last) + free(handle->rx_bytes_last); + if (handle->tx_pkts_last) + free(handle->tx_pkts_last); + if (handle->tx_bytes_last) + free(handle->tx_bytes_last); + if (handle->line_ids) + free(handle->line_ids); + if (handle->column_ids) + free(handle->column_ids); } free(handle); diff --git a/src/hos_client.h b/src/hos_client.h index a871f1cb..438b5be7 100644 --- a/src/hos_client.h +++ b/src/hos_client.h @@ -108,23 +108,23 @@ bool hos_verify_bucket(hos_client_handle handle, const char *bucket); *************************************************************************************/ int hos_create_bucket(hos_client_handle handle, const char *bucket); /************************************************************************************* - * 函数名： set_cache_size + * 函数名： hos_set_cache_size * 参数： hos_client_handle handle 非空句柄 * size_t cache_size append 模式每次追加的buffer大小 *************************************************************************************/ -void set_cache_size(hos_client_handle handle, size_t cache_size); +void hos_set_cache_size(hos_client_handle handle, size_t cache_size); /************************************************************************************* - * 函数名： set_cache_times + * 函数名： hos_set_cache_times * 参数： hos_client_handle handle 非空句柄 * size_t cache_times append 模式追加次数 *************************************************************************************/ -void set_cache_times(hos_client_handle handle, size_t cache_times); +void hos_set_cache_times(hos_client_handle handle, size_t cache_times); /************************************************************************************* - * 函数名： set_thread_sum + * 函数名： hos_set_thread_sum * 参数： hos_client_handle handle 非空句柄 * size_t thread_sum append 模式追加次数 *************************************************************************************/ -void set_thread_sum(hos_client_handle handle, size_t thread_sum); +void hos_set_thread_sum(hos_client_handle handle, size_t thread_sum); /************************************************************************************* * 函数名： hos_upload_async * 参数： hos_client_handle handle 非空句柄