actl.c

   1 /*
   2  * Copyright (c) 2016 Mohamed Aslan <maslan@sce.carleton.ca>
   3  *
   4  * Permission to use, copy, modify, and distribute this software for any
   5  * purpose with or without fee is hereby granted, provided that the above
   6  * copyright notice and this permission notice appear in all copies.
   7  *
   8  * THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
   9  * WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
  10  * MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
  11  * ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
  12  * WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
  13  * ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
  14  * OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
  15  */
  16
  17 #include <stdio.h>
  18 #include <stdlib.h>
  19 #include <string.h>
  20 #include <pthread.h>
  21 #include <err.h>
  22 #include <errno.h>
  23 #include <unistd.h>
  24 #include <sys/wait.h>
  25 #include <dlfcn.h>
  26
  27 #include <hashtab.h>
  28 #include <dht.h>
  29 #include <libof.h>
  30 #include <of10.h>
  31
  32 #include "actl.h"
  33 #include "conf.h"
  34 #include "utils.h"
  35 #include "model.h"
  36 #include "clustering.h"
  37
  38
  39 static struct actl_ctx          ctx;
  40
  41
  42 static void
  43 adapt(struct actl_ctx *c, double kpi)
  44 {
  45         c->target_kpi = kpi;
  46         c->adaptive = 1;
  47 }
  48
  49 static int
  50 put(struct actl_ctx *c, const char *key, const char *val)
  51 {
  52         int                     r, w;
  53         struct timespec         ts;
  54
  55         phi2rw(c->target_phi, c->conf->c_nreplicas, &r, &w);
  56         (void)clock_gettime(CLOCK_REALTIME, &ts);
  57         return dht_put_tunable(c->node, key, val, &ts, w);
  58 }
  59
  60 static int
  61 get(struct actl_ctx *c, const char *key, char **val)
  62 {
  63         int             r, w;
  64
  65         phi2rw(c->target_phi, c->conf->c_nreplicas, &r, &w);
  66         return dht_get_tunable(c->node, key, val, r);
  67 }
  68
  69 static void
  70 topo(struct actl_ctx *c, struct host_info ***hosts, int *n)
  71 {
  72         *hosts = c->conf->c_topo;
  73         *n = c->conf->c_topo_n;
  74 }
  75
  76 static void
  77 init_context(struct actl_ctx *c)
  78 {
  79         if (!c)
  80                 return;
  81         c->cntl = (struct of_controller *)xmalloc(sizeof(struct of_controller));
  82         c->node = (struct dht_node *)xmalloc(sizeof(struct dht_node));
  83         c->conf = (struct config *)xmalloc(sizeof(struct config));
  84         c->adaptive = 0;
  85         c->learning = 1;
  86         c->adapt = adapt;
  87         c->target_phi = 0;
  88         c->put = put;
  89         c->get = get;
  90         c->topo = topo;
  91         kmeans_seq_init(&c->kseq, 10);
  92 }
  93
  94 static void
  95 load_app(struct actl_ctx *c)
  96 {
  97         if (!c)
  98                 return;
  99         /* load application */
 100         c->conf->c_app_so = dlopen(c->conf->c_app_name, RTLD_NOW);
 101         if (!c->conf->c_app_so)
 102                 errx(1, "error loading application %s: %s", c->conf->c_app_name, dlerror());
 103         /* init */
 104         c->conf->c_app.a_init = dlsym(c->conf->c_app_so, "init");
 105         if (!(c->conf->c_app.a_init))
 106                 errx(1, "error loading symbol from %s: %s", c->conf->c_app_name, dlerror());
 107         /* main */
 108         c->conf->c_app.a_main = dlsym(c->conf->c_app_so, "main");
 109         if (!c->conf->c_app.a_main)
 110                 errx(1, "error loading symbol from %s: %s", c->conf->c_app_name, dlerror());
 111         /* handler */
 112         c->conf->c_app.a_handler = dlsym(c->conf->c_app_so, "handler");
 113         if (!c->conf->c_app.a_handler)
 114                 errx(1, "error loading symbol from %s: %s", c->conf->c_app_name, dlerror());
 115         /* handler */
 116         c->conf->c_app.a_kpi = dlsym(c->conf->c_app_so, "kpi");
 117         if (!c->conf->c_app.a_kpi)
 118                 errx(1, "error loading symbol from %s: %s", c->conf->c_app_name, dlerror());
 119 }
 120
 121 static void
 122 poll_kpi(struct of_controller *ctl)
 123 {
 124         double          cur_kpi, cur_phi;
 125
 126         if (ctx.conf->c_app_so && ctx.adaptive) {
 127                 cur_kpi = ctx.conf->c_app.a_kpi();
 128                 if (ctx.learning) {
 129                         kmeans_seq_insert(&ctx.kseq, cur_kpi, ctx.target_phi);
 130                         printf("[INFO] learn KPI = %lf -> Phi = %lf.\n", cur_kpi, ctx.target_phi);
 131                 } else {
 132                         cur_phi = kmeans_seq_find(&ctx.kseq, cur_kpi);
 133                         printf("[INFO] KPI = (target %lf, current %lf), Phi = %lf.\n", ctx.target_kpi, cur_kpi, cur_phi);
 134                         f_log(ctx.conf->c_logfp, "%lf, %lf, %lf, %lf\n", ctx.target_kpi, ctx.target_phi, cur_kpi, cur_phi);
 135                 }
 136         }
 137 }
 138
 139 static void
 140 learn_timer(struct of_controller *ctl)
 141 {
 142         if (ctx.conf->c_app_so && ctx.adaptive && ctx.learning) {
 143                 ctx.conf->c_ltime -= ctx.learn_perval;
 144                 if (ctx.conf->c_ltime <= 0) {
 145                         ctx.learning = 0;
 146                         printf("[INFO] learning finished.\n");
 147                         kmeans_seq_print(&ctx.kseq, "LEARN");
 148                 }
 149                 ctx.target_phi += 0.1;
 150         }
 151 }
 152
 153 static void *
 154 dht_thread(void *arg)
 155 {
 156         dht_event_loop(ctx.node);
 157         return NULL;
 158 }
 159
 160 static void *
 161 controller_thread(void *arg)
 162 {
 163         /* init default app */
 164         if (ctx.conf->c_app_so) {
 165                 optind = 0; /* make sure apps can use getopt(3) or getopt_long(3) */
 166                 ctx.conf->c_app.a_init(&ctx);
 167                 printf("[INFO] adaptation period is %d sec.\n", ctx.conf->c_atime);
 168                 ctx.conf->c_app.a_main(ctx.conf->c_app_argc, ctx.conf->c_app_argv);
 169         }
 170         ctx.learn_perval = ctx.conf->c_ltime / 10;
 171         ctx.cntl->timer(ctx.cntl, ctx.learn_perval * 1000, learn_timer);
 172         /* poll application-specific KPI */
 173         ctx.cntl->timer(ctx.cntl, ctx.conf->c_atime * 1000, poll_kpi);
 174         ctx.cntl->loop(ctx.cntl);
 175         return NULL;
 176 }
 177
 178 static void
 179 handler(struct of_controller *ctl, struct of_event *ev)
 180 {
 181 #ifdef DEBUG
 182         printf("actl: handler.\n");
 183 #endif
 184         if (ctx.conf->c_app_so)
 185                 ctx.conf->c_app.a_handler(ev);
 186 }
 187
 188 int
 189 main(int argc, char **argv)
 190 {
 191         char                    *homedir, *conf_file;
 192         int                      ch, flag_altconf = 0, flag_error = 0;
 193         pthread_t                dtid, ctid;
 194
 195
 196         /* init global context */
 197         init_context(&ctx);
 198
 199         /* parse command line options */
 200         while ((ch = getopt(argc, argv, "c:")) != -1) {
 201                 switch (ch) {
 202                 case 'c':
 203                         flag_altconf = 1;
 204                         conf_file = strdup(optarg);
 205                         break;
 206                 default:
 207                         flag_error = 1;
 208                         break;
 209                 }
 210         }
 211         argc -= optind;
 212         argv += optind;
 213
 214         if (flag_error) {
 215                 return EXIT_FAILURE; /* TODO: usage */
 216         }
 217         if (!flag_altconf) {
 218                 if ((homedir = getenv("HOME")) == NULL) {
 219                         /* TODO: look in current directory */
 220                         errx(1, "failed to read environmental variable \'$HOME\'");
 221                 }
 222                 asprintf(&conf_file, "%s/%s", homedir, CONFFILENAME);   /* XXX: null */
 223         }
 224         /* parse config file */
 225         if (parse_config(ctx.conf, conf_file))
 226                 errx(1, "error parsing config file.");
 227
 228         /* open log file */
 229         if (ctx.conf->c_logfile) {
 230                 ctx.conf->c_logfp = fopen(ctx.conf->c_logfile, "w");
 231                 if (!ctx.conf->c_logfp)
 232                         errx(1, "error creating log file.");
 233         }
 234
 235         /* load application */
 236         load_app(&ctx);
 237
 238 #ifdef DEBUG
 239         printf("config info...\n");
 240         printf("\tnode id \'%s\', control port \'%d\', switch port \'%d\'.\n",
 241                 ctx.conf->c_nodeid, ctx.conf->c_co_port, ctx.conf->c_sw_port);
 242         printf("\topenflow version \'%s\'.\n", ctx.conf->c_ofp_verstr);
 243         printf("\tnumber of replicas \'%d\'.\n", ctx.conf->c_nreplicas);
 244 #endif
 245
 246         /* initialize DHT */
 247         if (!dht_init(ctx.node, ctx.conf->c_nodeid, ctx.conf->c_co_port, ctx.conf->c_nreplicas, 0, NULL))
 248                 errx(1, "dht_init");
 249         else
 250                 printf("dht init successful.\n");
 251         pthread_create(&dtid, NULL, dht_thread, NULL);
 252
 253         /* initialize OF controller */
 254         if (of_controller_init(ctx.cntl, ctx.conf->c_sw_port, ctx.conf->c_ofp()))
 255                 errx(1, "init failed.");
 256         else
 257                 printf("controller init successful.\n");
 258         ctx.cntl->handler(ctx.cntl, handler);
 259         pthread_create(&ctid, NULL, controller_thread, NULL);
 260
 261         /* wait */
 262         pthread_join(dtid, NULL);
 263         pthread_join(ctid, NULL);
 264
 265         fclose(ctx.conf->c_logfp);
 266         return 0;
 267 }