-/* ------------------------------------------------
- Copyright 2014 AT&T Intellectual Property
- Licensed under the Apache License, Version 2.0 (the "License");
- you may not use this file except in compliance with the License.
- You may obtain a copy of the License at
-
- http://www.apache.org/licenses/LICENSE-2.0
-
- Unless required by applicable law or agreed to in writing, software
- distributed under the License is distributed on an "AS IS" BASIS,
- WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- See the License for the specific language governing permissions and
- limitations under the License.
- ------------------------------------------- */
-///////////////////////////////////////////////
-//// State, functions for adaptive smart sampling
-/// written by Irina Rozenbaum
-
-#include <iostream>
-#define max(x,y) ((x)>(y)?(x):(y))
-#define _BURST_SOOTHING_FACTOR 10.0
-
-struct SSstate {
- //unsigned long long int count; // count to sample small packets with certain probability
- double count;
- double gcount; // count for clean_with() function
- double fcount; // count for final_clean() function
- double z; //z is the threshold for a size of the packet
- double z_prev; // value of threshold from the previos cleaning phase
- double gamma; //tolerance parameter for emergency control over the
- //number of samples, should be >= 1
- int do_clean; // trigger cleaning phase
- int bcount; //count for number of packets that exceed threshold ,
- //need it for one of the variations of algorithms for
- //threshold adjustment
- int s_size; //need to remember sample size for _sfun_state_init()
- int final_z; //bool that indicated if z was adjusted to its filnal value
- //before the final clean
- int time; //timestamp from previously processed packet
- //used in detecting frequency of the clening phases with aggregation
-
- int packet_count; //count for number of packets per time window
- int do_clean_count; //count number of cleaning phases per time window
- bool do_sooth; //turn on _BURST_SOOTHING_FACTOR
-
- // flow sampling
- int count_closed; //bool indicates that counting phase is being triggered
- int count_closed_flows; //count for closed flows
- int count_notsampled_new; //count for the number of new groups that were not sampled during
- // the final sampling phase will be used in next time window to compute resonable
- // approximation for threshold before final sampling phase
-
- // for reporting
- int delay; //interval within which there was no packets that belong to the current flow
- //thus flow is considered to be closed.
- //parameter specified by the query
- int closing_now; //count for groups that are closing at the final cleraning phase
- //or were closed during the most recent counting phase (newly closed)
-
- //for debugging
- int how_many_cleaned;
-};
-
-
-// the function is called once at the initial initialization of the state
-void _sfun_state_clean_init_smart_sampling_state(void *s){
- struct SSstate *state = (struct SSstate *)s;
-
- state->count = 0;
- state->gcount = 0;
- state->fcount = 0;
- state->z = 200; //need to figure out good initial value for z
- // cout << "clean init Z: " << state->z << "\n";
- state->z_prev = 0;
- // cout << "clean init Z prev: " << state->z << "\n";
- state->gamma = 2;
- state->do_clean = 0;
- state->bcount = 0;
- state->s_size = 0;
- state->final_z = 0;
- state->time = 0;
- state->count_closed = 0;
- state->count_closed_flows = 0;
- state->count_notsampled_new = 0;
-
- state->packet_count = 0;
- state->do_clean_count = 0;
- state->do_sooth = true;
-
- state->how_many_cleaned = 0;
- state->delay = 0;
- state->closing_now = 0;
-};
-
-
-// the function will be called at the beginning of every time window
-void _sfun_state_dirty_init_smart_sampling_state(void *s_new, void *s_old, int curr_num_samples){
- struct SSstate *state_new = (struct SSstate *)s_new;
- struct SSstate *state_old = (struct SSstate *)s_old;
-
- // cout << "**flows were sampled: " << state_old->how_many_cleaned << "\n";
- // cout << "***dirty current num of samples: " << curr_num_samples << "\n";
- // cout << "dirty init Z old: " << state_old->z << "\n";
- if(curr_num_samples < state_old->s_size){
- // state_new->z = state_old->z*((max((double)curr_num_samples-(double)state_old->bcount,1))/((double)state_old->s_size-(double)state_old->bcount));
-
- //temp fix for early aggregation
- state_new->z = (state_old->z)/2;
-
- if(state_old->do_sooth)
- state_new->z = state_old->z / _BURST_SOOTHING_FACTOR;
- }
- else {
- if(curr_num_samples >= state_old->s_size){
- //cout << "yes\n";
- state_new->z = state_old->z*((double)curr_num_samples/(double)state_old->s_size);
-
- if(state_old->do_sooth)
- state_new->z = state_old->z / _BURST_SOOTHING_FACTOR;
- }
- }
-
- if(state_new->z <= 1.0)
- state_new->z = 1;
- // cout << "dirty init Z new: " << state_new->z << "\n";
-
- state_new->gamma = state_old->gamma;
- state_new->do_clean = state_old->do_clean;
- state_new->s_size = state_old->s_size;
- state_new->bcount = 0;
- state_new->gcount = 0;
- state_new->count = 0;
- state_new->fcount = 0;
- state_new->final_z = 0;
- state_new->time = 0;
- state_new->count_closed = 0;
- state_new->count_closed_flows = 0;
- state_new->count_notsampled_new = state_old->count_notsampled_new;
-
- state_new->packet_count = 0;
- state_new->do_clean_count = 0;
- state_new->do_sooth = true;
-
- state_new->how_many_cleaned = 0;
- state_new->delay = 0;
- state_new->closing_now = 0;
-
- // cout << "dirty init gamma: " << state_new->gamma << "\n";
- // cout << "dirty init do_clean: " << state_new->do_clean << "\n";
- // cout << "dirty init s_size: " << state_new->s_size << "\n";
- // cout << "dirty init bcount: " << state_new->bcount << "\n";
- // cout << "dirty init gcount: " << state_new->gcount << "\n";
- // cout << "dirty init count: " << state_new->count << "\n";
- // cout << "dirty init fcount: " << state_new->fcount << "\n";
- // cout << "dirty init final_z: " << state_new->final_z << "\n";
- // cout << "dirty init time " << state_new->time << "\n";
- // cout << "dirty init count_closed: " << state_new->count_closed << "\n";
- // cout << "dirty init count_closed_flows: " << state_new->count_closed_flows << "\n";
-
- // cout << "dirty init packet count: " << state_new->packet_count << "\n";
- // cout << "dirty init do_clean_count: " << state_new->do_clean_count << "\n";
-
-};
-
-// function is called in two cases:
-// adjustment of threshold before the cleaning phase is being triggered
-// adjustment of threshold at the window border for the old state, this new
-// threshold will be used by partial_flush() and flush() functions (final_clean())
-
-void _sfun_state_final_init_smart_sampling_state(void *s, int curr_num_samples){
- struct SSstate *state = (struct SSstate *)s;
-
- // cout << "final init Z old: " << state->z << "\n";
-
- if(state->final_z == 0){
-
- // just returned from the counting phase
- // case when the time window is changed right after the counting phase
- if(state->count_closed == 1){
- state->count_closed = 0;
- state->count_closed_flows = 0;
- state->how_many_cleaned = 0;
- }
-
- state->z_prev = state->z;
- // cout << "final current num of samples: " << curr_num_samples << "\n";
- // cout << "final count not sampled new: " << state->count_notsampled_new << "\n";
- // adjust count for current number of sampled based on statistics
- // gathered in the previous time window
- // otherwise we overestimate it
- if(state->count_notsampled_new != 0)
- curr_num_samples -= state->count_notsampled_new;
-
- if(curr_num_samples < state->s_size){
- //state->z = state->z*((max((double)curr_num_samples-(double)state->bcount,1))/((double)state->s_size-(double)state->bcount));
-
- //temp fix for early aggregation
- state->z = state->z/2;
- }
- else {
- if(curr_num_samples >= state->s_size){
- state->z = state->z*((double)curr_num_samples/(double)state->s_size);
- }
- }
-
- if(state->z <= 0)
- state->z = 1;
- // cout << "final init Z new: " << state->z << "\n";
-
- state->bcount = 0;
- state->final_z = 1;
- state->do_clean_count++;
- state->count_notsampled_new = 0;
-
- }
-
- // cout << "final init gamma: " << state->gamma << "\n";
- // cout << "final init do_clean: " << state->do_clean << "\n";
- // cout << "final init s_size: " << state->s_size << "\n";
- // cout << "final init bcount: " << state->bcount << "\n";
- // cout << "final init gcount: " << state->gcount << "\n";
- // cout << "final init count: " << state->count << "\n";
- // cout << "final init fcount: " << state->fcount << "\n";
- // cout << "final init final_z: " << state->final_z << "\n";
- // cout << "final init time " << state->time << "\n";
- // cout << "final init count_closed: " << state->count_closed << "\n";
- // cout << "final init count_closed_flows: " << state->count_closed_flows << "\n";
-
- // cout << "final init packet count: " << state->packet_count << "\n";
- // cout << "final init do_clean_count: " << state->do_clean_count << "\n";
-
-};
-
-int ssample(void *s,int curr_num_samples, unsigned long long int len, unsigned int sample_size){
- struct SSstate *state = (struct SSstate *)s;
-
- state->packet_count++;
-
- state->s_size = sample_size;
- int sampled = 0;
-
- //just returned from the cleaning phase
- if(state->do_clean == 1){
- state->gcount = 0;
- state->do_clean = 0;
- }
-
- //sampling
- if(len > state->z){
- state->bcount++;
- sampled=1;
- }
- else{
- state->count += len;
- if(state->count >= state->z){
- sampled=1;
- state->count -= state->z;
- }
- }
- return sampled;
-
-};
-
-int flow_ssample(void *s,int curr_num_samples, unsigned int sample_size){
- struct SSstate *state = (struct SSstate *)s;
-
- state->packet_count++;
-
- state->s_size = sample_size;
- int sampled = 0;
-
- //just returned from the counting phase
- if(state->count_closed == 1)
- // cout << "closed flows after counting phase: " << state->count_closed_flows << "\n";
-
- //just returned from the cleaning phase
- if(state->do_clean == 1){
- // cout << "flows were sampled: " << state->how_many_cleaned << "\n";
- state->how_many_cleaned = 0;
- state->gcount = 0;
- state->do_clean = 0;
- }
-
- return 1;
-
-};
-
-
-int ssfinal_clean(void *s, int curr_num_samples, unsigned long long int glen){
- struct SSstate *state = (struct SSstate *)s;
-
- state->do_sooth = true;
-
- // for ssample() where just returned from the clening phase
- state->do_clean = 1;
-
- int sampled = 0;
- double new_len = 0;
-
- if (glen < state->z_prev)
- new_len = state->z_prev;
- else
- new_len = glen;
-
- //no need to clean
- if(curr_num_samples <= state->s_size){
- return 1;
- }
- else{
- if(new_len > state->z){
- sampled = 1;
- state->bcount++;
- }
- else{
- state->fcount += new_len;
- if(state->fcount >= state->z){
- sampled = 1;
- state->fcount -= state->z;
- }
- //else{
- //state->scount--;
- //}
- }
-
- return sampled;
- }
-
-};
-
-int flow_ssfinal_clean(void *s, int curr_num_samples, unsigned int ccondition, unsigned int delay, unsigned int maxtime, unsigned long long int glen){
- struct SSstate *state = (struct SSstate *)s;
-
- state->do_sooth = true;
- // only for reporting
- state->delay = delay;
-
- //for ssample() where just returned from the clening phase
- state->do_clean = 1;
-
- int sampled = 0;
- double new_len = 0;
- // group is newly closed or not closed yet
- int new_group = 1;
-
- // the flow is closed
- if((ccondition == 1)||((state->time-maxtime)>=delay)){
-
- new_group = 0;
-
- // TF case or TT case for ccondition and delay
- if(ccondition == 1){
- if((state->time-maxtime) <= 1){ // 1 is time interval for count of closed flows
- // the flow was closed in the previous counting slot
- new_group = 1;
- }
- }
- //FT case for ccondition and delay
- else{
- if((state->time-maxtime) <= (delay+1)){ // 1 is time interval for count of closed flows
- // the flow was closed in the previous counting slot
- new_group = 1;
- }
- }
- // the flow was closed earlier than previous counting slot
- // it is closed and old and new_group = 0
- }
-
- //adjust size only for old closed groups
- if(new_group == 0){
- if (glen < state->z_prev)
- new_len = state->z_prev;
- else
- new_len = glen;
- }
- // newly closed group
- else{
- state->closing_now++;
- new_len = glen;
- }
-
-
- //no need to clean
- if(curr_num_samples <= state->s_size){
- state->count_notsampled_new = 0;
- return 1;
- }
- else{
- if(new_len > state->z){
- sampled = 1;
- state->bcount++;
- }
- else{
- state->fcount += new_len;
- if(state->fcount >= state->z){
- sampled = 1;
- state->fcount -= state->z;
- }
- //count number of not sampled newly closed groups
- //will use in the next time window at the final cleaning phase
- else{
- if(new_group == 1)
- state->count_notsampled_new++;
- }
-
- }
-
- state->how_many_cleaned += sampled;
-
- return sampled;
- }
-
-};
-
-
-int ssdo_clean(void *s, int curr_num_samples){
- struct SSstate *state = (struct SSstate *)s;
-
- //emergency control
- //bcount will be different after the cleaning phase
- if(curr_num_samples > (state->gamma*state->s_size)){
- state->z_prev = state->z;
- state->z=(double)state->gamma*state->z;
- state->do_clean = 1;
- state->bcount = 0;
- state->count = 0;
- state->gcount = 0;
-
- state->do_clean_count++;
- }
-
- return state->do_clean;
-};
-
-int flow_ssdo_clean(void *s, int curr_num_samples, unsigned int maxtime){
- struct SSstate *state = (struct SSstate *)s;
-
- // initialize timestamp with first seen packet from the current timewindow
- if(state->time == 0)
- state->time = maxtime;
-
- // detect next counting time slot
- if(state->time != maxtime){
- // cout << "need to count\n";
- state->time = maxtime;
- state->count_closed = 1;
- return 1;
- }
-
- //emergency control
- //bcount will be different after the cleaning phase
- //if(curr_num_samples > (state->gamma*state->s_size)){
- if(state->count_closed_flows > (state->gamma*state->s_size)){
- // cout << "need to clean, num closed flows: " << state->count_closed_flows << "\n";
- state->z_prev = state->z;
- // cout << "do clean Z old: " << state->z << "\n";
- state->z=(((double)state->count_closed_flows)/((double)state->s_size))*state->z;
- // cout << "do clean Z new: " << state->z << "\n";
- state->do_clean = 1;
- state->bcount = 0;
- state->count = 0;
- state->gcount = 0;
-
- state->do_clean_count++;
- }
-
- //just returned from the counting iteration
- if(state->count_closed == 1){
- state->count_closed = 0;
- // cout << "number of closed flows: " << state->count_closed_flows << "\n";
- state->count_closed_flows = 0;
- //initialize gcount since was used for sampling of new closed flows during counting phase
- state->gcount = 0;
- }
-
- return state->do_clean;
-};
-
-double ssthreshold(void *s, int curr_num_samples){
- struct SSstate *state = (struct SSstate *)s;
-
- return state->z;
-};
-
-int count_distinct(void *s, int curr_num_samples){
- return curr_num_samples;
-};
-
-int ssclean_with(void *s,int curr_num_samples, unsigned long long int glen){
- struct SSstate *state = (struct SSstate *)s;
-
- //cleaning condition
- int sampled = 0;
- double new_len = 0;
-
- if (glen < state->z_prev)
- new_len = state->z_prev;
- else
- new_len = glen;
-
- if(new_len > state->z){
- state->bcount++;
- sampled = 1;
- }
- else{
- state->gcount += new_len;
- if(state->gcount >= state->z){
- sampled = 1;
- state->gcount -= state->z;
- }
- //else{
- //state->scount--;
- //}
- }
-
- return sampled;
-};
-
-int flow_ssclean_with(void *s,int curr_num_samples, unsigned int ccondition, unsigned int delay, unsigned int maxtime, unsigned long long int glen){
- struct SSstate *state = (struct SSstate *)s;
-
- //cleaning condition
- int sampled = 0;
- double new_len = 0;
- int new_group = 0;
-
- //need to count closed flows
- if(state->count_closed == 1){
- //the flow is closed
- if((ccondition == 1)||((state->time-maxtime)>=delay)){
- state->count_closed_flows++;
- // TF case or TT case for ccondition and delay
- if(ccondition == 1){
- if((state->time-maxtime) <= 1) //1 is time interval for count of closed flows
- new_group = 1;
- }
- //FT case for ccondition and delay
- else{
- if((state->time-maxtime) <= (delay+1)) //1 is time interval for count of closed flows
- new_group = 1;
- }
- }
-
- // if flow is closed but old, no need to subsample it
- if (new_group == 0)
- return true;
-
- }
-
- // clean only closed flows FF case
- // the flow is still open
- if((ccondition == 0)&&((state->time-maxtime)<delay)){
- return true;
- }
-
- // use glen for a new group and z_prev for old one
- if(new_group == 0){
- if (glen < state->z_prev)
- new_len = state->z_prev;
- else
- new_len = glen;
- }
- //the group is new
- else{
- new_len = glen;
- }
-
- // at this point either flow is closed and old and we are at the cleaning phase
- // or flow is closed and new and we are at the counting phase
- if(new_len > state->z){
- state->bcount++;
- sampled = 1;
- }
- else{
- state->gcount += new_len;
- if(state->gcount >= state->z){
- sampled = 1;
- state->gcount -= state->z;
- }
- //new_group is not sampled during counting phase
- else{
- if(state->count_closed == 1)
- state->count_closed_flows--;
- }
-
- }
-
- if(state->do_clean == 1)
- state->how_many_cleaned += sampled;
-
- return sampled;
-};
-
-int packet_count(void *s, int curr_num_samples){
- struct SSstate *state = (struct SSstate *)s;
- return state->packet_count;
-};
-
-double gamma(void *s, int curr_num_samples){
- struct SSstate *state = (struct SSstate *)s;
- return state->gamma;
-};
-
-int do_clean_count(void *s, int curr_num_samples){
- struct SSstate *state = (struct SSstate *)s;
- return state->do_clean_count;
-};
-
-int delay(void *s, int curr_num_samples){
- struct SSstate *state = (struct SSstate *)s;
- return state->delay;
-};
-
-// number of groups which are newly closed (in the most recent
-// counting phase or not closed at all during the final
-// cleaning phase
-int newly_closed(void *s, int curr_num_samples){
- struct SSstate *state = (struct SSstate *)s;
- return state->closing_now;
-};
-
-
-
-
+/* ------------------------------------------------\r
+ Copyright 2014 AT&T Intellectual Property\r
+ Licensed under the Apache License, Version 2.0 (the "License");\r
+ you may not use this file except in compliance with the License.\r
+ You may obtain a copy of the License at\r
+ \r
+ http://www.apache.org/licenses/LICENSE-2.0\r
+ \r
+ Unless required by applicable law or agreed to in writing, software\r
+ distributed under the License is distributed on an "AS IS" BASIS,\r
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\r
+ See the License for the specific language governing permissions and\r
+ limitations under the License.\r
+ ------------------------------------------- */\r
+///////////////////////////////////////////////\r
+//// State, functions for adaptive smart sampling\r
+/// written by Irina Rozenbaum\r
+\r
+#include <iostream>\r
+#define max(x,y) ((x)>(y)?(x):(y))\r
+#define _BURST_SOOTHING_FACTOR 10.0\r
+\r
+struct SSstate {\r
+ //unsigned long long int count; // count to sample small packets with certain probability\r
+ double count;\r
+ double gcount; // count for clean_with() function\r
+ double fcount; // count for final_clean() function\r
+ double z; //z is the threshold for a size of the packet\r
+ double z_prev; // value of threshold from the previos cleaning phase\r
+ double gamma; //tolerance parameter for emergency control over the\r
+ //number of samples, should be >= 1\r
+ int do_clean; // trigger cleaning phase\r
+ int bcount; //count for number of packets that exceed threshold ,\r
+ //need it for one of the variations of algorithms for\r
+ //threshold adjustment\r
+ int s_size; //need to remember sample size for _sfun_state_init()\r
+ int final_z; //bool that indicated if z was adjusted to its filnal value\r
+ //before the final clean\r
+ int time; //timestamp from previously processed packet\r
+ //used in detecting frequency of the clening phases with aggregation\r
+ \r
+ int packet_count; //count for number of packets per time window\r
+ int do_clean_count; //count number of cleaning phases per time window\r
+ bool do_sooth; //turn on _BURST_SOOTHING_FACTOR\r
+ \r
+ // flow sampling\r
+ int count_closed; //bool indicates that counting phase is being triggered\r
+ int count_closed_flows; //count for closed flows\r
+ int count_notsampled_new; //count for the number of new groups that were not sampled during\r
+ // the final sampling phase will be used in next time window to compute resonable\r
+ // approximation for threshold before final sampling phase\r
+ \r
+ // for reporting\r
+ int delay; //interval within which there was no packets that belong to the current flow\r
+ //thus flow is considered to be closed.\r
+ //parameter specified by the query\r
+ int closing_now; //count for groups that are closing at the final cleraning phase\r
+ //or were closed during the most recent counting phase (newly closed)\r
+ \r
+ //for debugging\r
+ int how_many_cleaned;\r
+};\r
+\r
+\r
+// the function is called once at the initial initialization of the state\r
+void _sfun_state_clean_init_smart_sampling_state(void *s){\r
+ struct SSstate *state = (struct SSstate *)s;\r
+ \r
+ state->count = 0;\r
+ state->gcount = 0;\r
+ state->fcount = 0;\r
+ state->z = 200; //need to figure out good initial value for z\r
+ // cout << "clean init Z: " << state->z << "\n";\r
+ state->z_prev = 0;\r
+ // cout << "clean init Z prev: " << state->z << "\n";\r
+ state->gamma = 2;\r
+ state->do_clean = 0;\r
+ state->bcount = 0;\r
+ state->s_size = 0;\r
+ state->final_z = 0;\r
+ state->time = 0;\r
+ state->count_closed = 0;\r
+ state->count_closed_flows = 0;\r
+ state->count_notsampled_new = 0;\r
+ \r
+ state->packet_count = 0;\r
+ state->do_clean_count = 0;\r
+ state->do_sooth = true;\r
+ \r
+ state->how_many_cleaned = 0;\r
+ state->delay = 0;\r
+ state->closing_now = 0;\r
+};\r
+\r
+\r
+// the function will be called at the beginning of every time window\r
+void _sfun_state_dirty_init_smart_sampling_state(void *s_new, void *s_old, int curr_num_samples){\r
+ struct SSstate *state_new = (struct SSstate *)s_new;\r
+ struct SSstate *state_old = (struct SSstate *)s_old;\r
+ \r
+ // cout << "**flows were sampled: " << state_old->how_many_cleaned << "\n";\r
+ // cout << "***dirty current num of samples: " << curr_num_samples << "\n";\r
+ // cout << "dirty init Z old: " << state_old->z << "\n";\r
+ if(curr_num_samples < state_old->s_size){\r
+ // state_new->z = state_old->z*((max((double)curr_num_samples-(double)state_old->bcount,1))/((double)state_old->s_size-(double)state_old->bcount));\r
+ \r
+ //temp fix for early aggregation\r
+ state_new->z = (state_old->z)/2;\r
+ \r
+ if(state_old->do_sooth)\r
+ state_new->z = state_old->z / _BURST_SOOTHING_FACTOR;\r
+ }\r
+ else {\r
+ if(curr_num_samples >= state_old->s_size){\r
+ //cout << "yes\n";\r
+ state_new->z = state_old->z*((double)curr_num_samples/(double)state_old->s_size);\r
+ \r
+ if(state_old->do_sooth)\r
+ state_new->z = state_old->z / _BURST_SOOTHING_FACTOR;\r
+ }\r
+ }\r
+ \r
+ if(state_new->z <= 1.0)\r
+ state_new->z = 1;\r
+ // cout << "dirty init Z new: " << state_new->z << "\n";\r
+ \r
+ state_new->gamma = state_old->gamma;\r
+ state_new->do_clean = state_old->do_clean;\r
+ state_new->s_size = state_old->s_size;\r
+ state_new->bcount = 0;\r
+ state_new->gcount = 0;\r
+ state_new->count = 0;\r
+ state_new->fcount = 0;\r
+ state_new->final_z = 0;\r
+ state_new->time = 0;\r
+ state_new->count_closed = 0;\r
+ state_new->count_closed_flows = 0;\r
+ state_new->count_notsampled_new = state_old->count_notsampled_new;\r
+ \r
+ state_new->packet_count = 0;\r
+ state_new->do_clean_count = 0;\r
+ state_new->do_sooth = true;\r
+ \r
+ state_new->how_many_cleaned = 0;\r
+ state_new->delay = 0;\r
+ state_new->closing_now = 0;\r
+ \r
+ // cout << "dirty init gamma: " << state_new->gamma << "\n";\r
+ // cout << "dirty init do_clean: " << state_new->do_clean << "\n";\r
+ // cout << "dirty init s_size: " << state_new->s_size << "\n";\r
+ // cout << "dirty init bcount: " << state_new->bcount << "\n";\r
+ // cout << "dirty init gcount: " << state_new->gcount << "\n";\r
+ // cout << "dirty init count: " << state_new->count << "\n";\r
+ // cout << "dirty init fcount: " << state_new->fcount << "\n";\r
+ // cout << "dirty init final_z: " << state_new->final_z << "\n";\r
+ // cout << "dirty init time " << state_new->time << "\n";\r
+ // cout << "dirty init count_closed: " << state_new->count_closed << "\n";\r
+ // cout << "dirty init count_closed_flows: " << state_new->count_closed_flows << "\n";\r
+ \r
+ // cout << "dirty init packet count: " << state_new->packet_count << "\n";\r
+ // cout << "dirty init do_clean_count: " << state_new->do_clean_count << "\n";\r
+ \r
+};\r
+\r
+// function is called in two cases:\r
+// adjustment of threshold before the cleaning phase is being triggered\r
+// adjustment of threshold at the window border for the old state, this new\r
+// threshold will be used by partial_flush() and flush() functions (final_clean())\r
+\r
+void _sfun_state_final_init_smart_sampling_state(void *s, int curr_num_samples){\r
+ struct SSstate *state = (struct SSstate *)s;\r
+ \r
+ // cout << "final init Z old: " << state->z << "\n";\r
+ \r
+ if(state->final_z == 0){\r
+ \r
+ // just returned from the counting phase\r
+ // case when the time window is changed right after the counting phase\r
+ if(state->count_closed == 1){\r
+ state->count_closed = 0;\r
+ state->count_closed_flows = 0;\r
+ state->how_many_cleaned = 0;\r
+ }\r
+ \r
+ state->z_prev = state->z;\r
+ // cout << "final current num of samples: " << curr_num_samples << "\n";\r
+ // cout << "final count not sampled new: " << state->count_notsampled_new << "\n";\r
+ // adjust count for current number of sampled based on statistics\r
+ // gathered in the previous time window\r
+ // otherwise we overestimate it\r
+ if(state->count_notsampled_new != 0)\r
+ curr_num_samples -= state->count_notsampled_new;\r
+ \r
+ if(curr_num_samples < state->s_size){\r
+ //state->z = state->z*((max((double)curr_num_samples-(double)state->bcount,1))/((double)state->s_size-(double)state->bcount));\r
+ \r
+ //temp fix for early aggregation\r
+ state->z = state->z/2;\r
+ }\r
+ else {\r
+ if(curr_num_samples >= state->s_size){\r
+ state->z = state->z*((double)curr_num_samples/(double)state->s_size);\r
+ }\r
+ }\r
+ \r
+ if(state->z <= 0)\r
+ state->z = 1;\r
+ // cout << "final init Z new: " << state->z << "\n";\r
+ \r
+ state->bcount = 0;\r
+ state->final_z = 1;\r
+ state->do_clean_count++;\r
+ state->count_notsampled_new = 0;\r
+ \r
+ }\r
+ \r
+ // cout << "final init gamma: " << state->gamma << "\n";\r
+ // cout << "final init do_clean: " << state->do_clean << "\n";\r
+ // cout << "final init s_size: " << state->s_size << "\n";\r
+ // cout << "final init bcount: " << state->bcount << "\n";\r
+ // cout << "final init gcount: " << state->gcount << "\n";\r
+ // cout << "final init count: " << state->count << "\n";\r
+ // cout << "final init fcount: " << state->fcount << "\n";\r
+ // cout << "final init final_z: " << state->final_z << "\n";\r
+ // cout << "final init time " << state->time << "\n";\r
+ // cout << "final init count_closed: " << state->count_closed << "\n";\r
+ // cout << "final init count_closed_flows: " << state->count_closed_flows << "\n";\r
+ \r
+ // cout << "final init packet count: " << state->packet_count << "\n";\r
+ // cout << "final init do_clean_count: " << state->do_clean_count << "\n";\r
+ \r
+};\r
+\r
+int ssample(void *s,int curr_num_samples, unsigned long long int len, unsigned int sample_size){\r
+ struct SSstate *state = (struct SSstate *)s;\r
+ \r
+ state->packet_count++;\r
+ \r
+ state->s_size = sample_size;\r
+ int sampled = 0;\r
+ \r
+ //just returned from the cleaning phase\r
+ if(state->do_clean == 1){\r
+ state->gcount = 0;\r
+ state->do_clean = 0;\r
+ }\r
+ \r
+ //sampling\r
+ if(len > state->z){\r
+ state->bcount++;\r
+ sampled=1;\r
+ }\r
+ else{\r
+ state->count += len;\r
+ if(state->count >= state->z){\r
+ sampled=1;\r
+ state->count -= state->z;\r
+ }\r
+ }\r
+ return sampled;\r
+ \r
+};\r
+\r
+int flow_ssample(void *s,int curr_num_samples, unsigned int sample_size){\r
+ struct SSstate *state = (struct SSstate *)s;\r
+ \r
+ state->packet_count++;\r
+ \r
+ state->s_size = sample_size;\r
+ int sampled = 0;\r
+ \r
+ //just returned from the counting phase\r
+ if(state->count_closed == 1)\r
+ // cout << "closed flows after counting phase: " << state->count_closed_flows << "\n";\r
+ \r
+ //just returned from the cleaning phase\r
+ if(state->do_clean == 1){\r
+ // cout << "flows were sampled: " << state->how_many_cleaned << "\n";\r
+ state->how_many_cleaned = 0;\r
+ state->gcount = 0;\r
+ state->do_clean = 0;\r
+ }\r
+ \r
+ return 1;\r
+ \r
+};\r
+\r
+\r
+int ssfinal_clean(void *s, int curr_num_samples, unsigned long long int glen){\r
+ struct SSstate *state = (struct SSstate *)s;\r
+ \r
+ state->do_sooth = true;\r
+ \r
+ // for ssample() where just returned from the clening phase\r
+ state->do_clean = 1;\r
+ \r
+ int sampled = 0;\r
+ double new_len = 0;\r
+ \r
+ if (glen < state->z_prev)\r
+ new_len = state->z_prev;\r
+ else\r
+ new_len = glen;\r
+ \r
+ //no need to clean\r
+ if(curr_num_samples <= state->s_size){\r
+ return 1;\r
+ }\r
+ else{\r
+ if(new_len > state->z){\r
+ sampled = 1;\r
+ state->bcount++;\r
+ }\r
+ else{\r
+ state->fcount += new_len;\r
+ if(state->fcount >= state->z){\r
+ sampled = 1;\r
+ state->fcount -= state->z;\r
+ }\r
+ //else{\r
+ //state->scount--;\r
+ //}\r
+ }\r
+ \r
+ return sampled;\r
+ }\r
+ \r
+};\r
+\r
+int flow_ssfinal_clean(void *s, int curr_num_samples, unsigned int ccondition, unsigned int delay, unsigned int maxtime, unsigned long long int glen){\r
+ struct SSstate *state = (struct SSstate *)s;\r
+ \r
+ state->do_sooth = true;\r
+ // only for reporting\r
+ state->delay = delay;\r
+ \r
+ //for ssample() where just returned from the clening phase\r
+ state->do_clean = 1;\r
+ \r
+ int sampled = 0;\r
+ double new_len = 0;\r
+ // group is newly closed or not closed yet\r
+ int new_group = 1;\r
+ \r
+ // the flow is closed\r
+ if((ccondition == 1)||((state->time-maxtime)>=delay)){\r
+ \r
+ new_group = 0;\r
+ \r
+ // TF case or TT case for ccondition and delay\r
+ if(ccondition == 1){\r
+ if((state->time-maxtime) <= 1){ // 1 is time interval for count of closed flows\r
+ // the flow was closed in the previous counting slot\r
+ new_group = 1;\r
+ }\r
+ }\r
+ //FT case for ccondition and delay\r
+ else{\r
+ if((state->time-maxtime) <= (delay+1)){ // 1 is time interval for count of closed flows\r
+ // the flow was closed in the previous counting slot\r
+ new_group = 1;\r
+ }\r
+ }\r
+ // the flow was closed earlier than previous counting slot\r
+ // it is closed and old and new_group = 0\r
+ }\r
+ \r
+ //adjust size only for old closed groups\r
+ if(new_group == 0){\r
+ if (glen < state->z_prev)\r
+ new_len = state->z_prev;\r
+ else\r
+ new_len = glen;\r
+ }\r
+ // newly closed group\r
+ else{\r
+ state->closing_now++;\r
+ new_len = glen;\r
+ }\r
+ \r
+ \r
+ //no need to clean\r
+ if(curr_num_samples <= state->s_size){\r
+ state->count_notsampled_new = 0;\r
+ return 1;\r
+ }\r
+ else{\r
+ if(new_len > state->z){\r
+ sampled = 1;\r
+ state->bcount++;\r
+ }\r
+ else{\r
+ state->fcount += new_len;\r
+ if(state->fcount >= state->z){\r
+ sampled = 1;\r
+ state->fcount -= state->z;\r
+ }\r
+ //count number of not sampled newly closed groups\r
+ //will use in the next time window at the final cleaning phase\r
+ else{\r
+ if(new_group == 1)\r
+ state->count_notsampled_new++;\r
+ }\r
+ \r
+ }\r
+ \r
+ state->how_many_cleaned += sampled;\r
+ \r
+ return sampled;\r
+ }\r
+ \r
+};\r
+\r
+\r
+int ssdo_clean(void *s, int curr_num_samples){\r
+ struct SSstate *state = (struct SSstate *)s;\r
+ \r
+ //emergency control\r
+ //bcount will be different after the cleaning phase\r
+ if(curr_num_samples > (state->gamma*state->s_size)){\r
+ state->z_prev = state->z;\r
+ state->z=(double)state->gamma*state->z;\r
+ state->do_clean = 1;\r
+ state->bcount = 0;\r
+ state->count = 0;\r
+ state->gcount = 0;\r
+ \r
+ state->do_clean_count++;\r
+ }\r
+ \r
+ return state->do_clean;\r
+};\r
+\r
+int flow_ssdo_clean(void *s, int curr_num_samples, unsigned int maxtime){\r
+ struct SSstate *state = (struct SSstate *)s;\r
+ \r
+ // initialize timestamp with first seen packet from the current timewindow\r
+ if(state->time == 0)\r
+ state->time = maxtime;\r
+ \r
+ // detect next counting time slot\r
+ if(state->time != maxtime){\r
+ // cout << "need to count\n";\r
+ state->time = maxtime;\r
+ state->count_closed = 1;\r
+ return 1;\r
+ }\r
+ \r
+ //emergency control\r
+ //bcount will be different after the cleaning phase\r
+ //if(curr_num_samples > (state->gamma*state->s_size)){\r
+ if(state->count_closed_flows > (state->gamma*state->s_size)){\r
+ // cout << "need to clean, num closed flows: " << state->count_closed_flows << "\n";\r
+ state->z_prev = state->z;\r
+ // cout << "do clean Z old: " << state->z << "\n";\r
+ state->z=(((double)state->count_closed_flows)/((double)state->s_size))*state->z;\r
+ // cout << "do clean Z new: " << state->z << "\n";\r
+ state->do_clean = 1;\r
+ state->bcount = 0;\r
+ state->count = 0;\r
+ state->gcount = 0;\r
+ \r
+ state->do_clean_count++;\r
+ }\r
+ \r
+ //just returned from the counting iteration\r
+ if(state->count_closed == 1){\r
+ state->count_closed = 0;\r
+ // cout << "number of closed flows: " << state->count_closed_flows << "\n";\r
+ state->count_closed_flows = 0;\r
+ //initialize gcount since was used for sampling of new closed flows during counting phase\r
+ state->gcount = 0;\r
+ }\r
+ \r
+ return state->do_clean;\r
+};\r
+\r
+double ssthreshold(void *s, int curr_num_samples){\r
+ struct SSstate *state = (struct SSstate *)s;\r
+ \r
+ return state->z;\r
+};\r
+\r
+int count_distinct(void *s, int curr_num_samples){\r
+ return curr_num_samples;\r
+};\r
+\r
+int ssclean_with(void *s,int curr_num_samples, unsigned long long int glen){\r
+ struct SSstate *state = (struct SSstate *)s;\r
+ \r
+ //cleaning condition\r
+ int sampled = 0;\r
+ double new_len = 0;\r
+ \r
+ if (glen < state->z_prev)\r
+ new_len = state->z_prev;\r
+ else\r
+ new_len = glen;\r
+ \r
+ if(new_len > state->z){\r
+ state->bcount++;\r
+ sampled = 1;\r
+ }\r
+ else{\r
+ state->gcount += new_len;\r
+ if(state->gcount >= state->z){\r
+ sampled = 1;\r
+ state->gcount -= state->z;\r
+ }\r
+ //else{\r
+ //state->scount--;\r
+ //}\r
+ }\r
+ \r
+ return sampled;\r
+};\r
+\r
+int flow_ssclean_with(void *s,int curr_num_samples, unsigned int ccondition, unsigned int delay, unsigned int maxtime, unsigned long long int glen){\r
+ struct SSstate *state = (struct SSstate *)s;\r
+ \r
+ //cleaning condition\r
+ int sampled = 0;\r
+ double new_len = 0;\r
+ int new_group = 0;\r
+ \r
+ //need to count closed flows\r
+ if(state->count_closed == 1){\r
+ //the flow is closed\r
+ if((ccondition == 1)||((state->time-maxtime)>=delay)){\r
+ state->count_closed_flows++;\r
+ // TF case or TT case for ccondition and delay\r
+ if(ccondition == 1){\r
+ if((state->time-maxtime) <= 1) //1 is time interval for count of closed flows\r
+ new_group = 1;\r
+ }\r
+ //FT case for ccondition and delay\r
+ else{\r
+ if((state->time-maxtime) <= (delay+1)) //1 is time interval for count of closed flows\r
+ new_group = 1;\r
+ }\r
+ }\r
+ \r
+ // if flow is closed but old, no need to subsample it\r
+ if (new_group == 0)\r
+ return true;\r
+ \r
+ }\r
+ \r
+ // clean only closed flows FF case\r
+ // the flow is still open\r
+ if((ccondition == 0)&&((state->time-maxtime)<delay)){\r
+ return true;\r
+ }\r
+ \r
+ // use glen for a new group and z_prev for old one\r
+ if(new_group == 0){\r
+ if (glen < state->z_prev)\r
+ new_len = state->z_prev;\r
+ else\r
+ new_len = glen;\r
+ }\r
+ //the group is new\r
+ else{\r
+ new_len = glen;\r
+ }\r
+ \r
+ // at this point either flow is closed and old and we are at the cleaning phase\r
+ // or flow is closed and new and we are at the counting phase\r
+ if(new_len > state->z){\r
+ state->bcount++;\r
+ sampled = 1;\r
+ }\r
+ else{\r
+ state->gcount += new_len;\r
+ if(state->gcount >= state->z){\r
+ sampled = 1;\r
+ state->gcount -= state->z;\r
+ }\r
+ //new_group is not sampled during counting phase\r
+ else{\r
+ if(state->count_closed == 1)\r
+ state->count_closed_flows--;\r
+ }\r
+ \r
+ }\r
+ \r
+ if(state->do_clean == 1)\r
+ state->how_many_cleaned += sampled;\r
+ \r
+ return sampled;\r
+};\r
+\r
+int packet_count(void *s, int curr_num_samples){\r
+ struct SSstate *state = (struct SSstate *)s; \r
+ return state->packet_count;\r
+};\r
+\r
+double gamma(void *s, int curr_num_samples){\r
+ struct SSstate *state = (struct SSstate *)s; \r
+ return state->gamma;\r
+};\r
+\r
+int do_clean_count(void *s, int curr_num_samples){\r
+ struct SSstate *state = (struct SSstate *)s; \r
+ return state->do_clean_count;\r
+};\r
+\r
+int delay(void *s, int curr_num_samples){\r
+ struct SSstate *state = (struct SSstate *)s; \r
+ return state->delay;\r
+};\r
+\r
+// number of groups which are newly closed (in the most recent\r
+// counting phase or not closed at all during the final\r
+// cleaning phase\r
+int newly_closed(void *s, int curr_num_samples){\r
+ struct SSstate *state = (struct SSstate *)s; \r
+ return state->closing_now;\r
+};\r
+\r
+\r
+\r
+\r
Code Review / com / gs-lite.git / blobdiff