drl/ulogd_DRL.c

   1 /* See the DRL-LICENSE file for this file's software license. */
   2
   3 /*
   4  * ulogd output target for DRL: GRD and FPS
   5  *
   6  * Ken Yocum <kyocum@cs.ucsd.edu>
   7  *
   8  * Original shell of this code from ulogd_NETFLOW
   9  * Like that code, we keep track of per-slice data rates
  10  * out of this slice.  Thus we are rate limiting particular slices
  11  * across multiple boxes, ensuring that their outbound rate does not
  12  * exceed some fixed limit.
  13  *
  14  * Fabric:  static mesh
  15  *
  16  * Enforcer: linux drop percentage.
  17  *
  18  * This file reads packets from the netlink socket.  It updates all
  19  * the hashmaps which track how much data has arrived per flow.
  20  * It starts two threads for this limiter.
  21  * One thread handles periodic estimation.
  22  * The other thread handles communication with other limiters.
  23  *
  24  *
  25  * Code map
  26  *
  27  * ulogd_DRL: attach to netlink socket, accept packets.  replaces ratelimit.cc
  28  * util.c: generic hashing functions, flow comparisons, sundry items.
  29  * gossip.c: Recv gossip, send gossip.
  30  * peer_comm.c: Thread to listen for updates from other limiters.
  31  * estimate.c: Thread to calculate the local limits.
  32  *
  33  *
  34  * Ken Yocum <kyocum@cs.ucsd.edu>
  35  * 2007
  36  *
  37  * Some code appropriated from ulogd_NETFLOW:
  38  *
  39  * Mark Huang <mlhuang@cs.princeton.edu>
  40  * Copyright (C) 2004-2005 The Trustees of Princeton University
  41  *
  42  * Based on admindump.pl by Mic Bowman and Paul Brett
  43  * Copyright (c) 2002 Intel Corporation
  44  *
  45  */
  46
  47 /* Enable GNU glibc extensions */
  48 #define _GNU_SOURCE
  49
  50 #include <stdio.h>
  51 #include <stdlib.h>
  52
  53 /* va_start() and friends */
  54 #include <stdarg.h>
  55
  56 /* ispunct() */
  57 #include <ctype.h>
  58
  59 /* strstr() and friends */
  60 #include <string.h>
  61
  62 /* dirname() and basename() */
  63 #include <libgen.h>
  64
  65 /* fork() and wait() */
  66 #include <sys/types.h>
  67 #include <unistd.h>
  68 #include <sys/wait.h>
  69
  70 /* errno and assert() */
  71 #include <errno.h>
  72 #include <assert.h>
  73
  74 /* getopt_long() */
  75 #include <getopt.h>
  76
  77 /* time() and friends */
  78 #include <time.h>
  79 #include <sys/time.h>
  80
  81 /* inet_aton() */
  82 #include <sys/socket.h>
  83 #include <netinet/in.h>
  84 #include <arpa/inet.h>
  85
  86 /* ICMP definitions */
  87 #include <netinet/ip.h>
  88 #include <netinet/ip_icmp.h>
  89
  90 /* stat() */
  91 #include <sys/stat.h>
  92
  93 /* pthread_create() */
  94 #include <pthread.h>
  95
  96 /* flock() */
  97 #include <sys/file.h>
  98
  99 /* Signal definitions - so that we can catch SIGHUP and update config. */
 100 #include <signal.h>
 101
 102 #include <ulogd/ulogd.h>
 103 #include <ulogd/conffile.h>
 104
 105 /* Perhaps useful for files within vservers? */
 106 #if !defined(STANDALONE) && HAVE_LIBPROPER
 107 #include <proper/prop.h>
 108 #endif
 109
 110 /*
 111  * Jenkins hash support
 112  * lives in raterouter.h
 113  */
 114
 115 /* DRL specifics */
 116 #include "raterouter.h"
 117 #include "util.h"
 118 #include "ratetypes.h" /* needs util and pthread.h */
 119 #include "calendar.h"
 120 #include "logging.h"
 121
 122 /*
 123  * /etc/ulogd.conf configuration options
 124  * Add the config options for DRL.
 125  */
 126
 127 static config_entry_t partition = {
 128     .next = NULL,
 129     .key = "partition_set",
 130     .type = CONFIG_TYPE_INT,
 131     .options = CONFIG_OPT_NONE,
 132     .u = { .value = 0xfffffff },
 133 };
 134
 135 static config_entry_t netem_slice = {
 136     .next = &partition,
 137     .key = "netem_slice",
 138     .type = CONFIG_TYPE_STRING,
 139     .options = CONFIG_OPT_NONE,
 140     .u = { .string = "ALL" },
 141 };
 142
 143 static config_entry_t netem_loss = {
 144     .next = &netem_slice,
 145     .key = "netem_loss",
 146     .type = CONFIG_TYPE_INT,
 147     .options = CONFIG_OPT_NONE,
 148     .u = { .value = 0 },
 149 };
 150
 151 static config_entry_t netem_delay = {
 152     .next = &netem_loss,
 153     .key = "netem_delay",
 154     .type = CONFIG_TYPE_INT,
 155     .options = CONFIG_OPT_NONE,
 156     .u = { .value = 0 },
 157 };
 158
 159 static config_entry_t drl_configfile = {
 160     .next = &netem_delay,
 161     .key = "drl_configfile",
 162     .type = CONFIG_TYPE_STRING,
 163     .options = CONFIG_OPT_MANDATORY,
 164     .u = { .string = "drl.xml" },
 165 };
 166
 167 /** The administrative bandwidth limit (mbps) for the local node.  The node
 168  * will not set a limit higher than this, even when distributed capacity is
 169  * available.  Set to 0 for no limit. */
 170 static config_entry_t nodelimit = {
 171     .next = &drl_configfile,
 172     .key = "nodelimit",
 173     .type = CONFIG_TYPE_INT,
 174     .options = CONFIG_OPT_MANDATORY,
 175     .u = { .value = 0 },
 176 };
 177
 178 /** Determines the verbosity of logging. */
 179 static config_entry_t drl_loglevel = {
 180     .next = &nodelimit,
 181     .key = "drl_loglevel",
 182     .type = CONFIG_TYPE_INT,
 183     .options = CONFIG_OPT_MANDATORY,
 184     .u = { .value = LOG_WARN },
 185 };
 186
 187 /** The path of the logfile. */
 188 static config_entry_t drl_logfile = {
 189     .next = &drl_loglevel,
 190     .key = "drl_logfile",
 191     .type = CONFIG_TYPE_STRING,
 192     .options = CONFIG_OPT_MANDATORY,
 193     .u = { .string = "drl_logfile.log" },
 194 };
 195
 196 /** The choice of DRL protocol. */
 197 static config_entry_t policy = {
 198     .next = &drl_logfile,
 199     .key = "policy",
 200     .type = CONFIG_TYPE_STRING,
 201     .options = CONFIG_OPT_MANDATORY,
 202     .u = { .string = "GRD" },
 203 };
 204
 205 /** The estimate interval, in milliseconds. */
 206 static config_entry_t estintms = {
 207     .next = &policy,
 208     .key = "estintms",
 209     .type = CONFIG_TYPE_INT,
 210     .options = CONFIG_OPT_MANDATORY,
 211     .u = { .value = 100 },
 212 };
 213
 214 #define config_entries (&estintms)
 215
 216 /*
 217  * Debug functionality
 218  */
 219
 220 #ifdef DMALLOC
 221 #include <dmalloc.h>
 222 #endif
 223
 224 #define NIPQUAD(addr) \
 225     ((unsigned char *)&addr)[0], \
 226     ((unsigned char *)&addr)[1], \
 227     ((unsigned char *)&addr)[2], \
 228     ((unsigned char *)&addr)[3]
 229
 230 #define IPQUAD(addr) \
 231     ((unsigned char *)&addr)[3], \
 232     ((unsigned char *)&addr)[2], \
 233     ((unsigned char *)&addr)[1], \
 234     ((unsigned char *)&addr)[0]
 235
 236
 237
 238 /* Salt for the hash functions */
 239 static int salt;
 240
 241 /*
 242  * Hash slice name lookups on context ID.
 243  */
 244
 245 /* Special context IDs */
 246 #define UNKNOWN_XID -1
 247 #define ROOT_XID 0
 248
 249 enum {
 250     CONNECTION_REFUSED_XID = 65536, /* MAX_S_CONTEXT + 1 */
 251     ICMP_ECHOREPLY_XID,
 252     ICMP_UNREACH_XID,
 253 };
 254
 255
 256 /* globals */
 257 pthread_t estimate_thread;
 258 pthread_t signal_thread;
 259 pthread_t comm_thread;
 260 uint32_t local_ip = 0;
 261 limiter_t limiter;
 262 extern FILE *logfile;
 263 extern uint8_t system_loglevel;
 264 extern uint8_t do_enforcement;
 265
 266 /* From peer_comm.c - used to simulate partition. */
 267 extern int do_partition;
 268 extern int partition_set;
 269
 270 /* functions */
 271
 272 static inline uint32_t
 273 hash_flow(uint8_t protocol, uint32_t src_ip, uint16_t src_port, uint32_t dst_ip, uint16_t dst_port, uint32_t hash_max)
 274 {
 275     unsigned char mybytes[FLOWKEYSIZE];
 276     mybytes[0] = protocol;
 277     *(uint32_t*)(&(mybytes[1])) = src_ip;
 278     *(uint32_t*)(&(mybytes[5])) = dst_ip;
 279     *(uint32_t*)(&(mybytes[9])) = (src_port << 16) | dst_port;
 280     return jhash(mybytes,FLOWKEYSIZE,salt) & (hash_max - 1);
 281 }
 282
 283 uint32_t sampled_hasher(const key_flow *key) {
 284     /* Last arg is UINT_MAX because sampled flow keeps track of its own capacity. */
 285     return hash_flow(key->protocol, key->source_ip, key->source_port, key->dest_ip, key->dest_port, UINT_MAX);
 286 }
 287
 288 uint32_t standard_hasher(const key_flow *key) {
 289     return hash_flow(key->protocol, key->source_ip, key->source_port, key->dest_ip, key->dest_port, STD_FLOW_HASH_SIZE);
 290 }
 291
 292 uint32_t multiple_hasher(const key_flow *key) {
 293     return hash_flow(key->protocol, key->source_ip, key->source_port, key->dest_ip, key->dest_port, MUL_FLOW_HASH_SIZE);
 294 }
 295
 296 struct intr_id {
 297     char* name;
 298     ulog_iret_t *res;
 299 };
 300
 301 /* Interesting keys */
 302 enum {
 303     OOB_TIME_SEC = 0,
 304     OOB_MARK,
 305     IP_SADDR,
 306     IP_DADDR,
 307     IP_TOTLEN,
 308     IP_PROTOCOL,
 309     TCP_SPORT,
 310     TCP_DPORT,
 311     TCP_ACK,
 312     TCP_FIN,
 313     TCP_SYN,
 314     TCP_RST,
 315     UDP_SPORT,
 316     UDP_DPORT,
 317     ICMP_TYPE,
 318     ICMP_CODE,
 319     GRE_FLAG_KEY,
 320     GRE_VERSION,
 321     GRE_KEY,
 322     PPTP_CALLID,
 323 };
 324
 325 #define INTR_IDS (sizeof(intr_ids)/sizeof(intr_ids[0]))
 326 static struct intr_id intr_ids[] = {
 327     [OOB_TIME_SEC] = { "oob.time.sec", 0 },
 328     [OOB_MARK] = { "oob.mark", 0 },
 329     [IP_SADDR] = { "ip.saddr", 0 },
 330     [IP_DADDR] = { "ip.daddr", 0 },
 331     [IP_TOTLEN] = { "ip.totlen", 0 },
 332     [IP_PROTOCOL] = { "ip.protocol", 0 },
 333     [TCP_SPORT] = { "tcp.sport", 0 },
 334     [TCP_DPORT] { "tcp.dport", 0 },
 335     [TCP_ACK] = { "tcp.ack", 0 },
 336     [TCP_FIN] = { "tcp.fin", 0 },
 337     [TCP_SYN] = { "tcp.syn", 0 },
 338     [TCP_RST] = { "tcp.rst", 0 },
 339     [UDP_SPORT] = { "udp.sport", 0 },
 340     [UDP_DPORT] = { "udp.dport", 0 },
 341     [ICMP_TYPE] = { "icmp.type", 0 },
 342     [ICMP_CODE] = { "icmp.code", 0 },
 343     [GRE_FLAG_KEY] = { "gre.flag.key", 0 },
 344     [GRE_VERSION] = { "gre.version", 0 },
 345     [GRE_KEY] = { "gre.key", 0 },
 346     [PPTP_CALLID] = { "pptp.callid", 0 },
 347 };
 348
 349 #define GET_VALUE(x) intr_ids[x].res->value
 350
 351 #define DATE(t) ((t) / (24*60*60) * (24*60*60))
 352
 353 static int _output_drl(ulog_iret_t *res)
 354 {
 355     int xid;
 356     uint32_t src_ip, dst_ip;
 357     uint16_t src_port, dst_port;
 358     uint8_t protocol;
 359
 360     key_flow key;
 361     identity_t *ident;
 362     leaf_t *leaf;
 363
 364     protocol = GET_VALUE(IP_PROTOCOL).ui8;
 365     src_ip = GET_VALUE(IP_SADDR).ui32;
 366     dst_ip = GET_VALUE(IP_DADDR).ui32;
 367     xid = GET_VALUE(OOB_MARK).ui32;
 368
 369     switch (protocol) {
 370
 371         case IPPROTO_TCP:
 372             src_port = GET_VALUE(TCP_SPORT).ui16;
 373             dst_port = GET_VALUE(TCP_DPORT).ui16;
 374             break;
 375
 376         case IPPROTO_UDP:
 377             /* netflow had an issue with many udp flows and set
 378              * src_port=0 to handle it.  We don't.
 379              */
 380             src_port = GET_VALUE(UDP_SPORT).ui16;
 381
 382             /*
 383              * traceroutes create a large number of flows in the db
 384              * this is a quick hack to catch the most common form
 385              * of traceroute (basically we're mapping any UDP packet
 386              * in the 33435-33524 range to the "trace" port, 33524 is
 387              * 3 packets * nhops (30).
 388              */
 389             dst_port = GET_VALUE(UDP_DPORT).ui16;
 390             if (dst_port >= 33435 && dst_port <= 33524)
 391                 dst_port = 33435;
 392             break;
 393
 394         case IPPROTO_ICMP:
 395             src_port = GET_VALUE(ICMP_TYPE).ui8;
 396             dst_port = GET_VALUE(ICMP_CODE).ui8;
 397
 398             /*
 399              * We special case some of the ICMP traffic that the kernel
 400              * always generates. Since this is attributed to root, it
 401              * creates significant "noise" in the output. We want to be
 402              * able to quickly see that root is generating traffic.
 403              */
 404             if (xid == ROOT_XID) {
 405                 if (src_port == ICMP_ECHOREPLY)
 406                     xid = ICMP_ECHOREPLY_XID;
 407                 else if (src_port == ICMP_UNREACH)
 408                     xid = ICMP_UNREACH_XID;
 409             }
 410             break;
 411
 412         case IPPROTO_GRE:
 413             if (GET_VALUE(GRE_FLAG_KEY).b) {
 414                 if (GET_VALUE(GRE_VERSION).ui8 == 1) {
 415                     /* Get PPTP call ID */
 416                     src_port = GET_VALUE(PPTP_CALLID).ui16;
 417                 } else {
 418                     /* XXX Truncate GRE keys to 16 bits */
 419                     src_port = (uint16_t) GET_VALUE(GRE_KEY).ui32;
 420                 }
 421             } else {
 422                 /* No key available */
 423                 src_port = 0;
 424             }
 425             dst_port = 0;
 426             break;
 427
 428         default:
 429             /* This is the default key for packets from unsupported protocols */
 430             src_port = 0;
 431             dst_port = 0;
 432             break;
 433     }
 434
 435     key.protocol = protocol;
 436     key.source_ip = src_ip;
 437     key.dest_ip = dst_ip;
 438     key.source_port = src_port;
 439     key.dest_port = dst_port;
 440     key.packet_size = GET_VALUE(IP_TOTLEN).ui16;
 441     key.packet_time = (time_t) GET_VALUE(OOB_TIME_SEC).ui32;
 442
 443     pthread_rwlock_rdlock(&limiter.limiter_lock); /* CLUNK! */
 444
 445     leaf = (leaf_t *) map_search(limiter.stable_instance.leaf_map, &xid, sizeof(xid));
 446
 447     /* Even if the packet doesn't match any specific xid, it should still
 448      * count in the machine-type tables.  This catches root (xid == 0) and
 449      * unclassified (xid = fff) packets, which don't have map entries. */
 450     if (leaf == NULL) {
 451         ident = limiter.stable_instance.last_machine;
 452     } else {
 453         ident = leaf->parent;
 454     }
 455
 456     while (ident) {
 457         pthread_mutex_lock(&ident->table_mutex);
 458
 459         /* Update the identity's table. */
 460         ident->table_sample_function(ident->table, &key);
 461
 462 #ifdef SHADOW_ACCTING
 463
 464         /* Update the shadow perfect copy of the accounting table. */
 465         standard_table_sample((standard_flow_table) ident->shadow_table, &key);
 466
 467 #endif
 468
 469         pthread_mutex_unlock(&ident->table_mutex);
 470
 471         ident = ident->parent;
 472     }
 473
 474     pthread_rwlock_unlock(&limiter.limiter_lock); /* CLINK! */
 475
 476     return 0;
 477 }
 478
 479 /* get all key id's for the keys we are intrested in */
 480 static int get_ids(void)
 481 {
 482     int i;
 483     struct intr_id *cur_id;
 484
 485     for (i = 0; i < INTR_IDS; i++) {
 486         cur_id = &intr_ids[i];
 487         cur_id->res = keyh_getres(keyh_getid(cur_id->name));
 488         if (!cur_id->res) {
 489             ulogd_log(ULOGD_ERROR,
 490                     "Cannot resolve keyhash id for %s\n",
 491                     cur_id->name);
 492             return 1;
 493         }
 494     }
 495     return 0;
 496 }
 497
 498 static void free_identity(identity_t *ident) {
 499     if (ident) {
 500         free_comm(&ident->comm);
 501
 502         if (ident->table) {
 503             ident->table_destroy_function(ident->table);
 504         }
 505
 506         if (ident->loop_action) {
 507             ident->loop_action->valid = 0;
 508         }
 509
 510         if (ident->comm_action) {
 511             ident->comm_action->valid = 0;
 512         }
 513
 514         pthread_mutex_destroy(&ident->table_mutex);
 515
 516         free(ident);
 517     }
 518 }
 519
 520 static void free_identity_map(map_handle map) {
 521     identity_t *tofree = NULL;
 522
 523     map_reset_iterate(map);
 524     while ((tofree = (identity_t *) map_next(map))) {
 525         free_identity(tofree);
 526     }
 527
 528     free_map(map, 0);
 529 }
 530
 531 static void free_instance(drl_instance_t *instance) {
 532     if (instance->leaves)
 533         free(instance->leaves);
 534     if (instance->leaf_map)
 535         free_map(instance->leaf_map, 0);
 536     if (instance->ident_map)
 537         free_identity_map(instance->ident_map);
 538     if (instance->machines)
 539         free(instance->machines);
 540     if (instance->sets)
 541         free(instance->sets);
 542
 543     /* FIXME: Drain the calendar first and free all the entries. */
 544     if (instance->cal) {
 545         free(instance->cal);
 546     }
 547
 548     memset(instance, 0, sizeof(drl_instance_t));
 549 }
 550
 551 static void free_failed_config(parsed_configs configs, drl_instance_t *instance) {
 552     /* Free configs. */
 553     if (configs.machines)
 554         free_ident_list(configs.machines);
 555     if (configs.sets)
 556         free_ident_list(configs.sets);
 557
 558     /* Free instance. */
 559     if (instance)
 560         free_instance(instance);
 561 }
 562
 563 static identity_t *new_identity(ident_config *config) {
 564     identity_t *ident = malloc(sizeof(identity_t));
 565     remote_node_t *comm_nodes = malloc(sizeof(remote_node_t)*config->peer_count);
 566     ident_peer *peer = config->peers;
 567     int peer_slot = 0;
 568
 569     if (ident == NULL) {
 570         return NULL;
 571     }
 572
 573     if (comm_nodes == NULL) {
 574         free(ident);
 575         return NULL;
 576     }
 577
 578     memset(ident, 0, sizeof(identity_t));
 579     memset(comm_nodes, 0, config->peer_count * sizeof(remote_node_t));
 580
 581     ident->id = config->id;
 582     ident->limit = (uint32_t) (((double) config->limit * 1000.0) / 8.0);
 583     ident->fixed_ewma_weight = config->fixed_ewma_weight;
 584     ident->communication_intervals = config->communication_intervals;
 585     ident->mainloop_intervals = config->mainloop_intervals;
 586     ident->ewma_weight = pow(ident->fixed_ewma_weight,
 587                              (limiter.estintms/1000.0) * config->mainloop_intervals);
 588     ident->parent = NULL;
 589     ident->independent = config->independent;
 590
 591     pthread_mutex_init(&ident->table_mutex, NULL);
 592     switch (config->accounting) {
 593         case ACT_STANDARD:
 594             ident->table =
 595                 standard_table_create(standard_hasher, &ident->common);
 596
 597             /* Ugly function pointer casting.  Makes things sufficiently
 598              * generic, though. */
 599             ident->table_sample_function =
 600                 (int (*)(void *, const key_flow *)) standard_table_sample;
 601             ident->table_cleanup_function =
 602                 (int (*)(void *)) standard_table_cleanup;
 603             ident->table_update_function =
 604                 (void (*)(void *, struct timeval, double)) standard_table_update_flows;
 605             ident->table_destroy_function =
 606                 (void (*)(void *)) standard_table_destroy;
 607             break;
 608
 609         case ACT_MULTIPLE:
 610             ident->table =
 611                 multiple_table_create(multiple_hasher, MUL_INTERVAL_COUNT, &ident->common);
 612
 613             ident->table_sample_function =
 614                 (int (*)(void *, const key_flow *)) multiple_table_sample;
 615             ident->table_cleanup_function =
 616                 (int (*)(void *)) multiple_table_cleanup;
 617             ident->table_update_function =
 618                 (void (*)(void *, struct timeval, double)) multiple_table_update_flows;
 619             ident->table_destroy_function =
 620                 (void (*)(void *)) multiple_table_destroy;
 621             break;
 622
 623         case ACT_SAMPLEHOLD:
 624             ident->table = sampled_table_create(sampled_hasher,
 625                     ident->limit * IDENT_CLEAN_INTERVAL,
 626                     SAMPLEHOLD_PERCENTAGE, SAMPLEHOLD_OVERFACTOR, &ident->common);
 627
 628             ident->table_sample_function =
 629                 (int (*)(void *, const key_flow *)) sampled_table_sample;
 630             ident->table_cleanup_function =
 631                 (int (*)(void *)) sampled_table_cleanup;
 632             ident->table_update_function =
 633                 (void (*)(void *, struct timeval, double)) sampled_table_update_flows;
 634             ident->table_destroy_function =
 635                 (void (*)(void *)) sampled_table_destroy;
 636             break;
 637
 638         case ACT_SIMPLE:
 639             ident->table = simple_table_create(&ident->common);
 640
 641             ident->table_sample_function =
 642                 (int (*)(void *, const key_flow *)) simple_table_sample;
 643             ident->table_cleanup_function =
 644                 (int (*)(void *)) simple_table_cleanup;
 645             ident->table_update_function =
 646                 (void (*)(void *, struct timeval, double)) simple_table_update_flows;
 647             ident->table_destroy_function =
 648                 (void (*)(void *)) simple_table_destroy;
 649             break;
 650     }
 651
 652 #ifdef SHADOW_ACCTING
 653
 654     ident->shadow_table = standard_table_create(standard_hasher, &ident->shadow_common);
 655
 656     if (ident->shadow_table == NULL) {
 657         ident->table_destroy_function(ident->table);
 658         free(ident);
 659         return NULL;
 660     }
 661
 662 #endif
 663
 664     /* Make sure the table was allocated. */
 665     if (ident->table == NULL) {
 666         free(ident);
 667         return NULL;
 668     }
 669
 670     while (peer) {
 671         comm_nodes[peer_slot].addr = peer->ip;
 672         comm_nodes[peer_slot].port = htons(LIMITER_LISTEN_PORT);
 673         peer = peer->next;
 674         peer_slot += 1;
 675     }
 676
 677     if (new_comm(&ident->comm, config, comm_nodes)) {
 678         printlog(LOG_CRITICAL, "Failed to create communication structure.\n");
 679         return NULL;
 680     }
 681
 682     ident->comm.remote_nodes = comm_nodes;
 683
 684     return ident;
 685 }
 686
 687 /* Determines the validity of the parameters of one ident_config.
 688  *
 689  * 0 valid
 690  * 1 invalid
 691  */
 692 static int validate_config(ident_config *config) {
 693     /* Limit must be a positive integer. */
 694     if (config->limit < 1) {
 695         return 1;
 696     }
 697
 698     /* Commfabric must be a valid choice (COMM_MESH or COMM_GOSSIP). */
 699     if (config->commfabric != COMM_MESH &&
 700             config->commfabric != COMM_GOSSIP) {
 701         return 1;
 702     }
 703
 704     /* If commfabric is COMM_GOSSIP, this must be a positive integer. */
 705     if (config->commfabric == COMM_GOSSIP && config->branch < 1) {
 706         return 1;
 707     }
 708
 709     /* Accounting must be a valid choice (ACT_STANDARD, ACT_SAMPLEHOLD,
 710      * ACT_SIMPLE, ACT_MULTIPLE). */
 711     if (config->accounting != ACT_STANDARD &&
 712             config->accounting != ACT_SAMPLEHOLD &&
 713             config->accounting != ACT_SIMPLE &&
 714             config->accounting != ACT_MULTIPLE) {
 715         return 1;
 716     }
 717
 718     /* Ewma weight must be greater than or equal to zero. */
 719     if (config->fixed_ewma_weight < 0) {
 720         return 1;
 721     }
 722
 723     /* Note: Parsing stage requires that each ident has at least one peer. */
 724     return 0;
 725 }
 726
 727 /* 0 success
 728  * non-zero failure
 729  */
 730 static int validate_configs(parsed_configs configs, drl_instance_t *instance) {
 731
 732     ident_config *mlist = configs.machines;
 733     ident_config *slist = configs.sets;
 734     ident_config *tmp = NULL;
 735     int i = 0;
 736
 737     while (mlist) {
 738         /* ID must be non-zero and unique. */
 739         /* This is ugly and hackish, but this function will be called rarely.
 740          * I'm tired of trying to be clever. */
 741         if (mlist->id < 0) {
 742             printlog(LOG_CRITICAL, "Negative ident id: %d (%x) ?\n", mlist->id, mlist->id);
 743             return EINVAL;
 744         }
 745         tmp = mlist->next;
 746         while (tmp) {
 747             if (mlist->id == tmp->id) {
 748                 printlog(LOG_CRITICAL, "Duplicate ident id: %d (%x)\n", mlist->id, mlist->id);
 749                 return EINVAL;
 750             }
 751             tmp = tmp->next;
 752         }
 753         tmp = configs.sets;
 754         while (tmp) {
 755             if (mlist->id == tmp->id) {
 756                 printlog(LOG_CRITICAL, "Duplicate ident id: %d (%x)\n", mlist->id, mlist->id);
 757                 return EINVAL;
 758             }
 759             tmp = tmp->next;
 760         }
 761
 762         if (validate_config(mlist)) {
 763             printlog(LOG_CRITICAL, "Invalid ident parameters for id: %d (%x)\n", mlist->id, mlist->id);
 764             return EINVAL;
 765         }
 766
 767         if (mlist->independent) {
 768             printlog(LOG_CRITICAL, "Makes no sense to have independent machine node - setting independent to false.\n");
 769             mlist->independent = 0;
 770         }
 771
 772         mlist = mlist->next;
 773     }
 774
 775     instance->sets = malloc(configs.set_count * sizeof(identity_t *));
 776     if (instance->sets == NULL) {
 777         printlog(LOG_CRITICAL, "Not enough memory to allocate set identity collection.\n");
 778         return ENOMEM;
 779     }
 780
 781     memset(instance->sets, 0, configs.set_count * sizeof(identity_t *));
 782     instance->set_count = configs.set_count;
 783
 784     /* For sets, make sure that the hierarchy is valid. */
 785     while (slist) {
 786         ident_member *members = slist->members;
 787
 788         /* ID must be non-zero and unique. */
 789         if (slist->id < 0) {
 790             printlog(LOG_CRITICAL, "Negative ident id: %d (%x) ?\n", slist->id, slist->id);
 791             return EINVAL;
 792         }
 793         tmp = slist->next;
 794         while (tmp) {
 795             if (slist->id == tmp->id) {
 796                 printlog(LOG_CRITICAL, "Duplicate ident id: %d (%x)\n", slist->id, slist->id);
 797                 return EINVAL;
 798             }
 799             tmp = tmp->next;
 800         }
 801
 802         if (validate_config(slist)) {
 803             printlog(LOG_CRITICAL, "Invalid ident parameters for id: %d (%x)\n", slist->id, slist->id);
 804             return EINVAL;
 805         }
 806
 807         /* Allocate an identity_t for this set-type identity. */
 808         instance->sets[i] = new_identity(slist);
 809
 810         if (instance->sets[i] == NULL) {
 811             printlog(LOG_CRITICAL, "Not enough memory to allocate set identity.\n");
 812             return ENOMEM;
 813         }
 814
 815         /* Loop through children and look up each in leaf or ident map
 816          * depending on the type of child.  Set the child's parent pointer
 817          * to the identity we just created above, unless it is already set,
 818          * in which case we have an error. */
 819         while (members) {
 820             identity_t *child_ident = NULL;
 821             leaf_t *child_leaf = NULL;
 822
 823             switch (members->type) {
 824                 case MEMBER_XID:
 825                     child_leaf = map_search(instance->leaf_map, &members->value,
 826                                             sizeof(members->value));
 827                     if (child_leaf == NULL) {
 828                         printlog(LOG_CRITICAL, "xid: child leaf not found.\n");
 829                         return EINVAL;
 830                     }
 831                     if (child_leaf->parent != NULL) {
 832                         /* Error - This leaf already has a parent. */
 833                         printlog(LOG_CRITICAL, "xid: child already has a parent.\n");
 834                         return EINVAL;
 835                     }
 836                     child_leaf->parent = instance->sets[i];
 837                     break;
 838                 case MEMBER_GUID:
 839                     child_ident = map_search(instance->ident_map, &members->value,
 840                                              sizeof(members->value));
 841                     if (child_ident == NULL) {
 842                         printlog(LOG_CRITICAL, "guid: child identity not found.\n");
 843                         return EINVAL;
 844                     }
 845                     if (child_ident->parent != NULL) {
 846                         /* Error - This identity already has a parent. */
 847                         printlog(LOG_CRITICAL, "guid: child identity already has a parent.\n");
 848                         return EINVAL;
 849                     }
 850                     child_ident->parent = instance->sets[i];
 851                     break;
 852                 default:
 853                     /* Error - shouldn't be possible. */
 854                     break;
 855             }
 856             members = members->next;
 857         }
 858
 859         map_insert(instance->ident_map, &instance->sets[i]->id,
 860                    sizeof(instance->sets[i]->id), instance->sets[i]);
 861
 862         slist = slist->next;
 863         i++;
 864     }
 865     return 0;
 866 }
 867
 868 static int fill_set_leaf_pointer(drl_instance_t *instance, identity_t *ident) {
 869     int count = 0;
 870     identity_t *current_ident;
 871     leaf_t *current_leaf;
 872     leaf_t **leaves = malloc(instance->leaf_count * sizeof(leaf_t *));
 873     if (leaves == NULL) {
 874         return 1;
 875     }
 876
 877     map_reset_iterate(instance->leaf_map);
 878     while ((current_leaf = (leaf_t *) map_next(instance->leaf_map))) {
 879         current_ident = current_leaf->parent;
 880         while (current_ident != NULL && current_ident != instance->last_machine) {
 881             if (current_ident == ident) {
 882                 /* Found the ident we were looking for - add the leaf. */
 883                 leaves[count] = current_leaf;
 884                 count += 1;
 885                 break;
 886             }
 887             current_ident = current_ident->parent;
 888         }
 889     }
 890
 891     ident->leaves = leaves;
 892     ident->leaf_count = count;
 893
 894     return 0;
 895 }
 896
 897 static int init_identities(parsed_configs configs, drl_instance_t *instance) {
 898     int i, j;
 899     ident_config *config = configs.machines;
 900     leaf_t *leaf = NULL;
 901
 902     instance->cal = malloc(sizeof(struct ident_calendar) * SCHEDLEN);
 903
 904     if (instance->cal == NULL) {
 905         return ENOMEM;
 906     }
 907
 908     for (i = 0; i < SCHEDLEN; ++i) {
 909         TAILQ_INIT(instance->cal + i);
 910     }
 911     instance->cal_slot = 0;
 912
 913     instance->machines = malloc(configs.machine_count * sizeof(drl_instance_t *));
 914
 915     if (instance->machines == NULL) {
 916         return ENOMEM;
 917     }
 918
 919     memset(instance->machines, 0, configs.machine_count * sizeof(drl_instance_t *));
 920     instance->machine_count = configs.machine_count;
 921
 922     /* Allocate and add the machine identities. */
 923     for (i = 0; i < configs.machine_count; ++i) {
 924         identity_action *loop_action;
 925         identity_action *comm_action;
 926         instance->machines[i] = new_identity(config);
 927
 928         if (instance->machines[i] == NULL) {
 929             return ENOMEM;
 930         }
 931
 932         loop_action = malloc(sizeof(identity_action));
 933         comm_action = malloc(sizeof(identity_action));
 934
 935         if (loop_action == NULL || comm_action == NULL) {
 936             return ENOMEM;
 937         }
 938
 939         /* The first has no parent - it is the root.  All others have the
 940          * previous ident as their parent. */
 941         if (i == 0) {
 942             instance->machines[i]->parent = NULL;
 943         } else {
 944             instance->machines[i]->parent = instance->machines[i - 1];
 945         }
 946
 947         instance->last_machine = instance->machines[i];
 948
 949         /* Add the ident to the guid->ident map. */
 950         map_insert(instance->ident_map, &instance->machines[i]->id,
 951                    sizeof(instance->machines[i]->id), instance->machines[i]);
 952
 953         config = config->next;
 954
 955         memset(loop_action, 0, sizeof(identity_action));
 956         memset(comm_action, 0, sizeof(identity_action));
 957         loop_action->ident = instance->machines[i];
 958         loop_action->action = ACTION_MAINLOOP;
 959         loop_action->valid = 1;
 960         comm_action->ident = instance->machines[i];
 961         comm_action->action = ACTION_COMMUNICATE;
 962         comm_action->valid = 1;
 963
 964         instance->machines[i]->loop_action = loop_action;
 965         instance->machines[i]->comm_action = comm_action;
 966
 967         TAILQ_INSERT_TAIL(instance->cal + (instance->cal_slot & SCHEDMASK),
 968                           loop_action, calendar);
 969
 970         TAILQ_INSERT_TAIL(instance->cal + (instance->cal_slot & SCHEDMASK),
 971                           comm_action, calendar);
 972
 973         /* Setup the array of pointers to leaves.  This is easy for machines
 974          * because a machine node applies to every leaf. */
 975         instance->machines[i]->leaves =
 976             malloc(instance->leaf_count * sizeof(leaf_t *));
 977         if (instance->machines[i]->leaves == NULL) {
 978             return ENOMEM;
 979         }
 980         instance->machines[i]->leaf_count = instance->leaf_count;
 981         for (j = 0; j < instance->leaf_count; ++j) {
 982             instance->machines[i]->leaves[j] = &instance->leaves[j];
 983         }
 984     }
 985
 986     /* Connect the set subtree to the machines. Any set or leaf without a
 987      * parent will take the last machine as its parent. */
 988
 989     /* Leaves... */
 990     map_reset_iterate(instance->leaf_map);
 991     while ((leaf = (leaf_t *) map_next(instance->leaf_map))) {
 992         if (leaf->parent == NULL) {
 993             leaf->parent = instance->last_machine;
 994         }
 995     }
 996
 997     /* Sets... */
 998     for (i = 0; i < instance->set_count; ++i) {
 999         identity_action *loop_action;
1000         identity_action *comm_action;
1001
1002         if (instance->sets[i]->parent == NULL && instance->sets[i]->independent == 0) {
1003             instance->sets[i]->parent = instance->last_machine;
1004         }
1005
1006         loop_action = malloc(sizeof(identity_action));
1007         comm_action = malloc(sizeof(identity_action));
1008
1009         if (loop_action == NULL || comm_action == NULL) {
1010             return ENOMEM;
1011         }
1012
1013         memset(loop_action, 0, sizeof(identity_action));
1014         memset(comm_action, 0, sizeof(identity_action));
1015         loop_action->ident = instance->sets[i];
1016         loop_action->action = ACTION_MAINLOOP;
1017         loop_action->valid = 1;
1018         comm_action->ident = instance->sets[i];
1019         comm_action->action = ACTION_COMMUNICATE;
1020         comm_action->valid = 1;
1021
1022         instance->sets[i]->loop_action = loop_action;
1023         instance->sets[i]->comm_action = comm_action;
1024
1025         TAILQ_INSERT_TAIL(instance->cal + (instance->cal_slot & SCHEDMASK),
1026                           loop_action, calendar);
1027
1028         TAILQ_INSERT_TAIL(instance->cal + (instance->cal_slot & SCHEDMASK),
1029                           comm_action, calendar);
1030
1031         /* Setup the array of pointers to leaves.  This is harder for sets,
1032          * but this doesn't need to be super-efficient because it happens
1033          * rarely and it isn't on the critical path for reconfig(). */
1034         if (fill_set_leaf_pointer(instance, instance->sets[i])) {
1035             return ENOMEM;
1036         }
1037     }
1038
1039     /* Success. */
1040     return 0;
1041 }
1042
1043 static void print_instance(drl_instance_t *instance) {
1044     leaf_t *leaf = NULL;
1045     identity_t *ident = NULL;
1046
1047     if (system_loglevel == LOG_DEBUG) {
1048         map_reset_iterate(instance->leaf_map);
1049         while ((leaf = (leaf_t *) map_next(instance->leaf_map))) {
1050             printf("%x:", leaf->xid);
1051             ident = leaf->parent;
1052             while (ident) {
1053                 printf("%d:",ident->id);
1054                 ident = ident->parent;
1055             }
1056             printf("Leaf's parent pointer is %p\n", leaf->parent);
1057         }
1058
1059         printf("instance->last_machine is %p\n", instance->last_machine);
1060     }
1061 }
1062
1063 static int assign_htb_hierarchy(drl_instance_t *instance) {
1064     int i, j;
1065     int next_node = 0x100;
1066
1067     /* Chain machine nodes under 1:10. */
1068     for (i = 0; i < instance->machine_count; ++i) {
1069         if (instance->machines[i]->parent == NULL) {
1070             /* Top node. */
1071             instance->machines[i]->htb_parent = 0x10;
1072         } else {
1073             /* Pointerific! */
1074             instance->machines[i]->htb_parent =
1075                 instance->machines[i]->parent->htb_node;
1076         }
1077
1078         instance->machines[i]->htb_node = next_node;
1079         next_node += 1;
1080     }
1081
1082     next_node += 0x10;
1083
1084     /* Add set nodes under machine nodes. Iterate backwards to ensure parent is
1085      * already there. */
1086     for (j = (instance->set_count - 1); j >= 0; --j) {
1087         if (instance->sets[j]->parent == NULL) {
1088             /* Independent node - goes under 0x10 away from machine nodes. */
1089             instance->sets[j]->htb_parent = 0x10;
1090         } else {
1091             instance->sets[j]->htb_parent = instance->sets[j]->parent->htb_node;
1092         }
1093         instance->sets[j]->htb_node = next_node;
1094
1095         next_node += 1;
1096     }
1097
1098     return 0;
1099 }
1100
1101 /* Added this so that I could comment one line and kill off all of the
1102  * command execution. */
1103 static inline int execute_cmd(const char *cmd) {
1104     return system(cmd);
1105 }
1106
1107 static inline int add_htb_node(const char *iface, const uint32_t parent_major, const uint32_t parent_minor,
1108                                const uint32_t classid_major, const uint32_t classid_minor,
1109                                const uint64_t rate, const uint64_t ceil) {
1110     char cmd[300];
1111
1112     sprintf(cmd, "tc class add dev %s parent %x:%x classid %x:%x htb rate %llubit ceil %llubit",
1113             iface, parent_major, parent_minor, classid_major, classid_minor, rate, ceil);
1114     printlog(LOG_WARN, "INIT: HTB_cmd: %s\n", cmd);
1115
1116     return execute_cmd(cmd);
1117 }
1118
1119 static inline int add_htb_netem(const char *iface, const uint32_t parent_major,
1120                                 const uint32_t parent_minor, const uint32_t handle,
1121                                 const int loss, const int delay) {
1122     char cmd[300];
1123
1124     sprintf(cmd, "/sbin/tc qdisc del dev %s parent %x:%x handle %x pfifo", iface, parent_major,
1125             parent_minor, handle);
1126     printlog(LOG_DEBUG, "HTB_cmd: %s\n", cmd);
1127     if (execute_cmd(cmd))
1128         printlog(LOG_DEBUG, "HTB_cmd: Previous deletion did not succeed.\n");
1129
1130     sprintf(cmd, "/sbin/tc qdisc replace dev %s parent %x:%x handle %x netem loss %d%% delay %dms",
1131             iface, parent_major, parent_minor, handle, loss, delay);
1132     printlog(LOG_DEBUG, "HTB_cmd: %s\n", cmd);
1133     return execute_cmd(cmd);
1134 }
1135
1136 static int create_htb_hierarchy(drl_instance_t *instance) {
1137     char cmd[300];
1138     int i, j, k;
1139     uint64_t gigabit = 1024 * 1024 * 1024;
1140
1141     /* Nuke the hierarchy. */
1142     sprintf(cmd, "tc qdisc del dev eth0 root handle 1: htb");
1143     execute_cmd(cmd);
1144     printlog(LOG_DEBUG, "HTB_cmd: %s\n", cmd);
1145
1146     /* Re-initialize the basics. */
1147     sprintf(cmd, "tc qdisc add dev eth0 root handle 1: htb default 1fff");
1148     if (execute_cmd(cmd)) {
1149         return 1;
1150     }
1151     printlog(LOG_DEBUG, "HTB_cmd: %s\n", cmd);
1152
1153     if (add_htb_node("eth0", 1, 0, 1, 1, gigabit, gigabit))
1154         return 1;
1155
1156     /* Add back 1:10. (Nodelimit : kilobits/sec -> bits/second)*/
1157     if (limiter.nodelimit) {
1158         if (add_htb_node("eth0", 1, 1, 1, 0x10, 8, (uint64_t) limiter.nodelimit * 1024))
1159             return 1;
1160     } else {
1161         if (add_htb_node("eth0", 1, 1, 1, 0x10, 8, gigabit))
1162             return 1;
1163     }
1164
1165     /* Add machines. */
1166     for (i = 0; i < instance->machine_count; ++i) {
1167         if (add_htb_node("eth0", 1, instance->machines[i]->htb_parent, 1,
1168                          instance->machines[i]->htb_node, 8, instance->machines[i]->limit * 1024)) {
1169             return 1;
1170         }
1171     }
1172
1173 #define LIMITEXEMPT
1174
1175     /* Add back 1:20. */
1176 #ifdef LIMITEXEMPT
1177     if (instance->last_machine == NULL) {
1178         sprintf(cmd, "/sbin/tc class add dev eth0 parent 1:1 classid 1:20 htb rate 8bit ceil 1000mbit");
1179     } else {
1180         sprintf(cmd, "/sbin/tc class add dev eth0 parent 1:%x classid 1:20 htb rate 8bit ceil 1000mbit",
1181             instance->last_machine->htb_node);
1182     }
1183 #else
1184     sprintf(cmd, "/sbin/tc class add dev eth0 parent 1:1 classid 1:20 htb rate 8bit ceil 1000mbit");
1185 #endif
1186
1187     if (execute_cmd(cmd)) {
1188         return 1;
1189     }
1190     printlog(LOG_DEBUG, "HTB_cmd: %s\n", cmd);
1191
1192     /* Add sets. */
1193     for (j = (instance->set_count - 1); j >= 0; --j) {
1194         if (add_htb_node("eth0", 1, instance->sets[j]->htb_parent, 1,
1195                          instance->sets[j]->htb_node, 8, instance->sets[j]->limit * 1024)) {
1196             return 1;
1197         }
1198     }
1199
1200     /* Add leaves. FIXME: Set static sliver limit as ceil here! */
1201     for (k = 0; k < instance->leaf_count; ++k) {
1202         if (instance->leaves[k].parent == NULL) {
1203             if (add_htb_node("eth0", 1, 0x10, 1, (0x1000 | instance->leaves[k].xid), 8, gigabit))
1204                 return 1;
1205         } else {
1206             if (add_htb_node("eth0", 1, instance->leaves[k].parent->htb_node, 1, (0x1000 | instance->leaves[k].xid), 8, gigabit))
1207                 return 1;
1208         }
1209
1210         /* Add exempt node for the leaf under 1:20 as 1:2<xid> */
1211         if (add_htb_node("eth0", 1, 0x20, 1, (0x2000 | instance->leaves[k].xid), 8, gigabit))
1212             return 1;
1213     }
1214
1215     /* Add 1:1000 and 1:2000 */
1216     if (instance->last_machine == NULL) {
1217         if (add_htb_node("eth0", 1, 0x10, 1, 0x1000, 8, gigabit))
1218             return 1;
1219     } else {
1220         if (add_htb_node("eth0", 1, instance->last_machine->htb_node, 1, 0x1000, 8, gigabit))
1221             return 1;
1222     }
1223
1224     if (add_htb_node("eth0", 1, 0x20, 1, 0x2000, 8, gigabit))
1225         return 1;
1226
1227     /* Add 1:1fff and 1:2fff */
1228     if (instance->last_machine == NULL) {
1229         if (add_htb_node("eth0", 1, 0x10, 1, 0x1fff, 8, gigabit))
1230             return 1;
1231     } else {
1232         if (add_htb_node("eth0", 1, instance->last_machine->htb_node, 1, 0x1fff, 8, gigabit))
1233             return 1;
1234     }
1235
1236     if (add_htb_node("eth0", 1, 0x20, 1, 0x2fff, 8, gigabit))
1237         return 1;
1238
1239     /* Artifical delay or loss for experimentation. */
1240     if (netem_delay.u.value || netem_loss.u.value) {
1241         if (!strcmp(netem_slice.u.string, "ALL")) {
1242             /* By default, netem applies to all leaves. */
1243             if (add_htb_netem("eth0", 1, 0x1000, 0x1000, netem_loss.u.value, netem_delay.u.value))
1244                 return 1;
1245             if (add_htb_netem("eth0", 1, 0x1fff, 0x1fff, netem_loss.u.value, netem_delay.u.value))
1246                 return 1;
1247
1248             for (k = 0; k < instance->leaf_count; ++k) {
1249                 if (add_htb_netem("eth0", 1, (0x1000 | instance->leaves[k].xid),
1250                             (0x1000 | instance->leaves[k].xid), netem_loss.u.value, netem_delay.u.value)) {
1251                     return 1;
1252                 }
1253
1254                 //FIXME: add exempt delay/loss here on 0x2000 ... ?
1255             }
1256         } else {
1257             /* netem_slice is not the default ALL value.  Only apply netem
1258              * to the slice that is set in netem_slice.u.string. */
1259             uint32_t slice_xid;
1260
1261             sscanf(netem_slice.u.string, "%x", &slice_xid);
1262
1263             if (add_htb_netem("eth0", 1, slice_xid, slice_xid, netem_loss.u.value, netem_delay.u.value))
1264                 return 1;
1265         }
1266     }
1267
1268     return 0;
1269 }
1270
1271 static int setup_tc_grd(drl_instance_t *instance) {
1272     int i;
1273     char cmd[300];
1274
1275     for (i = 0; i < instance->leaf_count; ++i) {
1276         /* Delete the old pfifo qdisc that might have been there before. */
1277         sprintf(cmd, "/sbin/tc qdisc del dev eth0 parent 1:1%x handle 1%x pfifo",
1278                 instance->leaves[i].xid, instance->leaves[i].xid);
1279
1280         if (execute_cmd(cmd)) {
1281             printlog(LOG_DEBUG, "GRD: pfifo qdisc wasn't there!\n");
1282         }
1283
1284         /* Add the netem qdisc. */
1285 #ifdef DELAY40MS
1286         sprintf(cmd, "/sbin/tc qdisc replace dev eth0 parent 1:1%x handle 1%x netem loss 0 delay 40ms",
1287                 instance->leaves[i].xid, instance->leaves[i].xid);
1288 #else
1289         sprintf(cmd, "/sbin/tc qdisc replace dev eth0 parent 1:1%x handle 1%x netem loss 0 delay 0ms",
1290                 instance->leaves[i].xid, instance->leaves[i].xid);
1291 #endif
1292
1293         if (execute_cmd(cmd)) {
1294             return 1;
1295         }
1296     }
1297
1298     /* Do the same for 1000 and 1fff. */
1299     sprintf(cmd, "/sbin/tc qdisc del dev eth0 parent 1:1000 handle 1000 pfifo");
1300
1301     if (execute_cmd(cmd)) {
1302         printlog(LOG_DEBUG, "GRD: pfifo qdisc wasn't there!\n");
1303     }
1304
1305     /* Add the netem qdisc. */
1306 #ifdef DELAY40MS
1307     sprintf(cmd, "/sbin/tc qdisc replace dev eth0 parent 1:1000 handle 1000 netem loss 0 delay 40ms");
1308 #else
1309     sprintf(cmd, "/sbin/tc qdisc replace dev eth0 parent 1:1000 handle 1000 netem loss 0 delay 0ms");
1310 #endif
1311
1312     if (execute_cmd(cmd)) {
1313         return 1;
1314     }
1315
1316     sprintf(cmd, "/sbin/tc qdisc del dev eth0 parent 1:1fff handle 1fff pfifo");
1317
1318     if (execute_cmd(cmd)) {
1319         printlog(LOG_DEBUG, "GRD: pfifo qdisc wasn't there!\n");
1320     }
1321
1322     /* Add the netem qdisc. */
1323 #ifdef DELAY40MS
1324     sprintf(cmd, "/sbin/tc qdisc replace dev eth0 parent 1:1fff handle 1fff netem loss 0 delay 40ms");
1325 #else
1326     sprintf(cmd, "/sbin/tc qdisc replace dev eth0 parent 1:1fff handle 1fff netem loss 0 delay 0ms");
1327 #endif
1328
1329     if (execute_cmd(cmd)) {
1330         return 1;
1331     }
1332
1333     return 0;
1334 }
1335
1336 /* init_drl
1337  *
1338  * Initialize this limiter with options
1339  * Open UDP socket for peer communication
1340  */
1341 static int init_drl(void) {
1342     parsed_configs configs;
1343     struct sockaddr_in server_address;
1344
1345     memset(&limiter, 0, sizeof(limiter_t));
1346
1347     /* Setup logging. */
1348     system_loglevel = (uint8_t) drl_loglevel.u.value;
1349     logfile = fopen(drl_logfile.u.string, "w");
1350
1351     if (logfile == NULL) {
1352         printf("Couldn't open logfile - ");
1353         perror("fopen()");
1354         exit(EXIT_FAILURE);
1355     }
1356
1357     printlog(LOG_CRITICAL, "ulogd_DRL initializing . . .\n");
1358
1359     limiter.nodelimit = (uint32_t) (((double) nodelimit.u.value * 1000000.0) / 8.0);
1360
1361     init_hashing();  /* for all hash maps */
1362
1363     pthread_rwlock_init(&limiter.limiter_lock,NULL);
1364
1365     /* determine our local IP by iterating through interfaces */
1366     if ((limiter.ip = get_local_ip())==0) {
1367         printlog(LOG_CRITICAL,
1368                  "ulogd_DRL unable to aquire local IP address, not registering.\n");
1369         return (false);
1370     }
1371     limiter.localaddr = inet_addr(limiter.ip);
1372     limiter.port = htons(LIMITER_LISTEN_PORT);
1373     limiter.udp_socket = socket(PF_INET, SOCK_DGRAM, IPPROTO_UDP);
1374     if (limiter.udp_socket < 0) {
1375         printlog(LOG_CRITICAL, "Failed to create UDP socket().\n");
1376         return false;
1377     }
1378
1379     memset(&server_address, 0, sizeof(server_address));
1380     server_address.sin_family = AF_INET;
1381     server_address.sin_addr.s_addr = limiter.localaddr;
1382     server_address.sin_port = limiter.port;
1383
1384     if (bind(limiter.udp_socket, (struct sockaddr *) &server_address, sizeof(server_address)) < 0) {
1385         printlog(LOG_CRITICAL, "Failed to bind UDP socket.\n");
1386         return false;
1387     }
1388
1389     printlog(LOG_WARN, "     POLICY: %s\n",policy.u.string);
1390     if (strcasecmp(policy.u.string,"GRD") == 0) {
1391         limiter.policy = POLICY_GRD;
1392     } else if (strcasecmp(policy.u.string,"FPS") == 0) {
1393         limiter.policy = POLICY_FPS;
1394     } else {
1395         printlog(LOG_CRITICAL,
1396                  "Unknown DRL policy %s, aborting.\n",policy.u.string);
1397         return (false);
1398     }
1399
1400     limiter.estintms = estintms.u.value;
1401     if (limiter.estintms > 1000) {
1402         printlog(LOG_CRITICAL,
1403                  "DRL: sorry estimate intervals must be less than 1 second.");
1404         printlog(LOG_CRITICAL,
1405                  "  Simple source mods will allow larger intervals.  Using 1 second.\n");
1406         limiter.estintms = 1000;
1407     }
1408     printlog(LOG_WARN, "     Est interval: %dms\n",limiter.estintms);
1409
1410     /* Acquire the big limiter lock for writing.  Prevents pretty much
1411      * anything else from happening while the hierarchy is being changed. */
1412     pthread_rwlock_wrlock(&limiter.limiter_lock);
1413
1414     limiter.stable_instance.ident_map = allocate_map();
1415     if (limiter.stable_instance.ident_map == NULL) {
1416         printlog(LOG_CRITICAL, "Failed to allocate memory for identity map.\n");
1417         return false;
1418     }
1419
1420     if (get_eligible_leaves(&limiter.stable_instance)) {
1421         printlog(LOG_CRITICAL, "Failed to read eligigle leaves.\n");
1422         return false;
1423     }
1424
1425     if (parse_drl_config(drl_configfile.u.string, &configs)) {
1426         /* Parse error occured. Return non-zero to notify init_drl(). */
1427         printlog(LOG_CRITICAL, "Failed to parse the DRL configuration file (%s).\n",
1428             drl_configfile.u.string);
1429         return false;
1430     }
1431
1432     /* Validate identity hierarchy! */
1433     if (validate_configs(configs, &limiter.stable_instance)) {
1434         /* Clean up everything. */
1435         free_failed_config(configs, &limiter.stable_instance);
1436         printlog(LOG_CRITICAL, "Invalid DRL configuration file (%s).\n",
1437             drl_configfile.u.string);
1438         return false;
1439     }
1440
1441     if (init_identities(configs, &limiter.stable_instance)) {
1442         free_failed_config(configs, &limiter.stable_instance);
1443         printlog(LOG_CRITICAL, "Failed to initialize identities.\n");
1444         return false;
1445     }
1446
1447     /* At this point, we should be done with configs. */
1448     free_ident_list(configs.machines);
1449     free_ident_list(configs.sets);
1450
1451     print_instance(&limiter.stable_instance);
1452
1453     switch (limiter.policy) {
1454         case POLICY_FPS:
1455             if (assign_htb_hierarchy(&limiter.stable_instance)) {
1456                 free_instance(&limiter.stable_instance);
1457                 printlog(LOG_CRITICAL, "Failed to assign HTB hierarchy.\n");
1458                 return false;
1459             }
1460
1461             if (create_htb_hierarchy(&limiter.stable_instance)) {
1462                 free_instance(&limiter.stable_instance);
1463                 printlog(LOG_CRITICAL, "Failed to create HTB hierarchy.\n");
1464                 return false;
1465             }
1466         break;
1467
1468         case POLICY_GRD:
1469             if (setup_tc_grd(&limiter.stable_instance)) {
1470                 free_instance(&limiter.stable_instance);
1471                 printlog(LOG_CRITICAL, "Failed to initialize tc calls for GRD.\n");
1472                 return false;
1473             }
1474         break;
1475
1476         default:
1477         return false;
1478     }
1479
1480     partition_set = partition.u.value;
1481
1482     pthread_rwlock_unlock(&limiter.limiter_lock);
1483
1484     if (pthread_create(&limiter.udp_recv_thread, NULL, limiter_receive_thread, NULL)) {
1485         printlog(LOG_CRITICAL, "Unable to start UDP receive thread.\n");
1486         return false;
1487     }
1488
1489     printlog(LOG_WARN, "ulogd_DRL init finished.\n");
1490
1491     return true;
1492 }
1493
1494 static void reconfig() {
1495     parsed_configs configs;
1496
1497     printlog(LOG_DEBUG, "--Starting reconfig()--\n");
1498     flushlog();
1499
1500     memset(&configs, 0, sizeof(parsed_configs));
1501     memset(&limiter.new_instance, 0, sizeof(drl_instance_t));
1502
1503     limiter.new_instance.ident_map = allocate_map();
1504     if (limiter.new_instance.ident_map == NULL) {
1505         printlog(LOG_CRITICAL, "Failed to allocate ident_map during reconfig().\n");
1506         return;
1507     }
1508
1509     if (get_eligible_leaves(&limiter.new_instance)) {
1510         free_failed_config(configs, &limiter.new_instance);
1511         printlog(LOG_CRITICAL, "Failed to read leaves during reconfig().\n");
1512         return;
1513     }
1514
1515     if (parse_drl_config(drl_configfile.u.string, &configs)) {
1516         free_failed_config(configs, &limiter.new_instance);
1517         printlog(LOG_CRITICAL, "Failed to parse config during reconfig().\n");
1518         return;
1519     }
1520
1521     if (validate_configs(configs, &limiter.new_instance)) {
1522         free_failed_config(configs, &limiter.new_instance);
1523         printlog(LOG_CRITICAL, "Validation failed during reconfig().\n");
1524         pthread_rwlock_unlock(&limiter.limiter_lock);
1525         return;
1526     }
1527
1528     if (init_identities(configs, &limiter.new_instance)) {
1529         free_failed_config(configs, &limiter.new_instance);
1530         printlog(LOG_CRITICAL, "Initialization failed during reconfig().\n");
1531         pthread_rwlock_unlock(&limiter.limiter_lock);
1532         return;
1533     }
1534
1535     free_ident_list(configs.machines);
1536     free_ident_list(configs.sets);
1537
1538     print_instance(&limiter.new_instance);
1539
1540     /* Lock */
1541     pthread_rwlock_wrlock(&limiter.limiter_lock);
1542
1543     switch (limiter.policy) {
1544         case POLICY_FPS:
1545             if (assign_htb_hierarchy(&limiter.new_instance)) {
1546                 free_instance(&limiter.new_instance);
1547                 printlog(LOG_CRITICAL, "Failed to assign HTB hierarchy during reconfig().\n");
1548                 pthread_rwlock_unlock(&limiter.limiter_lock);
1549                 return;
1550             }
1551
1552             if (create_htb_hierarchy(&limiter.new_instance)) {
1553                 free_instance(&limiter.new_instance);
1554                 printlog(LOG_CRITICAL, "Failed to create HTB hierarchy during reconfig().\n");
1555
1556                 /* Re-create old instance. */
1557                 if (create_htb_hierarchy(&limiter.stable_instance)) {
1558                     /* Error reinstating the old one - big problem. */
1559                     printlog(LOG_CRITICAL, "Failed to reinstate HTB hierarchy during reconfig().\n");
1560                     printlog(LOG_CRITICAL, "Giving up...\n");
1561                     flushlog();
1562                     exit(EXIT_FAILURE);
1563                 }
1564
1565                 pthread_rwlock_unlock(&limiter.limiter_lock);
1566                 return;
1567             }
1568         break;
1569
1570         case POLICY_GRD:
1571             if (setup_tc_grd(&limiter.new_instance)) {
1572                 free_instance(&limiter.new_instance);
1573                 printlog(LOG_CRITICAL, "GRD tc calls failed during reconfig().\n");
1574
1575                 /* Try to re-create old instance. */
1576                 if (setup_tc_grd(&limiter.stable_instance)) {
1577                     printlog(LOG_CRITICAL, "Failed to reinstate old GRD qdiscs during reconfig().\n");
1578                     printlog(LOG_CRITICAL, "Giving up...\n");
1579                     flushlog();
1580                     exit(EXIT_FAILURE);
1581                 }
1582             }
1583         break;
1584
1585         default:
1586             /* Should be impossible. */
1587             printf("Pigs are flying?\n");
1588             exit(EXIT_FAILURE);
1589     }
1590
1591     /* Switch over new to stable instance. */
1592     free_instance(&limiter.stable_instance);
1593     memcpy(&limiter.stable_instance, &limiter.new_instance, sizeof(drl_instance_t));
1594
1595     /* Success! - Unlock */
1596     pthread_rwlock_unlock(&limiter.limiter_lock);
1597 }
1598
1599 static ulog_output_t drl_op = {
1600     .name = "drl",
1601     .output = &_output_drl,
1602     .signal = NULL, /* This appears to be broken. Using my own handler. */
1603     .init = NULL,
1604     .fini = NULL,
1605 };
1606
1607 /* Tests the amount of time it takes to call reconfig(). */
1608 static void time_reconfig(int iterations) {
1609     struct timeval start, end;
1610     int i;
1611
1612     gettimeofday(&start, NULL);
1613     for (i = 0; i < iterations; ++i) {
1614         reconfig();
1615     }
1616     gettimeofday(&end, NULL);
1617
1618     printf("%d reconfigs() took %d seconds and %d microseconds.\n",
1619            iterations, end.tv_sec - start.tv_sec, end.tv_usec - start.tv_usec);
1620     exit(0);
1621
1622     // Seems to take about 85ms / iteration
1623 }
1624
1625 static int stop_enforcement(drl_instance_t *instance) {
1626     char cmd[300];
1627     int i;
1628
1629     for (i = 0; i < instance->machine_count; ++i) {
1630         sprintf(cmd, "/sbin/tc class change dev eth0 parent 1:%x classid 1:%x htb rate 8bit ceil 100mbit",
1631                 instance->machines[i]->htb_parent,
1632                 instance->machines[i]->htb_node);
1633
1634         if (execute_cmd(cmd)) {
1635             return 1;
1636         }
1637     }
1638
1639     for (i = 0; i < instance->set_count; ++i) {
1640         sprintf(cmd, "/sbin/tc class change dev eth0 parent 1:%x classid 1:%x htb rate 8bit ceil 100mbit",
1641                 instance->sets[i]->htb_parent,
1642                 instance->sets[i]->htb_node);
1643
1644         if (execute_cmd(cmd)) {
1645             return 1;
1646         }
1647     }
1648
1649     return 0;
1650 }
1651
1652 static void *signal_thread_func(void *args) {
1653     int sig;
1654     int err;
1655     sigset_t sigs;
1656
1657     sigemptyset(&sigs);
1658     sigaddset(&sigs, SIGHUP);
1659     sigaddset(&sigs, SIGUSR1);
1660     sigaddset(&sigs, SIGUSR2);
1661     pthread_sigmask(SIG_BLOCK, &sigs, NULL);
1662
1663     while (1) {
1664         sigemptyset(&sigs);
1665         sigaddset(&sigs, SIGHUP);
1666         sigaddset(&sigs, SIGUSR1);
1667         sigaddset(&sigs, SIGUSR2);
1668
1669         err = sigwait(&sigs, &sig);
1670
1671         if (err) {
1672             printlog(LOG_CRITICAL, "sigwait() returned an error.\n");
1673             flushlog();
1674         }
1675
1676         switch (sig) {
1677             case SIGHUP:
1678                 printlog(LOG_WARN, "Caught SIGHUP - re-reading XML file.\n");
1679                 reconfig();
1680                 //time_reconfig(1000); /* instrumentation */
1681                 flushlog();
1682                 break;
1683             case SIGUSR1:
1684                 pthread_rwlock_wrlock(&limiter.limiter_lock);
1685                 if (do_enforcement) {
1686                     do_enforcement = 0;
1687                     stop_enforcement(&limiter.stable_instance);
1688                     printlog(LOG_CRITICAL, "--Switching enforcement off.--\n");
1689                 } else {
1690                     do_enforcement = 1;
1691                     printlog(LOG_CRITICAL, "--Switching enforcement on.--\n");
1692                 }
1693                 pthread_rwlock_unlock(&limiter.limiter_lock);
1694                 break;
1695             case SIGUSR2:
1696                 do_partition = !do_partition;
1697                 break;
1698             default:
1699                 /* Intentionally blank. */
1700                 break;
1701         }
1702     }
1703
1704 }
1705
1706 /* register output plugin with ulogd */
1707 static void _drl_reg_op(void)
1708 {
1709     ulog_output_t *op = &drl_op;
1710     sigset_t signal_mask;
1711
1712     sigemptyset(&signal_mask);
1713     sigaddset(&signal_mask, SIGHUP);
1714     sigaddset(&signal_mask, SIGUSR1);
1715     sigaddset(&signal_mask, SIGUSR2);
1716     pthread_sigmask(SIG_BLOCK, &signal_mask, NULL);
1717
1718     if (pthread_create(&signal_thread, NULL, &signal_thread_func, NULL) != 0) {
1719         printlog(LOG_CRITICAL, "Failed to create signal handling thread.\n");
1720         fprintf(stderr, "An error has occured starting ulogd_DRL.  Refer to your logfile (%s) for additional information.\n", drl_logfile.u.string);
1721         flushlog();
1722         exit(EXIT_FAILURE);
1723     }
1724
1725     if (!init_drl()) {
1726         printlog(LOG_CRITICAL, "Init failed. :(\n");
1727         fprintf(stderr, "An error has occured starting ulogd_DRL.  Refer to your logfile (%s) for additional information.\n", drl_logfile.u.string);
1728         flushlog();
1729         exit(EXIT_FAILURE);
1730     }
1731
1732     register_output(op);
1733
1734     /* start up the thread that will periodically estimate the
1735      * local rate and set the local limits
1736      * see estimate.c
1737      */
1738     if (pthread_create(&estimate_thread, NULL, (void*(*)(void*)) &handle_estimation, &limiter)!=0) {
1739         printlog(LOG_CRITICAL, "Couldn't start estimate thread.\n");
1740         fprintf(stderr, "An error has occured starting ulogd_DRL.  Refer to your logfile (%s) for additional information.\n", drl_logfile.u.string);
1741         exit(EXIT_FAILURE);
1742     }
1743 }
1744
1745 void _init(void)
1746 {
1747     /* have the opts parsed */
1748     config_parse_file("DRL", config_entries);
1749
1750     if (get_ids()) {
1751         ulogd_log(ULOGD_ERROR, "can't resolve all keyhash id's\n");
1752         exit(2);
1753     }
1754
1755     /* Seed the hash function */
1756     salt = getpid() ^ time(NULL);
1757
1758     _drl_reg_op();
1759 }