drl/estimate.c

   1 /* See the DRL-LICENSE file for this file's software license. */
   2
   3 /*
   4  * Thread to periodically calculate the estimated local limits
   5  * Barath Raghavan 2006/2007
   6  * Ken Yocum 2007
   7  * Kevin Webb 2007/2008
   8  */
   9
  10 #include <assert.h>
  11
  12 /** The size of the buffer we use to hold tc commands. */
  13 #define CMD_BUFFER_SIZE 200
  14
  15 /* DRL specifics */
  16 #include "raterouter.h"
  17 #include "util.h"
  18 #include "ratetypes.h" /* needs util and pthread.h */
  19 #include "logging.h"
  20
  21 static int underlimit_flowcount_count = 0;
  22 static int underlimit_normal_count = 0;
  23
  24 /**
  25  * Called for each identity each estimate interval.  Uses flow table information
  26  * to estimate the current aggregate rate and the rate of the individual flows
  27  * in the table.
  28  */
  29 static void estimate(identity_t *ident) {
  30     struct timeval now;
  31
  32     gettimeofday(&now, NULL);
  33
  34     pthread_mutex_lock(&ident->table_mutex); /* CLUNK ! */
  35
  36     ident->table_update_function(ident->table, now, ident->ewma_weight);
  37
  38     pthread_mutex_unlock(&ident->table_mutex); /* CLINK ! */
  39 }
  40
  41 /**
  42  * Determines the FPS weight allocation when the identity is under its current
  43  * local rate limit.
  44  */
  45 static double allocate_fps_under_limit(identity_t *ident, uint32_t local_rate, double peer_weights) {
  46     uint32_t target = local_rate;
  47     double ideal_weight;
  48     double total_weight = peer_weights + ident->last_localweight;
  49
  50     if (ident->flowstart) {
  51         target = local_rate*4;
  52         if (local_rate >= FLOW_START_THRESHOLD) {
  53             ident->flowstart = false;
  54         }
  55     }
  56     else {
  57         /* June 16, 2008 (KCW)
  58          * ident->flowstart gets set initially to one, but it is never set again.  However,
  59          * if a limiter gets flows and then the number of flows drops to zero, it has trouble
  60          * increasing the limit again. */
  61         if (local_rate < FLOW_START_THRESHOLD) {
  62             ident->flowstart = true;
  63         }
  64     }
  65
  66     if (target >= ident->limit) {
  67         ideal_weight = total_weight;
  68     } else if (target <= 0) {
  69         ideal_weight = 0; // no flows here
  70     } else {
  71         ideal_weight = ((double)target / (double)ident->limit) * total_weight;
  72     }
  73
  74 #if 0
  75     else if (peer_weights <= 0) {
  76 #if 0
  77         // doesn't matter what we pick as our weight, so pick 1 / N.
  78         ideal_weight = MAX_FLOW_SCALING_FACTOR / (remote_count(ident->i_handle) + 1);
  79 #endif
  80         ideal_weight = ((double)target / (double)ident->limit) * total_weight;
  81     } else {
  82 #if 0
  83         double divisor = (double) ident->limit - (double) target;
  84         ideal_weight = ((double) target * peer_weights) / divisor;
  85 #else
  86         ideal_weight = ((double)target / (double)ident->limit) * total_weight;
  87 #endif
  88     }
  89 #endif
  90
  91     return ideal_weight;
  92 }
  93
  94 /**
  95  * Determines the FPS weight allocation when the identity is over its current
  96  * local rate limit.
  97  */
  98 static double allocate_fps_over_limit(identity_t *ident) {
  99     double ideal_weight;
 100
 101     if (ident->common.max_flow_rate > 0) {
 102         ideal_weight = (double) ident->locallimit / (double) ident->common.max_flow_rate;
 103
 104         printlog(LOG_DEBUG, "%.3f  %d  %d  FlowCount, TotalRate, MaxRate\n",
 105                 ideal_weight, ident->common.rate, ident->common.max_flow_rate);
 106     } else {
 107         ideal_weight = 1;
 108     }
 109
 110     return ideal_weight;
 111 }
 112
 113 /**
 114  * Determines the amount of FPS weight to allocate to the identity during each
 115  * estimate interval.  Note that total_weight includes local weight.
 116  */
 117 static uint32_t allocate_fps(identity_t *ident, double total_weight) {
 118     common_accounting_t *ftable = &ident->common; /* Common flow table info */
 119     uint32_t local_rate = ftable->rate;
 120     uint32_t ideallocal = 0;
 121     double peer_weights; /* sum of weights of all other limiters  */
 122     double idealweight = 0;
 123     double last_portion = 0;
 124     double this_portion = 0;
 125
 126     static int dampen = 0;
 127     int dampen_increase = 0;
 128
 129     double ideal_under = 0;
 130     double ideal_over = 0;
 131
 132     int regime = 0;
 133
 134     /* two cases:
 135        1. the aggregate is < limit
 136        2. the aggregate is >= limit
 137        */
 138     peer_weights = total_weight - ident->last_localweight;
 139     if (peer_weights < 0) {
 140         peer_weights = 0;
 141     }
 142
 143     if (dampen == 1) {
 144         int64_t rate_delta =
 145             (int64_t) ftable->inst_rate - (int64_t) ftable->last_inst_rate;
 146         double threshold =
 147             (double) ident->limit * (double) LARGE_INCREASE_PERCENTAGE / 10;
 148
 149         if (rate_delta > threshold) {
 150             dampen_increase = 1;
 151             printlog(LOG_DEBUG, "DAMPEN: delta(%.3f) thresh(%.3f)\n",
 152                      rate_delta, threshold);
 153         }
 154     }
 155
 156     if (local_rate <= 0) {
 157         idealweight = 0;
 158     } else if (dampen_increase == 0 && (ident->locallimit <= 0 || local_rate < ident->locallimit || ident->flowstart)) {
 159         /* We're under the limit - all flows are bottlenecked. */
 160         idealweight = allocate_fps_under_limit(ident, local_rate, peer_weights);
 161         ideal_over = allocate_fps_over_limit(ident);
 162         ideal_under = idealweight;
 163
 164         if (ideal_over < idealweight) {
 165             idealweight = ideal_over;
 166             regime = 3;
 167             dampen = 2;
 168             underlimit_flowcount_count += 1;
 169         } else {
 170             regime = 1;
 171             dampen = 0;
 172             underlimit_normal_count += 1;
 173         }
 174
 175         /* Apply EWMA */
 176         ident->localweight = (ident->localweight * ident->ewma_weight +
 177                               idealweight * (1 - ident->ewma_weight));
 178
 179     } else {
 180         idealweight = allocate_fps_over_limit(ident);
 181
 182         /* Apply EWMA */
 183         ident->localweight = (ident->localweight * ident->ewma_weight +
 184                               idealweight * (1 - ident->ewma_weight));
 185
 186         /* This is the portion of the total weight in the system that was caused
 187          * by this limiter in the last interval. */
 188         last_portion = ident->last_localweight / total_weight;
 189
 190         /* This is the fraction of the total weight in the system that our
 191          * proposed value for idealweight would use. */
 192         this_portion = ident->localweight / (peer_weights + ident->localweight);
 193
 194         /* Dampen the large increase the first time... */
 195         if (dampen == 0 && (this_portion - last_portion > LARGE_INCREASE_PERCENTAGE)) {
 196             ident->localweight = ident->last_localweight + (LARGE_INCREASE_PERCENTAGE * total_weight);
 197             dampen = 1;
 198         } else {
 199             dampen = 2;
 200         }
 201
 202         ideal_under = allocate_fps_under_limit(ident, local_rate, peer_weights);
 203         ideal_over = idealweight;
 204
 205         regime = 2;
 206     }
 207
 208     /* Convert weight into a rate - add in our new local weight */
 209     ident->total_weight = total_weight = ident->localweight + peer_weights;
 210
 211     /* compute local allocation:
 212        if there is traffic elsewhere, use the weights
 213        otherwise do a L/n allocation */
 214     if (total_weight > 0) {
 215     //if (peer_weights > 0) {
 216         ideallocal = (uint32_t) (ident->localweight * ident->limit / total_weight);
 217     } else {
 218         ideallocal = ident->limit / (ident->comm.remote_node_count + 1);
 219     }
 220
 221     printlog(LOG_DEBUG, "%.3f ActualWeight\n", ident->localweight);
 222
 223     printlog(LOG_DEBUG, "%.3f %.3f %.3f %.3f  Under / Over / Actual / Rate\n",
 224             ideal_under / (ideal_under + peer_weights),
 225             ideal_over / (ideal_over + peer_weights),
 226             ident->localweight / (ident->localweight + peer_weights),
 227             (double) local_rate / (double) ident->limit);
 228
 229     printlog(LOG_DEBUG, "%.3f %.3f IdealUnd IdealOve\n",ideal_under,ideal_over);
 230
 231     printf("local_rate: %d, idealweight: %.3f, localweight: %.3f, total_weight: %.3f\n",
 232             local_rate, idealweight, ident->localweight, total_weight);
 233
 234     //printf("Dampen: %d, dampen_increase: %d, peer_weights: %.3f, regime: %d\n",
 235     //       dampen, dampen_increase, peer_weights, regime);
 236
 237     //printf("normal_count: %d, flowcount_count: %d\n", underlimit_normal_count, underlimit_flowcount_count);
 238
 239     if (regime == 3) {
 240         printlog(LOG_DEBUG, "MIN: min said to use flow counting, which was %.3f when other method said %.3f.\n",
 241                  ideal_over, ideal_under);
 242     }
 243
 244     printlog(LOG_DEBUG, "ideallocal is %d\n", ideallocal);
 245
 246     return(ideallocal);
 247 }
 248
 249 /**
 250  * Determines the local drop probability for a GRD identity every estimate
 251  * interval.
 252  */
 253 static double allocate_grd(identity_t *ident, double aggdemand) {
 254     double dropprob;
 255     double global_limit = (double) (ident->limit);
 256
 257     if (aggdemand > global_limit) {
 258         dropprob = (aggdemand-global_limit)/aggdemand;
 259     } else {
 260         dropprob = 0.0;
 261     }
 262
 263     printf("local rate: %d, aggregate demand: %.3f, drop prob: %.3f\n",
 264            ident->common.rate, aggdemand, dropprob);
 265
 266     return dropprob;
 267 }
 268
 269 /**
 270  * Given current estimates of local rate (weight) and remote rates (weights)
 271  * use GRD or FPS to calculate a new local limit.
 272  */
 273 static void allocate(limiter_t *limiter, identity_t *ident) {
 274     /* Represents aggregate rate for GRD and aggregate weight for FPS. */
 275     double comm_val = 0;
 276
 277     /* Read comm_val from comm layer. */
 278     if (limiter->policy == POLICY_FPS) {
 279         read_comm(&ident->comm, &comm_val,
 280                 ident->total_weight / (double) (ident->comm.remote_node_count + 1));
 281     } else {
 282         read_comm(&ident->comm, &comm_val,
 283                 (double) (ident->limit / (double) (ident->comm.remote_node_count + 1)));
 284     }
 285     printlog(LOG_DEBUG, "%.3f Aggregate weight/rate (FPS/GRD)\n", comm_val);
 286
 287     /* Experimental printing. */
 288     printlog(LOG_DEBUG, "%.3f \t Kbps used rate. ID:%d\n",
 289              (double) ident->common.rate / (double) 128, ident->id);
 290     ident->avg_bytes += ident->common.rate;
 291
 292     if (limiter->policy == POLICY_FPS) {
 293         ident->locallimit = allocate_fps(ident, comm_val);
 294         ident->last_localweight = ident->localweight;
 295
 296         /* Update other limiters with our weight by writing to comm layer. */
 297         write_local_value(&ident->comm, ident->localweight);
 298     } else {
 299         ident->locallimit = 0; /* Unused with GRD. */
 300         ident->last_drop_prob = ident->drop_prob;
 301         ident->drop_prob = allocate_grd(ident, comm_val);
 302
 303         /* Update other limiters with our rate by writing to comm layer. */
 304         write_local_value(&ident->comm, ident->common.rate);
 305     }
 306
 307     /* Update identity state. */
 308     ident->common.last_rate = ident->common.rate;
 309 }
 310
 311 /**
 312  * Traces all of the parent pointers of a leaf all the way to the root in
 313  * order to find the maximum drop probability in the chain.
 314  */
 315 static double find_leaf_drop_prob(leaf_t *leaf) {
 316     identity_t *current = leaf->parent;
 317     double result = 0;
 318
 319     assert(current);
 320
 321     while (current != NULL) {
 322         if (current->drop_prob > result) {
 323             result = current->drop_prob;
 324         }
 325         current = current->parent;
 326     }
 327
 328     return result;
 329 }
 330
 331 /**
 332  * This is called once per estimate interval to enforce the rate that allocate
 333  * has decided upon.  It makes calls to tc using system().
 334  */
 335 static void enforce(limiter_t *limiter, identity_t *ident) {
 336     char cmd[CMD_BUFFER_SIZE];
 337     int ret = 0;
 338     int i = 0;
 339
 340     switch (limiter->policy) {
 341         case POLICY_FPS:
 342
 343             /* TC treats limits of 0 (8bit) as unlimited, which causes the
 344              * entire rate limiting system to become unpredictable.  In
 345              * reality, we also don't want any limiter to be able to set its
 346              * limit so low that it chokes all of the flows to the point that
 347              * they can't increase.  Thus, when we're setting a low limit, we
 348              * make sure that it isn't too low by using the
 349              * FLOW_START_THRESHOLD. */
 350
 351             if (ident->locallimit < FLOW_START_THRESHOLD) {
 352                 ident->locallimit = FLOW_START_THRESHOLD * 2;
 353             }
 354
 355             /* Do not allow the node to set a limit higher than its
 356              * administratively assigned upper limit (bwcap). */
 357             if (limiter->nodelimit != 0 && ident->locallimit > limiter->nodelimit) {
 358                 ident->locallimit = limiter->nodelimit;
 359             }
 360
 361             printf("FPS: Setting local limit to %d\n", ident->locallimit);
 362             printlog(LOG_DEBUG, "%d Limit ID:%d\n", ident->locallimit, ident->id);
 363
 364             snprintf(cmd, CMD_BUFFER_SIZE,
 365                      "/sbin/tc class change dev eth0 parent 1:%x classid 1:%x htb rate 8bit ceil %dbps quantum 1600",
 366                      ident->htb_parent, ident->htb_node, ident->locallimit);
 367
 368             ret = system(cmd);
 369
 370             if (ret) {
 371                 /* FIXME: call failed.  What to do? */
 372             }
 373             break;
 374
 375         case POLICY_GRD:
 376             for (i = 0; i < ident->leaf_count; ++i) {
 377                 if (ident->drop_prob >= ident->leaves[i]->drop_prob) {
 378                     /* The new drop probability for this identity is greater
 379                      * than or equal to the leaf's current drop probability.
 380                      * We can safely use the larger value at this leaf
 381                      * immediately. */
 382                     ident->leaves[i]->drop_prob = ident->drop_prob;
 383                 } else if (ident->last_drop_prob < ident->leaves[i]->drop_prob) {
 384                     /* The old drop probability for this identity is less than
 385                      * the leaf's current drop probability.  This means that
 386                      * this identity couldn't have been the limiting ident,
 387                      * so nothing needs to be done because the old limiting
 388                      * ident is still the limiting factor. */
 389
 390                     /* Intentionally blank. */
 391                 } else {
 392                     /* If neither of the above are true, then...
 393                      * 1) The new drop probability for the identity is less
 394                      * than what it previously was, and
 395                      * 2) This ident may have had the maximum drop probability
 396                      * of all idents limiting this leaf, and therefore we need
 397                      * to follow the leaf's parents up to the root to find the
 398                      * new leaf drop probability safely. */
 399                     ident->leaves[i]->drop_prob =
 400                             find_leaf_drop_prob(ident->leaves[i]);
 401                 }
 402
 403                 /* Make the call to tc. */
 404 #ifdef DELAY40MS
 405                 snprintf(cmd, CMD_BUFFER_SIZE,
 406                          "/sbin/tc qdisc change dev eth0 parent 1:1%x handle 1%x netem loss %.4f delay 40ms",
 407                          ident->leaves[i]->xid, ident->leaves[i]->xid,
 408                          (100 * ident->leaves[i]->drop_prob));
 409 #else
 410                 snprintf(cmd, CMD_BUFFER_SIZE,
 411                          "/sbin/tc qdisc change dev eth0 parent 1:1%x handle 1%x netem loss %.4f delay 0ms",
 412                          ident->leaves[i]->xid, ident->leaves[i]->xid,
 413                          (100 * ident->leaves[i]->drop_prob));
 414 #endif
 415                 ret = system(cmd);
 416
 417                 if (ret) {
 418                     /* FIXME: call failed.  What to do? */
 419                 }
 420             }
 421
 422             break;
 423
 424         default:
 425             printlog(LOG_CRITICAL, "DRL enforce: unknown policy %d\n",limiter->policy);
 426             break;
 427     }
 428
 429     return;
 430 }
 431
 432 /**
 433  * This function is periodically called to clean the stable instance's flow
 434  * accounting tables for each identity.
 435  */
 436 static void clean(drl_instance_t *instance) {
 437     identity_t *ident = NULL;
 438
 439     map_reset_iterate(instance->ident_map);
 440     while ((ident = map_next(instance->ident_map)) != NULL) {
 441         pthread_mutex_lock(&ident->table_mutex);
 442
 443         ident->table_cleanup_function(ident->table);
 444
 445         pthread_mutex_unlock(&ident->table_mutex);
 446     }
 447
 448     /* Periodically flush the log file. */
 449     flushlog();
 450 }
 451
 452 static void print_averages(drl_instance_t *instance, int print_interval) {
 453     identity_t *ident = NULL;
 454
 455     map_reset_iterate(instance->ident_map);
 456     while ((ident = map_next(instance->ident_map)) != NULL) {
 457         ident->avg_bytes /= (double) print_interval;
 458         //printf("avg_bytes = %f, print_interval = %d\n", ident->avg_bytes, print_interval);
 459         printlog(LOG_DEBUG, "%.3f \t Avg rate. ID:%d\n",
 460                  ident->avg_bytes / 128, ident->id);
 461         //printf("%.3f \t Avg rate. ID:%d\n",
 462         //         ident->avg_bytes / 128, ident->id);
 463         ident->avg_bytes = 0;
 464     }
 465 }
 466
 467 /** Thread function to handle local rate estimation.
 468  *
 469  * None of our simple hashmap functions are thread safe, so we lock the limiter
 470  * with an rwlock to prevent another thread from attempting to modify the set
 471  * of identities.
 472  *
 473  * Each identity also has a private lock for its table.  This gets locked by
 474  * table-modifying functions such as estimate and clean.
 475  */
 476 void handle_estimation(void *arg) {
 477     limiter_t *limiter = (limiter_t *) arg;
 478     identity_t *ident = NULL;
 479     int clean_timer, clean_wait_intervals;
 480     useconds_t sleep_time = limiter->estintms * 1000;
 481     uint32_t cal_slot = 0;
 482     int print_interval = 1000 / (limiter->estintms);
 483
 484     sigset_t signal_mask;
 485
 486     sigemptyset(&signal_mask);
 487     sigaddset(&signal_mask, SIGHUP);
 488     pthread_sigmask(SIG_BLOCK, &signal_mask, NULL);
 489
 490     /* Determine the number of intervals we should wait before hitting the
 491      * specified clean interval. (Converts seconds -> intervals). */
 492     clean_wait_intervals = IDENT_CLEAN_INTERVAL * (1000.0 / limiter->estintms);
 493     clean_timer = clean_wait_intervals;
 494
 495     while (true) {
 496         /* Sleep according to the delay of the estimate interval. */
 497         usleep(sleep_time);
 498
 499         /* Grab the limiter lock for reading.  This prevents identities from
 500          * disappearing beneath our feet. */
 501         pthread_rwlock_rdlock(&limiter->limiter_lock);
 502
 503         cal_slot = limiter->stable_instance.cal_slot & SCHEDMASK;
 504
 505         /* Service all the identities that are scheduled to run during this
 506          * tick. */
 507         while (!TAILQ_EMPTY(limiter->stable_instance.cal + cal_slot)) {
 508             ident = TAILQ_FIRST(limiter->stable_instance.cal + cal_slot);
 509             TAILQ_REMOVE(limiter->stable_instance.cal + cal_slot, ident, calendar);
 510
 511             /* Update the ident's flow accouting table with the latest info. */
 512             estimate(ident);
 513
 514             /* Determine its share of the rate allocation. */
 515             allocate(limiter, ident);
 516
 517             /* Make tc calls to enforce the rate we decided upon. */
 518             enforce(limiter, ident);
 519
 520             /* Tell the comm library to propagate this identity's result for
 521              * this interval.*/
 522             send_update(&ident->comm, ident->id);
 523
 524             /* Add ident back to the queue at a future time slot. */
 525             TAILQ_INSERT_TAIL(limiter->stable_instance.cal +
 526                               ((cal_slot + ident->intervals) & SCHEDMASK),
 527                               ident, calendar);
 528         }
 529
 530         print_interval--;
 531         if (loglevel() == LOG_DEBUG && print_interval <= 0) {
 532             print_interval = 1000 / (limiter->estintms);
 533             print_averages(&limiter->stable_instance, print_interval);
 534         }
 535
 536         /* Check if enough intervals have passed for cleaning. */
 537         if (clean_timer <= 0) {
 538             clean(&limiter->stable_instance);
 539             clean_timer = clean_wait_intervals;
 540         } else {
 541             clean_timer--;
 542         }
 543
 544         limiter->stable_instance.cal_slot += 1;
 545
 546         pthread_rwlock_unlock(&limiter->limiter_lock);
 547     }
 548 }