gfwleak/pxz-hos-client-cpp-module
Archived
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

修改field_stat2

Browse Source
This commit is contained in:
pengxuanzheng
2020-11-02 17:55:02 +08:00
parent 6b2bcbab23
commit 9306b888cf
7 changed files with 302 additions and 165 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
example/demo/hos_write_complete.cpp
View File

@@ -70,18 +70,15 @@ int main(int argc, char *argv[])
char *bucket = argv[1]; char *bucket = argv[1];
char *object = argv[2]; char *object = argv[2];
int test_times = atoi(argv[3]); int test_times = atoi(argv[3]);
//int test_times = 10000;
char *buf = (char *)malloc(1024 * 1024 * 40); char *buf = (char *)malloc(1024 * 1024 * 40);
//char buf[1024 * 1024 * 4];
//char buf[1024 * 4];
size_t buf_size; size_t buf_size;
int mode = FILE_MODE; int mode = FILE_MODE;
size_t fd[10001] = {0}; size_t fd[10001] = {0};
userdata_t data = {&finished}; userdata_t data = {&finished};
//printf("finished:%lu\n", finished.tv_sec * 1000 * 1000 * 1000 + finished.tv_nsec);
file_to_buffer(object, buf, &buf_size); file_to_buffer(object, buf, &buf_size);
hos_init_api();
debuginfo("hos_client_init start ...\n"); 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.40.223:9098/hos/", "default", "default", 4000);
//hos_client_handle handle = hos_client_create("http://192.168.32.10:9098/hos/", "default", "default", 4); //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 #if 1
mode = FILE_MODE; mode = FILE_MODE;
#if 0
for (i = 0; i < test_times; i++)
{
}
#endif
debuginfo("hos_upload_file start ...\n"); debuginfo("hos_upload_file start ...\n");
clock_gettime(CLOCK_MONOTONIC, &start); clock_gettime(CLOCK_MONOTONIC, &start);
for (i = 0; i < test_times; i++) for (i = 0; i < test_times; i++)
@@ -165,6 +157,7 @@ int main(int argc, char *argv[])
} }
debuginfo("hos_client_close end ...\n"); debuginfo("hos_client_close end ...\n");
hos_shutdown_api();
free(buf); free(buf);
return 0; return 0;
} }

241
example/performance/HosClientPerformance.cpp
View File

@@ -13,10 +13,11 @@ extern "C"
#include<pthread.h> #include<pthread.h>
#include<dirent.h> #include<dirent.h>
#include<sys/stat.h> #include<sys/stat.h>
#include<math.h>
} }
#include"../../src/hos_client.h" #include"../../src/hos_client.h"
#define MAX_THREAD_NUM 16 #define MAX_THREAD_NUM 32
#ifndef MIN #ifndef MIN
#define MIN(a,b) ((a) > (b)) ? (b) : (a) #define MIN(a,b) ((a) > (b)) ? (b) : (a)
#endif #endif
@@ -30,7 +31,9 @@ typedef struct conf_s
char file[STRING_SIZE]; char file[STRING_SIZE];
size_t pool_size; size_t pool_size;
size_t thread_sum; size_t thread_sum;
size_t append_size; size_t size;
float append_size;
size_t slice;
int mode; int mode;
}conf_t; }conf_t;
@@ -51,6 +54,7 @@ static void configuration_init(conf_t *conf)
conf->append_size = 1024; conf->append_size = 1024;
conf->thread_sum = 1; conf->thread_sum = 1;
conf->mode = BUFF_MODE; conf->mode = BUFF_MODE;
conf->slice = 0;
return ; return ;
} }
@@ -136,14 +140,20 @@ static int upload_file(char *file, char *buff, int buff_len, int test_times, thr
size_t fd[3000]; size_t fd[3000];
conf_t *conf = &thread_info->conf; conf_t *conf = &thread_info->conf;
struct timespec tstart, tend, twrite; struct timespec tstart, tend, twrite;
size_t time_write, time_upload; long time_write = 0, time_upload = 0;
size_t len = strlen(performance_info); size_t len = strlen(performance_info);
char file_size[128]; 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++) for (i = 0; i < test_times; i++)
{ {
clock_gettime(CLOCK_MONOTONIC, &tstart);
fp = fopen(file, "w+"); fp = fopen(file, "w+");
if (fp == NULL) if (fp == NULL)
{ {
@@ -157,31 +167,44 @@ static int upload_file(char *file, char *buff, int buff_len, int test_times, thr
return -1; return -1;
} }
fclose(fp); fclose(fp);
clock_gettime(CLOCK_MONOTONIC, &twrite);
record[i] = calc_time(tstart, twrite);
time_write += record[i];
} }
clock_gettime(CLOCK_MONOTONIC, &twrite); //clock_gettime(CLOCK_MONOTONIC, &twrite);
time_write = calc_time(tstart, twrite); //time_write = calc_time(tstart, twrite);
time_write /= test_times; time_write /= test_times;
//上传文件 //上传文件
clock_gettime(CLOCK_MONOTONIC, &tstart); //clock_gettime(CLOCK_MONOTONIC, &tstart);
for (i = 0; i < test_times; i++) 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); 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) if (hos_write(fd[i], file, 0, thread_info->thread_num) != HOS_CLIENT_OK)
{ {
printf("error:hos_write file:%s\n", file); printf("error:hos_write file:%s\n", file);
return -1; return -1;
} }
clock_gettime(CLOCK_MONOTONIC, &tend);
time = calc_time(tstart, tend);
time_upload += time;
record[i] += time;
} }
clock_gettime(CLOCK_MONOTONIC, &tend); //clock_gettime(CLOCK_MONOTONIC, &tend);
time_upload = calc_time(tstart, tend); //time_upload = calc_time(tstart, tend);
time_upload /= test_times; time_upload /= test_times;
sprintf(file_size, "%dk", buff_len / 1024); average = time_write + time_upload;
sprintf(&performance_info[len], "%-20lu%-20s%-20lu%-20lu%-20lu\n", for (i = 0; i < test_times; i++)
thread_info->thread_num, file_size, time_write, time_upload, time_write + time_upload); {
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; 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]; char append_size[128];
size_t success_cnt = 0; size_t success_cnt = 0;
int ret = 0; int ret = 0;
int drop = 0;
double variance = 0.00;
double average = 0.00;
long record[1000] = {0};
#if 0 if (conf->slice)
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)
{ {
fd = hos_open_fd(thread_info->handle, conf->bucket, conf->object, callback, NULL, thread_info->thread_num, conf->mode); 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++) for (i = 0; i < test_times; i++)
{ {
j = 0;
clock_gettime(CLOCK_MONOTONIC, &tstart);
while(1) while(1)
{ {
tmp = j * conf->append_size; tmp = j * conf->slice;
rest = buff_len - tmp; rest = buff_len - tmp;
if (rest < conf->append_size) if (rest < conf->slice)
{ {
hos_write(fd, &buff[tmp], rest, thread_info->thread_num); hos_write(fd, &buff[tmp], rest, thread_info->thread_num);
break; 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++; j++;
} }
clock_gettime(CLOCK_MONOTONIC, &ttmp);
record[i] = calc_time(tstart, ttmp);
average += record[i];
} }
clock_gettime(CLOCK_MONOTONIC, &ttmp); average /= test_times;
time = calc_time(tstart, ttmp);
time /= 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++) 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); variance += pow((record[i] - average), 2);
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);
}
} }
clock_gettime(CLOCK_MONOTONIC, &ttmp); variance /= test_times;
time = calc_time(tstart, ttmp);
if (success_cnt)
time /= success_cnt;
else
time /= test_times;
sprintf(file_size, "%dk", buff_len / 1024); 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); len = strlen(performance_info);
sprintf(&performance_info[len], "%-20lu%-20s%-20d%-20d%-20lu\n", sprintf(&performance_info[len], "%-20lu%-20s%-20s%-20d%-20lf%-20lf\n",
thread_info->thread_num, file_size, 0, 0, time); 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); hos_close_fd(fd, thread_info->thread_num);
return 0; return 0;
@@ -415,12 +456,12 @@ int main(int argc, char *argv[])
#endif #endif
configuration_init(&conf); 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) switch(ch)
{ {
case 'a': case 'a':
conf.append_size = 1024 * atoi(optarg); conf.append_size = 1024 * atof(optarg);
break; break;
case 'e': case 'e':
//endpoint //endpoint
@@ -461,6 +502,12 @@ int main(int argc, char *argv[])
case 'k': case 'k':
conf.append_size = 1024 * atoi(optarg); conf.append_size = 1024 * atoi(optarg);
break; break;
case 's':
conf.size = atoi(optarg);
break;
case 'S':
conf.slice = atoi(optarg);
break;
case 'h': case 'h':
default: default:
printf("usage: HosClientPerformance \n[-e set endpoint] \n[-b set bucket] \n" printf("usage: HosClientPerformance \n[-e set endpoint] \n[-b set bucket] \n"
@@ -492,8 +539,8 @@ int main(int argc, char *argv[])
return -1; return -1;
} }
set_thread_sum(handle, conf.thread_sum); hos_set_thread_sum(handle, conf.thread_sum);
set_cache_size(handle, conf.append_size); hos_set_cache_size(handle, conf.append_size);
//创建bucket //创建bucket
if (hos_create_bucket(handle, conf.bucket)) if (hos_create_bucket(handle, conf.bucket))
@@ -529,7 +576,7 @@ int main(int argc, char *argv[])
} }
CPU_ZERO(&mask); 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) if (pthread_setaffinity_np(thread[thread_num], sizeof(mask), &mask) != 0)
{ {
printf("warning:could not set CPU affinity, continuing...\n"); printf("warning:could not set CPU affinity, continuing...\n");

2
example/performance/file/c.sh
View File

@@ -4,7 +4,7 @@
# Created Time: Tue 13 Oct 2020 06:28:57 PM CST # Created Time: Tue 13 Oct 2020 06:28:57 PM CST
######################################################################### #########################################################################
#!/bin/bash #!/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 num=0

14
example/performance/test.sh
View File

@@ -4,14 +4,22 @@
# Created Time: Tue 13 Oct 2020 02:28:31 PM CST # Created Time: Tue 13 Oct 2020 02:28:31 PM CST
######################################################################### #########################################################################
#!/bin/bash #!/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 num=0
while((${num} < 37)) while((${num} < 37))
do do
#./HosClientPerformance -a ${size[$num]} #./HosClientPerformance -a ${size[$num]}
echo "./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/1M.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++" let "num++"
done done

2
src/CMakeLists.txt
View File

@@ -9,7 +9,7 @@ link_directories(${SUPPORT_INSTALL_PREFIX})
link_directories(${CMAKE_INSTALL_PREFIX}/lib) link_directories(${CMAKE_INSTALL_PREFIX}/lib)
set(CMKAE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -shared -fPIC") set(CMKAE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -shared -fPIC")
add_library(${lib_name}_shared SHARED hos_client.cpp hos_hash.cpp) 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}) set_target_properties(${lib_name}_shared PROPERTIES OUTPUT_NAME ${lib_name})
add_dependencies(${lib_name}_shared aws-sdk-cpp-master-static) add_dependencies(${lib_name}_shared aws-sdk-cpp-master-static)

185
src/hos_client.cpp
View File

@@ -42,10 +42,14 @@ typedef struct hos_client_handle_s
#define HOS_FS2_STOP 2 #define HOS_FS2_STOP 2
int *line_ids; int *line_ids;
int *column_ids; int *column_ids;
size_t tx_pkts; int *tx_pkts;
size_t tx_bytes; int *tx_bytes;
size_t rx_pkts; int *rx_pkts;
size_t rx_bytes; int *rx_bytes;
int *tx_pkts_last;
int *tx_bytes_last;
int *rx_pkts_last;
int *rx_bytes_last;
}hos_client_handle_t; }hos_client_handle_t;
hos_info_t *hash_hos_info[MAX_HOS_CLIENT_THREAD_NUM]; 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; client->cache_size = cache_size;
return ; 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; client->cache_times = cache_times;
return ; 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; client->thread_sum = thread_sum;
return ; return ;
@@ -145,7 +149,7 @@ hos_client_handle hos_client_create(const char *endpoint, const char *accesskeyi
config.endpointOverride = endpoint; config.endpointOverride = endpoint;
config.verifySSL = false; config.verifySSL = false;
config.enableEndpointDiscovery = true; config.enableEndpointDiscovery = true;
config.executor = std::shared_ptr<Aws::Utils::Threading::PooledThreadExecutor>(std::make_shared<Aws::Utils::Threading::PooledThreadExecutor>(pool_size));//支持线程池 config.executor = std::shared_ptr<Aws::Utils::Threading::PooledThreadExecutor>(std::make_shared<Aws::Utils::Threading::PooledThreadExecutor>(pool_size, Aws::Utils::Threading::OverflowPolicy::REJECT_IMMEDIATELY));//支持线程池
handle->S3Client = new Aws::S3::S3Client(credentials, config, Aws::Client::AWSAuthV4Signer::PayloadSigningPolicy::Never, false); handle->S3Client = new Aws::S3::S3Client(credentials, config, Aws::Client::AWSAuthV4Signer::PayloadSigningPolicy::Never, false);
handle->options = options; 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) static void *fs2_statistics(void *ptr)
{ {
hos_client_handle handle = (hos_client_handle)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) while(1)
{ {
@@ -174,17 +187,53 @@ static void *fs2_statistics(void *ptr)
{ {
break; 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); rx_pkts_sum = 0;
FS_operate(handle->fs2_handle, handle->line_ids[1], handle->column_ids[1], FS_OP_SET, handle->tx_bytes * 10); rx_bytes_sum = 0;
FS_operate(handle->fs2_handle, handle->line_ids[1], handle->column_ids[2], FS_OP_SET, handle->rx_pkts * 10); tx_pkts_sum = 0;
FS_operate(handle->fs2_handle, handle->line_ids[1], handle->column_ids[3], FS_OP_SET, handle->rx_bytes * 10); 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); 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; screen_stat_handle_t fs2_handle = NULL;
const char *app_name = "hos-sdk-client-cpp"; 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; int value = 0;
char buff[128]; char buff[128];
int i = 0;
fs2_handle = FS_create_handle(); 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; value = 1;
FS_set_para(fs2_handle, CREATE_THREAD, &value, sizeof(value)); 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, METRIS_FORMAT, &format, sizeof(format));
FS_set_para(fs2_handle, STAT_CYCLE, &value, sizeof(value));
value = 4096; value = 4096;
FS_set_para(fs2_handle, MAX_STAT_FIELD_NUM, &value, sizeof(value)); FS_set_para(fs2_handle, MAX_STAT_FIELD_NUM, &value, sizeof(value));
if (server_ip == NULL) 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)); 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 //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"); 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"); 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); 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"); snprintf(buff, sizeof(buff), "total");
line_ids[0] = 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), "speed/s"); snprintf(buff, sizeof(buff), "rate");
line_ids[1] = FS_register(fs2_handle, FS_STYLE_LINE, FS_CALC_CURRENT, buff); line_ids[2 * handle->thread_sum + 1] = FS_register(fs2_handle, FS_STYLE_LINE, FS_CALC_CURRENT, buff);
handle->fs2_handle = fs2_handle; handle->fs2_handle = fs2_handle;
handle->line_ids = line_ids; handle->line_ids = line_ids;
handle->column_ids = column_ids; handle->column_ids = column_ids;
handle->fs2_status = HOS_FS2_START; 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); FS_start(fs2_handle);
pthread_create(&handle->fs2_thread, NULL, fs2_statistics, 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; 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); add_hos_info(&hash_hos_info[thread_id], &info);
fd_info[thread_id][fd] = 1; 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; handle = (hos_client_handle)hos_info->handle;
//field_stat2 record //field_stat2 record
handle->rx_pkts++; if (handle->fs2_handle)
handle->rx_bytes += stream_len; {
handle->rx_pkts[thread_id]++;
handle->rx_bytes[thread_id] += stream_len;
}
Aws::S3::S3Client& S3Client = *(handle->S3Client); 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) if (ret)
{ {
handle->tx_pkts++; if (handle->fs2_handle)
if (hos_info->mode & BUFF_MODE)
{ {
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 }else
{ {
handle->tx_bytes += stream_len; handle->tx_bytes[thread_id] += buffer.st_size;
} }
}else
{
handle->tx_bytes += buffer.st_size;
} }
while (flag == 1) while (flag == 1)
{ {
@@ -562,19 +639,31 @@ int hos_client_destory(hos_client_handle handle)
{ {
delete_all(&hash_hos_info[i]); delete_all(&hash_hos_info[i]);
} }
handle->fs2_status = HOS_FS2_STOP;
pthread_join(handle->fs2_thread, NULL);
if (handle->fs2_handle) if (handle->fs2_handle)
{ {
handle->fs2_status = HOS_FS2_STOP;
pthread_join(handle->fs2_thread, NULL);
FS_stop(&handle->fs2_handle); FS_stop(&handle->fs2_handle);
} if (handle->rx_pkts)
if (handle->line_ids) free(handle->rx_pkts);
{ if (handle->rx_bytes)
free(handle->line_ids); free(handle->rx_bytes);
} if (handle->tx_pkts)
if (handle->column_ids) free(handle->tx_pkts);
{ if (handle->tx_bytes)
free(handle->column_ids); 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); free(handle);

12
src/hos_client.h
View File

@@ -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); int hos_create_bucket(hos_client_handle handle, const char *bucket);
/************************************************************************************* /*************************************************************************************
* 函数名： set_cache_size * 函数名： hos_set_cache_size
* 参数： hos_client_handle handle 非空句柄 * 参数： hos_client_handle handle 非空句柄
* size_t cache_size append 模式每次追加的buffer大小 * 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 非空句柄 * 参数： hos_client_handle handle 非空句柄
* size_t cache_times append 模式追加次数 * 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 非空句柄 * 参数： hos_client_handle handle 非空句柄
* size_t thread_sum append 模式追加次数 * 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_upload_async
* 参数： hos_client_handle handle 非空句柄 * 参数： hos_client_handle handle 非空句柄