1 /* See the DRL-LICENSE file for this file's software license. */
10 #include "common_accounting.h"
11 #include "samplehold.h"
13 static int match(const key_flow *key, const sampled_flow *flow) {
14 if (flow->state != FLOW_USED)
17 if (key->source_ip != flow->source_ip)
20 if (key->dest_ip != flow->dest_ip)
23 if (key->source_port != flow->source_port)
26 if (key->dest_port != flow->dest_port)
29 if (key->protocol != flow->protocol)
35 static void get_key(key_flow *key, sampled_flow *flow) {
36 key->source_ip = flow->source_ip;
37 key->dest_ip = flow->dest_ip;
38 key->source_port = flow->source_port;
39 key->dest_port = flow->dest_port;
40 key->protocol = flow->protocol;
45 static void move_flow(sampled_flow *dest, sampled_flow *src) {
46 memmove(dest, src, sizeof(sampled_flow));
47 memset(src, 0, sizeof(sampled_flow));
50 uint32_t sampled_table_size(const sampled_flow_table table) {
57 * max_bytes is the maximum number of bytes that can pass though DURING THE
58 * MEASUREMENT INTERVAL. So, if you can have 100 Mbit/s and your measurement
59 * interval is 1/10 of a second, your max_bytes is 10Mbit because that's all
60 * you can transfer in 1/10 of a second.
62 * flow_percentage is the percentage of max_bytes that is considered an
65 * oversampling factor is a knob that tunes how accurate our results are at
66 * the cost of additional state/memory.
68 sampled_flow_table sampled_table_create(uint32_t (*hash_function)(const key_flow *key), const uint32_t max_bytes, const uint32_t flow_percentage, const uint32_t oversampling_factor, common_accounting_t *common) {
69 sampled_flow_table table = malloc(sizeof(struct sampled_flow_table));
70 double base_size = (double) 100 / (double) flow_percentage;
76 table->capacity = (uint32_t) ((base_size * oversampling_factor) * 1.03);
78 table->hash_function = hash_function;
79 table->sample_prob = (double) (((double) table->capacity / (double) max_bytes) * (double) RANDOM_GRANULARITY);
80 table->threshold = (double) ((double) flow_percentage / 100) * max_bytes;
82 table->largest = NULL;
83 table->backing = malloc(sizeof(sampled_flow) * table->capacity);
85 if (table->backing == NULL) {
90 memset(table->backing, 0, sizeof(sampled_flow) * table->capacity);
94 table->common = common;
95 gettimeofday(&table->common->last_update, NULL);
100 void sampled_table_destroy(sampled_flow_table table) {
101 free(table->backing);
105 sampled_flow *sampled_table_lookup(sampled_flow_table table, const key_flow *key) {
106 uint32_t hash = table->hash_function(key) % table->capacity;
107 uint32_t location = hash;
110 if (table->backing[location].state == FLOW_FREE) {
111 /* It ain't here... */
115 if (match(key, &table->backing[location])) {
117 return &table->backing[location];
121 if (location == table->capacity) {
124 } while (location != hash);
129 int sampled_table_sample(sampled_flow_table table, const key_flow *key) {
130 sampled_flow *lookup = sampled_table_lookup(table, key);
134 /* First we update the common accouting information so that we have accurate
135 * aggregate information. */
136 table->common->bytes_since += key->packet_size;
138 /* Below here we're dealing with individual flows. */
140 /* It's already in the table, update it. */
141 if (lookup != NULL) {
142 lookup->bytes += key->packet_size;
146 /* It's not in the table, probabilistically sample it. */
147 packet_prob = table->sample_prob * (double) key->packet_size;
148 random_number = rand() % RANDOM_GRANULARITY;
150 if (random_number < packet_prob) {
151 /* It's being sampled - add it to the table. */
152 uint32_t hash = table->hash_function(key) % table->capacity;
153 uint32_t location = hash;
156 if (table->backing[location].state == FLOW_FREE ||
157 table->backing[location].state == FLOW_DELETED) {
158 lookup = &table->backing[location];
163 if (location == table->capacity) {
166 } while (location != hash);
168 if (lookup == NULL) {
169 /* Table is full!?! */
170 printf("Full table!\n");
176 lookup->bytes = key->packet_size;
177 lookup->source_ip = key->source_ip;
178 lookup->dest_ip = key->dest_ip;
179 lookup->source_port = key->source_port;
180 lookup->dest_port = key->dest_port;
181 lookup->protocol = key->protocol;
182 lookup->state = FLOW_USED;
183 lookup->last_bytes = 0;
186 gettimeofday(&lookup->last_update, NULL);
195 int sampled_table_cleanup(sampled_flow_table table) {
197 * 1) Remove "small" flows from the table.
198 * 2) Compact the table so that the remaining flows are closer to their
200 * 3) Reset the state of deleted flows to free.
203 /* How it might work...
204 * 1) Scan through the backing array.
205 * 2) If the flow is small, memset it to 0.
206 * It it's large, add it to a linked list.
207 * 3) For all items in the linked list, hash them and put them in the
211 /* For now though, we're going to do it the inefficient way and loop
212 * through the backing twice.
217 /* Clear small items. */
218 for (i = 0; i < table->capacity; ++i) {
219 if (table->backing[i].state == FLOW_USED && table->backing[i].bytes > table->threshold) {
220 /* It gets to stick around. */
223 memset(&table->backing[i], 0, sizeof(sampled_flow));
227 /* Compact the table and put things closer to their hash locations. */
228 for (i = 0; i < table->capacity; ++i) {
229 if (table->backing[i].state == FLOW_USED) {
233 get_key(&key, &table->backing[i]);
234 hash = table->hash_function(&key) % table->capacity;
237 /* Already in the best place */
238 table->backing[i].bytes = 0;
239 table->backing[i].last_bytes = 0;
240 table->backing[i].rate = 0;
242 uint32_t location = hash;
245 if (table->backing[location].state == FLOW_FREE) {
246 move_flow(&table->backing[location], &table->backing[i]);
247 table->backing[location].bytes = 0;
248 table->backing[location].last_bytes = 0;
249 table->backing[location].rate = 0;
254 if (location == table->capacity) {
257 } while (location != hash);
262 table->largest = NULL;
267 void sampled_table_update_flows(sampled_flow_table table, struct timeval now, double ewma_weight) {
269 uint32_t largest_rate = 0;
270 uint32_t rate_delta = 0;
271 double time_delta = 0;
272 double unweighted_rate = 0;
274 /* Update common aggregate information. */
275 time_delta = timeval_subtract(now, table->common->last_update);
277 if (time_delta <= 0) {
280 unweighted_rate = table->common->bytes_since / time_delta;
283 table->common->last_inst_rate = table->common->inst_rate;
284 table->common->inst_rate = unweighted_rate;
286 table->common->last_rate = table->common->rate;
288 /* If the rate is zero, then we don't know anything yet. Don't apply EWMA
290 if (table->common->rate == 0) {
291 table->common->rate = unweighted_rate;
293 table->common->rate = table->common->rate * ewma_weight +
294 unweighted_rate * (1 - ewma_weight);
297 table->common->bytes_since = 0;
298 table->common->last_update = now;
300 /* Update per-flow information. */
301 table->largest = &table->backing[i];
302 largest_rate = table->backing[i].rate;
304 for (i = 0; i < table->capacity; ++i) {
305 if (table->backing[i].state == FLOW_USED) {
306 rate_delta = table->backing[i].bytes - table->backing[i].last_bytes;
307 time_delta = timeval_subtract(now, table->backing[i].last_update);
309 /* Calculate the unweighted rate. Be careful not to divide by
310 * something silly. */
311 if (time_delta <= 0) {
314 unweighted_rate = rate_delta / time_delta;
317 if (table->backing[i].rate == 0) {
318 table->backing[i].rate = unweighted_rate;
320 table->backing[i].rate = (table->backing[i].rate * ewma_weight +
321 unweighted_rate * (1 - ewma_weight));
324 table->backing[i].last_bytes = table->backing[i].bytes;
325 table->backing[i].last_update = now;
327 if (table->backing[i].rate > largest_rate) {
328 largest_rate = table->backing[i].rate;
329 table->largest = &table->backing[i];
334 table->common->max_flow_rate = largest_rate;
337 sampled_flow *sampled_table_largest(sampled_flow_table table) {
338 return table->largest;