diff options
| -rw-r--r-- | CRDT/ftbt_gtest.cpp | 224 | ||||
| -rw-r--r-- | CRDT/util/fair_index.cpp | 64 | ||||
| -rw-r--r-- | CRDT/util/fair_index.h | 12 | ||||
| -rw-r--r-- | CRDT/util/flow_tokens.cpp | 59 | ||||
| -rw-r--r-- | CRDT/util/flow_tokens.h | 5 |
5 files changed, 273 insertions, 91 deletions
diff --git a/CRDT/ftbt_gtest.cpp b/CRDT/ftbt_gtest.cpp index c7713e9..6e01e3a 100644 --- a/CRDT/ftbt_gtest.cpp +++ b/CRDT/ftbt_gtest.cpp @@ -3,6 +3,8 @@ #include "ransampl.h" #include <cstddef> #include <cstdio> +#include <cstdlib> +#include <cstring> #include <gtest/gtest.h> #include <unistd.h> #include <uuid/uuid.h> @@ -10,6 +12,11 @@ #include "util/flow_tokens.h" #include "util/fair_index.h" +#define FAIR_INDEX_CIR 1*1000*1000 //1000 bit/ms, 1M/s +#define FAIR_INDEX_CBS 10*1000 +#define FAIR_INDEX_REQ FAIR_INDEX_CIR/1000 +#define FAIR_INDEX_MIMIC_DURATION_US 100*1000*1000 + void ftb_throttler_sync(struct ftb_throttler *list[], size_t n) { char *blob=NULL; @@ -148,11 +155,6 @@ TEST(FTBT, Merge) ftb_throttler_free(bucket[1]); } -#define FAIR_INDEX_CIR 1*1024*1000 //1024 bit/ms -#define FAIR_INDEX_CBS 10*1024 -#define FAIR_INDEX_REQ FAIR_INDEX_CIR/1000 -#define FAIR_INDEX_MIMIC_DURATION_US 1*1000*1000 - TEST(FTBT, SingleReplica) { int class_num = 24; @@ -172,19 +174,16 @@ TEST(FTBT, SingleReplica) step.tv_sec=0; step.tv_usec=(suseconds_t)1000; - struct test_records ftbf_class_results[class_num], ftbf_result; - unsigned char class_weight[class_num]; double prob[class_num]; - memset(&ftbf_result, 0, sizeof(ftbf_result)); - memset(ftbf_class_results, 0, sizeof(ftbf_class_results)); - memset(class_weight, 0, sizeof(class_weight)); + struct fstb_class pareto_classes[class_num]; memset(prob, 0, sizeof(prob)); - + memset(pareto_classes, 0, sizeof(pareto_classes)); for(i = 0; i < (size_t)class_num; i++) { - class_weight[i] = 1; + pareto_classes[i].class_id = i+1; + pareto_classes[i].weight = 1; } - set_flow_prob(prob, class_num, ftype); + long double sum_w = set_flow_prob(prob, class_num, ftype); ransampl_ws *ws = ransampl_init(prob, class_num); int round = FAIR_INDEX_MIMIC_DURATION_US/1000; srandom(17); @@ -194,19 +193,13 @@ TEST(FTBT, SingleReplica) for(k = 0; k < (size_t)class_num; ++k) { j = ransampl_draw(ws); - tokens = 0.2*FAIR_INDEX_REQ+rand()%5; + tokens = (long long)(sum_w*FAIR_INDEX_REQ/class_num); requested += tokens; - int rtn = ftb_throttler_control(bucket, now, FTBT_CMD_CONSUME_FLEXIBLE, tokens, j, class_weight[j]); - ftbf_result.cnt++; - ftbf_class_results[j].cnt++; + pareto_classes[j].demand_tokens += tokens; + int rtn = ftb_throttler_control(bucket, now, FTBT_CMD_CONSUME_NORMAL, tokens, j, pareto_classes[j].weight); if(rtn > 0) - { - ftbf_result.succ_cnt++; - ftbf_class_results[j].succ_cnt++; - }else - { - ftbf_result.fail_cnt++; - ftbf_class_results[j].fail_cnt++; + { + pareto_classes[j].allocated_tokens += tokens; } } } @@ -218,36 +211,45 @@ TEST(FTBT, SingleReplica) EXPECT_LE(consumed, requested); double accuracy=(double)consumed/MIN(refilled, requested); - float jain_fair_index = calculate_jain_fair_index(&ftbf_result, ftbf_class_results, class_weight, class_num); - printf("accuracy:%f, upper_limit:%lld, refilled:%lld, requested:%lld, consumed:%lld, jain_fair_index:%f\n", + float fair_index = max_min_fairness_index(refilled, pareto_classes, class_num); + int print=1; + if(print) + { + printf("class\tweight\tdemand\tallocated\tideal\r\n"); + long duration_s = FAIR_INDEX_MIMIC_DURATION_US/1000/1000; + for(size_t i=0; i<(size_t)class_num; i++) + { + printf("%lld\t%lld\t%lld\t%lld\t%lld\r\n", pareto_classes[i].class_id, + pareto_classes[i].weight, + pareto_classes[i].demand_tokens/duration_s, + pareto_classes[i].allocated_tokens/duration_s, + pareto_classes[i].ideal_tokens/duration_s); + } + printf("CIR %d fairness index %f\r\n", FAIR_INDEX_CIR, fair_index); + } + printf("accuracy:%f, upper_limit:%lld, refilled:%lld, requested:%lld, consumed:%lld\n", accuracy, upper_limit, refilled, requested, - consumed, - jain_fair_index); + consumed); ransampl_free(ws); ftb_throttler_free(bucket); - EXPECT_NEAR(jain_fair_index, 1, 0.01); + EXPECT_NEAR(fair_index, 1, 0.1); } -TEST(FTBT, MultiReplica) +TEST(FTBT, SingleReplicaOneHeavy) { int class_num = 24; - int replica_num = 3; - enum flow_type ftype = PARETO; - size_t i,j,k; + enum flow_type ftype = ONEHEAVY; uuid_t uuid; + uuid_generate(uuid); struct timeval start; gettimeofday(&start, NULL); struct ftb_throttler_info *info = ALLOC(struct ftb_throttler_info, 1); - struct ftb_throttler *buckets[replica_num]; - for(i = 0; i < (size_t)replica_num; ++i) - { - uuid_generate(uuid); - buckets[i]=ftb_throttler_new(uuid, start, FAIR_INDEX_CIR, FAIR_INDEX_CBS, FAIR_INDEX_CIR/10); - } - + struct ftb_throttler *bucket=NULL; + bucket=ftb_throttler_new(uuid, start, FAIR_INDEX_CIR, FAIR_INDEX_CBS, FAIR_INDEX_CIR/10); + size_t i,j,k; long long tokens=0; long long consumed=0, requested=0, upper_limit=0, refilled=0; struct timeval step, now; @@ -255,19 +257,16 @@ TEST(FTBT, MultiReplica) step.tv_sec=0; step.tv_usec=(suseconds_t)1000; - struct test_records ftbf_class_results[class_num], ftbf_result; - unsigned char class_weight[class_num]; double prob[class_num]; - memset(&ftbf_result, 0, sizeof(ftbf_result)); - memset(ftbf_class_results, 0, sizeof(ftbf_class_results)); - memset(class_weight, 0, sizeof(class_weight)); + struct fstb_class one_heavy_classes[class_num]; memset(prob, 0, sizeof(prob)); - + memset(one_heavy_classes, 0, sizeof(one_heavy_classes)); for(i = 0; i < (size_t)class_num; i++) { - class_weight[i] = 1; + one_heavy_classes[i].class_id = i+1; + one_heavy_classes[i].weight = 1; } - set_flow_prob(prob, class_num, ftype); + long double sum_w = set_flow_prob(prob, class_num, ftype); ransampl_ws *ws = ransampl_init(prob, class_num); int round = FAIR_INDEX_MIMIC_DURATION_US/1000; srandom(17); @@ -277,52 +276,141 @@ TEST(FTBT, MultiReplica) for(k = 0; k < (size_t)class_num; ++k) { j = ransampl_draw(ws); - tokens = 0.2*FAIR_INDEX_REQ+rand()%5; + tokens = (long long)(sum_w*FAIR_INDEX_REQ/class_num); requested += tokens; - int rtn = ftb_throttler_control(buckets[i%replica_num], now, FTBT_CMD_CONSUME_FLEXIBLE, tokens, j, class_weight[j]); - ftbf_result.cnt++; - ftbf_class_results[j].cnt++; + one_heavy_classes[j].demand_tokens += tokens; + int rtn = ftb_throttler_control(bucket, now, FTBT_CMD_CONSUME_NORMAL, tokens, j, one_heavy_classes[j].weight); if(rtn > 0) { - ftbf_result.succ_cnt++; - ftbf_class_results[j].succ_cnt++; - }else - { - ftbf_result.fail_cnt++; - ftbf_class_results[j].fail_cnt++; + one_heavy_classes[j].allocated_tokens += tokens; } - ftb_throttler_sync(buckets, replica_num); } } upper_limit=FAIR_INDEX_CBS+FAIR_INDEX_CIR/1000*timeval_delta_ms(start, now); - ftb_throttler_info(buckets[0], info); + ftb_throttler_info(bucket, info); refilled=info->refilled; consumed=info->consumed; EXPECT_LE(consumed, requested); double accuracy=(double)consumed/MIN(refilled, requested); - float jain_fair_index = calculate_jain_fair_index(&ftbf_result, ftbf_class_results, class_weight, class_num); - printf("accuracy:%f, upper_limit:%lld, refilled:%lld, requested:%lld, consumed:%lld, jain_fair_index:%f\n", + float fair_index = max_min_fairness_index(refilled, one_heavy_classes, class_num); + int print=1; + if(print) + { + printf("class\tweight\tdemand\tallocated\tideal\r\n"); + long duration_s = FAIR_INDEX_MIMIC_DURATION_US/1000/1000; + for(size_t i=0; i<(size_t)class_num; i++) + { + printf("%lld\t%lld\t%lld\t%lld\t%lld\r\n", one_heavy_classes[i].class_id, + one_heavy_classes[i].weight, + one_heavy_classes[i].demand_tokens/duration_s, + one_heavy_classes[i].allocated_tokens/duration_s, + one_heavy_classes[i].ideal_tokens/duration_s); + } + printf("CIR %d fairness index %f\r\n", FAIR_INDEX_CIR, fair_index); + } + printf("accuracy:%f, upper_limit:%lld, refilled:%lld, requested:%lld, consumed:%lld\n", accuracy, upper_limit, refilled, requested, - consumed, - jain_fair_index); + consumed); ransampl_free(ws); - for(i = 0; i < (size_t)replica_num; ++i) + ftb_throttler_free(bucket); + EXPECT_NEAR(fair_index, 1, 0.1); +} + +TEST(FTBT, SingleReplicaShake) +{ + int class_num = 5; + enum flow_type ftype = UNIFORM; + uuid_t uuid; + uuid_generate(uuid); + struct timeval start; + gettimeofday(&start, NULL); + struct ftb_throttler_info *info = ALLOC(struct ftb_throttler_info, 1); + struct ftb_throttler *bucket=NULL; + bucket=ftb_throttler_new(uuid, start, FAIR_INDEX_CIR, FAIR_INDEX_CBS, FAIR_INDEX_CIR/10); + size_t i,j,k; + long long tokens=0; + long long consumed=0, requested=0, upper_limit=0, refilled=0; + struct timeval step, now; + memcpy(&now, &start, sizeof(now)); + step.tv_sec=0; + step.tv_usec=(suseconds_t)1000; + + double prob[class_num]; + struct fstb_class shake_classes[class_num]; + memset(prob, 0, sizeof(prob)); + memset(shake_classes, 0, sizeof(shake_classes)); + for(i = 0; i < (size_t)class_num; i++) { - ftb_throttler_free(buckets[i]); + shake_classes[i].class_id = i+1; + shake_classes[i].weight = 1; + } + long double sum_w = set_flow_prob(prob, class_num, ftype); + ransampl_ws *ws = ransampl_init(prob, class_num); + int round = FAIR_INDEX_MIMIC_DURATION_US/1000; + srandom(17); + for(i=0; i<(size_t)round; i++) + { + timeradd(&now, &step, &now); + for(k = 0; k < (size_t)class_num; ++k) + { + j = ransampl_draw(ws); + if((size_t)(random()%class_num) > j) + continue; + tokens = (long long)(sum_w*FAIR_INDEX_REQ/class_num); + requested += tokens; + shake_classes[j].demand_tokens += tokens; + int rtn = ftb_throttler_control(bucket, now, FTBT_CMD_CONSUME_NORMAL, tokens, j, shake_classes[j].weight); + if(rtn > 0) + { + shake_classes[j].allocated_tokens += tokens; + } + } } - EXPECT_NEAR(jain_fair_index, 1, 0.01); -} + upper_limit=FAIR_INDEX_CBS+FAIR_INDEX_CIR/1000*timeval_delta_ms(start, now); + ftb_throttler_info(bucket, info); + refilled=info->refilled; + consumed=info->consumed; + EXPECT_LE(consumed, requested); + double accuracy=(double)consumed/MIN(refilled, requested); + + float fair_index = max_min_fairness_index(refilled, shake_classes, class_num); + int print=1; + if(print) + { + printf("class\tweight\tdemand\tallocated\tideal\r\n"); + long duration_s = FAIR_INDEX_MIMIC_DURATION_US/1000/1000; + for(size_t i=0; i<(size_t)class_num; i++) + { + printf("%lld\t%lld\t%lld\t%lld\t%lld\r\n", shake_classes[i].class_id, + shake_classes[i].weight, + shake_classes[i].demand_tokens/duration_s, + shake_classes[i].allocated_tokens/duration_s, + shake_classes[i].ideal_tokens/duration_s); + } + printf("CIR %d fairness index %f\r\n", FAIR_INDEX_CIR, fair_index); + } + printf("accuracy:%f, upper_limit:%lld, refilled:%lld, requested:%lld, consumed:%lld\n", + accuracy, + upper_limit, + refilled, + requested, + consumed); + ransampl_free(ws); + ftb_throttler_free(bucket); + EXPECT_NEAR(fair_index, 1, 0.1); +} int main(int argc, char ** argv) { int ret=0; ::testing::InitGoogleTest(&argc, argv); + // ::testing::GTEST_FLAG(filter) = "FTBT.SingleReplicaShake"; ret=RUN_ALL_TESTS(); return ret; } diff --git a/CRDT/util/fair_index.cpp b/CRDT/util/fair_index.cpp index 3724a88..d936d3d 100644 --- a/CRDT/util/fair_index.cpp +++ b/CRDT/util/fair_index.cpp @@ -28,3 +28,67 @@ float calculate_jain_fair_index(struct test_records *result, struct test_records printf("total:%d, succ:%d, fail:%d\n", result->cnt, result->succ_cnt, result->fail_cnt); return (float)(pow(succ_sum,2))/(float)(class_num*succ_square_sum); } + +int cmp_fstb_class(const void *a, const void *b) +{ + struct fstb_class *ra=(struct fstb_class*)a; + struct fstb_class *rb=(struct fstb_class*)b; + return (int)(ra->demand_tokens-rb->demand_tokens); +} + +float max_min_fairness_index(long long available_tokens, struct fstb_class * classes, size_t n_class) +{ + qsort(classes, n_class, sizeof(struct fstb_class), cmp_fstb_class); + long long total_weight=0; + for(size_t i=0; i<n_class; i++) + { + total_weight+=classes[i].weight; + } + long long left_tokens=available_tokens; + long long left_weight=total_weight; + size_t n_satisfied=0; + + while(n_satisfied<n_class && left_tokens/left_weight>0 ) + { + long long share=left_tokens/left_weight; + for(size_t i=0; i<n_class; i++) + { + long long my_share=classes[i].weight*share; + if(classes[i].demand_tokens == classes[i].ideal_tokens) + { + continue; + } + else if(classes[i].demand_tokens - classes[i].ideal_tokens <= my_share) + { + left_tokens -= (classes[i].demand_tokens - classes[i].ideal_tokens); + classes[i].ideal_tokens=classes[i].demand_tokens; + } + else + { + left_tokens -= my_share; + classes[i].ideal_tokens+=my_share; + + } + } + left_weight=0; + n_satisfied=0; + for(size_t i=0; i<n_class; i++) + { + if(classes[i].demand_tokens == classes[i].ideal_tokens) + { + n_satisfied++; + continue; + } + left_weight += classes[i].weight; + + } + } + double index=0; + for(size_t i=0; i<n_class; i++) + { + index += pow((double)(classes[i].ideal_tokens-classes[i].allocated_tokens)/classes[i].ideal_tokens, 2); + } + index=1-sqrt(index/n_class); + + return index; +}
\ No newline at end of file diff --git a/CRDT/util/fair_index.h b/CRDT/util/fair_index.h index cf3d9aa..e64103a 100644 --- a/CRDT/util/fair_index.h +++ b/CRDT/util/fair_index.h @@ -7,4 +7,14 @@ struct test_records int fail_cnt; }; -float calculate_jain_fair_index(struct test_records *result, struct test_records *class_result, unsigned char *class_weight, int class_num);
\ No newline at end of file +struct fstb_class +{ + long long class_id; + long long weight; + long long requested_CIR; + long long demand_tokens; + long long allocated_tokens; + long long ideal_tokens; +}; +float calculate_jain_fair_index(struct test_records *result, struct test_records *class_result, unsigned char *class_weight, int class_num); +float max_min_fairness_index(long long available_tokens, struct fstb_class * classes, size_t n_class);
\ No newline at end of file diff --git a/CRDT/util/flow_tokens.cpp b/CRDT/util/flow_tokens.cpp index a1057e2..fae761d 100644 --- a/CRDT/util/flow_tokens.cpp +++ b/CRDT/util/flow_tokens.cpp @@ -1,6 +1,7 @@ #include "flow_tokens.h" #include "math.h" #include "../crdt_utils.h" +#include <cstdio> long long get_request_tokens(long long class_num, long long CIR, size_t index, long long step_us, enum traffic_type type) { @@ -8,7 +9,7 @@ long long get_request_tokens(long long class_num, long long CIR, size_t index, l double average_ratio2, average_ratio8, total_ratio2, total_ratio8; long long standard = CIR*step_us/1000; // long long sdf = floor((long double)standard * 0.05); //floating 5% - size_t occupy8_replica_num = floor(class_num * 0.8); + size_t occupy8_class_num = floor(class_num * 0.8); // long long rand_sdf = random()%sdf; long long rand_sdf = 0; size_t scope_flag = FALSE; @@ -27,13 +28,13 @@ long long get_request_tokens(long long class_num, long long CIR, size_t index, l //average 0.67*CIR per node total_ratio2 = (0.67*class_num) / 5; total_ratio8 = (0.67*class_num) - total_ratio2; - average_ratio2 = total_ratio2 / occupy8_replica_num; - average_ratio8 = total_ratio8 / (class_num-occupy8_replica_num); - if (index < occupy8_replica_num && scope_flag) { + average_ratio2 = total_ratio2 / occupy8_class_num; + average_ratio8 = total_ratio8 / (class_num-occupy8_class_num); + if (index < occupy8_class_num && scope_flag) { request_size = (long long)floor((long double)standard * average_ratio2) + rand_sdf; - } else if (index < occupy8_replica_num && !scope_flag) { + } else if (index < occupy8_class_num && !scope_flag) { request_size = (long long)floor((long double)standard * average_ratio2) - rand_sdf; - } else if (index >= occupy8_replica_num && scope_flag) { + } else if (index >= occupy8_class_num && scope_flag) { request_size = (long long)floor((long double)standard * average_ratio8) + rand_sdf; } else { request_size = (long long)floor((long double)standard * average_ratio8) - rand_sdf; @@ -43,13 +44,13 @@ long long get_request_tokens(long long class_num, long long CIR, size_t index, l //average 2*CIR per node total_ratio2 = (2.0*class_num) / 5; total_ratio8 = (2.0*class_num) - total_ratio2; - average_ratio2 = total_ratio2 / occupy8_replica_num; - average_ratio8 = total_ratio8 / (class_num-occupy8_replica_num); - if (index < occupy8_replica_num && scope_flag) { + average_ratio2 = total_ratio2 / occupy8_class_num; + average_ratio8 = total_ratio8 / (class_num-occupy8_class_num); + if (index < occupy8_class_num && scope_flag) { request_size = (long long)floor((long double)standard * average_ratio2) + rand_sdf; - } else if (index < occupy8_replica_num && !scope_flag) { + } else if (index < occupy8_class_num && !scope_flag) { request_size = (long long)floor((long double)standard * average_ratio2) - rand_sdf; - } else if (index >= occupy8_replica_num && scope_flag) { + } else if (index >= occupy8_class_num && scope_flag) { request_size = (long long)floor((long double)standard * average_ratio8) + rand_sdf; } else { request_size = (long long)floor((long double)standard * average_ratio8) - rand_sdf; @@ -69,24 +70,42 @@ long long get_request_tokens(long long class_num, long long CIR, size_t index, l return request_size; } -void set_flow_prob(double *prob, int replica_num, enum flow_type ftype) +double set_flow_prob(double *prob, int class_num, enum flow_type ftype) { int i; + double sum_w = 0.0, sum_p = 0.0; + double probabilities[class_num]; + double a = 0.9; //Pareto形参 if(ftype == UNIFORM) { - for(i = 0; i < replica_num; i++) + for(i = 0; i < class_num; i++) { - prob[i] = 1; + prob[i] = 2.0/class_num; + sum_w += prob[i]; } - }else{ - int eight_replica_num = floor(replica_num * 0.8); - for(i = 0; i < eight_replica_num; ++i) + }else if(ftype == PARETO){ + for (i = 1; i <= class_num; ++i) { - prob[i] = ceil(1000*2.0/eight_replica_num); + probabilities[i - 1] = (a * pow(1, a)) / pow(i, a + 1); + sum_p += probabilities[i - 1]; + } + + for (i = 0; i < class_num; ++i) { + probabilities[i] /= sum_p; + prob[class_num-i-1] = probabilities[i] * 2.0; + // printf("[%d]=%lf\n ", class_num-i-1, prob[class_num-i-1]); + sum_w += prob[class_num-i-1]; } - for(; i < replica_num; ++i) + }else if(ftype == ONEHEAVY){ + for(i = 0; i < class_num-1; i++) { - prob[i] = floor(1000*8.0/(replica_num-eight_replica_num)); + prob[i] = 1.0/class_num; + sum_w += prob[i]; } + prob[i] = 0.5; + sum_w += prob[i]; + }else{ + sum_w = 0; } + return sum_w; }
\ No newline at end of file diff --git a/CRDT/util/flow_tokens.h b/CRDT/util/flow_tokens.h index 08a0e39..d08fd5d 100644 --- a/CRDT/util/flow_tokens.h +++ b/CRDT/util/flow_tokens.h @@ -12,9 +12,10 @@ enum traffic_type enum flow_type { UNIFORM, - PARETO + PARETO, + ONEHEAVY, }; long long get_request_tokens(long long class_num, long long CIR, size_t index, long long step_us, enum traffic_type type); -void set_flow_prob(double *prob, int replica_num, enum flow_type ftype); +double set_flow_prob(double *prob, int class_num, enum flow_type ftype); |